diff --git a/README.md b/README.md
index e489396..b3c6f99 100644
--- a/README.md
+++ b/README.md
@@ -154,6 +154,88 @@ See [examples/datatool_dashboard.py](examples/datatool_dashboard.py) for a full
 
 To regenerate the screenshots after UI changes, see [docs/generate_screenshots.py](docs/generate_screenshots.py).
 
+## Server-side downsampling
+
+Large time series are downsampled on the server so the plot only transports the
+points the current zoom level can show. This is driven by `PlotConfig.downsample`:
+
+```python
+from opensemantic.base.view._config import DashboardConfig, PlotConfig, DownsampleConfig
+
+config = DashboardConfig(
+    plot=PlotConfig(
+        downsample=DownsampleConfig(
+            enabled=True,      # downsample when the backend supports it
+            max_points=2000,   # target points per channel
+            method="auto",     # auto | sample | average | minmax
+            edge_anchors=True, # keep the window's first/last real datapoints
+        )
+    )
+)
+```
+
+It only engages on a PostgREST/TimescaleDB backend (the `downsample_tool_channel`
+RPC). On a SQLite/local backend, the RPC being absent, or any error, the read
+silently falls back to the full-resolution path - downsampling never breaks a
+read. The `DataToolView` plot also reloads at a finer resolution when you zoom in.
+
+Strategies (N = number of buckets):
+
+- `sample` (default): one real datapoint nearest each bucket center. Schema
+  agnostic; works for scalar and composite channels. N rows.
+- `average`: structure-preserving deep average per bucket (every numeric leaf
+  averaged, non-numeric keys carried), bucket-center timestamp. N rows.
+- `minmax`: the real min and max datapoint of every numeric sub-characteristic
+  per bucket. Scalar: 2N real rows; composite: the real per-leaf extremes. Best
+  at preserving spikes.
+- `auto`: `minmax` for numeric channels, `sample` for text channels.
+
+`average`/`minmax` skip non-numeric leaves (comments, labels) and fall back to
+`sample` when a channel has no numeric leaf at all.
+
+**Unit normalization caveat:** `average` and `minmax` compare and combine the
+bare stored numbers per leaf, so they are only correct when all stored values of
+a leaf share the same unit. The archive stores base-unit-normalized values, but
+data ingested without normalization (mixed units in one channel) will produce
+wrong `average`/`minmax` results. `sample` returns whole real rows and is
+unaffected.
+
+The RPC lives in pgstack's `postgres/config/optional/100_init_tsdb_schema.sql`
+(uses only core, Apache-2 TimescaleDB; no toolkit dependency). It is created at
+database init; apply it manually (`psql -f` / pgAdmin) on an already-running
+cluster. To measure the speedup, see
+[benchmarks/bench_downsample.py](benchmarks/bench_downsample.py).
+
+[examples/downsample_demo.py](examples/downsample_demo.py) is an interactive
+demo: four channels carry the same 100k-point signal (with narrow spikes), one
+per strategy - the channels are named `raw`, `sample`, `average`, `minmax`.
+Selecting `raw` loads slowly with full detail; `sample`/`average` load fast but
+drop the spikes; `minmax` keeps them. Box-zoom into a flat stretch and click
+"Load current range" to re-fetch that window at finer resolution - the hidden
+spikes reappear on `sample`/`average`. Needs a running pgstack with the RPC
+applied; seed once with `python examples/downsample_demo.py`, then
+`panel serve examples/downsample_demo.py`.
+
+![Downsampling demo](docs/downsample_demo.gif)
+
+![Strategy comparison](docs/screenshot_downsample_raw.png)
+
+*Full window, all four channels: `raw` and `minmax` keep the spikes; `sample`
+and `average` smooth them away at the coarse full-window resolution.*
+
+![Zoomed, before reload](docs/screenshot_downsample_zoom_select.png)
+
+*Box-zoomed around a spike on the `sample` channel: only the coarse full-window
+points are shown, so the spike is still hidden.*
+
+![Zoom reveals the peak](docs/screenshot_downsample_zoom.png)
+
+*After "Load current range": the window is re-fetched at finer buckets and the
+hidden spike reappears. The toolbar reset returns to the full window.*
+
+To regenerate these, see
+[docs/generate_downsample_screenshots.py](docs/generate_downsample_screenshots.py).
+
 ## ProcessObjectView (Process/Object Dashboard UI)
 
 Where `DataToolView` is centered on data tools, `ProcessObjectView` is centered
