diff --git a/backend/tests/runner/README.md b/backend/tests/runner/README.md
new file mode 100644
index 00000000..6fc54c72
--- /dev/null
+++ b/backend/tests/runner/README.md
@@ -0,0 +1,58 @@
+# Test runner
+
+An interactive test runner: discover tests via pytest's own collector, cherry-pick
+them in a Textual tree, and run them while a live Rich dashboard streams pass/fail.
+
+```
+python -m tests.runner            # discover → picker → run
+python -m tests.runner -k ingest  # skip the picker, run a -k selection
+python -m tests.runner tests/api  # run specific paths
+```
+
+## Two-venv design
+
+The runner is split across two interpreters so it can be dropped into any repo
+without polluting that repo's test environment:
+
+| Process | venv | Imports | Needs |
+| --- | --- | --- | --- |
+| Runner (parent) | runner venv | `typer`, `rich`, `textual` | the UI libs only — **never** pytest |
+| Worker (subprocess) | test venv | `pytest`, `coverage` | pytest, pytest-asyncio, coverage + the project's own deps |
+
+Discovery and execution both run in the **test venv** via subprocess. The worker
+([`_worker.py`](./_worker.py)) registers a thin pytest plugin that frames every
+collection/run event (and, in `-s` mode, each line of streamed output) as JSON
+onto a pipe. The parent ([`run.py`](./run.py)) reads those events and owns all
+Rich rendering. See [`events.py`](./events.py) for the wire protocol.
+
+```
+parent (rich)  <—— JSON events over a pipe FD ——  worker (pytest, test venv)
+```
+
+## Setup
+
+1. Runner venv (the UI):
+
+   ```
+   python -m venv .runner-venv
+   .runner-venv/bin/pip install -e tests/runner
+   ```
+
+2. Test venv (where tests actually run) — your project's existing venv with
+   `pytest`, `pytest-asyncio`, and `coverage` installed.
+
+3. Point the runner at the test venv in [`config.json`](./config.json).
+
+## config.json
+
+All path/venv coupling lives here so no code edits are needed per repo:
+
+| key | meaning | default |
+| --- | --- | --- |
+| `repo_root` | cwd for pytest; resolved relative to this folder | `../..` (repo root) |
+| `test_paths` | default search roots when no paths are given | `["tests/unit", "tests/api"]` |
+| `venv_python` | interpreter the **tests** run in; relative to `repo_root` | the runner's own interpreter |
+| `coverage_source` | packages measured under `--cov` and the report filter | `["backend"]` |
+
+If `venv_python` is unset or missing on disk, the runner falls back to the
+current interpreter (single-venv mode), reproducing the original behavior.
diff --git a/backend/tests/runner/__init__.py b/backend/tests/runner/__init__.py
new file mode 100644
index 00000000..ff1bc5d2
--- /dev/null
+++ b/backend/tests/runner/__init__.py
@@ -0,0 +1,8 @@
+"""Crisp interactive test runner for product-analytics-v1.
+
+A small Typer CLI that discovers tests via pytest's own collector, lets you
+cherry-pick them in a Textual tree, and runs them in-process while a Rich
+dashboard streams pass/fail live.
+
+Entry point:  python -m tests.runner
+"""
diff --git a/backend/tests/runner/__main__.py b/backend/tests/runner/__main__.py
new file mode 100644
index 00000000..3f954599
--- /dev/null
+++ b/backend/tests/runner/__main__.py
@@ -0,0 +1,88 @@
+"""Typer entrypoint:  python -m tests.runner
+
+Default (no args): discover tests → interactive Textual picker → run selection.
+With paths / -k / --no-pick: skip the picker and run directly.
+"""
+
+from __future__ import annotations
+
+from typing import List, Optional
+
+import typer
+from rich.console import Console
+
+from tests.runner.discovery import discover
+from tests.runner.picker import run_picker
+from tests.runner.run import RunOptions, run_tests
+
+console = Console()
+
+
+def main(
+    paths: Optional[List[str]] = typer.Argument(
+        None, help="pytest paths or node IDs to target (skips the picker)."
+    ),
+    keyword: Optional[str] = typer.Option(
+        None, "-k", help="Only tests matching this keyword expression (skips the picker)."
+    ),
+    cov: bool = typer.Option(False, "--cov", help="Measure and report coverage of backend/."),
+    exitfirst: bool = typer.Option(
+        False, "-x", "--exitfirst", help="Stop after the first failure."
+    ),
+    last_failed: bool = typer.Option(
+        False, "--lf", "--last-failed", help="Run only the tests that failed last time."
+    ),
+    failed_first: bool = typer.Option(
+        False, "--ff", "--failed-first", help="Run last-failed tests first, then the rest."
+    ),
+    verbose: bool = typer.Option(
+        False, "-v", "--verbose", help="Richer failure output (untruncated assertions)."
+    ),
+    no_capture: bool = typer.Option(
+        False, "-s", "--no-capture", help="Show test stdout / print() output live (pytest -s)."
+    ),
+    show_output: bool = typer.Option(
+        False, "-O", "--show-output", help="Show captured output for passing tests too (not just failures)."
+    ),
+    pick: bool = typer.Option(
+        True, "--pick/--no-pick", help="Open the interactive picker (default on)."
+    ),
+) -> None:
+    try:
+        node_ids = discover(paths, keyword)
+    except RuntimeError as exc:
+        console.print(f"[red]discovery failed:[/]\n{exc}")
+        raise typer.Exit(2)
+
+    if not node_ids:
+        console.print("[yellow]No tests found.[/]")
+        raise typer.Exit(5)
+
+    # CLI flags seed the picker's toggles and drive the non-interactive run.
+    opts = RunOptions(
+        cov=cov,
+        exitfirst=exitfirst,
+        last_failed=last_failed,
+        failed_first=failed_first,
+        verbose=verbose,
+        no_capture=no_capture,
+        show_output=show_output,
+    )
+
+    interactive = pick and not paths and not keyword
+    if interactive:
+        picked = run_picker(node_ids, opts)
+        if picked is None:
+            console.print("[dim]cancelled[/]")
+            raise typer.Exit(0)
+        node_ids, opts = picked
+        if not node_ids:
+            console.print("[yellow]Nothing selected.[/]")
+            raise typer.Exit(0)
+
+    code = run_tests(node_ids, opts)
+    raise typer.Exit(code)
+
+
+if __name__ == "__main__":
+    typer.run(main)
diff --git a/backend/tests/runner/_worker.py b/backend/tests/runner/_worker.py
new file mode 100644
index 00000000..dd169e8a
--- /dev/null
+++ b/backend/tests/runner/_worker.py
@@ -0,0 +1,171 @@
+"""Subprocess entry point that runs pytest in the *test* venv.
+
+Spawned by the parent runner as::
+
+    <venv_python> -m tests.runner._worker <event_fd> <options_json>
+
+with ``cwd`` set to the repo root (PEP 420 makes ``tests.runner`` importable
+without an ``__init__.py``, the same trick discovery uses). It registers a thin
+pytest plugin that frames every collection/run event as JSON onto ``event_fd``,
+optionally streams ``-s`` output through a gutter shim onto the same FD, and (for
+``--cov``) measures coverage and emits the computed rows. The parent owns all
+Rich rendering — this process imports neither rich nor textual.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import os
+import sys
+from io import StringIO
+
+import pytest
+
+from tests.runner import events
+
+
+class _GutterStream:
+    """stdout/stderr shim used in ``-s`` mode.
+
+    Buffers writes and forwards each completed line as an OUTPUT event so the
+    parent can rail it under the current test's header. Mirrors the old
+    in-process shim, but emits framed events instead of printing.
+    """
+
+    def __init__(self, fd: int) -> None:
+        self._fd = fd
+        self._buf = ""
+
+    def write(self, s) -> int:
+        if isinstance(s, bytes):
+            s = s.decode("utf-8", "replace")
+        self._buf += s
+        while "\n" in self._buf:
+            line, self._buf = self._buf.split("\n", 1)
+            events.emit(self._fd, {"type": events.OUTPUT, "line": line})
+        return len(s)
+
+    def flush(self) -> None:
+        if self._buf:
+            events.emit(self._fd, {"type": events.OUTPUT, "line": self._buf})
+            self._buf = ""
+
+    def isatty(self) -> bool:
+        return False
+
+    def writable(self) -> bool:
+        return True
+
+    def fileno(self):  # some libraries probe this; signal "not a real fd"
+        raise io.UnsupportedOperation("fileno")
+
+
+class _EventEmitter:
+    """pytest plugin: forward collection/run hooks as framed events."""
+
+    def __init__(self, fd: int) -> None:
+        self._fd = fd
+
+    def pytest_collection_finish(self, session) -> None:
+        events.emit(
+            self._fd,
+            {"type": events.COLLECTION, "items": [item.nodeid for item in session.items]},
+        )
+
+    def pytest_runtest_logstart(self, nodeid, location) -> None:
+        events.emit(self._fd, {"type": events.LOGSTART, "nodeid": nodeid})
+
+    def pytest_runtest_logreport(self, report) -> None:
+        events.emit(
+            self._fd,
+            {
+                "type": events.LOGREPORT,
+                "nodeid": report.nodeid,
+                "when": report.when,
+                "passed": bool(report.passed),
+                "failed": bool(report.failed),
+                "skipped": bool(report.skipped),
+                "duration": float(getattr(report, "duration", 0.0) or 0.0),
+                "capstdout": report.capstdout or "",
+                "capstderr": report.capstderr or "",
+                "longreprtext": report.longreprtext or "",
+            },
+        )
+
+    def pytest_runtest_logfinish(self, nodeid, location) -> None:
+        events.emit(self._fd, {"type": events.LOGFINISH, "nodeid": nodeid})
+
+
+def _emit_coverage(fd: int, cov, coverage_source: list[str], repo_root: str) -> None:
+    """Compute the per-file coverage table and emit it (rendering is the parent's job)."""
+    try:
+        cov.stop()
+        cov.save()
+        total = cov.report(file=StringIO())
+    except Exception as exc:  # pragma: no cover - defensive
+        events.emit(fd, {"type": events.COVERAGE, "error": str(exc)})
+        return
+
+    needles = [f"{os.sep}{name}{os.sep}" for name in coverage_source]
+    data = cov.get_data()
+    rows = []
+    for path in data.measured_files():
+        if not any(n in path for n in needles):
+            continue
+        try:
+            _, statements, _, missing, _ = cov.analysis2(path)
+        except Exception:
+            continue
+        n = len(statements)
+        if n == 0:
+            continue
+        miss = len(missing)
+        pct = (n - miss) / n * 100
+        rel = os.path.relpath(path, repo_root)
+        rows.append([rel, n, miss, pct])
+
+    events.emit(fd, {"type": events.COVERAGE, "rows": rows, "total": float(total)})
+
+
+def main(argv: list[str]) -> int:
+    fd = int(argv[0])
+    opts = json.loads(argv[1])
+
+    pytest_args = opts.get("pytest_args", [])
+    no_capture = bool(opts.get("no_capture"))
+    cov = bool(opts.get("cov"))
+    coverage_source = opts.get("coverage_source") or []
+    repo_root = opts.get("repo_root") or os.getcwd()
+
+    cov_obj = None
+    if cov:
+        import coverage
+
+        cov_obj = coverage.Coverage(source=coverage_source)
+        cov_obj.start()
+
+    emitter = _EventEmitter(fd)
+    args = ["-o", "addopts=", "-p", "no:terminal", *pytest_args]
+
+    saved_out, saved_err = sys.stdout, sys.stderr
+    if no_capture:
+        sys.stdout = _GutterStream(fd)
+        sys.stderr = _GutterStream(fd)
+    try:
+        code = int(pytest.main(args, plugins=[emitter]))
+    finally:
+        if no_capture:
+            sys.stdout.flush()
+            sys.stderr.flush()
+            sys.stdout, sys.stderr = saved_out, saved_err
+
+    if cov_obj is not None:
+        _emit_coverage(fd, cov_obj, coverage_source, repo_root)
+
+    events.emit(fd, {"type": events.DONE, "code": code})
+    return code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))
diff --git a/backend/tests/runner/config.json b/backend/tests/runner/config.json
new file mode 100644
index 00000000..bb6b01b3
--- /dev/null
+++ b/backend/tests/runner/config.json
@@ -0,0 +1,6 @@
+{
+  "repo_root": "../..",
+  "test_paths": ["tests/unit", "tests/api"],
+  "venv_python": "backend/.venv/bin/python",
+  "coverage_source": ["backend"]
+}
diff --git a/backend/tests/runner/config.py b/backend/tests/runner/config.py
new file mode 100644
index 00000000..9df7a77a
--- /dev/null
+++ b/backend/tests/runner/config.py
@@ -0,0 +1,80 @@
+"""Runner configuration: paths + the test venv interpreter.
+
+A single ``config.json`` next to this module centralizes everything that used to
+be hard-coded in ``discovery.py`` / ``run.py`` so the runner can be dropped into
+another repo by editing data instead of code:
+
+    repo_root        where pytest runs (cwd); test_paths are relative to it
+    test_paths       default search roots when no paths are given
+    venv_python      the interpreter the *tests* run in (separate from the
+                     runner's own venv); falls back to the current interpreter
+    coverage_source  packages measured under --cov, and the report path filter
+
+All paths in the JSON are resolved relative to this directory (the runner
+folder) unless absolute. Missing keys fall back to today's behavior.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+
+RUNNER_DIR = Path(__file__).resolve().parent
+DEFAULT_CONFIG_PATH = RUNNER_DIR / "config.json"
+
+# Defaults that reproduce the pre-config behavior of the runner.
+_DEFAULT_REPO_ROOT = RUNNER_DIR.parents[1]  # <repo>/tests/runner -> <repo>
+_DEFAULT_TEST_PATHS = ["tests/unit", "tests/api"]
+_DEFAULT_COVERAGE_SOURCE = ["backend"]
+
+
+@dataclass(frozen=True)
+class Config:
+    repo_root: Path
+    test_paths: list[str]
+    venv_python: str
+    coverage_source: list[str]
+
+
+def _resolve(base: Path, value: str) -> Path:
+    p = Path(value).expanduser()
+    return p if p.is_absolute() else (base / p).resolve()
+
+
+def load_config(path: Path | None = None) -> Config:
+    """Load and resolve the runner config. Unknown/missing keys use defaults."""
+    cfg_path = path or DEFAULT_CONFIG_PATH
+    raw: dict = {}
+    if cfg_path.is_file():
+        raw = json.loads(cfg_path.read_text())
+
+    repo_root = (
+        _resolve(RUNNER_DIR, raw["repo_root"])
+        if raw.get("repo_root")
+        else _DEFAULT_REPO_ROOT
+    )
+
+    test_paths = raw.get("test_paths") or list(_DEFAULT_TEST_PATHS)
+    coverage_source = raw.get("coverage_source") or list(_DEFAULT_COVERAGE_SOURCE)
+
+    # The test interpreter is resolved against repo_root. If it is unset or does
+    # not exist, fall back to the interpreter running the runner so the tool
+    # still works in a single-venv setup. We deliberately do NOT call
+    # ``.resolve()`` here: a venv's ``python`` is a symlink to the base
+    # interpreter, and following it would dereference away the venv.
+    venv_python = sys.executable
+    if raw.get("venv_python"):
+        candidate = Path(raw["venv_python"]).expanduser()
+        if not candidate.is_absolute():
+            candidate = repo_root / candidate
+        if candidate.exists():
+            venv_python = str(candidate)
+
+    return Config(
+        repo_root=repo_root,
+        test_paths=test_paths,
+        venv_python=venv_python,
+        coverage_source=coverage_source,
+    )
diff --git a/backend/tests/runner/discovery.py b/backend/tests/runner/discovery.py
new file mode 100644
index 00000000..891efba6
--- /dev/null
+++ b/backend/tests/runner/discovery.py
@@ -0,0 +1,56 @@
+"""Test discovery by delegating to pytest's own collector.
+
+We never parse test files ourselves — we ask pytest to collect and emit node
+IDs. This matches pytest exactly (handles asyncio_mode=auto, parametrization,
+classes, markers) instead of guessing from decorators like the old runner did.
+
+Collection runs in the configured *test* venv (``config.venv_python``), because
+collecting imports the test modules and therefore needs the project's full test
+dependencies — not the runner's own venv.
+"""
+
+from __future__ import annotations
+
+import subprocess
+
+from tests.runner.config import load_config
+
+_CONFIG = load_config()
+
+# Kept for backwards-compatible imports (e.g. run.py uses it for relpath).
+REPO_ROOT = _CONFIG.repo_root
+
+# Default search roots come from config; we avoid scanning tests/runner itself.
+DEFAULT_PATHS: list[str] = _CONFIG.test_paths
+
+
+def discover(paths: list[str] | None = None, keyword: str | None = None) -> list[str]:
+    """Return pytest node IDs for the given paths (optionally -k filtered).
+
+    Raises RuntimeError if pytest collection itself errored.
+    """
+    search = paths or DEFAULT_PATHS
+    cmd = [
+        _CONFIG.venv_python,
+        "-m",
+        "pytest",
+        "-o",
+        "addopts=",  # drop the global -q so node IDs print one per line
+        "--collect-only",
+        "-q",
+        *search,
+    ]
+    if keyword:
+        cmd += ["-k", keyword]
+
+    proc = subprocess.run(
+        cmd, capture_output=True, text=True, cwd=str(REPO_ROOT)
+    )
+    # Collection errors (import errors, bad -k) → surface stderr/stdout.
+    if proc.returncode not in (0, 5):  # 5 = "no tests collected"
+        raise RuntimeError(
+            f"pytest collection failed (exit {proc.returncode}):\n"
+            f"{proc.stdout}\n{proc.stderr}".strip()
+        )
+
+    return [line.strip() for line in proc.stdout.splitlines() if "::" in line]
diff --git a/backend/tests/runner/events.py b/backend/tests/runner/events.py
new file mode 100644
index 00000000..83f10897
--- /dev/null
+++ b/backend/tests/runner/events.py
@@ -0,0 +1,31 @@
+"""Wire protocol shared by the parent runner and the subprocess worker.
+
+The worker (running in the *test* venv) writes one JSON object per line to an
+inherited pipe FD; the parent (running in the *runner* venv) reads them and
+drives the Rich dashboard. This module is deliberately dependency-free (only the
+stdlib) so it imports cleanly in *both* venvs — no rich, textual, or pytest.
+
+Ordering guarantee: both control events and streamed ``-s`` output are written
+through :func:`emit` to the *same* FD from the *same* process, so an OUTPUT
+event always lands between its test's LOGSTART and LOGFINISH.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+
+# Event types (the "type" field of every framed message).
+COLLECTION = "collection"  # {"items": [nodeid, ...]}
+LOGSTART = "logstart"      # {"nodeid": ...}
+LOGREPORT = "logreport"    # {nodeid, when, passed, failed, skipped, duration, capstdout, capstderr, longreprtext}
+LOGFINISH = "logfinish"    # {"nodeid": ...}
+OUTPUT = "output"          # {"line": ...}  (streamed -s test output)
+COVERAGE = "coverage"      # {"rows": [[rel, stmts, miss, pct], ...], "total": float}
+DONE = "done"              # {"code": int}
+
+
+def emit(fd: int, obj: dict) -> None:
+    """Write one framed event to ``fd``. Unbuffered to preserve ordering."""
+    data = (json.dumps(obj) + "\n").encode("utf-8")
+    os.write(fd, data)
diff --git a/backend/tests/runner/picker.py b/backend/tests/runner/picker.py
new file mode 100644
index 00000000..154036c9
--- /dev/null
+++ b/backend/tests/runner/picker.py
@@ -0,0 +1,333 @@
+"""Interactive Textual tree picker.
+
+Renders the discovered test tree with checkboxes and a row of run-option
+toggles (coverage, stop-on-first-fail, last-failed, failed-first, verbose, and a
+``-k`` keyword expression). Returns ``(node_ids, RunOptions)`` for the selection
+or ``None`` if cancelled. Selecting a branch selects all descendant leaves.
+Every row always shows a box: empty when unselected, checked when selected, and
+a half-filled box when only some descendants are selected.
+
+The tree keybindings live on a ``Screen`` (not the ``App``) so they go quiet
+while the keyword modal is open and you can freely type expressions like
+``ingest and not batch`` without the single-key toggles firing.
+"""
+
+from __future__ import annotations
+
+from rich.text import Text
+from textual.app import App, ComposeResult
+from textual.binding import Binding
+from textual.containers import Vertical
+from textual.screen import ModalScreen, Screen
+from textual.widgets import Footer, Header, Input, Label, Static, Tree
+from textual.widgets.tree import TreeNode
+
+from tests.runner.run import RunOptions
+from tests.runner.tree import TNode, build_tree
+
+# Always-visible ballot-box glyphs (matched empty/checked pair) + a partial box.
+_BOX_EMPTY = "\u2610"     # ☐
+_BOX_FULL = "\u2611"      # ☑
+_BOX_PARTIAL = "\u25a3"   # ▣
+
+
+class KeywordScreen(ModalScreen[str | None]):
+    """Modal text prompt for the pytest ``-k`` expression."""
+
+    CSS = """
+    KeywordScreen { align: center middle; }
+    #kw-box {
+        width: 64;
+        height: auto;
+        padding: 1 2;
+        background: $panel;
+        border: round $accent;
+    }
+    #kw-box Label { margin-bottom: 1; }
+    """
+
+    BINDINGS = [Binding("escape", "cancel", "Cancel")]
+
+    def __init__(self, current: str) -> None:
+        super().__init__()
+        self._current = current
+
+    def compose(self) -> ComposeResult:
+        with Vertical(id="kw-box"):
+            yield Label("pytest -k expression  (enter to apply, blank to clear)")
+            yield Input(
+                value=self._current,
+                placeholder="e.g. ingest and not batch",
+                id="kw-input",
+            )
+
+    def on_mount(self) -> None:
+        self.query_one(Input).focus()
+
+    def on_input_submitted(self, event: Input.Submitted) -> None:
+        self.dismiss(event.value.strip())
+
+    def action_cancel(self) -> None:
+        self.dismiss(None)
+
+
+class PickerScreen(Screen):
+    CSS = """
+    Tree { padding: 0 1; }
+    #status { padding: 0 1; }
+    """
+
+    BINDINGS = [
+        Binding("space", "toggle_select", "Select", priority=True),
+        Binding("right", "expand_node", "Expand", priority=True),
+        Binding("left", "collapse_node", "Collapse", priority=True),
+        Binding("a", "toggle_all", "All", priority=True),
+        Binding("enter", "run", "Run", priority=True),
+        Binding("c", "toggle_cov", "cov"),
+        Binding("x", "toggle_x", "stop"),
+        Binding("l", "toggle_lf", "last-fail"),
+        Binding("f", "toggle_ff", "fail-first"),
+        Binding("v", "toggle_v", "verbose"),
+        Binding("s", "toggle_s", "stdout"),
+        Binding("o", "toggle_o", "output"),
+        Binding("k", "keyword", "keyword"),
+        Binding("q", "cancel", "Quit", priority=True),
+        Binding("escape", "cancel", "Quit", priority=True),
+    ]
+
+    def __init__(self, root: TNode, opts: RunOptions) -> None:
+        super().__init__()
+        self._root = root
+        self.selected: set[str] = set()
+        self._leaf_id: dict[TreeNode, str] = {}
+        self._base_label: dict[TreeNode, str] = {}
+        self._kind: dict[TreeNode, str] = {}
+        # Run-option state, seeded from the CLI flags.
+        self.cov = opts.cov
+        self.exitfirst = opts.exitfirst
+        self.last_failed = opts.last_failed
+        self.failed_first = opts.failed_first
+        self.verbose = opts.verbose
+        self.no_capture = opts.no_capture
+        self.show_output = opts.show_output
+        self.keyword = opts.keyword
+
+    def compose(self) -> ComposeResult:
+        yield Header(show_clock=False)
+        yield Static(id="status")
+        tree: Tree = Tree(self._root.label, id="tree")
+        tree.root.expand()
+        self._base_label[tree.root] = self._root.label
+        self._kind[tree.root] = self._root.kind
+        self._build(tree.root, self._root)
+        self._refresh_labels(tree)
+        yield tree
+        yield Footer()
+
+    def on_mount(self) -> None:
+        self.app.title = "test picker"
+        self.app.sub_title = (
+            f"{_BOX_EMPTY} empty   {_BOX_FULL} selected   {_BOX_PARTIAL} partial"
+        )
+        self._refresh_status()
+
+    # --- tree construction -------------------------------------------------
+    def _build(self, widget: TreeNode, tnode: TNode) -> None:
+        for child in tnode.children:
+            if child.is_leaf:
+                leaf = widget.add_leaf(child.label)
+                self._base_label[leaf] = child.label
+                self._kind[leaf] = child.kind
+                if child.node_id:
+                    self._leaf_id[leaf] = child.node_id
+            else:
+                branch = widget.add(child.label, expand=True)
+                self._base_label[branch] = child.label
+                self._kind[branch] = child.kind
+                self._build(branch, child)
+
+    # --- selection helpers -------------------------------------------------
+    def _leaves_under(self, node: TreeNode) -> set[str]:
+        if node in self._leaf_id:
+            return {self._leaf_id[node]}
+        out: set[str] = set()
+        for child in node.children:
+            out |= self._leaves_under(child)
+        return out
+
+    def _all_nodes(self, node: TreeNode) -> list[TreeNode]:
+        nodes = [node]
+        for child in node.children:
+            nodes.extend(self._all_nodes(child))
+        return nodes
+
+    def _refresh_labels(self, tree: Tree) -> None:
+        for node in self._all_nodes(tree.root):
+            base = self._base_label.get(node, str(node.label))
+            ids = self._leaves_under(node)
+            sel = ids & self.selected
+            if ids and sel == ids:
+                glyph, style = _BOX_FULL, "bold green"
+            elif sel:
+                glyph, style = _BOX_PARTIAL, "bold yellow"
+            else:
+                glyph, style = _BOX_EMPTY, "grey62"
+            label = Text.assemble((f"{glyph} ", style), (base, ""))
+            node.set_label(label)
+
+    # --- run-option status line -------------------------------------------
+    def _refresh_status(self) -> None:
+        def mark(on: bool) -> str:
+            return "[bold green]on[/]" if on else "[dim]off[/]"
+
+        kw = self.keyword or "[dim]—[/]"
+        status = "  ".join(
+            [
+                f"[bold]\\[c][/] cov {mark(self.cov)}",
+                f"[bold]\\[x][/] stop {mark(self.exitfirst)}",
+                f"[bold]\\[l][/] --lf {mark(self.last_failed)}",
+                f"[bold]\\[f][/] --ff {mark(self.failed_first)}",
+                f"[bold]\\[v][/] -v {mark(self.verbose)}",
+                f"[bold]\\[s][/] -s {mark(self.no_capture)}",
+                f"[bold]\\[o][/] out {mark(self.show_output)}",
+                f"[bold]\\[k][/] -k {kw}",
+            ]
+        )
+        self.query_one("#status", Static).update(status)
+
+    def options(self) -> RunOptions:
+        return RunOptions(
+            cov=self.cov,
+            exitfirst=self.exitfirst,
+            last_failed=self.last_failed,
+            failed_first=self.failed_first,
+            verbose=self.verbose,
+            no_capture=self.no_capture,
+            show_output=self.show_output,
+            keyword=self.keyword,
+        )
+
+    # --- selection actions -------------------------------------------------
+    def action_toggle_select(self) -> None:
+        tree = self.query_one(Tree)
+        node = tree.cursor_node
+        if node is None:
+            return
+        ids = self._leaves_under(node)
+        if ids and ids <= self.selected:
+            self.selected -= ids
+        else:
+            self.selected |= ids
+        self._refresh_labels(tree)
+
+    def action_expand_node(self) -> None:
+        tree = self.query_one(Tree)
+        node = tree.cursor_node
+        if node is None or not node.allow_expand:
+            return
+        if not node.is_expanded:
+            node.expand()
+        elif node.children:
+            # Already expanded → step into the first child (like pressing down).
+            tree.move_cursor(node.children[0])
+
+    def action_collapse_node(self) -> None:
+        tree = self.query_one(Tree)
+        node = tree.cursor_node
+        if node is None:
+            return
+        # On an expanded directory, collapse the directory itself.
+        if (
+            node is not tree.root
+            and self._kind.get(node) == "dir"
+            and node.allow_expand
+            and node.is_expanded
+        ):
+            node.collapse()
+            return
+        # Anywhere else (file, function, case, or a collapsed dir): collapse the
+        # section that contains this node and move the cursor onto it.
+        parent = node.parent
+        if parent is None:
+            return
+        if parent is not tree.root and parent.allow_expand and parent.is_expanded:
+            parent.collapse()
+        tree.move_cursor(parent)
+
+    def action_toggle_all(self) -> None:
+        tree = self.query_one(Tree)
+        everything = self._leaves_under(tree.root)
+        self.selected = set() if self.selected >= everything else set(everything)
+        self._refresh_labels(tree)
+
+    # --- run-option actions ------------------------------------------------
+    def action_toggle_cov(self) -> None:
+        self.cov = not self.cov
+        self._refresh_status()
+
+    def action_toggle_x(self) -> None:
+        self.exitfirst = not self.exitfirst
+        self._refresh_status()
+
+    def action_toggle_lf(self) -> None:
+        self.last_failed = not self.last_failed
+        if self.last_failed:
+            self.failed_first = False  # --lf and --ff are mutually exclusive
+        self._refresh_status()
+
+    def action_toggle_ff(self) -> None:
+        self.failed_first = not self.failed_first
+        if self.failed_first:
+            self.last_failed = False
+        self._refresh_status()
+
+    def action_toggle_v(self) -> None:
+        self.verbose = not self.verbose
+        self._refresh_status()
+
+    def action_toggle_s(self) -> None:
+        self.no_capture = not self.no_capture
+        self._refresh_status()
+
+    def action_toggle_o(self) -> None:
+        self.show_output = not self.show_output
+        self._refresh_status()
+
+    def action_keyword(self) -> None:
+        def apply(value: str | None) -> None:
+            if value is not None:  # None = cancelled, leave keyword untouched
+                self.keyword = value or None
+                self._refresh_status()
+
+        self.app.push_screen(KeywordScreen(self.keyword or ""), apply)
+
+    # --- finish ------------------------------------------------------------
+    def action_run(self) -> None:
+        chosen = set(self.selected)
+        if not chosen:
+            node = self.query_one(Tree).cursor_node
+            if node is not None:
+                chosen = self._leaves_under(node)
+        self.app.exit((sorted(chosen), self.options()))
+
+    def action_cancel(self) -> None:
+        self.app.exit(None)
+
+
+class TestPicker(App):
+    def __init__(self, root: TNode, opts: RunOptions) -> None:
+        super().__init__()
+        self._root = root
+        self._opts = opts
+
+    def on_mount(self) -> None:
+        self.push_screen(PickerScreen(self._root, self._opts))
+
+
+def run_picker(
+    node_ids: list[str], opts: RunOptions
+) -> tuple[list[str], RunOptions] | None:
+    """Launch the picker. Returns ``(node_ids, options)`` or None if cancelled."""
+    root = build_tree(node_ids)
+    app = TestPicker(root, opts)
+    return app.run()
diff --git a/backend/tests/runner/pyproject.toml b/backend/tests/runner/pyproject.toml
new file mode 100644
index 00000000..2561f3ae
--- /dev/null
+++ b/backend/tests/runner/pyproject.toml
@@ -0,0 +1,27 @@
+# Dependency manifest for the test runner's OWN venv.
+#
+# These are the UI libraries the runner process needs. They are intentionally
+# separate from the test venv: the runner imports typer/rich/textual but never
+# pytest, and the worker subprocess imports pytest but never rich/textual.
+#
+# Install into a dedicated runner venv, e.g.:
+#     python -m venv .runner-venv
+#     .runner-venv/bin/pip install -e tests/runner
+#
+# The *test* venv (pointed at by config.json -> "venv_python") is a separate
+# environment that must provide: pytest, pytest-asyncio, coverage, plus the
+# project's own runtime/test dependencies.
+
+[project]
+name = "test-runner"
+version = "0.1.0"
+description = "Interactive subprocess test runner with a live Rich dashboard."
+requires-python = ">=3.9"
+dependencies = [
+    "typer",    # CLI entrypoint (__main__.py)
+    "rich",     # live dashboard (run.py)
+    "textual",  # interactive tree picker (picker.py)
+]
+
+[tool.setuptools]
+packages = ["tests.runner"]
diff --git a/backend/tests/runner/run.py b/backend/tests/runner/run.py
new file mode 100644
index 00000000..2c674951
--- /dev/null
+++ b/backend/tests/runner/run.py
@@ -0,0 +1,581 @@
+"""Subprocess pytest execution with a live Rich dashboard.
+
+pytest runs in a separate process (the configured *test* venv) via
+``tests.runner._worker``. That worker streams framed events over a pipe; this
+module consumes them, mutates a :class:`Dashboard`, and renders a live Rich
+panel plus a final summary and (optionally) a coverage table. Keeping execution
+out-of-process is what lets the runner's own venv stay free of pytest and the
+project's test dependencies.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import time
+from collections import defaultdict
+from dataclasses import dataclass
+
+from rich import box
+from rich.console import Console, Group
+from rich.live import Live
+from rich.panel import Panel
+from rich.progress_bar import ProgressBar
+from rich.spinner import Spinner
+from rich.table import Table
+from rich.text import Text
+
+from tests.runner import events
+from tests.runner.config import load_config
+
+_CONFIG = load_config()
+
+# Cap per-test captured output so one chatty test can't flood the scrollback.
+_MAX_OUTPUT_LINES = 40
+
+_OUTCOME_GLYPH: dict[str, tuple[str, str]] = {
+    "passed": ("\u2713", "green"),       # ✓
+    "failed": ("\u2717", "bold red"),    # ✗
+    "skipped": ("\u21b7", "yellow"),     # ↷
+    "error": ("\u2717", "bold magenta"),  # ✗ (setup/teardown)
+}
+
+
+def _short(nodeid: str, width: int = 80) -> str:
+    """Keep the spinner's current-test label on a single line."""
+    return nodeid if len(nodeid) <= width else "…" + nodeid[-(width - 1):]
+
+
+@dataclass
+class RunOptions:
+    """Execution options shared between the CLI, the picker, and run_tests."""
+
+    cov: bool = False          # measure backend/ coverage
+    exitfirst: bool = False    # -x  : stop after the first failure
+    last_failed: bool = False  # --lf: run only last-failed tests
+    failed_first: bool = False  # --ff: run last-failed first, then the rest
+    verbose: bool = False      # -v  : richer failure output (untruncated asserts)
+    no_capture: bool = False   # -s  : stream test stdout / print() output live
+    show_output: bool = False  # -O  : show captured output for passing tests too
+    keyword: str | None = None  # -k  : only tests matching this expression
+
+
+class Dashboard:
+    """Event consumer + Rich renderer (runs in the parent / runner venv).
+
+    Equivalent to the old in-process collector, but instead of pytest hooks it
+    is driven by :meth:`handle_event` over events streamed from the worker. All
+    counters, per-file tallies, and the two live layouts are unchanged:
+
+    * ``-s`` ON: output streams live, so each test gets a header rule + a
+      persistent ``✓/✗ nodeid duration`` scrollback line bracketing its prints.
+    * ``-s`` OFF: a per-file table with live ``passed/failed/total`` counters is
+      pinned at the bottom; captured output for failures (or all tests under
+      ``-O``) still prints above it.
+
+    The dashboard renders itself (``__rich__``) so Live's auto-refresh animates
+    the spinner and ticks the elapsed clock without us pushing snapshots.
+    """
+
+    def __init__(self, opts: RunOptions) -> None:
+        self.opts = opts
+        self.total = 0
+        self.passed = 0
+        self.failed = 0
+        self.skipped = 0
+        self.errors = 0
+        self.current = ""
+        self.failures: list[tuple[str, str]] = []
+        self.live: Live | None = None
+        self._start = time.time()
+        self._current_start = self._start
+        # Per-test accumulator across setup/call/teardown phases, keyed by nodeid.
+        self._pending: dict[str, dict] = {}
+        self._spinner = Spinner("dots", style="cyan")
+        # Per-file tallies for the -s-off view (files in collection order).
+        self._file_order: list[str] = []
+        self._file_total: dict[str, int] = defaultdict(int)
+        self._file_pass: dict[str, int] = defaultdict(int)
+        self._file_fail: dict[str, int] = defaultdict(int)
+        self._file_skip: dict[str, int] = defaultdict(int)
+        self._current_file = ""
+        # -s streaming: track the open file bracket so we can rail + close it.
+        self._streamed_file: str | None = None
+        self._file_open = False
+
+    @staticmethod
+    def _file_of(nodeid: str) -> str:
+        return nodeid.split("::", 1)[0]
+
+    # --- event dispatch ----------------------------------------------------
+    def handle_event(self, evt: dict) -> None:
+        etype = evt.get("type")
+        if etype == events.COLLECTION:
+            self._on_collection(evt["items"])
+        elif etype == events.LOGSTART:
+            self._on_logstart(evt["nodeid"])
+        elif etype == events.LOGREPORT:
+            self._on_logreport(evt)
+        elif etype == events.LOGFINISH:
+            self._on_logfinish(evt["nodeid"])
+        elif etype == events.OUTPUT:
+            self._emit_stream_line(evt["line"])
+
+    def _on_collection(self, items: list[str]) -> None:
+        self.total = len(items)
+        for nodeid in items:
+            f = self._file_of(nodeid)
+            if f not in self._file_total:
+                self._file_order.append(f)
+            self._file_total[f] += 1
+        self._refresh()
+
+    def _on_logstart(self, nodeid: str) -> None:
+        self.current = nodeid
+        self._current_file = self._file_of(nodeid)
+        self._current_start = time.time()
+        # In -s mode, output streams live; print a file header box (once per
+        # file) and a per-test header box so the prints that follow are
+        # attributable. The streamed output itself is left unboxed.
+        if self.opts.no_capture and self.live is not None:
+            self._emit_stream_headers(nodeid)
+        self._refresh()
+
+    def _on_logreport(self, evt: dict) -> None:
+        nodeid = evt["nodeid"]
+        when = evt["when"]
+        entry = self._pending.setdefault(
+            nodeid, {"outcome": "passed", "duration": 0.0, "out": ""}
+        )
+        entry["duration"] += evt.get("duration", 0.0) or 0.0
+        # Take captured output from the authoritative phase only: the call, or
+        # setup when setup fails/skips (no call then). pytest's teardown report
+        # re-includes the call's capstdout, so summing every phase would
+        # duplicate the test's output.
+        if when == "call" or (when == "setup" and (evt["failed"] or evt["skipped"])):
+            out = (evt.get("capstdout") or "") + (evt.get("capstderr") or "")
+            if out:
+                entry["out"] = out
+
+        longrepr = evt.get("longreprtext") or ""
+        if when == "setup":
+            if evt["skipped"]:
+                entry["outcome"] = "skipped"
+                self.skipped += 1
+            elif evt["failed"]:
+                entry["outcome"] = "error"
+                self.errors += 1
+                self.failures.append((nodeid + " (setup)", longrepr))
+        elif when == "call":
+            if evt["passed"]:
+                self.passed += 1
+            elif evt["failed"]:
+                entry["outcome"] = "failed"
+                self.failed += 1
+                self.failures.append((nodeid, longrepr))
+            elif evt["skipped"]:
+                entry["outcome"] = "skipped"
+                self.skipped += 1
+        elif when == "teardown" and evt["failed"]:
+            if entry["outcome"] == "passed":
+                entry["outcome"] = "error"
+            self.errors += 1
+            self.failures.append((nodeid + " (teardown)", longrepr))
+        self._refresh()
+
+    def _on_logfinish(self, nodeid: str) -> None:
+        entry = self._pending.pop(nodeid, None)
+        self.current = ""
+        if entry is None:
+            self._refresh()
+            return
+
+        # Update per-file tallies for the -s-off table.
+        f = self._file_of(nodeid)
+        outcome = entry["outcome"]
+        if outcome == "passed":
+            self._file_pass[f] += 1
+        elif outcome in ("failed", "error"):
+            self._file_fail[f] += 1
+        elif outcome == "skipped":
+            self._file_skip[f] += 1
+
+        if self.live is not None:
+            if self.opts.no_capture:
+                # Streamed mode: persistent per-test line bracketing the output.
+                self._emit_test_line(nodeid, entry)
+            else:
+                # File-counter mode: no per-test line, but still surface
+                # captured output for failures (or all tests under -O).
+                self._emit_output_panel(nodeid, entry)
+        self._refresh()
+
+    # --- streaming (-s) nested brackets -----------------------------------
+    # Two levels: a blue outer bracket per .py file wraps the grey inner
+    # bracket of each test, so tests chunk visually by their parent file.
+    #
+    #     ╔═ test_persist.py ═╗   (file header box)
+    #     ┃ ╭ test_one          (┃ = blue file rail; ╭ opens the test)
+    #     ┃ │ ...streamed output (railed + gutter-wrapped)
+    #     ┃ ╰ ✓ 11ms            (closes the test)
+    #     ┃
+    #     ┗━━━                   (blue foot closes the file)
+    _OUTER = "\u2503 "   # "┃ "
+    _FOOT = "\u2517" + "\u2501" * 3  # "┗━━━"
+    _OPEN = "\u256d "    # "╭ "
+    _RAIL = "\u2502 "    # "│ "
+    _CLOSE = "\u2570 "   # "╰ "
+    _OUTER_STYLE = "blue"
+
+    def _outer(self) -> Text:
+        return Text(self._OUTER, style=self._OUTER_STYLE)
+
+    def _emit_stream_headers(self, nodeid: str) -> None:
+        f = self._file_of(nodeid)
+        if f != self._streamed_file:
+            self._close_file_bracket()  # close the previous file's blue bracket
+            self._streamed_file = f
+            self.live.console.print(self._file_header(f))
+            self._file_open = True
+        name = nodeid.split("::", 1)[1] if "::" in nodeid else nodeid
+        self.live.console.print(
+            self._outer() + Text.assemble((self._OPEN, "grey50"), (name, "bold"))
+        )
+
+    def _close_file_bracket(self) -> None:
+        if self.live is not None and self._file_open:
+            self.live.console.print(Text(self._FOOT, style=self._OUTER_STYLE))
+            self.live.console.print()
+        self._file_open = False
+
+    @staticmethod
+    def _file_header(f: str):
+        return Panel(
+            Text(f, style="bold blue"),
+            box=box.DOUBLE,
+            border_style="blue",
+            expand=False,
+            padding=(0, 1),
+        )
+
+    def _emit_stream_line(self, line: str) -> None:
+        """Forward one line of test output, railed inside both brackets."""
+        if self.live is None:
+            return
+        if line == "":
+            self.live.console.print(self._outer() + Text("\u2502", style="grey50"))
+            return
+        inner = Text(self._RAIL, style="grey50")
+        text = Text.from_ansi(line)
+        prefix_w = len(self._OUTER) + len(self._RAIL)
+        width = max(self.live.console.size.width - prefix_w, 1)
+        for seg in text.wrap(self.live.console, width):
+            self.live.console.print(self._outer() + inner + seg)
+
+    # --- scrollback emitters ----------------------------------------------
+    def _emit_test_line(self, nodeid: str, entry: dict) -> None:
+        outcome = entry["outcome"]
+        glyph, style = _OUTCOME_GLYPH.get(outcome, ("\u2022", "white"))
+        dur = entry["duration"]
+        timing = f"{dur * 1000:.0f}ms" if dur < 1 else f"{dur:.2f}s"
+        # Close the test bracket with the result, then a rail-only spacer line.
+        self.live.console.print(
+            self._outer()
+            + Text.assemble((self._CLOSE, style), (f"{glyph} ", style), (timing, "dim"))
+        )
+        self.live.console.print(self._outer())
+
+    def _emit_output_panel(self, nodeid: str, entry: dict) -> None:
+        outcome = entry["outcome"]
+        _, style = _OUTCOME_GLYPH.get(outcome, ("\u2022", "white"))
+        out = entry["out"].rstrip()
+        show = self.opts.show_output or outcome in ("failed", "error")
+        if out and show:
+            self.live.console.print(self._output_panel(nodeid, out, style))
+
+    def _output_panel(self, nodeid: str, out: str, style: str) -> Panel:
+        lines = out.splitlines()
+        hidden = len(lines) - _MAX_OUTPUT_LINES
+        if hidden > 0:
+            body = Text("\n".join(lines[-_MAX_OUTPUT_LINES:]))
+            body.append(f"\n… {hidden} earlier line(s) hidden", style="dim italic")
+        else:
+            body = Text(out)
+        return Panel(
+            body,
+            title=f"output · {nodeid}",
+            title_align="left",
+            border_style="grey42",
+            padding=(0, 1),
+        )
+
+    # --- rendering ---------------------------------------------------------
+    @property
+    def done(self) -> int:
+        return self.passed + self.failed + self.skipped + self.errors
+
+    def _refresh(self) -> None:
+        if self.live is not None:
+            try:
+                self.live.refresh()
+            except Exception:
+                pass
+
+    def __rich__(self):
+        return self.render_progress()
+
+    def render_progress(self):
+        # -s off → per-file counter table; -s on → overall bar + streaming spinner.
+        if self.opts.no_capture:
+            return self._render_streaming()
+        return self._render_by_file()
+
+    def _current_line(self):
+        """The animated 'currently running' spinner line (shared by both views)."""
+        if self.current:
+            elapsed = time.time() - self._current_start
+            self._spinner.update(
+                text=Text.assemble(
+                    (_short(self.current), "cyan"), (f"  {elapsed:0.1f}s", "cyan dim")
+                )
+            )
+            return self._spinner
+        if self.total and self.done >= self.total:
+            return Text("finished", style="green dim")
+        return Text("collecting…", style="cyan dim")
+
+    def _render_streaming(self):
+        bar = ProgressBar(total=max(self.total, 1), completed=self.done, width=46)
+        counts = Text.assemble(
+            ("  passed ", "dim"), (f"{self.passed}", "bold green"),
+            ("   failed ", "dim"), (f"{self.failed}", "bold red"),
+            ("   skipped ", "dim"), (f"{self.skipped}", "bold yellow"),
+            ("   errors ", "dim"), (f"{self.errors}", "bold magenta"),
+        )
+        progress_line = Text.assemble(
+            (f"{self.done}", "bold"), (f"/{self.total}  ", "dim"),
+        )
+        return Panel(
+            Group(Group(progress_line, bar), counts, self._current_line()),
+            title="[bold]running tests[/bold]",
+            border_style="cyan",
+        )
+
+    def _render_by_file(self):
+        if not self._file_order:
+            return Panel(
+                Text("collecting…", style="cyan dim"),
+                title="[bold]running tests[/bold]",
+                border_style="cyan",
+            )
+        table = Table(show_header=False, box=None, pad_edge=False, expand=False)
+        table.add_column(width=1)                 # running marker
+        table.add_column(no_wrap=True)            # file
+        table.add_column(justify="right")         # passed
+        table.add_column(justify="right")         # failed
+        table.add_column(justify="right")         # done/total
+        table.add_column(width=18)                # progress bar
+        for f in self._file_order:
+            total = self._file_total.get(f, 0)
+            passed = self._file_pass.get(f, 0)
+            failed = self._file_fail.get(f, 0)
+            skipped = self._file_skip.get(f, 0)
+            done = passed + failed + skipped
+            complete = total > 0 and done >= total
+            running = f == self._current_file and not complete and bool(self.current)
+
+            if running:
+                marker = Text("\u25b6", style="cyan")          # ▶
+            elif complete:
+                marker = Text("\u2713" if failed == 0 else "\u2717",
+                              style="green" if failed == 0 else "bold red")
+            else:
+                marker = Text(" ")
+
+            name_style = "bold red" if failed else ("green" if complete else
+                                                    "cyan" if running else "")
+            bar_style = "red" if failed else "green"
+            table.add_row(
+                marker,
+                Text(f, style=name_style),
+                Text(f"{passed}\u2713", style="green" if passed else "dim"),
+                Text(f"{failed}\u2717", style="bold red" if failed else "dim"),
+                Text(f"{done}/{total}", style="dim"),
+                ProgressBar(
+                    total=max(total, 1), completed=done, width=18,
+                    complete_style=bar_style, finished_style=bar_style,
+                ),
+            )
+        return Panel(
+            Group(table, self._current_line()),
+            title="[bold]running tests[/bold]",
+            border_style="cyan",
+        )
+
+    def render_summary(self, console: Console) -> None:
+        elapsed = time.time() - self._start
+        ok = self.failed == 0 and self.errors == 0
+        table = Table(show_header=False, box=None, pad_edge=False)
+        table.add_row(Text("passed", style="green"), Text(str(self.passed), style="bold green"))
+        if self.failed:
+            table.add_row(Text("failed", style="red"), Text(str(self.failed), style="bold red"))
+        if self.skipped:
+            table.add_row(Text("skipped", style="yellow"), Text(str(self.skipped), style="bold yellow"))
+        if self.errors:
+            table.add_row(Text("errors", style="magenta"), Text(str(self.errors), style="bold magenta"))
+        table.add_row(Text("time", style="dim"), Text(f"{elapsed:.2f}s", style="dim"))
+
+        verdict = "[bold green]PASSED[/]" if ok else "[bold red]FAILED[/]"
+        console.print(
+            Panel(table, title=verdict, border_style="green" if ok else "red")
+        )
+
+        for nodeid, longrepr in self.failures:
+            console.print(
+                Panel(
+                    Text(longrepr or "(no traceback captured)"),
+                    title=f"[red]{nodeid}[/]",
+                    border_style="red",
+                )
+            )
+
+
+def render_coverage(console: Console, rows: list, total: float) -> None:
+    """Render the coverage table from rows computed by the worker."""
+    table = Table(title="coverage", title_style="bold", header_style="dim")
+    table.add_column("file")
+    table.add_column("stmts", justify="right")
+    table.add_column("miss", justify="right")
+    table.add_column("cover", justify="right")
+
+    for rel, n, miss, pct in sorted(rows, key=lambda r: r[3]):
+        colour = "green" if pct >= 90 else "yellow" if pct >= 70 else "red"
+        table.add_row(rel, str(n), str(miss), Text(f"{pct:.0f}%", style=colour))
+
+    console.print(table)
+    if total is not None:
+        tcolour = "green" if total >= 90 else "yellow" if total >= 70 else "red"
+        console.print(Text.assemble(("TOTAL  ", "bold"), (f"{total:.0f}%", f"bold {tcolour}")))
+
+
+def _build_pytest_args(node_ids: list[str], opts: RunOptions) -> list[str]:
+    """Translate RunOptions into the pytest flags handed to the worker."""
+    args: list[str] = []
+    if opts.exitfirst:
+        args.append("-x")
+    if opts.last_failed:
+        args.append("--lf")
+    if opts.failed_first:
+        args.append("--ff")
+    if opts.verbose:
+        # `-v` is owned by pytest's terminal plugin, which the worker disables
+        # via `-p no:terminal`. The part of verbosity that still matters for our
+        # custom dashboard is untruncated assertion diffs in the failure panels,
+        # which this core ini option controls independently.
+        args += ["-o", "verbosity_assertions=2"]
+    if opts.no_capture:
+        # Disable pytest's output capture so test stdout / print() reaches the
+        # worker's gutter shim, which frames each line back to us as an event.
+        args.append("-s")
+    if opts.keyword:
+        args += ["-k", opts.keyword]
+    args += node_ids
+    return args
+
+
+def run_tests(node_ids: list[str], opts: RunOptions | None = None) -> int:
+    """Run the given node IDs in the test venv (subprocess). Returns the exit code."""
+    if opts is None:
+        opts = RunOptions()
+    if not node_ids:
+        node_ids = list(_CONFIG.test_paths)
+
+    console = Console(file=sys.stdout)
+    dashboard = Dashboard(opts)
+
+    worker_opts = {
+        "pytest_args": _build_pytest_args(node_ids, opts),
+        "no_capture": opts.no_capture,
+        "cov": opts.cov,
+        "coverage_source": _CONFIG.coverage_source,
+        "repo_root": str(_CONFIG.repo_root),
+    }
+
+    # The worker writes framed events to `w`; we read them from `r`. Its stderr
+    # goes to a temp file (not a pipe) so a chatty/​crashing worker can never
+    # deadlock against our event read.
+    r, w = os.pipe()
+    err_file = tempfile.TemporaryFile(mode="w+")
+    proc = subprocess.Popen(
+        [_CONFIG.venv_python, "-m", "tests.runner._worker", str(w), json.dumps(worker_opts)],
+        cwd=str(_CONFIG.repo_root),
+        pass_fds=(w,),
+        stdout=subprocess.DEVNULL,
+        stderr=err_file,
+    )
+    os.close(w)  # parent keeps only the read end
+
+    exit_code: int | None = None
+    saw_collection = False
+    cov_rows: list | None = None
+    cov_total: float | None = None
+    cov_error: str | None = None
+
+    with Live(
+        dashboard,
+        console=console,
+        refresh_per_second=12,
+        redirect_stdout=False,
+        redirect_stderr=False,
+    ) as live:
+        dashboard.live = live
+        with os.fdopen(r, "r", buffering=1) as stream:
+            for line in stream:
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    evt = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                etype = evt.get("type")
+                if etype == events.COLLECTION:
+                    saw_collection = True
+                elif etype == events.COVERAGE:
+                    cov_rows = evt.get("rows")
+                    cov_total = evt.get("total")
+                    cov_error = evt.get("error")
+                    continue
+                elif etype == events.DONE:
+                    exit_code = int(evt.get("code", 1))
+                    continue
+                dashboard.handle_event(evt)
+        if opts.no_capture:
+            dashboard._close_file_bracket()  # close the last file bracket
+        live.refresh()
+
+    proc.wait()
+    if exit_code is None:
+        exit_code = proc.returncode
+
+    dashboard.render_summary(console)
+    if cov_error:
+        console.print(f"[yellow]coverage unavailable: {cov_error}[/]")
+    elif cov_rows is not None:
+        render_coverage(console, cov_rows, cov_total)
+
+    # If the worker died before collecting anything (e.g. an import error in a
+    # test module, or a missing test venv), surface its stderr.
+    if not saw_collection and exit_code != 0:
+        err_file.seek(0)
+        stderr_text = err_file.read().strip()
+        if stderr_text:
+            console.print(
+                Panel(Text(stderr_text), title="[red]worker stderr[/]", border_style="red")
+            )
+    err_file.close()
+
+    return int(exit_code)
diff --git a/backend/tests/runner/tree.py b/backend/tests/runner/tree.py
new file mode 100644
index 00000000..b808151f
--- /dev/null
+++ b/backend/tests/runner/tree.py
@@ -0,0 +1,79 @@
+"""Turn a flat list of pytest node IDs into a selectable tree.
+
+Node ID shapes handled:
+    tests/unit/test_tokens.py::test_roundtrip
+    tests/unit/test_route_classifier.py::test_models[gemini-3-flash]   (parametrized)
+
+Directories and files become inner nodes; a parametrized function becomes an
+inner node whose leaves are its individual cases; a plain function is a leaf.
+Only leaves carry a runnable node_id.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class TNode:
+    label: str
+    kind: str  # 'dir' | 'file' | 'func' | 'case'
+    node_id: str | None = None  # set only on leaves (func without params, or case)
+    children: list["TNode"] = field(default_factory=list)
+
+    @property
+    def is_leaf(self) -> bool:
+        return not self.children
+
+    def _find_or_add(self, label: str, kind: str) -> "TNode":
+        for child in self.children:
+            if child.label == label and child.kind == kind:
+                return child
+        child = TNode(label=label, kind=kind)
+        self.children.append(child)
+        return child
+
+    def leaf_ids(self) -> list[str]:
+        if self.is_leaf:
+            return [self.node_id] if self.node_id else []
+        out: list[str] = []
+        for child in self.children:
+            out.extend(child.leaf_ids())
+        return out
+
+
+def build_tree(node_ids: list[str], root_label: str = "tests") -> TNode:
+    root = TNode(label=root_label, kind="dir")
+    for nid in node_ids:
+        file_part, _, test_part = nid.partition("::")
+        segments = [s for s in file_part.split("/") if s]
+        # Drop a leading "tests" segment so it nests under the single root.
+        if segments and segments[0] == root_label:
+            segments = segments[1:]
+
+        cursor = root
+        for seg in segments[:-1]:
+            cursor = cursor._find_or_add(seg, "dir")
+        if segments:
+            cursor = cursor._find_or_add(segments[-1], "file")
+
+        if "[" in test_part:
+            func_name = test_part.split("[", 1)[0]
+            func_node = cursor._find_or_add(func_name, "func")
+            func_node.children.append(
+                TNode(label=test_part, kind="case", node_id=nid)
+            )
+        else:
+            cursor.children.append(
+                TNode(label=test_part, kind="func", node_id=nid)
+            )
+    _sort(root)
+    return root
+
+
+def _sort(node: TNode) -> None:
+    # Dirs first, then files, then funcs/cases; alphabetical within a kind.
+    order = {"dir": 0, "file": 1, "func": 2, "case": 3}
+    node.children.sort(key=lambda c: (order.get(c.kind, 9), c.label))
+    for child in node.children:
+        _sort(child)