diff --git a/benchmarks/bench_downsample.py b/benchmarks/bench_downsample.py
new file mode 100644
index 0000000..7dd577b
--- /dev/null
+++ b/benchmarks/bench_downsample.py
@@ -0,0 +1,219 @@
+"""Benchmark the server-side downsampling RPC against a live pgstack.
+
+Seeds escalating series sizes for a scalar and a composite channel, then
+times a full-resolution read against each downsampling strategy and reports
+wall time, rows returned and approximate payload size, so the speedup vs
+full-resolution and the relative cost of the deep aggregates are visible.
+
+Gated on the same env vars as the integration tests; prints a skip notice
+otherwise:
+
+    TEST_PGRST_URL, TEST_PGRST_JWT_SECRET  (required)
+    TEST_PGRST_JWT_ROLE, TEST_PGRST_SCHEMA (optional)
+    BENCH_SIZES   comma-separated point counts (default "10000,100000")
+    BENCH_MAXPTS  target points per downsampled read (default 2000)
+    BENCH_REPEAT  timed repeats per case (default 3)
+    TEST_PG_DSN   optional libpq DSN; if psycopg is installed an
+                  EXPLAIN ANALYZE of one RPC call is printed.
+
+Run:  python benchmarks/bench_downsample.py
+"""
+
+import asyncio
+import datetime as dt
+import json
+import os
+import statistics
+import sys
+import time
+from pathlib import Path
+
+# Load tests/.env if present, mirroring tests/conftest.py, so the benchmark
+# can be run with the same configuration as the integration tests.
+_env_path = Path(__file__).resolve().parent.parent / "tests" / ".env"
+if _env_path.exists():
+    try:
+        from dotenv import load_dotenv
+
+        load_dotenv(_env_path)
+    except ImportError:
+        pass
+
+_URL = os.environ.get("TEST_PGRST_URL")
+_SECRET = os.environ.get("TEST_PGRST_JWT_SECRET")
+_ROLE = os.environ.get("TEST_PGRST_JWT_ROLE", "api_user")
+_SCHEMA = os.environ.get("TEST_PGRST_SCHEMA", "api")
+_SIZES = [int(s) for s in os.environ.get("BENCH_SIZES", "10000,100000").split(",")]
+_MAXPTS = int(os.environ.get("BENCH_MAXPTS", "2000"))
+_REPEAT = int(os.environ.get("BENCH_REPEAT", "3"))
+_PG_DSN = os.environ.get("TEST_PG_DSN")
+
+WRITE_CHUNK = 5000
+
+
+def _make_db():
+    import jwt
+    from postgrest import AsyncPostgrestClient
+
+    from opensemantic.base import PostgrestTimeSeriesDatabaseController
+    from opensemantic.core import Label
+
+    token = jwt.encode({"role": _ROLE}, _SECRET, algorithm="HS256")
+    client = AsyncPostgrestClient(
+        base_url=_URL,
+        schema=_SCHEMA,
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    db = PostgrestTimeSeriesDatabaseController(
+        name="bench", label=[Label(text="Bench")], buffered=False
+    )
+    db.set_client(client)
+    return db
+
+
+def _make_controller(db):
+    """A DataToolController with a scalar and a composite channel, bound to db."""
+    from uuid import uuid4
+
+    from opensemantic import compute_scoped_uuid
+    from opensemantic.base.v1 import DataChannel, DataToolController
+    from opensemantic.core.v1 import Label as LabelV1
+
+    parent = uuid4()
+    ctrl = DataToolController(
+        uuid=str(parent),
+        name="Bench",
+        label=[LabelV1(text="Bench")],
+        data_channels=[
+            DataChannel(
+                uuid=str(compute_scoped_uuid(parent, "scalar")),
+                osw_id="placeholder",
+                name="scalar",
+                label=[LabelV1(text="scalar")],
+            ),
+            DataChannel(
+                uuid=str(compute_scoped_uuid(parent, "composite")),
+                osw_id="placeholder",
+                name="composite",
+                label=[LabelV1(text="composite")],
+            ),
+        ],
+    )
+    ctrl.archive_database = db
+    return ctrl
+
+
+async def _timed_read(db, tool_id, ch, start, end, method):
+    from opensemantic.base import DownsampleParams, ReadToolChannelRawParams
+
+    ds = None
+    if method != "raw":
+        ds = DownsampleParams(max_points=_MAXPTS, method=method)
+    times = []
+    rows = []
+    for _ in range(_REPEAT):
+        t0 = time.perf_counter()
+        rows = await db.read_tool_channel_raw(
+            ReadToolChannelRawParams(
+                tool_osw_id=tool_id,
+                channel_osw_id=ch,
+                start=start,
+                end=end,
+                downsample=ds,
+            )
+        )
+        times.append((time.perf_counter() - t0) * 1000.0)
+    payload_kb = len(json.dumps(rows)) / 1024.0
+    return statistics.median(times), len(rows), payload_kb
+
+
+def _explain(tool_id, ch, start, end):
+    if not _PG_DSN:
+        return
+    try:
+        import psycopg
+    except Exception:
+        print("\n(psycopg not installed; skipping EXPLAIN ANALYZE)")
+        return
+    sql = (
+        "EXPLAIN (ANALYZE, BUFFERS) "
+        "SELECT * FROM api.downsample_tool_channel(%s, %s, %s, %s, %s, NULL, 'minmax')"
+    )
+    print("\nEXPLAIN ANALYZE (minmax):")
+    try:
+        with psycopg.connect(_PG_DSN) as conn, conn.cursor() as cur:
+            cur.execute(sql, (tool_id, ch, start, end, _MAXPTS))
+            for (line,) in cur.fetchall():
+                print("  " + line)
+    except Exception as e:
+        print(f"  EXPLAIN failed: {e}")
+
+
+async def _run():
+    from opensemantic.base import DeleteToolParams
+    from opensemantic.base._demo_data import seed_channel_series
+
+    db = _make_db()
+    header = (
+        f"{'size':>9} {'channel':>9} {'method':>8} {'rows':>8} {'ms':>9} {'KB':>10}"
+    )
+    print(header)
+    print("-" * len(header))
+    base = dt.datetime(2023, 1, 1, tzinfo=dt.timezone.utc)
+    for idx, n in enumerate(_SIZES):
+        is_last = idx == len(_SIZES) - 1
+        ctrl = _make_controller(db)
+        tool_id = ctrl.get_osw_id()
+        ch_scalar = ctrl.get_channel_by_name("scalar").get_osw_id().split("#")[-1]
+        ch_comp = ctrl.get_channel_by_name("composite").get_osw_id().split("#")[-1]
+
+        def _value(channel, i, _n=n):
+            if channel.name == "composite":
+                return {
+                    "temperature": {"value": float(i)},
+                    "humidity": {"value": float(_n - i)},
+                }
+            return {"value": float(i)}
+
+        try:
+            await seed_channel_series(
+                ctrl, n_points=n, base_ts=base, value_fn=_value, chunk_size=WRITE_CHUNK
+            )
+            await asyncio.sleep(1.0)  # let PostgREST settle after the writes
+            start, end = base, base + dt.timedelta(seconds=n - 1)
+            for label, ch in (("scalar", ch_scalar), ("composite", ch_comp)):
+                for method in ("raw", "sample", "average", "minmax"):
+                    ms, rows, kb = await _timed_read(
+                        db, tool_id, ch, start, end, method
+                    )
+                    print(
+                        f"{n:>9} {label:>9} {method:>8} {rows:>8} "
+                        f"{ms:>9.1f} {kb:>10.1f}"
+                    )
+            if is_last:
+                # EXPLAIN while the tool still exists (largest size).
+                _explain(tool_id, ch_scalar, start, end)
+        finally:
+            try:
+                await db.delete_tool(DeleteToolParams(tool_osw_id=tool_id))
+            except Exception:
+                pass
+
+
+def main():
+    if not (_URL and _SECRET):
+        print(
+            "Skipping benchmark: set TEST_PGRST_URL and TEST_PGRST_JWT_SECRET "
+            "to run against a live pgstack."
+        )
+        return 0
+    try:
+        asyncio.run(_run())
+    except Exception as e:
+        print(f"Benchmark failed (is pgstack up and the RPC applied?): {e}")
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/docs/_screenshot_utils.py b/docs/_screenshot_utils.py
new file mode 100644
index 0000000..584b282
--- /dev/null
+++ b/docs/_screenshot_utils.py
@@ -0,0 +1,91 @@
+"""Shared Playwright/Panel helpers for the docs screenshot generators.
+
+Used by generate_downsample_screenshots.py and generate_process_screenshots.py
+so the Wunderbaum-checkbox clicking, frame capture and panel-serve lifecycle
+live in one place.
+"""
+
+import glob
+import io
+import os
+import subprocess
+import sys
+import time
+
+import imageio.v3 as iio
+
+DOCS_DIR = os.path.dirname(os.path.abspath(__file__))
+PACKAGE_DIR = os.path.dirname(DOCS_DIR)
+
+
+def all_wb_shadows_js():
+    """JS defining allWbShadows(root): every Wunderbaum shadow root, in order."""
+    return """
+    function allWbShadows(root) {
+        const out = [];
+        function rec(r) {
+            for (const el of r.querySelectorAll('*')) {
+                if (el.shadowRoot) {
+                    if (el.shadowRoot.querySelectorAll('.wb-row').length > 0)
+                        out.push(el.shadowRoot);
+                    rec(el.shadowRoot);
+                }
+            }
+        }
+        rec(root);
+        return out;
+    }
+    """
+
+
+def click_tree_checkbox(page, tree_idx, cb_idx):
+    """Click the cb_idx-th checkbox of the tree_idx-th Wunderbaum (0 = first)."""
+    page.evaluate(
+        f"""() => {{
+        {all_wb_shadows_js()}
+        const sh = allWbShadows(document)[{tree_idx}];
+        if (sh) {{
+            const cbs = sh.querySelectorAll('i.wb-checkbox');
+            if (cbs[{cb_idx}]) cbs[{cb_idx}].click();
+        }}
+    }}"""
+    )
+
+
+def capture(page, frames, delay=800):
+    """Wait `delay` ms, then append a screenshot to `frames`."""
+    page.wait_for_timeout(delay)
+    frames.append(iio.imread(io.BytesIO(page.screenshot())))
+
+
+def start_server(example, port, env=None, clean_sqlite=False, settle=12):
+    """Start ``panel serve <example> --port <port>`` and return the process.
+
+    ``clean_sqlite`` first removes leftover ``*_db.sqlite`` in the package dir
+    (so a SQLite demo reseeds fresh). ``settle`` is the seconds to wait for the
+    first module execution before returning.
+    """
+    if clean_sqlite:
+        for db in glob.glob(os.path.join(PACKAGE_DIR, "*_db.sqlite")):
+            os.remove(db)
+    logf = open(os.path.join(PACKAGE_DIR, f"_gen_server_{port}.log"), "w")
+    proc_env = dict(os.environ)
+    if env:
+        proc_env.update(env)
+    proc = subprocess.Popen(
+        [sys.executable, "-m", "panel", "serve", example, "--port", str(port)],
+        stdout=logf,
+        stderr=subprocess.STDOUT,
+        cwd=PACKAGE_DIR,
+        env=proc_env,
+    )
+    time.sleep(settle)
+    return proc
+
+
+def stop_server(proc):
+    proc.terminate()
+    try:
+        proc.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        proc.kill()
diff --git a/docs/downsample_demo.gif b/docs/downsample_demo.gif
new file mode 100644
index 0000000..345b190
Binary files /dev/null and b/docs/downsample_demo.gif differ
diff --git a/docs/generate_downsample_screenshots.py b/docs/generate_downsample_screenshots.py
new file mode 100644
index 0000000..ac22e57
--- /dev/null
+++ b/docs/generate_downsample_screenshots.py
@@ -0,0 +1,315 @@
+"""Generate screenshots and a GIF for the downsampling demo README.
+
+Serves examples/downsample_demo.py (which reads from a running pgstack with the
+downsampling RPC applied and pre-seeded data) and drives the UI with Playwright.
+
+The script tells the whole story on a single full-window server:
+  1. Full window: select each channel (one strategy each). minmax keeps the
+     spikes, sample/average lose them at the coarse full-window resolution, raw
+     shows full detail (slow 100k load).
+  2. Zoom range: narrow the shared x-range to a window around the first spike
+     (this is what a horizontal box-zoom produces) - the spikes are still
+     missing because only the already-loaded coarse points are shown.
+  3. Load current range: click the button - the zoomed window is re-fetched at
+     finer buckets and the hidden spikes reappear on sample/average.
+
+Steps 2-3 exercise the real interactive feature (set zoom range -> "Load
+current range" re-fetch), so this script doubles as an end-to-end validation of
+the zoom path: if the zoomed window were loaded at the wrong position (e.g. a
+timezone offset bug), the spike would not land inside the narrowed axis.
+
+Prerequisites:
+    pip install playwright imageio
+    playwright install chromium
+    # a running pgstack on localhost:3000 with the demo data seeded
+    #   (run `python examples/downsample_demo.py` once to seed)
+
+Usage:
+    python docs/generate_downsample_screenshots.py
+"""
+
+import datetime as dt
+import os
+
+import imageio.v3 as iio
+from _screenshot_utils import (
+    capture,
+    click_tree_checkbox,
+    start_server,
+    stop_server,
+)
+from playwright.sync_api import sync_playwright
+
+DOCS_DIR = os.path.dirname(os.path.abspath(__file__))
+PACKAGE_DIR = os.path.dirname(DOCS_DIR)
+EXAMPLE = os.path.join(PACKAGE_DIR, "examples", "downsample_demo.py")
+VIEWPORT = {"width": 1400, "height": 1300}
+
+# Tree checkbox order: 0 = DataTool root, then channels in declaration order
+# (raw, sample, average, minmax).
+CB_RAW = 1
+CB_SAMPLE = 2
+CB_AVERAGE = 3
+CB_MINMAX = 4
+
+# Demo series constants (mirror examples/downsample_demo.py) used to compute the
+# zoom window directly, instead of reading possibly-stale Bokeh range models.
+BASE_TS = dt.datetime(2024, 1, 1, tzinfo=dt.timezone.utc)
+# Zoom window in data seconds, around the first spike (at 12 000 s) so the finer
+# buckets reveal it on sample/average after "Load current range".
+ZOOM_SEC = (4_000, 20_000)
+
+LOAD_RANGE_LABEL = "Load current range"
+
+
+def axis_ms(sec):
+    """Epoch-ms x-axis position of data second ``sec``.
+
+    The plot draws timestamps in local wall time (naive), so a box-zoom range is
+    expressed as the local-wall-clock epoch. Mirror that: take the UTC instant
+    ``BASE_TS + sec``, read it as local wall time, and return that as an epoch.
+    """
+    utc = BASE_TS + dt.timedelta(seconds=sec)
+    naive_local = utc.astimezone().replace(tzinfo=None)
+    return naive_local.replace(tzinfo=dt.timezone.utc).timestamp() * 1000.0
+
+
+def click_button(page, label):
+    """Click the first <button> whose text contains label (across shadow DOMs)."""
+    return page.evaluate(
+        """(l) => {
+            function f(r) {
+                for (const b of r.querySelectorAll('button')) {
+                    if ((b.textContent || '').includes(l)) { b.click(); return true; }
+                }
+                for (const e of r.querySelectorAll('*')) {
+                    if (e.shadowRoot && f(e.shadowRoot)) return true;
+                }
+                return false;
+            }
+            return f(document);
+        }""",
+        label,
+    )
+
+
+def read_time_xrange(page):
+    """Return [start, end] (epoch ms) of the shared datetime x-range, or None.
+
+    The time axis is the DataRange1d with epoch-ms bounds (>1e11); the value
+    axis has small bounds and is filtered out.
+    """
+    return page.evaluate(
+        """() => {
+            const docs = (window.Bokeh && window.Bokeh.documents) || [];
+            for (const doc of docs) {
+                const models = doc._all_models
+                    ? Array.from(doc._all_models.values()) : [];
+                for (const m of models) {
+                    if (m.type === 'DataRange1d' && m.start != null
+                        && m.end != null && m.end > 1e11) {
+                        return [m.start, m.end];
+                    }
+                }
+            }
+            return null;
+        }"""
+    )
+
+
+def set_time_xrange(page, start, end):
+    """Set every datetime DataRange1d to [start, end] (simulates a box-zoom)."""
+    page.evaluate(
+        """([s, e]) => {
+            const docs = (window.Bokeh && window.Bokeh.documents) || [];
+            for (const doc of docs) {
+                const models = doc._all_models
+                    ? Array.from(doc._all_models.values()) : [];
+                for (const m of models) {
+                    if (m.type === 'DataRange1d' && m.end > 1e11) {
+                        m.setv({start: s, end: e});
+                    }
+                }
+            }
+        }""",
+        [start, end],
+    )
+
+
+def click_reset_tool(page):
+    """Real pointer click on the Bokeh reset tool icon (across shadow DOMs).
+
+    figure.on_event(Reset) does not propagate through Panel, and a synthetic
+    el.click() does not trigger the Bokeh pointer handler, so locate the reset
+    icon and dispatch a real mouse down/up on its center.
+    """
+    box = page.evaluate(
+        """() => {
+            function f(r) {
+                for (const el of r.querySelectorAll('.bk-tool-icon-reset')) {
+                    const b = el.getBoundingClientRect();
+                    return {x: b.x + b.width / 2, y: b.y + b.height / 2};
+                }
+                for (const e of r.querySelectorAll('*')) {
+                    if (e.shadowRoot) { const g = f(e.shadowRoot); if (g) return g; }
+                }
+                return null;
+            }
+            return f(document);
+        }"""
+    )
+    if not box:
+        return False
+    page.mouse.move(box["x"], box["y"])
+    page.mouse.down()
+    page.wait_for_timeout(60)
+    page.mouse.up()
+    return True
+
+
+def max_line_points(page):
+    """Largest point count across line data sources (0 if no data yet)."""
+    return page.evaluate(
+        """() => {
+            let mx = 0;
+            const docs = window.Bokeh.documents || [];
+            for (const d of docs)
+                for (const m of (d._all_models
+                    ? Array.from(d._all_models.values()) : []))
+                    if (m.type === 'ColumnDataSource' && m.data
+                        && m.data.x && m.data.x.length)
+                        mx = Math.max(mx, m.data.x.length);
+            return mx;
+        }"""
+    )
+
+
+def wait_for_data(page, min_points=100, timeout_ms=30000, settle_ms=1500):
+    """Poll until a line source has >= min_points, then let it settle.
+
+    The demo loads asynchronously (lazy PostgREST client, serialized reloads);
+    capturing or zooming before the load settles races with a late figure
+    rebuild that reverts the view. Returns the final point count.
+    """
+    waited = 0
+    while waited < timeout_ms:
+        if max_line_points(page) >= min_points:
+            page.wait_for_timeout(settle_ms)
+            return max_line_points(page)
+        page.wait_for_timeout(500)
+        waited += 500
+    return max_line_points(page)
+
+
+def set_zoom_range(page, zs, ze, tries=4):
+    """Set the zoom window and confirm no figure rebuild reverted it."""
+    for _ in range(tries):
+        set_time_xrange(page, zs, ze)
+        page.wait_for_timeout(700)
+        cur = read_time_xrange(page)
+        if cur and abs(cur[0] - zs) < 1e6 and abs(cur[1] - ze) < 1e6:
+            return True
+    return False
+
+
+def main():
+    frames = []
+    compare_frame = raw_frame = zoom_select_frame = zoom_frame = None
+
+    proc = start_server(EXAMPLE, 5013)
+    try:
+        with sync_playwright() as p:
+            page = p.chromium.launch(headless=True).new_page(viewport=VIEWPORT)
+            page.goto(
+                "http://localhost:5013/downsample_demo",
+                timeout=120000,
+                wait_until="domcontentloaded",
+            )
+            page.wait_for_timeout(9000)
+            capture(page, frames, 800)  # empty
+
+            # -- Pass 1: full-window comparison, one strategy per channel --
+            # minmax keeps the spikes; sample/average lose them at the coarse
+            # full-window resolution; raw shows full detail (slow 100k load).
+            print("Pass 1: full-window comparison...")
+            click_tree_checkbox(page, 0, CB_MINMAX)  # spikes kept
+            capture(page, frames, 5000)
+            click_tree_checkbox(page, 0, CB_SAMPLE)  # spikes lost
+            capture(page, frames, 5000)
+            click_tree_checkbox(page, 0, CB_AVERAGE)  # spikes smoothed
+            print(f"  loaded points: {wait_for_data(page)}")
+            capture(page, frames, 1500)
+            compare_frame = frames[-1]
+
+            click_tree_checkbox(page, 0, CB_RAW)  # full detail, slow
+            raw_pts = wait_for_data(page, min_points=50000, timeout_ms=90000)
+            print(f"  raw points: {raw_pts}")
+            capture(page, frames, 1500)
+            raw_frame = frames[-1]
+
+            # -- Pass 2: interactive zoom on a single channel --
+            # Driven on one figure (sample) on a fresh session: with a single
+            # x-range there are no linked/stale-range races, so the box-zoom ->
+            # "Load current range" -> reset flow is captured reliably and the
+            # spike-reveal is unambiguous.
+            print("Pass 2: interactive zoom on 'sample'...")
+            page.goto(
+                "http://localhost:5013/downsample_demo",
+                timeout=120000,
+                wait_until="domcontentloaded",
+            )
+            page.wait_for_timeout(9000)
+            click_tree_checkbox(page, 0, CB_SAMPLE)
+            print(f"  sample points: {wait_for_data(page)}")
+            capture(page, frames, 1500)  # full window, spike hidden in coarse data
+
+            # Zoom range (what a horizontal box-zoom produces); spikes still
+            # missing because only the coarse full-window points are shown.
+            zs, ze = axis_ms(ZOOM_SEC[0]), axis_ms(ZOOM_SEC[1])
+            print(f"  zoom x-range (epoch ms): [{zs:.0f}, {ze:.0f}]")
+            print(f"  zoom set confirmed: {set_zoom_range(page, zs, ze)}")
+            capture(page, frames, 2500)
+            zoom_select_frame = frames[-1]
+
+            # Load current range: re-fetch the zoomed window at finer buckets;
+            # the hidden spike reappears.
+            clicked = click_button(page, LOAD_RANGE_LABEL)
+            print(f"  '{LOAD_RANGE_LABEL}' clicked: {clicked}")
+            if not clicked:
+                raise RuntimeError(f"button '{LOAD_RANGE_LABEL}' not found")
+            print(f"  reloaded points: {wait_for_data(page)}")
+            capture(page, frames, 1500)
+            zoom_frame = frames[-1]
+
+            # Toolbar reset returns to the full window.
+            reset_ok = click_reset_tool(page)
+            print(f"  reset tool clicked: {reset_ok}")
+            wait_for_data(page)
+            capture(page, frames, 2000)
+    finally:
+        stop_server(proc)
+
+    iio.imwrite(
+        os.path.join(DOCS_DIR, "downsample_demo.gif"), frames, duration=1400, loop=0
+    )
+    print(f"downsample_demo.gif: {len(frames)} frames")
+    if compare_frame is not None:
+        iio.imwrite(
+            os.path.join(DOCS_DIR, "screenshot_downsample_compare.png"), compare_frame
+        )
+    if raw_frame is not None:
+        iio.imwrite(os.path.join(DOCS_DIR, "screenshot_downsample_raw.png"), raw_frame)
+    if zoom_select_frame is not None:
+        iio.imwrite(
+            os.path.join(DOCS_DIR, "screenshot_downsample_zoom_select.png"),
+            zoom_select_frame,
+        )
+    if zoom_frame is not None:
+        iio.imwrite(
+            os.path.join(DOCS_DIR, "screenshot_downsample_zoom.png"), zoom_frame
+        )
+    print("Screenshots saved")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/docs/generate_process_screenshots.py b/docs/generate_process_screenshots.py
index ae484bd..3f99c03 100644
--- a/docs/generate_process_screenshots.py
+++ b/docs/generate_process_screenshots.py
@@ -8,14 +8,15 @@
     python docs/generate_process_screenshots.py
 """
 
-import glob
-import io
 import os
-import subprocess
-import sys
-import time
 
 import imageio.v3 as iio
+from _screenshot_utils import (
+    capture,
+    click_tree_checkbox,
+    start_server,
+    stop_server,
+)
 from playwright.sync_api import sync_playwright
 
 DOCS_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -26,41 +27,6 @@
 VIEWPORT = {"width": 1400, "height": 900}
 
 
-def all_wb_shadows_js():
-    """JS that returns all Wunderbaum shadow roots in DOM order."""
-    return """
-    function allWbShadows(root) {
-        const out = [];
-        function rec(r) {
-            for (const el of r.querySelectorAll('*')) {
-                if (el.shadowRoot) {
-                    if (el.shadowRoot.querySelectorAll('.wb-row').length > 0)
-                        out.push(el.shadowRoot);
-                    rec(el.shadowRoot);
-                }
-            }
-        }
-        rec(root);
-        return out;
-    }
-    """
-
-
-def click_tree_checkbox(page, tree_idx, cb_idx):
-    """Click the cb_idx-th checkbox of the tree_idx-th Wunderbaum (0=objects)."""
-    page.evaluate(
-        f"""() => {{
-        {all_wb_shadows_js()}
-        const shadows = allWbShadows(document);
-        const sh = shadows[{tree_idx}];
-        if (sh) {{
-            const cbs = sh.querySelectorAll('i.wb-checkbox');
-            if (cbs[{cb_idx}]) cbs[{cb_idx}].click();
-        }}
-    }}"""
-    )
-
-
 def switch_temperature_unit(page):
     """Switch the Temperature unit dropdown to Celsius."""
     page.evaluate(
@@ -94,40 +60,9 @@ def switch_temperature_unit(page):
     )
 
 
-def capture(page, frames, delay=500):
-    page.wait_for_timeout(delay)
-    buf = page.screenshot()
-    frames.append(iio.imread(io.BytesIO(buf)))
-
-
-def start_server():
-    """Start the Panel server, cleaning old archive DBs for fresh data."""
-    for db in glob.glob(os.path.join(PACKAGE_DIR, "*_db.sqlite")):
-        os.remove(db)
-    proc = subprocess.Popen(
-        [sys.executable, "-m", "panel", "serve", EXAMPLE, "--port", str(PORT)],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
-        cwd=PACKAGE_DIR,
-    )
-    # Let the import + first module execution settle. panel serve re-runs the
-    # script per session (storing the demo data each time), so the first page
-    # load is slow - handled by a long goto timeout below.
-    time.sleep(10)
-    return proc
-
-
-def stop_server(proc):
-    proc.terminate()
-    try:
-        proc.wait(timeout=5)
-    except subprocess.TimeoutExpired:
-        proc.kill()
-
-
 def main():
     print("Starting Panel server...")
-    proc = start_server()
+    proc = start_server(EXAMPLE, PORT, clean_sqlite=True, settle=10)
     try:
         frames = []
         with sync_playwright() as p:
diff --git a/docs/process_demo.gif b/docs/process_demo.gif
index b179ede..41b6339 100644
Binary files a/docs/process_demo.gif and b/docs/process_demo.gif differ
diff --git a/docs/screenshot_downsample_compare.png b/docs/screenshot_downsample_compare.png
new file mode 100644
index 0000000..8b2c8ae
Binary files /dev/null and b/docs/screenshot_downsample_compare.png differ
diff --git a/docs/screenshot_downsample_raw.png b/docs/screenshot_downsample_raw.png
new file mode 100644
index 0000000..e91712c
Binary files /dev/null and b/docs/screenshot_downsample_raw.png differ
diff --git a/docs/screenshot_downsample_zoom.png b/docs/screenshot_downsample_zoom.png
new file mode 100644
index 0000000..d657a28
Binary files /dev/null and b/docs/screenshot_downsample_zoom.png differ
diff --git a/docs/screenshot_downsample_zoom_select.png b/docs/screenshot_downsample_zoom_select.png
new file mode 100644
index 0000000..704ad47
Binary files /dev/null and b/docs/screenshot_downsample_zoom_select.png differ
diff --git a/docs/screenshot_process_fanout.png b/docs/screenshot_process_fanout.png
index 5046c7c..2c7bc9c 100644
Binary files a/docs/screenshot_process_fanout.png and b/docs/screenshot_process_fanout.png differ
diff --git a/docs/screenshot_process_heating.png b/docs/screenshot_process_heating.png
index 74c72e8..75b915b 100644
Binary files a/docs/screenshot_process_heating.png and b/docs/screenshot_process_heating.png differ
diff --git a/docs/screenshot_process_overlay.png b/docs/screenshot_process_overlay.png
index 05d2b49..f71ebd5 100644
Binary files a/docs/screenshot_process_overlay.png and b/docs/screenshot_process_overlay.png differ
diff --git a/docs/screenshot_process_units.png b/docs/screenshot_process_units.png
index 2b78d64..9471cf6 100644
Binary files a/docs/screenshot_process_units.png and b/docs/screenshot_process_units.png differ
diff --git a/examples/downsample_demo.py b/examples/downsample_demo.py
new file mode 100644
index 0000000..b5db733
--- /dev/null
+++ b/examples/downsample_demo.py
@@ -0,0 +1,296 @@
+"""Example: server-side downsampling, one strategy per channel.
+
+Four channels carry the *same* 100,000-point signal (a slow baseline with a
+few narrow spikes). Each channel is named after - and plotted with - a different
+downsampling strategy, so the trade-offs are visible side by side:
+
+    raw      no downsampling: full detail, slow to load
+    sample   one real point per bucket: spikes often lost
+    average  bucket mean: spikes smoothed away
+    minmax   real per-bucket extremes: spikes preserved
+
+What to try:
+  - Check "raw": note the longer load (100k points transported).
+  - Check the others: they load fast (~2000 points), but the spikes are missing
+    on "sample" and "average", while "minmax" keeps them.
+  - Box-zoom into a flat stretch (drag horizontally), then click "Load current
+    range": the zoomed window is re-fetched at finer resolution and the hidden
+    spikes reappear on "sample"/"average". Box-zoom alone is just visual; the
+    toolbar "reset" returns to the full view.
+
+Requires a running pgstack (TimescaleDB + PostgREST) with the downsampling RPC
+applied. Configure via env vars (defaults match the local dev stack):
+
+    DEMO_PGRST_URL          (default http://localhost:3000)
+    DEMO_PGRST_JWT_SECRET   (default reallyreallyreallyreallyverysafe)
+    DEMO_PGRST_JWT_ROLE     (default api_user)
+    DEMO_PGRST_SCHEMA       (default api)
+
+Seed the data once, then serve:
+    python examples/downsample_demo.py          # seeds 100k x 4 channels
+    panel serve examples/downsample_demo.py --dev
+"""
+
+import asyncio
+import datetime as dt
+import math
+import os
+from typing import Optional
+from uuid import NAMESPACE_URL, uuid5
+
+import panel as pn
+
+from opensemantic import compute_scoped_uuid
+from opensemantic.base._demo_data import already_seeded, seed_channel_series
+from opensemantic.base.v1 import (
+    DataChannel,
+    DataToolController,
+    PostgrestTimeSeriesDatabaseController,
+)
+from opensemantic.base.view import DataToolView
+from opensemantic.base.view._config import DashboardConfig, DownsampleConfig, PlotConfig
+from opensemantic.characteristics.quantitative.v1 import Characteristic
+from opensemantic.core.v1 import Label
+
+pn.extension()
+
+N_POINTS = 100_000
+MAX_POINTS = 2000
+# Narrow spikes (a few points wide): lost by sample/average at the full-window
+# resolution (~50 s buckets), but revealed once you zoom in (finer buckets).
+SPIKE_CENTERS = (12_000, 37_000, 62_000, 87_000)
+SPIKE_WIDTH = 5
+SPIKE_AMPLITUDE = 35.0
+
+# Channels are named after the downsampling strategy used to plot them.
+# "raw" means no downsampling (method=None).
+METHOD_BY_CHANNEL = {
+    "raw": None,
+    "sample": "sample",
+    "average": "average",
+    "minmax": "minmax",
+}
+
+_URL = os.environ.get("DEMO_PGRST_URL", "http://localhost:3000")
+_SECRET = os.environ.get("DEMO_PGRST_JWT_SECRET", "reallyreallyreallyreallyverysafe")
+_ROLE = os.environ.get("DEMO_PGRST_JWT_ROLE", "api_user")
+_SCHEMA = os.environ.get("DEMO_PGRST_SCHEMA", "api")
+
+BASE_TS = dt.datetime(2024, 1, 1, tzinfo=dt.timezone.utc)
+
+
+class _Signal(Characteristic):
+    """Base for the demo's plain numeric characteristics."""
+
+    value: Optional[float] = None
+
+
+# One distinct characteristic per channel so each lands in its OWN plot
+# (grouping is by characteristic). The underlying values are identical; only
+# the downsampling strategy differs, so the per-plot effect is easy to see.
+class RawSignal(_Signal):
+    class Config:
+        schema_extra = {"uuid": "00000000-0000-0000-0000-0000000000d1", "title": "raw"}
+
+    type: Optional[list] = ["Category:OSW000000000000000000000000000000d1"]
+
+
+class SampleSignal(_Signal):
+    class Config:
+        schema_extra = {
+            "uuid": "00000000-0000-0000-0000-0000000000d2",
+            "title": "sample",
+        }
+
+    type: Optional[list] = ["Category:OSW000000000000000000000000000000d2"]
+
+
+class AverageSignal(_Signal):
+    class Config:
+        schema_extra = {
+            "uuid": "00000000-0000-0000-0000-0000000000d3",
+            "title": "average",
+        }
+
+    type: Optional[list] = ["Category:OSW000000000000000000000000000000d3"]
+
+
+class MinmaxSignal(_Signal):
+    class Config:
+        schema_extra = {
+            "uuid": "00000000-0000-0000-0000-0000000000d4",
+            "title": "minmax",
+        }
+
+    type: Optional[list] = ["Category:OSW000000000000000000000000000000d4"]
+
+
+_CHARACTERISTICS = {
+    "raw": RawSignal,
+    "sample": SampleSignal,
+    "average": AverageSignal,
+    "minmax": MinmaxSignal,
+}
+
+
+try:
+    from oold.model import _types
+
+    for _c in _CHARACTERISTICS.values():
+        _types[_c.get_cls_iri()] = _c
+except ImportError:
+    pass
+
+
+class DownsampleDemoView(DataToolView):
+    """DataToolView that forces a fixed downsampling method per channel name."""
+
+    def _downsample_for(self, channel):
+        method = METHOD_BY_CHANNEL.get(getattr(channel, "name", ""), "sample")
+        if method is None:
+            return None, None, None  # raw / full resolution
+        ds = self._config.plot.downsample
+        return ds.max_points, method, ds.edge_anchors
+
+    async def _load_and_plot(self):
+        # Construct the PostgREST client on the server's event loop on first
+        # use, then run the normal load. Serialize concurrent loads (a fast
+        # selection plus the zoom-reload callback can otherwise overlap and
+        # render a stale, collapsed figure); re-run once if one was requested
+        # while loading.
+        _ensure_client(self._controllers[0].archive_database)
+        if getattr(self, "_loading", False):
+            self._reload_pending = True
+            return
+        self._loading = True
+        try:
+            await super()._load_and_plot()
+            while getattr(self, "_reload_pending", False):
+                self._reload_pending = False
+                await super()._load_and_plot()
+        finally:
+            self._loading = False
+
+
+def _make_client():
+    import jwt
+    from postgrest import AsyncPostgrestClient
+
+    token = jwt.encode({"role": _ROLE}, _SECRET, algorithm="HS256")
+    return AsyncPostgrestClient(
+        base_url=_URL,
+        schema=_SCHEMA,
+        headers={"Authorization": f"Bearer {token}"},
+    )
+
+
+def _make_db():
+    # No client here: the async client must be constructed on the event loop
+    # that will use it (see _ensure_client). Building it at import time, off any
+    # running loop, makes its anyio primitives bind to the wrong loop and every
+    # request fails with "Server disconnected" under panel serve.
+    return PostgrestTimeSeriesDatabaseController(
+        name="downsample_demo",
+        label=[Label(text="Downsample Demo DB")],
+        buffered=False,
+    )
+
+
+def _ensure_client(db):
+    """Lazily attach the PostgREST client (call from the running loop)."""
+    driver = getattr(db, "_driver", None)
+    if driver is not None and driver.client is None:
+        db.set_client(_make_client())
+
+
+def _signal(i: int) -> float:
+    """Slow baseline plus a narrow spike at a few index ranges."""
+    v = 10.0 + 5.0 * math.sin(2.0 * math.pi * i / 8000.0)
+    if any(c <= i < c + SPIKE_WIDTH for c in SPIKE_CENTERS):
+        v += SPIKE_AMPLITUDE
+    return v
+
+
+def _build_controller():
+    parent = uuid5(NAMESPACE_URL, "OSW-DownsampleDemoTool")
+    channels = [
+        DataChannel(
+            uuid=str(compute_scoped_uuid(parent, name)),
+            osw_id="placeholder",
+            name=name,
+            label=[Label(text=name)],
+            characteristic=_CHARACTERISTICS[name].get_cls_iri(),
+        )
+        for name in METHOD_BY_CHANNEL
+    ]
+    ctrl = DataToolController(
+        uuid=str(parent),
+        name="DownsampleDemo",
+        label=[Label(text="Downsample Demo Tool")],
+        data_channels=channels,
+        auto_archive=False,
+    )
+    return ctrl
+
+
+async def _seed_if_needed(ctrl):
+    _ensure_client(ctrl.archive_database)
+    if await already_seeded(ctrl):
+        print(f"Tool {ctrl.get_osw_id()} already seeded; skipping.")
+        return
+    print(f"Seeding {N_POINTS} points x {len(ctrl.get_all_channels())} channels...")
+    written = await seed_channel_series(
+        ctrl,
+        n_points=N_POINTS,
+        base_ts=BASE_TS,
+        value_fn=lambda ch, i: {"value": _signal(i)},
+    )
+    print(f"Seeded {written} rows for {ctrl.get_osw_id()}")
+
+
+# Build synchronously with NO network I/O at import time. ``panel serve``
+# re-runs this module per session; doing async PostgREST I/O here (under
+# nest_asyncio) is unreliable and would blank the app. The reads happen
+# lazily on the server's own event loop when a channel is selected.
+controller = _build_controller()
+controller.archive_database = _make_db()
+
+config = DashboardConfig(
+    lang="en",
+    plot=PlotConfig(
+        auto_fetch=True,
+        row_limit=N_POINTS,
+        downsample=DownsampleConfig(enabled=True, max_points=MAX_POINTS),
+    ),
+)
+
+view = DownsampleDemoView(
+    controllers=[controller],
+    config=config,
+    title="Downsampling Demo (raw / sample / average / minmax)",
+)
+# One plot per characteristic (channel). The plots stack at full height and the
+# main content area scrolls, so all four strategies can be compared by scrolling.
+# Time window. Defaults to the full series; DEMO_WINDOW="start_sec,end_sec"
+# (seconds from BASE_TS) narrows it - used to capture the "zoomed in" view
+# where sample/average reload at finer buckets and the spikes reappear.
+_win = os.environ.get("DEMO_WINDOW")
+if _win:
+    _a, _b = (int(x) for x in _win.split(","))
+    view.set_time_range(
+        BASE_TS + dt.timedelta(seconds=_a),
+        BASE_TS + dt.timedelta(seconds=_b),
+        fetch=False,
+    )
+else:
+    view.set_time_range(BASE_TS, BASE_TS + dt.timedelta(seconds=N_POINTS), fetch=False)
+
+view.servable()
+
+
+if __name__ == "__main__":
+    # Seed the demo data once, before serving:
+    #   python examples/downsample_demo.py
+    #   panel serve examples/downsample_demo.py --dev
+    asyncio.run(_seed_if_needed(controller))
+    print("Seeding complete. Now run: panel serve examples/downsample_demo.py")
diff --git a/src/opensemantic/base/_controller.py b/src/opensemantic/base/_controller.py
index 72eafd0..7a82208 100644
--- a/src/opensemantic/base/_controller.py
+++ b/src/opensemantic/base/_controller.py
@@ -9,8 +9,9 @@
 
 from pydantic import ConfigDict, PrivateAttr
 
-from opensemantic.base._controller_mixin import (
+from opensemantic.base._controller_mixin import (  # noqa: F401 (re-export)
     DataToolMixin,
+    DownsampleParams,
     TSDCMixin,
 )
 from opensemantic.base._model import Database as _Database
@@ -74,32 +75,6 @@ async def write_tool_channel_raw(
         ):
             return await self._driver.write(params.tool_osw_id, params.data)
 
-        async def read_tool_channel_raw(
-            self, params: TSDCMixin.ReadToolChannelRawParams
-        ):
-            filters = None
-            if params.filter:
-                filters = [
-                    {
-                        "column": (
-                            f.column.value
-                            if isinstance(f.column, TSDCMixin.FilterColumn)
-                            else f.column
-                        ),
-                        "operator": f.operator.value,
-                        "criteria": f.criteria,
-                    }
-                    for f in params.filter
-                ]
-            return await self._driver.read(
-                tool_osw_id=params.tool_osw_id,
-                channel_osw_id=params.channel_osw_id,
-                start=params.start,
-                end=params.end,
-                filters=filters,
-                limit=params.limit,
-            )
-
         async def delete_by_ids(self, tool_osw_id: str, ids: List[int]):
             return await self._driver.delete_by_ids(tool_osw_id, ids)
 
@@ -155,32 +130,6 @@ async def write_tool_channel_raw(
         ):
             return await self._driver.write(params.tool_osw_id, params.data)
 
-        async def read_tool_channel_raw(
-            self, params: TSDCMixin.ReadToolChannelRawParams
-        ):
-            filters = None
-            if params.filter:
-                filters = [
-                    {
-                        "column": (
-                            f.column.value
-                            if isinstance(f.column, TSDCMixin.FilterColumn)
-                            else f.column
-                        ),
-                        "operator": f.operator.value,
-                        "criteria": f.criteria,
-                    }
-                    for f in params.filter
-                ]
-            return await self._driver.read(
-                tool_osw_id=params.tool_osw_id,
-                channel_osw_id=params.channel_osw_id,
-                start=params.start,
-                end=params.end,
-                filters=filters,
-                limit=params.limit,
-            )
-
         async def get_tool_config(self):
             return await self._driver.get_tool_config()
 
@@ -192,3 +141,18 @@ async def start_offline_sync(self):
 
 except ImportError:
     pass
+
+
+# Re-export the param/result data classes so callers construct them from the
+# user-facing module (``from opensemantic.base import StoreChannelDataBulkParams``)
+# without reaching into the mixin classes. ``DownsampleParams`` is already a
+# module-level class and is re-exported via the import above.
+StoreChannelDataParams = DataToolMixin.StoreChannelDataParams
+StoreChannelSeriesParams = DataToolMixin.StoreChannelSeriesParams
+StoreChannelDataBulkParams = DataToolMixin.StoreChannelDataBulkParams
+LoadChannelDataParams = DataToolMixin.LoadChannelDataParams
+ChannelDataPoint = DataToolMixin.ChannelDataPoint
+CreateToolParams = TSDCMixin.CreateToolParams
+DeleteToolParams = TSDCMixin.DeleteToolParams
+WriteToolChannelRawParams = TSDCMixin.WriteToolChannelRawParams
+ReadToolChannelRawParams = TSDCMixin.ReadToolChannelRawParams
diff --git a/src/opensemantic/base/_controller_mixin.py b/src/opensemantic/base/_controller_mixin.py
index 4851345..45a2ed9 100644
--- a/src/opensemantic/base/_controller_mixin.py
+++ b/src/opensemantic/base/_controller_mixin.py
@@ -20,6 +20,37 @@
 _logger = logging.getLogger(__name__)
 
 
+class DownsampleParams(BaseModel):
+    """Server-side downsampling request (PostgREST/TimescaleDB backend).
+
+    Grouped into one sub-object so the read/load params stay tidy. Honored by
+    backends that support it; others ignore or approximate it. All-None (or a
+    None ``DownsampleParams``) means no downsampling - full resolution.
+    """
+
+    max_points: Optional[int] = None
+    """Target point count: the read is bucketed into ~max_points buckets."""
+    bin_size: Optional[str] = None
+    """Explicit bucket width as a Postgres interval string (e.g. '5 seconds');
+    overrides max_points when set."""
+    method: Optional[str] = None
+    """Strategy: 'sample', 'average' or 'minmax'. None disables downsampling.
+
+    Note: 'average'/'minmax' aggregate the bare stored numbers per leaf and are
+    only correct for unit-normalized (base-unit) data."""
+    edge_anchors: Optional[bool] = None
+    """Include the window's first/last real datapoints as endpoints
+    (server default: True)."""
+
+    def is_active(self) -> bool:
+        """True if any downsampling is requested."""
+        return (
+            self.max_points is not None
+            or self.bin_size is not None
+            or bool(self.method)
+        )
+
+
 class DataToolMixin(BaseController):
     """Generic controller mixin for DataTool models.
 
@@ -365,14 +396,44 @@ class StoreChannelDataParams(BaseModel):
         timestamp: Optional[dt.datetime] = None
         """Timestamp for the data point. Defaults to now(UTC)."""
 
+    class StoreChannelSeriesParams(BaseModel):
+        model_config = ConfigDict(arbitrary_types_allowed=True)
+        channel: Any = None
+        """DataChannel instance or channel name (str)."""
+        timestamps: List[dt.datetime] = []
+        """Timestamps, parallel to ``values``."""
+        values: List[Any] = []
+        """Raw dicts/scalars or Characteristic instances, parallel to
+        ``timestamps``."""
+
+    class StoreChannelDataBulkParams(BaseModel):
+        model_config = ConfigDict(arbitrary_types_allowed=True)
+        series: List["DataToolMixin.StoreChannelSeriesParams"] = []
+        """One entry per channel, each with parallel timestamps/values."""
+        chunk_size: int = 5000
+        """Rows per write_tool_channel_raw batch."""
+        ensure_tool: bool = True
+        """Create the tool table first if it does not exist."""
+        schema_reload_wait: Optional[float] = None
+        """Seconds to wait after creating the tool (PostgREST reloads its
+        schema cache after the DDL). None auto-selects 1.5 for PostgREST and
+        0 for local SQLite."""
+
     class LoadChannelDataParams(BaseModel):
         model_config = ConfigDict(arbitrary_types_allowed=True)
         channel: Union[str, List[str], Any, None] = None
         """Channel name (str), list of names, DataChannel instance,
         list of DataChannel instances, or None (all channels)."""
         start: Optional[dt.datetime] = None
+        """Start of the time range (inclusive). None reads from the earliest
+        stored point."""
         end: Optional[dt.datetime] = None
+        """End of the time range (inclusive). None reads up to the latest
+        stored point."""
         limit: Optional[int] = None
+        """Maximum number of rows to return. None means no limit."""
+        downsample: Optional[DownsampleParams] = None
+        """Optional server-side downsampling request. None = full resolution."""
         typed: bool = True
         """If True, deserialize values using channel characteristic or
         target_schema. If False, return raw dicts (faster)."""
@@ -616,6 +677,84 @@ async def store_channel_data(
             )
         )
 
+    async def _ensure_tool_exists(self, tool_osw_id: str) -> bool:
+        """Create the tool table if it does not exist. Returns True if created.
+
+        For PostgREST the CREATE triggers a schema-cache reload, so callers
+        should wait briefly before the first write (see store_channel_data_bulk).
+        """
+        try:
+            existing = await self.archive_database.get_tools_list()
+        except Exception:
+            existing = []
+        if tool_osw_id in existing:
+            return False
+        await self.archive_database.create_tool(
+            TSDCMixin.CreateToolParams(tool_osw_id=tool_osw_id)
+        )
+        return True
+
+    async def store_channel_data_bulk(
+        self, params: "DataToolMixin.StoreChannelDataBulkParams"
+    ) -> int:
+        """Store many channel values at once (efficient for large series).
+
+        Each entry in ``params.series`` carries a channel (name or instance)
+        with parallel ``timestamps``/``values`` arrays. Values are converted
+        exactly like store_channel_data (Characteristic -> base unit; dict or
+        scalar otherwise) and written in ``chunk_size`` batches via
+        write_tool_channel_raw. When ``ensure_tool`` the tool table is created
+        first if missing (with a schema-reload wait for PostgREST). Returns the
+        number of points written.
+        """
+        if self.archive_database is None:
+            raise ValueError("No archive database configured")
+        tool_osw_id = self.get_osw_id()
+
+        if params.ensure_tool and await self._ensure_tool_exists(tool_osw_id):
+            driver = getattr(self.archive_database, "_driver", None)
+            is_pgrst = getattr(driver, "client", None) is not None
+            wait = (
+                params.schema_reload_wait
+                if params.schema_reload_wait is not None
+                else (1.5 if is_pgrst else 0.0)
+            )
+            if wait > 0:
+                await asyncio.sleep(wait)
+
+        written = 0
+        batch: List[dict] = []
+
+        async def _flush():
+            nonlocal written, batch
+            if not batch:
+                return
+            await self.archive_database.write_tool_channel_raw(
+                TSDCMixin.WriteToolChannelRawParams(tool_osw_id=tool_osw_id, data=batch)
+            )
+            written += len(batch)
+            batch = []
+
+        for series in params.series:
+            channel = self._resolve_channel(series.channel)
+            ch_osw_id = channel.get_osw_id()
+            if "#" in ch_osw_id:
+                ch_osw_id = ch_osw_id.split("#")[-1]
+            n = min(len(series.timestamps), len(series.values))
+            for i in range(n):
+                data = self._value_to_store_data(series.values[i], channel)
+                batch.append(
+                    {
+                        "ts": series.timestamps[i].isoformat(),
+                        "ch": ch_osw_id,
+                        "data": data,
+                    }
+                )
+                if len(batch) >= params.chunk_size:
+                    await _flush()
+        await _flush()
+        return written
+
     async def load_channel_data(
         self,
         params: "DataToolMixin.LoadChannelDataParams",
@@ -670,6 +809,7 @@ async def load_channel_data(
                 start=params.start,
                 end=params.end,
                 limit=params.limit,
+                downsample=params.downsample,
             )
         )
 
@@ -888,11 +1028,50 @@ class ReadToolChannelRawParams(BaseModel):
         """Filters for reading data"""
         limit: Optional[int] = None
         """Limit the number of returned rows"""
+        downsample: Optional[DownsampleParams] = None
+        """Optional server-side downsampling request. None = full resolution."""
 
-    @abstractmethod
     async def read_tool_channel_raw(self, params: "TSDCMixin.ReadToolChannelRawParams"):
-        """Retrieves data for a tool within a time range."""
-        pass
+        """Retrieve data for a tool within a time range.
+
+        Shared by every driver-backed controller (v1 and v2, Local and
+        PostgREST): it builds the filter list and forwards all parameters -
+        including the optional server-side downsampling parameters
+        (``max_points`` / ``bin_size`` / ``downsample_method`` /
+        ``edge_anchors``) - to the backend driver via ``self._driver.read``.
+        """
+        driver = getattr(self, "_driver", None)
+        if driver is None:
+            raise NotImplementedError(
+                f"{type(self).__name__} has no _driver to read from"
+            )
+        filters = None
+        if params.filter:
+            filters = [
+                {
+                    "column": (
+                        f.column.value
+                        if isinstance(f.column, TSDCMixin.FilterColumn)
+                        else f.column
+                    ),
+                    "operator": f.operator.value,
+                    "criteria": f.criteria,
+                }
+                for f in params.filter
+            ]
+        ds = params.downsample
+        return await driver.read(
+            tool_osw_id=params.tool_osw_id,
+            channel_osw_id=params.channel_osw_id,
+            start=params.start,
+            end=params.end,
+            filters=filters,
+            limit=params.limit,
+            max_points=ds.max_points if ds else None,
+            bin_size=ds.bin_size if ds else None,
+            downsample_method=ds.method if ds else None,
+            edge_anchors=ds.edge_anchors if ds else None,
+        )
 
     def read_tool_channel_raw_sync(self, params: "TSDCMixin.ReadToolChannelRawParams"):
         return asyncio.run(self.read_tool_channel_raw(params=params))
diff --git a/src/opensemantic/base/_demo_data.py b/src/opensemantic/base/_demo_data.py
new file mode 100644
index 0000000..5d91229
--- /dev/null
+++ b/src/opensemantic/base/_demo_data.py
@@ -0,0 +1,66 @@
+"""Seeding helpers for demos, benchmarks and tests.
+
+Generate a synthetic time series and store it through the high-level
+``DataToolController`` bulk API (``store_channel_data_bulk``), so demo, bench
+and test seeding share one code path instead of each duplicating the low-level
+``create_tool`` + ``write_tool_channel_raw`` loop.
+"""
+
+import datetime as dt
+from typing import Any, Callable, List, Optional
+
+from opensemantic.base import (
+    LoadChannelDataParams,
+    StoreChannelDataBulkParams,
+    StoreChannelSeriesParams,
+)
+
+
+async def seed_channel_series(
+    controller: Any,
+    *,
+    n_points: int,
+    base_ts: dt.datetime,
+    value_fn: Callable[[Any, int], Any],
+    channels: Optional[List[Any]] = None,
+    seconds_step: float = 1.0,
+    chunk_size: int = 5000,
+) -> int:
+    """Seed ``n_points`` per channel via the high-level bulk store.
+
+    ``value_fn(channel, i)`` returns the value stored for ``channel`` at index
+    ``i`` (a raw dict/scalar or a Characteristic instance); the timestamp is
+    ``base_ts + i * seconds_step`` seconds. ``channels`` defaults to all of the
+    controller's channels. The tool table is created if missing. Returns the
+    number of points written.
+    """
+    if channels is None:
+        channels = controller.get_all_channels()
+    timestamps = [
+        base_ts + dt.timedelta(seconds=i * seconds_step) for i in range(n_points)
+    ]
+    series = [
+        StoreChannelSeriesParams(
+            channel=ch,
+            timestamps=timestamps,
+            values=[value_fn(ch, i) for i in range(n_points)],
+        )
+        for ch in channels
+    ]
+    return await controller.store_channel_data_bulk(
+        StoreChannelDataBulkParams(series=series, chunk_size=chunk_size)
+    )
+
+
+async def already_seeded(controller: Any) -> bool:
+    """True if the controller's tool exists and has at least one stored row."""
+    try:
+        tools = await controller.archive_database.get_tools_list()
+    except Exception:
+        return False
+    if controller.get_osw_id() not in tools:
+        return False
+    rows = await controller.load_channel_data(
+        LoadChannelDataParams(limit=1, typed=False)
+    )
+    return bool(rows)
diff --git a/src/opensemantic/base/_drivers.py b/src/opensemantic/base/_drivers.py
index 33c3475..e43c69b 100644
--- a/src/opensemantic/base/_drivers.py
+++ b/src/opensemantic/base/_drivers.py
@@ -22,6 +22,25 @@
 _logger = logging.getLogger(__name__)
 
 
+def _stride_decimate(rows: list, max_points: int) -> list:
+    """Reduce ``rows`` to about ``max_points`` by even striding.
+
+    Always keeps the first and last row so the series still spans its
+    full window. Used as the SQLite fallback when server-side
+    downsampling is unavailable.
+    """
+    n = len(rows)
+    if max_points <= 2 or n <= max_points:
+        return rows
+    step = n / float(max_points)
+    picked = [rows[min(int(i * step), n - 1)] for i in range(max_points)]
+    if picked[0] is not rows[0]:
+        picked[0] = rows[0]
+    if picked[-1] is not rows[-1]:
+        picked[-1] = rows[-1]
+    return picked
+
+
 class DatabaseDriver:
     """Base class with shared buffer management.
 
@@ -48,8 +67,18 @@ async def read(
         end: Optional[Any] = None,
         filters: Optional[list] = None,
         limit: Optional[int] = None,
+        max_points: Optional[int] = None,
+        bin_size: Optional[str] = None,
+        downsample_method: Optional[str] = None,
+        edge_anchors: Optional[bool] = None,
     ) -> list:
-        """Read rows from the backend."""
+        """Read rows from the backend.
+
+        The trailing ``max_points`` / ``bin_size`` / ``downsample_method`` /
+        ``edge_anchors`` parameters request server-side downsampling and are
+        honored by backends that support it (PostgREST/TimescaleDB); other
+        backends ignore or approximate them.
+        """
 
     @abstractmethod
     async def create_tool(self, tool_osw_id: str):
@@ -189,6 +218,10 @@ async def read(
         end: Optional[Any] = None,
         filters: Optional[list] = None,
         limit: Optional[int] = None,
+        max_points: Optional[int] = None,
+        bin_size: Optional[str] = None,
+        downsample_method: Optional[str] = None,
+        edge_anchors: Optional[bool] = None,
     ) -> list:
         query, query_params = build_sqlite_read_query(
             tool_osw_id=tool_osw_id,
@@ -200,7 +233,12 @@ async def read(
         )
         _logger.debug("Executing query: %s with params: %s", query, query_params)
         rows = await self.fetchall(query, tuple(query_params))
-        return parse_sqlite_rows(rows)
+        parsed = parse_sqlite_rows(rows)
+        # SQLite has no time_bucket; approximate downsampling with a simple
+        # row-stride decimation that keeps the first and last point.
+        if max_points and len(parsed) > max_points:
+            parsed = _stride_decimate(parsed, max_points)
+        return parsed
 
     async def delete_by_ids(self, tool_osw_id: str, ids: List[int]):
         if not ids:
@@ -364,8 +402,28 @@ async def read(
         end: Optional[Any] = None,
         filters: Optional[list] = None,
         limit: Optional[int] = None,
+        max_points: Optional[int] = None,
+        bin_size: Optional[str] = None,
+        downsample_method: Optional[str] = None,
+        edge_anchors: Optional[bool] = None,
     ) -> list:
         self._ensure_client()
+        # Route to the server-side downsampling RPC when requested. Any
+        # failure (RPC missing on an older pgstack, network error, ...)
+        # falls through to the plain full-resolution read below.
+        if max_points is not None or bin_size is not None or downsample_method:
+            rows = await self._read_downsampled(
+                tool_osw_id,
+                channel_osw_id,
+                start,
+                end,
+                max_points,
+                bin_size,
+                downsample_method,
+                edge_anchors,
+            )
+            if rows is not None:
+                return rows
         if channel_osw_id:
             query = (
                 self.client.table(tool_osw_id)
@@ -399,6 +457,47 @@ async def read(
         res = await query.execute()
         return res.data if res.data else []
 
+    async def _read_downsampled(
+        self,
+        tool_osw_id: str,
+        channel_osw_id: Optional[str],
+        start: Optional[Any],
+        end: Optional[Any],
+        max_points: Optional[int],
+        bin_size: Optional[str],
+        downsample_method: Optional[str],
+        edge_anchors: Optional[bool],
+    ) -> Optional[list]:
+        """Call the server-side downsampling RPC.
+
+        Returns the bucketed rows (same ``{"ts","ch","data"}`` shape as the
+        plain read), or ``None`` when the RPC is unavailable/failed so the
+        caller can fall back to a full-resolution read.
+        """
+        params: Dict[str, Any] = {"osw_tool": tool_osw_id}
+        if channel_osw_id:
+            params["ch_id"] = channel_osw_id
+        if start is not None:
+            params["ts_start"] = start.isoformat()
+        if end is not None:
+            params["ts_end"] = end.isoformat()
+        if max_points is not None:
+            params["max_points"] = max_points
+        if bin_size is not None:
+            params["bin_size"] = bin_size
+        if downsample_method:
+            params["method"] = downsample_method
+        if edge_anchors is not None:
+            params["edge_anchors"] = edge_anchors
+        try:
+            res = await self.client.rpc("downsample_tool_channel", params).execute()
+            return res.data if res.data else []
+        except Exception as e:
+            _logger.debug(
+                "downsample RPC unavailable (%s); falling back to full read", e
+            )
+            return None
+
     async def _flush_offline_buffer(self):
         """Background task that syncs offline-buffered data to remote."""
         if not self._local_db:
diff --git a/src/opensemantic/base/v1/_controller.py b/src/opensemantic/base/v1/_controller.py
index 9d07aa1..66a36ee 100644
--- a/src/opensemantic/base/v1/_controller.py
+++ b/src/opensemantic/base/v1/_controller.py
@@ -9,8 +9,9 @@
 
 from pydantic import ConfigDict, PrivateAttr
 
-from opensemantic.base._controller_mixin import (
+from opensemantic.base._controller_mixin import (  # noqa: F401 (re-export)
     DataToolMixin,
+    DownsampleParams,
     TSDCMixin,
 )
 from opensemantic.base.v1._model import Database as _Database
@@ -74,32 +75,6 @@ async def write_tool_channel_raw(
         ):
             return await self._driver.write(params.tool_osw_id, params.data)
 
-        async def read_tool_channel_raw(
-            self, params: TSDCMixin.ReadToolChannelRawParams
-        ):
-            filters = None
-            if params.filter:
-                filters = [
-                    {
-                        "column": (
-                            f.column.value
-                            if isinstance(f.column, TSDCMixin.FilterColumn)
-                            else f.column
-                        ),
-                        "operator": f.operator.value,
-                        "criteria": f.criteria,
-                    }
-                    for f in params.filter
-                ]
-            return await self._driver.read(
-                tool_osw_id=params.tool_osw_id,
-                channel_osw_id=params.channel_osw_id,
-                start=params.start,
-                end=params.end,
-                filters=filters,
-                limit=params.limit,
-            )
-
         async def delete_by_ids(self, tool_osw_id: str, ids: List[int]):
             return await self._driver.delete_by_ids(tool_osw_id, ids)
 
@@ -155,32 +130,6 @@ async def write_tool_channel_raw(
         ):
             return await self._driver.write(params.tool_osw_id, params.data)
 
-        async def read_tool_channel_raw(
-            self, params: TSDCMixin.ReadToolChannelRawParams
-        ):
-            filters = None
-            if params.filter:
-                filters = [
-                    {
-                        "column": (
-                            f.column.value
-                            if isinstance(f.column, TSDCMixin.FilterColumn)
-                            else f.column
-                        ),
-                        "operator": f.operator.value,
-                        "criteria": f.criteria,
-                    }
-                    for f in params.filter
-                ]
-            return await self._driver.read(
-                tool_osw_id=params.tool_osw_id,
-                channel_osw_id=params.channel_osw_id,
-                start=params.start,
-                end=params.end,
-                filters=filters,
-                limit=params.limit,
-            )
-
         async def get_tool_config(self):
             return await self._driver.get_tool_config()
 
@@ -192,3 +141,18 @@ async def start_offline_sync(self):
 
 except ImportError:
     pass
+
+
+# Re-export the param/result data classes so callers construct them from the
+# user-facing module (``from opensemantic.base.v1 import StoreChannelDataBulkParams``)
+# without reaching into the mixin classes. ``DownsampleParams`` is already a
+# module-level class and is re-exported via the import above.
+StoreChannelDataParams = DataToolMixin.StoreChannelDataParams
+StoreChannelSeriesParams = DataToolMixin.StoreChannelSeriesParams
+StoreChannelDataBulkParams = DataToolMixin.StoreChannelDataBulkParams
+LoadChannelDataParams = DataToolMixin.LoadChannelDataParams
+ChannelDataPoint = DataToolMixin.ChannelDataPoint
+CreateToolParams = TSDCMixin.CreateToolParams
+DeleteToolParams = TSDCMixin.DeleteToolParams
+WriteToolChannelRawParams = TSDCMixin.WriteToolChannelRawParams
+ReadToolChannelRawParams = TSDCMixin.ReadToolChannelRawParams
diff --git a/src/opensemantic/base/view/_base_view.py b/src/opensemantic/base/view/_base_view.py
index a5c23a0..9cf1736 100644
--- a/src/opensemantic/base/view/_base_view.py
+++ b/src/opensemantic/base/view/_base_view.py
@@ -82,9 +82,10 @@ def _on_unit_change(self, group_key: str, event):
     # -- Plot / log / config cards --
 
     def _build_plot(self):
-        self._plot_col = pn.Column(
-            sizing_mode="stretch_width", scroll=True, max_height=600
-        )
+        # No height cap / inner scroll here: the plots stack at their natural
+        # height and the main content area (.content) scrolls as one. A fixed
+        # max_height + scroll=True would add a second, nested scrollbar.
+        self._plot_col = pn.Column(sizing_mode="stretch_width")
         self._plot_card = pn.Card(
             self._plot_col,
             title=_t("time_series", self.lang),
diff --git a/src/opensemantic/base/view/_channel_utils.py b/src/opensemantic/base/view/_channel_utils.py
index 07e5ee7..4f93d48 100644
--- a/src/opensemantic/base/view/_channel_utils.py
+++ b/src/opensemantic/base/view/_channel_utils.py
@@ -297,6 +297,21 @@ def resolve_value_type(channel: Any) -> str:
     return "number"
 
 
+def resolve_downsample_method(channel: Any, config_method: str = "auto") -> str:
+    """Pick the downsampling strategy for a channel.
+
+    When ``config_method`` is an explicit strategy ('sample'/'average'/
+    'minmax') it is returned as-is. When it is 'auto' (or empty), numeric
+    channels (quantity/number/composite) use 'minmax' (spike-safe, real
+    points) and text/unknown channels use 'sample'.
+    """
+    if config_method and config_method != "auto":
+        return config_method
+    if resolve_value_type(channel) in ("quantity", "number", "composite"):
+        return "minmax"
+    return "sample"
+
+
 def _is_characteristic_subclass(cls: type) -> bool:
     """Check if a class is a Characteristic subclass (v1 or v2)."""
     if not isinstance(cls, type):
@@ -605,6 +620,7 @@ def get_selected_channels(
     "archive": {"en": "Archive", "de": "Archiv"},
     "live": {"en": "Live", "de": "Live"},
     "clear_cache": {"en": "Clear Cache", "de": "Cache leeren"},
+    "load_range": {"en": "Load current range", "de": "Aktuellen Bereich laden"},
 }
 
 
diff --git a/src/opensemantic/base/view/_config.py b/src/opensemantic/base/view/_config.py
index 3b6153e..b3dd2a6 100644
--- a/src/opensemantic/base/view/_config.py
+++ b/src/opensemantic/base/view/_config.py
@@ -3,9 +3,7 @@
 from enum import Enum
 from typing import List
 
-from pydantic import ConfigDict, Field
-
-from opensemantic import OswBaseModel
+from pydantic import BaseModel, ConfigDict, Field
 
 
 class LangCode(str, Enum):
@@ -19,7 +17,69 @@ class GroupingMode(str, Enum):
     SUB = "group-subcharacteristics"
 
 
-class TreeConfig(OswBaseModel):
+class DownsampleMethod(str, Enum):
+    AUTO = "auto"
+    SAMPLE = "sample"
+    AVERAGE = "average"
+    MINMAX = "minmax"
+
+
+class DownsampleConfig(BaseModel):
+    """Server-side downsampling configuration.
+
+    When enabled and the backend supports it (PostgREST/TimescaleDB), the
+    plot requests about ``max_points`` points per channel instead of the
+    full series. ``average`` and ``minmax`` aggregate the bare stored
+    numbers and are only correct for unit-normalized data; ``auto`` picks
+    ``minmax`` for numeric channels and ``sample`` for text channels.
+    """
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "title": "DownsampleConfig",
+            "defaultProperties": [
+                "enabled",
+                "max_points",
+                "method",
+                "edge_anchors",
+            ],
+        }
+    )
+
+    enabled: bool = Field(
+        True,
+        title="Downsample enabled",
+        json_schema_extra={"title*": {"de": "Downsampling aktiviert"}},
+    )
+    max_points: int = Field(
+        2000,
+        title="Max points",
+        description="Target number of points per channel",
+        ge=2,
+        json_schema_extra={
+            "title*": {"de": "Maximale Punkte"},
+            "description*": {"de": "Zielanzahl Punkte pro Kanal"},
+        },
+    )
+    method: DownsampleMethod = Field(
+        DownsampleMethod.AUTO,
+        title="Method",
+        json_schema_extra={"title*": {"de": "Methode"}},
+    )
+    edge_anchors: bool = Field(
+        True,
+        title="Edge anchors",
+        description="Keep the window's first/last real datapoints as endpoints",
+        json_schema_extra={
+            "title*": {"de": "Randpunkte"},
+            "description*": {
+                "de": "Ersten/letzten realen Datenpunkt des Fensters behalten"
+            },
+        },
+    )
+
+
+class TreeConfig(BaseModel):
     """Wunderbaum TreeGrid configuration."""
 
     model_config = ConfigDict(
@@ -27,7 +87,7 @@ class TreeConfig(OswBaseModel):
     )
 
 
-class PlotConfig(OswBaseModel):
+class PlotConfig(BaseModel):
     """Time series plot configuration."""
 
     model_config = ConfigDict(
@@ -38,6 +98,7 @@ class PlotConfig(OswBaseModel):
                 "auto_fetch",
                 "row_limit",
                 "cache_enabled",
+                "downsample",
             ],
         }
     )
@@ -72,9 +133,14 @@ class PlotConfig(OswBaseModel):
         title="Cache enabled",
         json_schema_extra={"title*": {"de": "Cache aktiviert"}},
     )
+    downsample: DownsampleConfig = Field(
+        default_factory=DownsampleConfig,
+        title="Downsample",
+        json_schema_extra={"title*": {"de": "Downsampling"}},
+    )
 
 
-class LiveConfig(OswBaseModel):
+class LiveConfig(BaseModel):
     """Live subscription configuration."""
 
     model_config = ConfigDict(
@@ -108,7 +174,7 @@ class LiveConfig(OswBaseModel):
     )
 
 
-class DashboardConfig(OswBaseModel):
+class DashboardConfig(BaseModel):
     """Root dashboard configuration."""
 
     model_config = ConfigDict(
diff --git a/src/opensemantic/base/view/_data_cache.py b/src/opensemantic/base/view/_data_cache.py
index 11005e3..565d4a8 100644
--- a/src/opensemantic/base/view/_data_cache.py
+++ b/src/opensemantic/base/view/_data_cache.py
@@ -37,6 +37,10 @@ async def get_data(
         end: Optional[dt.datetime] = None,
         limit: int = 10000,
         typed: bool = True,
+        max_points: Optional[int] = None,
+        bin_size: Optional[str] = None,
+        method: Optional[str] = None,
+        edge_anchors: Optional[bool] = None,
     ) -> List[Any]:
         """Return ChannelDataPoint list for [start, end].
 
@@ -56,6 +60,11 @@ async def get_data(
             Maximum rows per backend request.
         typed
             If True, values are typed Characteristic instances.
+        max_points, bin_size, method, edge_anchors
+            Server-side downsampling parameters. When any of max_points /
+            bin_size / method is set the gap-merge cache is bypassed (a
+            downsampled read is resolution-dependent and must not poison the
+            full-resolution cache) and the backend is queried directly.
 
         Returns
         -------
@@ -66,8 +75,21 @@ async def get_data(
 
         ch_uuid = channel.uuid
 
-        if not self.enabled:
-            return await self._fetch(controller, channel, start, end, limit, typed)
+        downsampling = max_points is not None or bin_size is not None or bool(method)
+
+        if not self.enabled or downsampling:
+            return await self._fetch(
+                controller,
+                channel,
+                start,
+                end,
+                limit,
+                typed,
+                max_points,
+                bin_size,
+                method,
+                edge_anchors,
+            )
 
         covered = self._intervals.get(ch_uuid, [])
         gaps = _compute_gaps(start, end, covered)
@@ -92,10 +114,25 @@ async def _fetch(
         end: dt.datetime,
         limit: int,
         typed: bool,
+        max_points: Optional[int] = None,
+        bin_size: Optional[str] = None,
+        method: Optional[str] = None,
+        edge_anchors: Optional[bool] = None,
     ) -> List[Any]:
         """Fetch data via controller.load_channel_data()."""
-        from opensemantic.base._controller_mixin import DataToolMixin
+        from opensemantic.base._controller_mixin import (
+            DataToolMixin,
+            DownsampleParams,
+        )
 
+        downsample = None
+        if max_points is not None or bin_size is not None or method:
+            downsample = DownsampleParams(
+                max_points=max_points,
+                bin_size=bin_size,
+                method=method,
+                edge_anchors=edge_anchors,
+            )
         points = await controller.load_channel_data(
             DataToolMixin.LoadChannelDataParams(
                 channel=channel,
@@ -103,6 +140,7 @@ async def _fetch(
                 end=end,
                 limit=limit,
                 typed=typed,
+                downsample=downsample,
             )
         )
         # Normalize timestamps to UTC-aware and sort
diff --git a/src/opensemantic/base/view/_datatool_dashboard.py b/src/opensemantic/base/view/_datatool_dashboard.py
index 3b53f60..12f8eac 100644
--- a/src/opensemantic/base/view/_datatool_dashboard.py
+++ b/src/opensemantic/base/view/_datatool_dashboard.py
@@ -18,7 +18,14 @@
 from typing import Any, Dict, List, Optional, Tuple
 
 import panel as pn
-from bokeh.models import ColumnDataSource, DatetimeTickFormatter
+from bokeh.events import Reset
+from bokeh.models import (
+    BoxZoomTool,
+    ColumnDataSource,
+    CustomJS,
+    DataRange1d,
+    DatetimeTickFormatter,
+)
 from bokeh.plotting import figure as bk_figure
 from panelini import Panelini
 from panelini.panels.wunderbaum import Wunderbaum
@@ -31,6 +38,7 @@
     get_display_label,
     get_selected_channels,
     group_channels_by_characteristic,
+    resolve_downsample_method,
     resolve_value_type,
 )
 from opensemantic.base.view._config import DashboardConfig
@@ -47,6 +55,14 @@ def get_unit_enum_from_value(value: Any) -> Any:
 
 _logger = logging.getLogger(__name__)
 
+# Wunderbaum's grid sizes its container to a fixed 800px (inline width), which
+# overflows the sidebar and shows a horizontal scrollbar. Force it to fill the
+# available width instead. !important beats the inline style (which has none).
+_TREE_GRID_CSS = (
+    ".tree-container.wunderbaum { width: 100% !important; }\n"
+    ".wunderbaum-wrapper { overflow-x: hidden !important; }"
+)
+
 
 class DataToolView(BaseDataView):
     """Archive-mode DataTool view.
@@ -92,6 +108,19 @@ def __init__(
         self._cached_data: Dict[str, List] = {}
         self._composite_parents: Dict[str, Tuple[Any, str]] = {}
 
+        # Zoom-driven downsampling state. _zoom_window overrides the picker
+        # window when "Load current range" re-fetches the box-zoomed range;
+        # _shared_x_range is the linked x-range the button reads.
+        self._zoom_window: Optional[Tuple[dt.datetime, dt.datetime]] = None
+        self._shared_x_range = None
+        # Reset bridge. figure.on_event(Reset) does not propagate through
+        # Panel's Bokeh pane, but model property changes sync reliably. A
+        # CustomJS on the Reset event bumps this source's data; its
+        # server-side on_change (_on_reset_bridge) reloads the full window
+        # after a "Load current range" zoom-reload.
+        self._reset_bridge = ColumnDataSource(data={"n": [0]})
+        self._reset_bridge.on_change("data", self._on_reset_bridge)
+
         # Cache
         self._cache = ChannelDataCache(enabled=self._config.plot.cache_enabled)
 
@@ -116,14 +145,15 @@ def _build_tree(self):
         self._tree = Wunderbaum(
             source=source,
             columns=[
-                {"id": "*", "title": _t("channel", self.lang), "width": "200px"},
+                {"id": "*", "title": _t("channel", self.lang), "width": "190px"},
                 {
                     "id": "characteristic",
                     "title": _t("characteristic", self.lang),
-                    "width": "150px",
+                    "width": "130px",
                 },
             ],
             options={"checkbox": True, "selectMode": "hier"},
+            stylesheets=[_TREE_GRID_CSS],
         )
         # Watch the source param for checkbox changes (emitSource syncs selected state)
         self._tree.param.watch(self._on_source_change, ["source"])
@@ -136,6 +166,8 @@ def _build_tree(self):
     def _on_source_change(self, *args):
         """Handle tree source changes (checkbox toggling)."""
         try:
+            # A selection change loads the full picker window, not a zoom slice.
+            self._zoom_window = None
             self._update_selection()
             self._update_unit_controls()
             if self._config.plot.auto_fetch:
@@ -178,6 +210,14 @@ def _build_controls(self):
         )
         self._load_button.on_click(self._on_load_click)
 
+        # Re-fetch the currently visible (box-zoomed) x-range at finer
+        # resolution, keeping the zoom. Box-zoom itself is purely visual.
+        self._load_range_button = pn.widgets.Button(
+            name=_t("load_range", self.lang),
+            button_type="default",
+        )
+        self._load_range_button.on_click(self._on_load_range)
+
         self._auto_fetch_cb = pn.widgets.Checkbox(
             name=_t("auto_fetch", self.lang),
             value=self._config.plot.auto_fetch,
@@ -207,6 +247,7 @@ def _build_controls(self):
             self._start_picker,
             self._end_picker,
             self._load_button,
+            self._load_range_button,
             self._auto_fetch_cb,
             self._row_limit_input,
             self._clear_cache_button,
@@ -218,6 +259,8 @@ def _on_load_click(self, event):
         self._trigger_load()
 
     def _on_time_change(self, event):
+        # A manual time-range edit overrides any active zoom slice.
+        self._zoom_window = None
         if self._config.plot.auto_fetch and self._selected:
             self._trigger_load()
 
@@ -296,9 +339,14 @@ def _rebuild_ui_labels(self):
 
     async def _load_and_plot(self):
         """Load data for all selected channels and update plots."""
-        start = self._start_picker.value
-        # Use current time as end when auto-fetching so new data is included
-        end = dt.datetime.now() if self._auto_fetch_cb.value else self._end_picker.value
+        if self._zoom_window is not None:
+            start, end = self._zoom_window
+        else:
+            # Respect the pickers. auto_fetch only controls whether a change
+            # (selection/time) auto-triggers a load, not the window itself;
+            # live "follow now" is the LiveDataToolView's job, not this view's.
+            start = self._start_picker.value
+            end = self._end_picker.value
         if start is None or end is None:
             return
 
@@ -320,18 +368,50 @@ async def _load_and_plot(self):
                     if parent_ch.uuid in loaded_parents:
                         continue
                     loaded_parents.add(parent_ch.uuid)
+                    mp, method, edge = self._downsample_for(parent_ch)
                     rows = await self._cache.get_data(
-                        ctrl, parent_ch, start, end, limit
+                        ctrl,
+                        parent_ch,
+                        start,
+                        end,
+                        limit,
+                        max_points=mp,
+                        method=method,
+                        edge_anchors=edge,
                     )
                     self._cached_data[parent_ch.uuid] = rows
                 else:
-                    rows = await self._cache.get_data(ctrl, ch, start, end, limit)
+                    mp, method, edge = self._downsample_for(ch)
+                    rows = await self._cache.get_data(
+                        ctrl,
+                        ch,
+                        start,
+                        end,
+                        limit,
+                        max_points=mp,
+                        method=method,
+                        edge_anchors=edge,
+                    )
                     self._cached_data[ch.uuid] = rows
             except Exception as e:
                 _logger.error("Error loading %s/%s: %s", ctrl.name, ch.name, e)
 
         self._refresh_plot()
 
+    def _downsample_for(self, channel):
+        """Return ``(max_points, method, edge_anchors)`` for a channel.
+
+        Default applies the dashboard config, auto-resolving the method per
+        channel type. Override to vary the strategy per channel (e.g. force
+        a different method, or raw/no downsampling, for specific channels).
+        Return ``(None, None, None)`` to read the channel at full resolution.
+        """
+        ds = self._config.plot.downsample
+        if not ds.enabled:
+            return None, None, None
+        method = resolve_downsample_method(channel, ds.method.value)
+        return ds.max_points, method, ds.edge_anchors
+
     def _build_figure(self):
         """Build Bokeh figures - one per characteristic group."""
         self._plot_col.clear()
@@ -349,6 +429,14 @@ def _build_figure(self):
         if not plot_groups:
             return
 
+        # One shared x-range for every figure, so panning, zooming or resetting
+        # any plot moves all of them together (the y-ranges stay independent).
+        # Sharing it at figure creation - rather than reassigning f.x_range
+        # after the panes render - is what guarantees the link: a deferred
+        # reassignment races the render and leaves some plots unlinked.
+        shared_x = DataRange1d()
+
+        figs = []
         color_idx = 0
         for group_key, channels, vtype in plot_groups:
             axis_label = self._get_axis_label(group_key)
@@ -356,8 +444,17 @@ def _build_figure(self):
                 height=250,
                 sizing_mode="stretch_width",
                 x_axis_type="datetime",
+                x_range=shared_x,
                 y_axis_label=axis_label,
+                tools="pan,wheel_zoom,reset,save",
             )
+            # x-only box zoom as the active drag tool: dragging zooms the time
+            # axis only (y stays auto so a downsampled spike isn't clipped).
+            # The zoom is purely visual; use the "Load current range" button to
+            # re-fetch that window at a finer resolution.
+            box_zoom = BoxZoomTool(dimensions="width")
+            fig.add_tools(box_zoom)
+            fig.toolbar.active_drag = box_zoom
             fig.xaxis.formatter = DatetimeTickFormatter(
                 seconds="%H:%M:%S",
                 minutes="%H:%M",
@@ -383,8 +480,75 @@ def _build_figure(self):
                     color_idx += 1
 
             fig.legend.click_policy = "hide"
+            figs.append(fig)
+
+        # Keep the shared range for "Load current range" and bridge the Reset
+        # event: figure.on_event(Reset) does not propagate through Panel's Bokeh
+        # pane, so a CustomJS on the Reset event bumps _reset_bridge, whose
+        # server-side on_change reloads. Done while the figures are still
+        # detached - mutating models already in the live document from this
+        # async load task raises a document-lock error.
+        self._shared_x_range = shared_x
+        reset_cb = CustomJS(
+            args={"bridge": self._reset_bridge},
+            code="bridge.data = {n: [(bridge.data.n[0] || 0) + 1]};",
+        )
+        for fig in figs:
+            fig.js_on_event(Reset, reset_cb)
+
+        for fig in figs:
             self._plot_col.append(pn.pane.Bokeh(fig, sizing_mode="stretch_width"))
 
+    def _current_xrange_window(self):
+        """Return (start, end) *naive* datetimes of the current plot x-range.
+
+        The shared x-range start/end (epoch ms) reflect the user's latest
+        pan/zoom, synced from the browser. The data is plotted in local wall
+        time (see _extract_trace_data), so these epochs are the local
+        wall-clock values - returned as naive datetimes so _load_and_plot
+        converts them naive->UTC exactly like the date pickers do (treating
+        them as local). Returning UTC-aware here would shift the loaded window
+        by the local/UTC offset. Returns None if no zoom range is set.
+        """
+        xr = self._shared_x_range
+        if xr is None or xr.start is None or xr.end is None:
+            return None
+        try:
+            start = dt.datetime.fromtimestamp(
+                xr.start / 1000.0, dt.timezone.utc
+            ).replace(tzinfo=None)
+            end = dt.datetime.fromtimestamp(xr.end / 1000.0, dt.timezone.utc).replace(
+                tzinfo=None
+            )
+        except Exception:
+            return None
+        return (start, end) if end > start else None
+
+    def _on_load_range(self, event):
+        """Re-fetch the currently visible (zoomed) x-range at finer resolution."""
+        window = self._current_xrange_window()
+        if window is None:
+            return
+        self._zoom_window = window
+        self._trigger_load()
+
+    def _on_reset_bridge(self, attr, old, new):
+        """Toolbar reset (bridged via CustomJS -> source data change).
+
+        figure.on_event(Reset) does not reach the server through Panel's Bokeh
+        pane, so the Reset event is bridged through this source's data change
+        (see _build_figure). After "Load current range" the figures hold only
+        the
+        zoomed window, so Bokeh's own reset would only return to that window;
+        clear the zoom and reload the full (picker) window. For a plain visual
+        box-zoom (nothing was reloaded) there is no zoom window, and Bokeh's
+        reset already restores the full view, so we do nothing.
+        """
+        if self._zoom_window is None:
+            return
+        self._zoom_window = None
+        self._trigger_load()
+
     def _extract_trace_data(self, ch: Any, group_key: str) -> Tuple[List, List]:
         """Extract timestamps and values from cached ChannelDataPoints.
 
@@ -556,11 +720,11 @@ def main_cards(self):
     def set_time_range(self, start, end, fetch: bool = True):
         """Set an explicit time range and (optionally) reload the data.
 
-        Accepts tz-aware or naive datetimes (naive is treated as UTC). Disables
-        auto-fetch, which otherwise pins the end to ``now``, so an arbitrary
-        historical window can be shown.
+        Accepts tz-aware or naive datetimes (naive is treated as UTC). The
+        window is honored as-is: auto-fetch no longer pins the end to ``now``
+        (it only controls whether changes auto-trigger a load), so an arbitrary
+        historical window can be shown without disabling auto-fetch.
         """
-        self._auto_fetch_cb.value = False
         self._start_picker.value = self._to_picker_value(start)
         self._end_picker.value = self._to_picker_value(end)
         if fetch:
diff --git a/src/opensemantic/base/view/_process_dashboard.py b/src/opensemantic/base/view/_process_dashboard.py
index 44de0de..0364041 100644
--- a/src/opensemantic/base/view/_process_dashboard.py
+++ b/src/opensemantic/base/view/_process_dashboard.py
@@ -32,6 +32,7 @@
     _t,
     get_display_label,
     group_channels_by_characteristic,
+    resolve_downsample_method,
     resolve_value_type,
 )
 from opensemantic.base.view._config import DashboardConfig
@@ -297,6 +298,10 @@ def _on_config_editor_change(self, event):
 
     async def _load_and_plot(self):
         limit = self._config.plot.row_limit
+        ds = self._config.plot.downsample
+        max_points = ds.max_points if ds.enabled else None
+        edge_anchors = ds.edge_anchors if ds.enabled else None
+        method_cfg = ds.method.value if ds.enabled else None
         self._traces = []
 
         for obj_entry in self._selected_objects:
@@ -306,8 +311,22 @@ async def _load_and_plot(self):
                     end = _as_utc(getattr(proc, "end_date_time", None))
                     if start is None:
                         continue
+                    method = (
+                        resolve_downsample_method(ch, method_cfg)
+                        if method_cfg
+                        else None
+                    )
                     try:
-                        points = await self._cache.get_data(ctrl, ch, start, end, limit)
+                        points = await self._cache.get_data(
+                            ctrl,
+                            ch,
+                            start,
+                            end,
+                            limit,
+                            max_points=max_points,
+                            method=method,
+                            edge_anchors=edge_anchors,
+                        )
                     except Exception as e:
                         _logger.error(
                             "Error loading %s/%s: %s",
diff --git a/tests/test_controller.py b/tests/test_controller.py
index 2442616..0dc0d99 100644
--- a/tests/test_controller.py
+++ b/tests/test_controller.py
@@ -991,6 +991,45 @@ async def _test():
     _cleanup_archive(ctrl)
 
 
+def test_store_channel_data_bulk():
+    """Bulk-store many points across two channels; load returns them all."""
+    import datetime
+
+    ctrl = _make_controller_with_channels()
+
+    async def _test():
+        base = datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc)
+        n = 2500  # 2 channels * 2500 = 5000 rows, chunked below
+        ts = [base + datetime.timedelta(seconds=i) for i in range(n)]
+        written = await ctrl.store_channel_data_bulk(
+            ctrl.StoreChannelDataBulkParams(
+                series=[
+                    ctrl.StoreChannelSeriesParams(
+                        channel="pressure",
+                        timestamps=ts,
+                        values=[{"value": float(i)} for i in range(n)],
+                    ),
+                    ctrl.StoreChannelSeriesParams(
+                        channel="temperature",
+                        timestamps=ts,
+                        values=[{"value": float(i)} for i in range(n)],
+                    ),
+                ],
+                chunk_size=1000,
+            )
+        )
+        assert written == 2 * n
+        # pressure has no characteristic -> raw dicts back
+        rows = await ctrl.load_channel_data(
+            ctrl.LoadChannelDataParams(channel="pressure")
+        )
+        assert len(rows) == n
+        assert sorted(r.value["value"] for r in rows) == [float(i) for i in range(n)]
+
+    asyncio.run(_test())
+    _cleanup_archive(ctrl)
+
+
 def test_store_channel_data_typed():
     """Store Temperature instance, verify base unit conversion."""
     import datetime
diff --git a/tests/test_downsample.py b/tests/test_downsample.py
new file mode 100644
index 0000000..09f4800
--- /dev/null
+++ b/tests/test_downsample.py
@@ -0,0 +1,354 @@
+"""Unit tests for server-side downsampling param threading and fallback.
+
+These tests use no live database. The PostgREST path is exercised with a
+fake client; the SQLite path uses a temporary file. End-to-end behaviour
+against a real TimescaleDB is covered (and gated) in
+``test_downsample_integration.py``.
+"""
+
+import asyncio
+import datetime as dt
+import os
+import tempfile
+from types import SimpleNamespace
+from uuid import uuid4
+
+import opensemantic.base.view._channel_utils as cu
+from opensemantic.base._drivers import (
+    LocalDatabaseDriver,
+    PostgrestDatabaseDriver,
+    _stride_decimate,
+)
+from opensemantic.base.view._data_cache import ChannelDataCache
+
+# -- Fake PostgREST client -------------------------------------------------
+
+
+class _FakeResult:
+    def __init__(self, data):
+        self.data = data
+
+
+class _FakeRpc:
+    def __init__(self, result, error):
+        self._result = result
+        self._error = error
+
+    async def execute(self):
+        if self._error is not None:
+            raise self._error
+        return _FakeResult(self._result)
+
+
+class _FakeQuery:
+    """Chainable stand-in for the table().select()... builder."""
+
+    def __init__(self, rows):
+        self._rows = rows
+
+    def select(self, *a, **k):
+        return self
+
+    def eq(self, *a, **k):
+        return self
+
+    def order(self, *a, **k):
+        return self
+
+    def gte(self, *a, **k):
+        return self
+
+    def lte(self, *a, **k):
+        return self
+
+    def filter(self, *a, **k):
+        return self
+
+    def limit(self, *a, **k):
+        return self
+
+    async def execute(self):
+        return _FakeResult(self._rows)
+
+
+class _FakeClient:
+    def __init__(self, rpc_result=None, rpc_error=None, table_rows=None):
+        self.rpc_result = rpc_result
+        self.rpc_error = rpc_error
+        self.table_rows = table_rows or []
+        self.rpc_calls = []
+
+    def rpc(self, name, params):
+        self.rpc_calls.append((name, params))
+        return _FakeRpc(self.rpc_result, self.rpc_error)
+
+    def table(self, name):
+        return _FakeQuery(self.table_rows)
+
+
+# -- _stride_decimate ------------------------------------------------------
+
+
+def test_stride_decimate_bounds_and_endpoints():
+    rows = list(range(1000))
+    out = _stride_decimate(rows, 100)
+    assert len(out) == 100
+    assert out[0] == 0
+    assert out[-1] == 999
+
+
+def test_stride_decimate_noop_when_small():
+    rows = list(range(10))
+    assert _stride_decimate(rows, 100) == rows
+
+
+# -- PostgREST driver routing ----------------------------------------------
+
+
+def test_postgrest_read_routes_to_rpc():
+    rpc_rows = [{"ts": "2023-01-01T00:00:00+00:00", "ch": "ch1", "data": {"value": 1}}]
+    client = _FakeClient(rpc_result=rpc_rows)
+    drv = PostgrestDatabaseDriver()
+    drv.set_client(client)
+
+    start = dt.datetime(2023, 1, 1, tzinfo=dt.timezone.utc)
+    end = dt.datetime(2023, 1, 2, tzinfo=dt.timezone.utc)
+    rows = asyncio.run(
+        drv.read(
+            "OSWtool",
+            channel_osw_id="ch1",
+            start=start,
+            end=end,
+            max_points=500,
+            downsample_method="minmax",
+            edge_anchors=True,
+        )
+    )
+
+    assert rows == rpc_rows
+    assert len(client.rpc_calls) == 1
+    name, params = client.rpc_calls[0]
+    assert name == "downsample_tool_channel"
+    assert params["osw_tool"] == "OSWtool"
+    assert params["ch_id"] == "ch1"
+    assert params["ts_start"] == start.isoformat()
+    assert params["ts_end"] == end.isoformat()
+    assert params["max_points"] == 500
+    assert params["method"] == "minmax"
+    assert params["edge_anchors"] is True
+
+
+def test_postgrest_read_falls_back_on_rpc_error():
+    table_rows = [
+        {"ts": "2023-01-01T00:00:00+00:00", "ch": "ch1", "data": {"value": 2}}
+    ]
+    client = _FakeClient(
+        rpc_error=RuntimeError("no such function"), table_rows=table_rows
+    )
+    drv = PostgrestDatabaseDriver()
+    drv.set_client(client)
+
+    rows = asyncio.run(
+        drv.read(
+            "OSWtool", channel_osw_id="ch1", max_points=500, downsample_method="minmax"
+        )
+    )
+
+    # RPC failed -> the plain table read result is returned instead.
+    assert rows == table_rows
+    assert len(client.rpc_calls) == 1
+
+
+def test_postgrest_read_without_downsample_skips_rpc():
+    table_rows = [{"ts": "2023-01-01T00:00:00+00:00", "ch": "ch1", "data": {}}]
+    client = _FakeClient(table_rows=table_rows)
+    drv = PostgrestDatabaseDriver()
+    drv.set_client(client)
+
+    rows = asyncio.run(drv.read("OSWtool", channel_osw_id="ch1"))
+
+    assert rows == table_rows
+    assert client.rpc_calls == []
+
+
+# -- SQLite stride fallback ------------------------------------------------
+
+
+def test_local_read_decimates():
+    path = tempfile.mktemp(suffix=".sqlite")
+    try:
+        drv = LocalDatabaseDriver(path)
+        base = dt.datetime(2023, 1, 1, tzinfo=dt.timezone.utc)
+        data = [
+            {
+                "ts": (base + dt.timedelta(seconds=i)).isoformat(),
+                "ch": "ch1",
+                "data": {"value": i},
+            }
+            for i in range(1000)
+        ]
+        asyncio.run(drv.write("OSWtool", data))
+
+        rows = asyncio.run(drv.read("OSWtool", max_points=100))
+        assert len(rows) == 100
+        values = {r["data"]["value"] for r in rows}
+        assert 0 in values and 999 in values
+
+        # Without max_points the full series is returned.
+        full = asyncio.run(drv.read("OSWtool"))
+        assert len(full) == 1000
+    finally:
+        if os.path.exists(path):
+            os.remove(path)
+
+
+# -- load_channel_data forwarding ------------------------------------------
+
+
+class _RecordingDriver:
+    """Driver stub that records the kwargs of read()."""
+
+    def __init__(self):
+        self.read_kwargs = None
+
+    async def read(self, **kwargs):
+        self.read_kwargs = kwargs
+        return []
+
+
+def test_load_channel_data_threads_downsample_params():
+    from opensemantic.base import (
+        DataToolController,
+        LocalTimeSeriesDatabaseController,
+    )
+    from opensemantic.base._controller_mixin import DataToolMixin, DownsampleParams
+    from opensemantic.base._model import DataChannel
+    from opensemantic.core import Label
+
+    ch = DataChannel(uuid=str(uuid4()), osw_id="temp", name="ch1")
+    db = LocalTimeSeriesDatabaseController(
+        name="rec",
+        label=[Label(text="rec")],
+        db_path=tempfile.mktemp(suffix=".sqlite"),
+    )
+    rec = _RecordingDriver()
+    db._driver = rec
+
+    tool = DataToolController(name="t", label=[Label(text="t")], data_channels=[ch])
+    tool.archive_database = db
+
+    asyncio.run(
+        tool.load_channel_data(
+            DataToolMixin.LoadChannelDataParams(
+                channel="ch1",
+                downsample=DownsampleParams(
+                    max_points=750,
+                    bin_size="5 seconds",
+                    method="average",
+                    edge_anchors=False,
+                ),
+            )
+        )
+    )
+
+    # The mixin unpacks the downsample sub-object into the driver's kwargs.
+    assert rec.read_kwargs is not None
+    assert rec.read_kwargs["max_points"] == 750
+    assert rec.read_kwargs["bin_size"] == "5 seconds"
+    assert rec.read_kwargs["downsample_method"] == "average"
+    assert rec.read_kwargs["edge_anchors"] is False
+
+
+# -- Cache bypass ----------------------------------------------------------
+
+
+class _RecordingController:
+    def __init__(self):
+        self.calls = []
+
+    async def load_channel_data(self, params):
+        self.calls.append(params)
+        pt = SimpleNamespace(
+            timestamp=dt.datetime(2023, 1, 1, tzinfo=dt.timezone.utc),
+            channel=params.channel,
+            value={"value": 1},
+        )
+        return [pt]
+
+
+def test_cache_bypassed_when_downsampling():
+    cache = ChannelDataCache(enabled=True)
+    ctrl = _RecordingController()
+    ch = SimpleNamespace(uuid="u1")
+    start = dt.datetime(2023, 1, 1, tzinfo=dt.timezone.utc)
+    end = dt.datetime(2023, 1, 2, tzinfo=dt.timezone.utc)
+
+    asyncio.run(
+        cache.get_data(
+            ctrl,
+            ch,
+            start,
+            end,
+            100,
+            max_points=500,
+            method="minmax",
+            edge_anchors=True,
+        )
+    )
+    asyncio.run(
+        cache.get_data(
+            ctrl,
+            ch,
+            start,
+            end,
+            100,
+            max_points=500,
+            method="minmax",
+            edge_anchors=True,
+        )
+    )
+
+    # Both calls hit the backend (no interval caching for downsampled reads).
+    assert len(ctrl.calls) == 2
+    assert cache._intervals == {}
+    params = ctrl.calls[0]
+    assert params.downsample.max_points == 500
+    assert params.downsample.method == "minmax"
+    assert params.downsample.edge_anchors is True
+
+
+def test_cache_caches_full_resolution_reads():
+    cache = ChannelDataCache(enabled=True)
+    ctrl = _RecordingController()
+    ch = SimpleNamespace(uuid="u2")
+    start = dt.datetime(2023, 1, 1, tzinfo=dt.timezone.utc)
+    end = dt.datetime(2023, 1, 2, tzinfo=dt.timezone.utc)
+
+    asyncio.run(cache.get_data(ctrl, ch, start, end, 100))
+    asyncio.run(cache.get_data(ctrl, ch, start, end, 100))
+
+    # Second full-resolution read is served from cache.
+    assert len(ctrl.calls) == 1
+    assert cache._intervals.get("u2")
+
+
+# -- Per-channel auto resolver ---------------------------------------------
+
+
+def test_auto_resolver_picks_minmax_for_numeric(monkeypatch):
+    for vt in ("quantity", "number", "composite"):
+        monkeypatch.setattr(cu, "resolve_value_type", lambda ch, _vt=vt: _vt)
+        assert cu.resolve_downsample_method(object(), "auto") == "minmax"
+
+
+def test_auto_resolver_picks_sample_for_text(monkeypatch):
+    for vt in ("text", "unknown"):
+        monkeypatch.setattr(cu, "resolve_value_type", lambda ch, _vt=vt: _vt)
+        assert cu.resolve_downsample_method(object(), "auto") == "sample"
+
+
+def test_auto_resolver_explicit_passthrough(monkeypatch):
+    monkeypatch.setattr(cu, "resolve_value_type", lambda ch: "text")
+    # An explicit method ignores the channel type.
+    assert cu.resolve_downsample_method(object(), "average") == "average"
+    assert cu.resolve_downsample_method(object(), "minmax") == "minmax"
diff --git a/tests/test_downsample_integration.py b/tests/test_downsample_integration.py
new file mode 100644
index 0000000..3cfb3f9
--- /dev/null
+++ b/tests/test_downsample_integration.py
@@ -0,0 +1,331 @@
+"""Integration tests for the downsample_tool_channel RPC.
+
+These run against a live pgstack (TimescaleDB + PostgREST) with the updated
+``100_init_tsdb_schema.sql`` applied. They are skipped unless TEST_PGRST_URL
+and TEST_PGRST_JWT_SECRET are set, and self-skip when the server or the RPC
+is unavailable.
+
+All coroutines run on a single module-scoped event loop so the shared async
+PostgREST client (an httpx.AsyncClient) is not torn down between calls.
+"""
+
+import asyncio
+import datetime as dt
+import os
+
+import pytest
+
+from opensemantic.base import (
+    DeleteToolParams,
+    DownsampleParams,
+    ReadToolChannelRawParams,
+)
+
+_PGRST_URL = os.environ.get("TEST_PGRST_URL")
+_PGRST_JWT_SECRET = os.environ.get("TEST_PGRST_JWT_SECRET")
+_PGRST_JWT_ROLE = os.environ.get("TEST_PGRST_JWT_ROLE", "api_user")
+_PGRST_SCHEMA = os.environ.get("TEST_PGRST_SCHEMA", "api")
+_PGRST_CONFIGURED = bool(_PGRST_URL and _PGRST_JWT_SECRET)
+
+pytestmark = pytest.mark.skipif(
+    not _PGRST_CONFIGURED,
+    reason="TEST_PGRST_URL and TEST_PGRST_JWT_SECRET not set",
+)
+
+WINDOW_SECONDS = 2000
+N_POINTS = 2000
+MAX_POINTS = 200  # -> 10 s buckets, ~10 points each
+
+
+def _make_db():
+    import jwt
+    from postgrest import AsyncPostgrestClient
+
+    from opensemantic.base import PostgrestTimeSeriesDatabaseController
+    from opensemantic.core import Label
+
+    token = jwt.encode({"role": _PGRST_JWT_ROLE}, _PGRST_JWT_SECRET, algorithm="HS256")
+    client = AsyncPostgrestClient(
+        base_url=_PGRST_URL,
+        schema=_PGRST_SCHEMA,
+        headers={"Authorization": f"Bearer {token}"},
+    )
+    db = PostgrestTimeSeriesDatabaseController(
+        name="downsample_test",
+        label=[Label(text="Downsample Test")],
+        buffered=False,
+    )
+    db.set_client(client)
+    return db
+
+
+def _read(
+    seeded,
+    ch,
+    start=None,
+    end=None,
+    max_points=None,
+    bin_size=None,
+    downsample_method=None,
+    edge_anchors=None,
+):
+    ds = None
+    if max_points is not None or bin_size is not None or downsample_method:
+        ds = DownsampleParams(
+            max_points=max_points,
+            bin_size=bin_size,
+            method=downsample_method,
+            edge_anchors=edge_anchors,
+        )
+    return seeded["run"](
+        seeded["db"].read_tool_channel_raw(
+            ReadToolChannelRawParams(
+                tool_osw_id=seeded["tool_id"],
+                channel_osw_id=ch,
+                start=start,
+                end=end,
+                downsample=ds,
+            )
+        )
+    )
+
+
+def _make_controller(db):
+    """A DataToolController with scalar, composite and text channels, bound to db."""
+    from uuid import uuid4
+
+    from opensemantic import compute_scoped_uuid
+    from opensemantic.base.v1 import DataChannel, DataToolController
+    from opensemantic.core.v1 import Label
+
+    parent = uuid4()
+    ctrl = DataToolController(
+        uuid=str(parent),
+        name="DownsampleTest",
+        label=[Label(text="Test")],
+        data_channels=[
+            DataChannel(
+                uuid=str(compute_scoped_uuid(parent, name)),
+                osw_id="placeholder",
+                name=name,
+                label=[Label(text=name)],
+            )
+            for name in ("scalar", "composite", "text")
+        ],
+    )
+    ctrl.archive_database = db
+    return ctrl
+
+
+@pytest.fixture(scope="module")
+def seeded():
+    """Create a tool with a scalar, a composite, and a text channel.
+
+    Seeds via the shared high-level bulk helper (seed_channel_series). One
+    event loop is used for setup, all reads and teardown.
+    """
+    from opensemantic.base._demo_data import seed_channel_series
+
+    loop = asyncio.new_event_loop()
+
+    def run(coro):
+        return loop.run_until_complete(coro)
+
+    db = _make_db()
+    ctrl = _make_controller(db)
+    tool_id = ctrl.get_osw_id()
+    base = dt.datetime(2023, 1, 1, tzinfo=dt.timezone.utc)
+
+    def short(name):
+        return ctrl.get_channel_by_name(name).get_osw_id().split("#")[-1]
+
+    def _value(channel, i):
+        if channel.name == "composite":
+            return {
+                "temperature": {"value": float(i)},
+                "humidity": {"value": float(N_POINTS - i)},
+                "comment": "ok",
+            }
+        if channel.name == "text":
+            return {"text": "hello"}
+        return {"value": float(i)}
+
+    try:
+        run(seed_channel_series(ctrl, n_points=N_POINTS, base_ts=base, value_fn=_value))
+    except Exception as e:
+        loop.close()
+        pytest.skip(f"pgstack unavailable: {e}")
+
+    # Probe the RPC now the tool + data exist; skip the whole module if absent.
+    try:
+        run(
+            db._driver.client.rpc(
+                "downsample_tool_channel", {"osw_tool": tool_id}
+            ).execute()
+        )
+    except Exception as e:
+        loop.close()
+        pytest.skip(f"downsample_tool_channel RPC unavailable: {e}")
+    run(asyncio.sleep(1.0))
+
+    yield {
+        "run": run,
+        "db": db,
+        "tool_id": tool_id,
+        "ch_scalar": short("scalar"),
+        "ch_comp": short("composite"),
+        "ch_text": short("text"),
+        "start": base,
+        "end": base + dt.timedelta(seconds=WINDOW_SECONDS - 1),
+    }
+
+    try:
+        run(db.delete_tool(DeleteToolParams(tool_osw_id=tool_id)))
+    except Exception:
+        pass
+    loop.close()
+
+
+def _monotonic(rows):
+    ts = [r["ts"] for r in rows]
+    return ts == sorted(ts)
+
+
+def test_sample_count_and_monotonic(seeded):
+    rows = _read(
+        seeded,
+        seeded["ch_scalar"],
+        start=seeded["start"],
+        end=seeded["end"],
+        max_points=MAX_POINTS,
+        downsample_method="sample",
+    )
+    # ~one point per bucket, plus up to 2 edge anchors.
+    assert 0 < len(rows) <= MAX_POINTS + 4
+    assert _monotonic(rows)
+
+
+def test_average_count_and_values(seeded):
+    rows = _read(
+        seeded,
+        seeded["ch_scalar"],
+        start=seeded["start"],
+        end=seeded["end"],
+        max_points=MAX_POINTS,
+        downsample_method="average",
+    )
+    assert 0 < len(rows) <= MAX_POINTS + 4
+    assert _monotonic(rows)
+    # Averaged values stay within the seeded range [0, N_POINTS).
+    vals = [r["data"]["value"] for r in rows if "value" in r["data"]]
+    assert vals and all(0.0 <= float(v) <= float(N_POINTS) for v in vals)
+
+
+def test_minmax_returns_more_rows_than_sample(seeded):
+    sample = _read(
+        seeded,
+        seeded["ch_scalar"],
+        start=seeded["start"],
+        end=seeded["end"],
+        max_points=MAX_POINTS,
+        downsample_method="sample",
+    )
+    minmax = _read(
+        seeded,
+        seeded["ch_scalar"],
+        start=seeded["start"],
+        end=seeded["end"],
+        max_points=MAX_POINTS,
+        downsample_method="minmax",
+    )
+    # Scalar minmax yields about two rows per bucket.
+    assert len(minmax) >= len(sample)
+    assert _monotonic(minmax)
+    # All returned rows are real stored datapoints (integer-valued here).
+    for r in minmax:
+        assert float(r["data"]["value"]).is_integer()
+
+
+def test_composite_sample_preserves_structure(seeded):
+    rows = _read(
+        seeded,
+        seeded["ch_comp"],
+        start=seeded["start"],
+        end=seeded["end"],
+        max_points=MAX_POINTS,
+        downsample_method="sample",
+    )
+    assert rows
+    d = rows[0]["data"]
+    assert "temperature" in d and "value" in d["temperature"]
+    assert "humidity" in d and "value" in d["humidity"]
+
+
+def test_composite_average_deep_averages_leaves(seeded):
+    rows = _read(
+        seeded,
+        seeded["ch_comp"],
+        start=seeded["start"],
+        end=seeded["end"],
+        max_points=MAX_POINTS,
+        downsample_method="average",
+    )
+    assert rows
+    d = rows[len(rows) // 2]["data"]
+    # Deep-averaged numeric leaves present; non-numeric 'comment' carried.
+    assert isinstance(d["temperature"]["value"], (int, float))
+    assert isinstance(d["humidity"]["value"], (int, float))
+    assert d.get("comment") == "ok"
+
+
+def test_composite_minmax_returns_real_rows(seeded):
+    sample = _read(
+        seeded,
+        seeded["ch_comp"],
+        start=seeded["start"],
+        end=seeded["end"],
+        max_points=MAX_POINTS,
+        downsample_method="sample",
+    )
+    minmax = _read(
+        seeded,
+        seeded["ch_comp"],
+        start=seeded["start"],
+        end=seeded["end"],
+        max_points=MAX_POINTS,
+        downsample_method="minmax",
+    )
+    assert len(minmax) >= len(sample)
+    assert _monotonic(minmax)
+    # Composite minmax rows are real stored rows with full structure.
+    d = minmax[0]["data"]
+    assert "temperature" in d and "humidity" in d
+
+
+def test_text_channel_falls_back_to_sample(seeded):
+    rows = _read(
+        seeded,
+        seeded["ch_text"],
+        start=seeded["start"],
+        end=seeded["end"],
+        max_points=MAX_POINTS,
+        downsample_method="minmax",
+    )
+    # No numeric leaf -> falls back to sample; rows are the real text rows.
+    assert rows
+    assert all(r["data"].get("text") == "hello" for r in rows)
+
+
+def test_edge_anchors_present(seeded):
+    with_anchors = _read(
+        seeded,
+        seeded["ch_scalar"],
+        start=seeded["start"],
+        end=seeded["end"],
+        max_points=MAX_POINTS,
+        downsample_method="sample",
+        edge_anchors=True,
+    )
+    # First/last returned rows are the window's first/last real datapoints.
+    assert float(with_anchors[0]["data"]["value"]) == 0.0
+    assert float(with_anchors[-1]["data"]["value"]) == float(WINDOW_SECONDS - 1)