diff --git a/CHANGELOG.md b/CHANGELOG.md
index 56dca7b..d04728d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -55,6 +55,10 @@ The first release, an alpha: the plugin API can still change before 0.1.0.
   in another app.
 - `chunkmirage.serve` and `Server`: an app served from Python on a port bound first, with a
   callback once it accepts connections, in a background thread or this one.
+- `chunkmirage materialize` and `chunkmirage.materialize`: a dataset, or a box of it,
+  computed and written into a store (a directory, `gs://`, `s3://`) as a frontend serves it,
+  zarr v3, zarr v2 or N5, so it reads back as a stored dataset of its own; the metadata
+  first, chunks several at once, empty ones left out.
 
 ### Sources
 
diff --git a/docs/concepts/formats.md b/docs/concepts/formats.md
index 23c80d5..47109ef 100644
--- a/docs/concepts/formats.md
+++ b/docs/concepts/formats.md
@@ -73,6 +73,31 @@ view as zarr v2 with OME-Zarr 0.4 (`virtual/<page>/zarr2/<view>/`), the same chu
 readers that predate zarr v3's final spec: GDAL 3.8, as the fires demo runs it in the page
 (gdal3.js), opens a level as a zarr array and translates it to a GeoTIFF.
 
+## A dataset written to a store (Python and the command line)
+
+```bash
+chunkmirage materialize SOURCE OUT [--op ...] [--format zarr3|zarr|n5] [--level N ...] [--box lo:hi]
+```
+
+computes SOURCE through its ops and writes it to OUT (a directory, or a `file://`, `gs://`
+or `s3://` URL, through tensorstore) as a frontend serves it: zarr v3 with OME-Zarr 0.5
+(the default), zarr v2 with OME-Zarr 0.4 and consolidated metadata, or N5. The metadata
+and every chunk are byte for byte what a client fetches, so the store opens wherever the
+served dataset did and reads back as a stored source of its own: a stage computed once (a
+registration's pass, a model's output over a region) is then read from disk instead of
+computed again. Every level's metadata is written first, so a store whose writing stopped
+part way is still the dataset, zero where no chunk was written yet; then the chunks of the
+levels asked for (all by default), several at once, each of nothing but zeros left out
+(`--keep-empty` writes them). `--box` takes level 0's voxels along the last axes, `lo:hi`
+each comma separated in C order (`0,0,0:64,512,512`), and every level writes the same
+region on its own voxels, rounded out to whole chunks. From Python,
+`chunkmirage.materialize(pipeline, where, format=, levels=, box=, compressor=,
+skip_empty=, progress=)` does the same for any pipeline, another dataset's chain read
+through `dataset://` among them (`registry.get(name)`), and returns the counts
+(`{chunks, written, skipped, bytes}`). Inside a running server it competes with clients'
+requests as any work asked for from Python does (never dropped, in arrival order), and
+its chunks pass through the server's cache.
+
 ## A region written to a store (browser engine)
 
 The browser engine also writes a view, or a box of it, into a store as files
diff --git a/docs/contributing.md b/docs/contributing.md
index 3a0cad3..6426b5a 100644
--- a/docs/contributing.md
+++ b/docs/contributing.md
@@ -114,7 +114,7 @@ Packages build on chunkmirage through its plugin API, which is stable:
 * **Schema.** A parameter model listed under `chunkmirage.models` appears in
   `chunkmirage schema` under the entry point's name, for the browser engine's types.
 * **Serving.** `PipelineSpec`, `Pipeline`, `create_app` and its parameters, `serve` and
-  `Server`, `gpu.set_device`, `demand.set_gather_threads`,
+  `Server`, `materialize`, `gpu.set_device`, `demand.set_gather_threads`,
   `DatasetRegistry` (`add`, `get`, `resolve`, `refresh`, `remove`, `subscribe`) and its
   resolver, the `chunkmirage.routes` entry point, the REST routes in the
   [API reference](reference/api.md), the served URL layout, and the CLI's flags.
diff --git a/docs/design.md b/docs/design.md
index e27227e..e147a0e 100644
--- a/docs/design.md
+++ b/docs/design.md
@@ -275,7 +275,7 @@ or astronomy? is a ratio of two bands geo?): if it has an algorithm in it, it go
 
 | package | holds |
 | ------- | ----- |
-| `chunkmirage` | the pipeline, cache, demand queue, server, REST API and CLI; the frontends; readers of stored formats (zarr, N5, precomputed, HDF5); `scene://` and the transforms (the pipeline's own resampler uses them), and that resampler on a GPU (the same code with CuPy for numpy, `gpu.py`, loaded only when a device is set: interpolation is what the core's own sources do, so a faster way to do it stays with them; CuPy rather than PyTorch because its calls are numpy's and scipy's, so one code path serves both and the CPU's tests cover it); the sources that compose other sources (`stack://`, `flip://`, `warp://`); meshes, a frontend's way of serving a surface; and two ops: `downsample`, which the pipeline runs to make the pyramid of an op bound to one resolution, and `python`, code carried in the spec |
+| `chunkmirage` | the pipeline, cache, demand queue, server, REST API and CLI; the frontends, and `materialize`, which writes a dataset as they serve it (their own metadata and encoded chunks, so a store reads back as the dataset did); readers of stored formats (zarr, N5, precomputed, HDF5); `scene://` and the transforms (the pipeline's own resampler uses them), and that resampler on a GPU (the same code with CuPy for numpy, `gpu.py`, loaded only when a device is set: interpolation is what the core's own sources do, so a faster way to do it stays with them; CuPy rather than PyTorch because its calls are numpy's and scipy's, so one code path serves both and the CPU's tests cover it); the sources that compose other sources (`stack://`, `flip://`, `warp://`); meshes, a frontend's way of serving a surface; and two ops: `downsample`, which the pipeline runs to make the pyramid of an op bound to one resolution, and `python`, code carried in the spec |
 | `chunkmirage_contrib` | every other op (filters, thresholds, morphology, labelling, detectors, terrain); the computed sources (`synthetic://`, `register://`, `stitch://`, GeoTIFF reading); the registration solver, stitching and tracking; and, in `web/src/contrib/`, the browser pages built on them |
 
 Contrib reaches the core only through the entry points any other package would use:
diff --git a/docs/getting-started.md b/docs/getting-started.md
index 5e02848..5451333 100644
--- a/docs/getting-started.md
+++ b/docs/getting-started.md
@@ -129,3 +129,10 @@ arr = ts.open({"driver": "zarr3",
                "kvstore": "http://localhost:8000/em-thresh/zarr3/s0/"}).result()
 arr[0:64, 0:64, 0:64].read().result()
 ```
+
+To keep a result instead, write it to disk as the server serves it, a zarr of its own that
+any tool reads (see [formats](concepts/formats.md#a-dataset-written-to-a-store-python-and-the-command-line)):
+
+```bash
+chunkmirage materialize /path/to/data.zarr/em/fibsem-uint8 em-thresh.zarr --op threshold:low=120
+```
diff --git a/docs/reference/api.md b/docs/reference/api.md
index f000b82..29eee42 100644
--- a/docs/reference/api.md
+++ b/docs/reference/api.md
@@ -143,6 +143,12 @@ server.stop()
 `Server(app, sock, ...)` takes a socket bound already (`netutil.bind_socket`), when the
 address must be known before the app is built.
 
+`chunkmirage.materialize(pipeline, where, *, format="zarr3", levels=None, box=None,
+compressor=None, skip_empty=True, progress=None)` writes a pipeline (or a box of it) into
+a store as a frontend serves it, which reads back as a stored dataset: the Python side of
+`chunkmirage materialize`
+([formats](../concepts/formats.md#a-dataset-written-to-a-store-python-and-the-command-line)).
+
 Two process-wide settings `chunkmirage serve` makes from its flags, for an application
 serving its own: `chunkmirage.gpu.set_device("cpu" | "auto" | "cuda" | "cuda:N")` (where
 `warp://` and `scene://` resample, `--device`), and `chunkmirage.demand.set_gather_threads(n)`
diff --git a/docs/reference/cli.md b/docs/reference/cli.md
index cb57abb..a17441d 100644
--- a/docs/reference/cli.md
+++ b/docs/reference/cli.md
@@ -49,6 +49,30 @@ the control page (`/ui`, with the token when there is one), the control API URL
 `--python-viewer`, the viewer URL; the port and thread count when they differ from what
 was asked for.
 
+## `materialize`
+
+```bash
+chunkmirage materialize SOURCE OUT [--op ...] [--format zarr3] [--level N ...] [--box lo:hi]
+```
+
+Computes SOURCE through its ops and writes it to OUT as a frontend serves it, a stored
+dataset of its own ([formats](../concepts/formats.md#a-dataset-written-to-a-store-python-and-the-command-line)).
+Prints the counts as JSON (`chunks`, `written`, `skipped`, `bytes`, `seconds`) and each
+batch's progress on stderr.
+
+| flag | default | meaning |
+| ---- | ------- | ------- |
+| `OUT` | | a directory, or a `file://`, `gs://` or `s3://` URL |
+| `--op`, `-o` | none | op spec, as for `serve` (repeatable) |
+| `--format` | `zarr3` | `zarr3` (OME-Zarr 0.5), `zarr` (zarr v2, OME-Zarr 0.4, consolidated) or `n5` |
+| `--level`, `-l` | every level | a level whose chunks to write (repeatable); every level's metadata is written |
+| `--box` | all of it | level 0's voxels along the last axes, `lo:hi` each comma separated in C order; every level writes the same region, rounded out to whole chunks |
+| `--chunk`, `--select` | as for `serve` | output chunk shape; non-spatial axes pinned |
+| `--compressor` | blosc (zarr), gzip (n5) | as the format takes it |
+| `--keep-empty` | off | also write chunks of nothing but zeros (left out otherwise: a reader takes a missing chunk as the fill value) |
+| `--device` | `cpu` | where `warp://` and `scene://` resample, as for `serve` |
+| `--cache-gb` | `2.0` | in-process chunk cache |
+
 ## `ops`
 
 Lists registered ops with halo, cache flag and parameters.
diff --git a/docs/roadmap.md b/docs/roadmap.md
index e123376..22fbfd6 100644
--- a/docs/roadmap.md
+++ b/docs/roadmap.md
@@ -36,9 +36,12 @@ describe shipped features as future work.
    viewer becomes the job scheduler. (Ordering the work by what clients ask for, and
    dropping what they stop waiting for, has shipped; see
    [caching](concepts/caching.md#order-of-work-and-requests-given-up-on). Inference ops
-   would join its queues. A first piece has shipped in the browser engine: an explicit
-   write of a box of a view into a store, a folder the visitor picks or memory, as an
-   OME-Zarr, its chunks computed in the background behind the viewer's
+   would join its queues. A first piece has shipped in both engines: an explicit write of
+   a dataset, or a box of it, into a store as the frontends serve it, from Python and the
+   command line (`chunkmirage materialize`,
+   [formats](concepts/formats.md#a-dataset-written-to-a-store-python-and-the-command-line)),
+   and in the browser into a folder the visitor picks or memory, its chunks computed in the
+   background behind the viewer's
    ([formats](concepts/formats.md#a-region-written-to-a-store-browser-engine)). The cache on
    disk itself, in either engine, its sharing across workers and the filler that grows it
    from what is browsed are still to do.)
diff --git a/src/chunkmirage/__init__.py b/src/chunkmirage/__init__.py
index 747bd91..ab67c15 100644
--- a/src/chunkmirage/__init__.py
+++ b/src/chunkmirage/__init__.py
@@ -1,10 +1,11 @@
 """chunkmirage: spoof chunked array formats over HTTP with on-the-fly processing."""
 
 from chunkmirage.core import ArrayInfo, Box
+from chunkmirage.materialize import materialize
 from chunkmirage.pipeline import Pipeline, PipelineSpec
 from chunkmirage.server import create_app
 from chunkmirage.serving import Server, serve
 from chunkmirage.sources import open_source
 
-__all__ = ["ArrayInfo", "Box", "Pipeline", "PipelineSpec", "Server", "create_app", "open_source", "serve"]
+__all__ = ["ArrayInfo", "Box", "Pipeline", "PipelineSpec", "Server", "create_app", "materialize", "open_source", "serve"]
 __version__ = "0.1.0a1"
diff --git a/src/chunkmirage/cli.py b/src/chunkmirage/cli.py
index d77151b..24a4f19 100644
--- a/src/chunkmirage/cli.py
+++ b/src/chunkmirage/cli.py
@@ -397,6 +397,52 @@ def schema(
         typer.echo(dumps(), nl=False)
 
 
+@app.command()
+def materialize(
+    source: str = typer.Argument(..., help="what to compute, as for serve"),
+    out: str = typer.Argument(..., help="where to write it: a directory, or a file://, gs:// or s3:// URL"),
+    op: list[str] = typer.Option([], "--op", "-o", help="op spec, as for serve (repeatable)"),
+    format: str = typer.Option("zarr3", help="zarr3 (OME-Zarr 0.5), zarr (zarr v2, OME-Zarr 0.4) or n5"),
+    level: list[int] = typer.Option([], "--level", "-l", help="a level to write (repeatable; default every level)"),
+    box: str | None = typer.Option(
+        None,
+        help="the region, in level 0's voxels along the last axes: lo:hi, each comma separated in "
+        "C order, e.g. 0,0,0:64,512,512 (every level writes the same region; default all of it)",
+    ),
+    chunk: str | None = typer.Option(None, help="output chunk shape, as for serve"),
+    select: str | None = typer.Option(None, help="pin non-spatial axes, as for serve"),
+    compressor: str | None = typer.Option(None, help="as the format takes it (default: blosc for zarr, gzip for n5)"),
+    keep_empty: bool = typer.Option(False, "--keep-empty", help="also write chunks of nothing but zeros"),
+    device: str = typer.Option("cpu", help="where warp:// and scene:// resample, as for serve"),
+    cache_gb: float = typer.Option(2.0, help="in-process chunk cache size"),
+):
+    """Compute SOURCE through its ops and write it to OUT as FORMAT serves it, read back as a
+    stored dataset of its own."""
+    import json
+    import time
+
+    from chunkmirage import gpu
+    from chunkmirage.core import Box
+    from chunkmirage.materialize import materialize as write
+
+    gpu.set_device(device)
+    registry = build_registry(source, "processed", op, chunk, cache_gb=cache_gb, select=select)
+    region = None
+    if box:
+        lo, _, hi = box.partition(":")
+        region = Box(tuple(int(v) for v in lo.split(",")), tuple(int(v) for v in hi.split(",")))
+    started = time.perf_counter()
+
+    def report(c):
+        typer.echo(f"s{c['level']}: {c['written'] + c['skipped']}/{c['chunks']} chunks", err=True)
+
+    counts = write(
+        registry.get("processed"), out, format=format, levels=level or None, box=region,
+        compressor=compressor, skip_empty=not keep_empty, progress=report,
+    )
+    typer.echo(json.dumps({**counts, "seconds": round(time.perf_counter() - started, 1)}))
+
+
 @app.command()
 def inspect(source: str):
     """Print what chunkmirage sees when opening SOURCE."""
diff --git a/src/chunkmirage/frontends/base.py b/src/chunkmirage/frontends/base.py
index 61eab1e..a6a9a1d 100644
--- a/src/chunkmirage/frontends/base.py
+++ b/src/chunkmirage/frontends/base.py
@@ -61,6 +61,12 @@ def compute(self, pipeline: Pipeline, req: ChunkRequest) -> bytes:
     root_keys: ClassVar[tuple[str, ...]] = ()  # metadata keys of the group
     level_keys: ClassVar[tuple[str, ...]] = ()  # metadata keys of each level
 
+    def chunk_key(self, level: int, index: tuple[int, ...]) -> str | None:
+        """The path chunk ``index`` of ``level`` is served at, which ``resolve`` maps back to
+        it; None for a format whose chunks are not files of their own (what ``materialize``
+        cannot write)."""
+        return None
+
     def listing(self, pipeline: Pipeline, path: str) -> list[str] | None:
         """What a directory listing of ``path`` shows: the group's metadata keys and one
         directory per level, or a level's metadata keys (never its chunks); None if ``path``
diff --git a/src/chunkmirage/frontends/n5.py b/src/chunkmirage/frontends/n5.py
index c66746f..24e6b2a 100644
--- a/src/chunkmirage/frontends/n5.py
+++ b/src/chunkmirage/frontends/n5.py
@@ -88,6 +88,9 @@ def _level_attrs(self, pipeline: Pipeline, lvl: int) -> dict:
             },
         }
 
+    def chunk_key(self, level: int, index) -> str:
+        return f"s{level}/" + "/".join(str(i) for i in index[::-1])  # N5's axes are x first
+
     def encode(self, info: ArrayInfo, index, block: np.ndarray) -> bytes:
         # mode 0 = default; dims are F-order (x first) = reversed C shape; data big-endian.
         header = struct.pack(">HH", 0, block.ndim) + struct.pack(
diff --git a/src/chunkmirage/frontends/zarr2.py b/src/chunkmirage/frontends/zarr2.py
index 35cb637..14f9eeb 100644
--- a/src/chunkmirage/frontends/zarr2.py
+++ b/src/chunkmirage/frontends/zarr2.py
@@ -78,6 +78,9 @@ def _consolidated(self, pipeline: Pipeline) -> dict:
             meta[f"s{lvl}/.zattrs"] = {}
         return {"zarr_consolidated_format": 1, "metadata": meta}
 
+    def chunk_key(self, level: int, index) -> str:
+        return f"s{level}/" + self.separator.join(str(i) for i in index)
+
     def encode(self, info: ArrayInfo, index, block: np.ndarray) -> bytes:
         full = self.pad_to_full(info, index, block)
         raw = np.ascontiguousarray(full).astype(full.dtype.newbyteorder("<"), copy=False).tobytes()
diff --git a/src/chunkmirage/frontends/zarr3.py b/src/chunkmirage/frontends/zarr3.py
index e2aac66..6cd1b3a 100644
--- a/src/chunkmirage/frontends/zarr3.py
+++ b/src/chunkmirage/frontends/zarr3.py
@@ -84,6 +84,9 @@ def _array_meta(self, info: ArrayInfo) -> dict:
             "dimension_names": list(info.axes),
         }
 
+    def chunk_key(self, level: int, index) -> str:
+        return f"s{level}/c/" + "/".join(str(i) for i in index)
+
     def encode(self, info: ArrayInfo, index, block: np.ndarray) -> bytes:
         full = self.pad_to_full(info, index, block)
         raw = np.ascontiguousarray(full).astype(full.dtype.newbyteorder("<"), copy=False).tobytes()
diff --git a/src/chunkmirage/materialize.py b/src/chunkmirage/materialize.py
new file mode 100644
index 0000000..55164ee
--- /dev/null
+++ b/src/chunkmirage/materialize.py
@@ -0,0 +1,116 @@
+"""A dataset written to a store as the server serves it: a pipeline's levels, or a box of
+them, computed and written as the files a frontend serves (zarr v3 with OME-Zarr 0.5, zarr
+v2 with OME-Zarr 0.4, or N5), the metadata and each chunk byte for byte what a client would
+fetch, so the store reads back as the dataset did, a stored source of its own. A stage
+computed once (a registration's pass, a model's output over a region) can then be read
+from disk instead of computed again. The browser engine's ``Engine.materialize``
+(``web/src/materialize.ts``) writes a view's region the same way.
+
+The metadata goes first, every level's, so a store whose writing stopped part way is still
+the dataset, zero where no chunk was written yet. Chunks are computed several at once
+(``demand.gather``), and one of nothing but zeros (the fill value) is left out unless
+asked for. The store is anything tensorstore writes: a directory, or a ``file://``,
+``gs://`` or ``s3://`` URL.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Callable, Sequence
+from functools import partial
+
+import numpy as np
+
+from chunkmirage import demand
+from chunkmirage.core import Box
+from chunkmirage.frontends import get_frontend
+from chunkmirage.frontends.base import Metadata
+from chunkmirage.pipeline import Pipeline
+
+FORMATS = ("zarr3", "zarr", "n5")
+BATCH = 256  # chunks asked for at once: the pool's work, and what one wait holds
+
+
+def _kvstore_url(where: str) -> str:
+    if "://" not in where:
+        where = "file://" + os.path.abspath(where)
+    return where if where.endswith("/") else where + "/"
+
+
+def level_box(pipeline: Pipeline, level: int, box: Box | None) -> Box:
+    """``box`` (voxels of level 0 along the last axes it has; the leading ones whole) on
+    ``level``'s voxels, rounded out and clipped to the level; the whole level if None."""
+    info = pipeline.info(level)
+    if box is None:
+        return Box((0,) * info.ndim, info.shape)
+    k, base = box.ndim, pipeline.info(0)
+    if k > info.ndim:
+        raise ValueError(f"a box of {k} axes for a dataset of {info.ndim}")
+    ratio = [float(b) / float(v) for b, v in zip(base.voxel_size[-k:], info.voxel_size[-k:])]
+    lo = [int(np.floor(a * r)) for a, r in zip(box.start, ratio)]
+    hi = [int(np.ceil(b * r)) for b, r in zip(box.stop, ratio)]
+    lead = info.ndim - k
+    return Box((0,) * lead + tuple(lo), info.shape[:lead] + tuple(hi)).clip(info.shape)
+
+
+def materialize(
+    pipeline: Pipeline,
+    where: str,
+    *,
+    format: str = "zarr3",
+    levels: Sequence[int] | None = None,
+    box: Box | None = None,
+    compressor: str | None = None,
+    skip_empty: bool = True,
+    progress: Callable[[dict], None] | None = None,
+) -> dict:
+    """Write ``pipeline`` into the store at ``where`` as ``format`` serves it: ``zarr3``
+    (OME-Zarr 0.5), ``zarr`` (zarr v2, OME-Zarr 0.4) or ``n5``; ``compressor`` as the
+    frontend takes it (blosc by default for zarr, gzip for N5). ``levels``: which levels'
+    chunks to write (all by default; every level's metadata is written whatever it is).
+    ``box``: voxels of level 0 along its last axes, each level the same region on its own
+    voxels, rounded out to whole chunks. ``progress`` is told the counts after each batch
+    of chunks. Returns ``{chunks, written, skipped, bytes}``."""
+    import tensorstore as ts
+
+    if format not in FORMATS:
+        raise ValueError(f"format {format!r}: one of {', '.join(FORMATS)}")
+    fe = get_frontend(format, **({} if compressor is None else {"compressor": compressor}))
+    kv = ts.KvStore.open(_kvstore_url(where)).result()
+    levels = list(range(pipeline.num_levels)) if levels is None else [int(lv) for lv in levels]
+    for lv in levels:
+        if not 0 <= lv < pipeline.num_levels:
+            raise ValueError(f"no level {lv}: the dataset has {pipeline.num_levels}")
+
+    # the metadata first: from then on the store is the dataset, zero where nothing is yet
+    keys = [*fe.root_keys, *(f"s{i}/{k}" for i in range(pipeline.num_levels) for k in fe.level_keys)]
+    for key in keys:
+        meta = fe.resolve(pipeline, key)
+        if isinstance(meta, Metadata):
+            kv.write(key, meta.body).result()
+
+    counts = {"chunks": 0, "written": 0, "skipped": 0, "bytes": 0}
+
+    def one(level: int, index: tuple[int, ...]) -> int | None:
+        """Chunk ``index`` of ``level`` computed and written: its size, or None if empty."""
+        block = pipeline.chunk(level, index)
+        if skip_empty and not block.any():
+            return None
+        data = fe.encode(pipeline.info(level), index, block)
+        kv.write(fe.chunk_key(level, index), data).result()
+        return len(data)
+
+    for level in levels:
+        info = pipeline.info(level)
+        todo = list(info.chunks_covering(level_box(pipeline, level, box)))
+        counts["chunks"] += len(todo)
+        for start in range(0, len(todo), BATCH):
+            for size in demand.gather([partial(one, level, idx) for idx in todo[start : start + BATCH]]):
+                if size is None:
+                    counts["skipped"] += 1
+                else:
+                    counts["written"] += 1
+                    counts["bytes"] += size
+            if progress is not None:
+                progress({"level": level, **counts})
+    return counts
diff --git a/tests/test_materialize.py b/tests/test_materialize.py
new file mode 100644
index 0000000..fd5b267
--- /dev/null
+++ b/tests/test_materialize.py
@@ -0,0 +1,45 @@
+"""A dataset written to a store as the server serves it, read back as a stored dataset."""
+
+import json
+
+import numpy as np
+import pytest
+from typer.testing import CliRunner
+
+from chunkmirage import Pipeline, PipelineSpec, materialize, open_source
+from chunkmirage.cli import app
+from chunkmirage.core import Box
+
+IMG = "synthetic://julia?shape=32,48,48&chunk=16,16,16&levels=2&seed=2&voxel_size=1"
+# moved 24 voxels along x: the last third of every row reads past the image, zeros
+SHIFTED = f"warp://{IMG}?affine=1,0,0,0,0,1,0,0,0,0,1,24"
+
+
+def _whole(p, level):
+    return p.read(level, Box((0,) * p.info(level).ndim, p.info(level).shape))
+
+
+@pytest.mark.parametrize("fmt", ["zarr3", "zarr", "n5"])
+def test_a_dataset_written_reads_back_as_it_was_served(tmp_path, fmt):
+    p = Pipeline.from_spec(PipelineSpec(source=SHIFTED))
+    counts = materialize(p, str(tmp_path / "out"), format=fmt)
+    assert counts["written"] + counts["skipped"] == counts["chunks"] == 18 + 4
+    assert counts["skipped"] > 0, "the chunks past the image are zeros, not written"
+    back = open_source(str(tmp_path / "out"))
+    assert len(back.levels) == p.num_levels
+    for level in range(p.num_levels):
+        got, want = back.levels[level].info, p.info(level)
+        assert (got.shape, got.voxel_size, got.dtype) == (want.shape, want.voxel_size, want.dtype)
+        np.testing.assert_array_equal(back.levels[level].read(Box((0, 0, 0), got.shape)), _whole(p, level))
+
+
+def test_a_box_of_one_level_from_the_command_line(tmp_path):
+    out = tmp_path / "box.zarr"
+    result = CliRunner().invoke(app, ["materialize", IMG, str(out), "--level", "1", "--box", "0,0,0:16,16,16"])
+    assert result.exit_code == 0, result.output
+    assert json.loads(result.stdout.strip().splitlines()[-1])["chunks"] == 1  # level 1's voxels 0-8: one chunk
+    back = open_source(str(out)).levels
+    p = Pipeline.from_spec(PipelineSpec(source=IMG))
+    got = back[1].read(Box((0, 0, 0), back[1].info.shape))
+    np.testing.assert_array_equal(got[:16, :16, :16], _whole(p, 1)[:16, :16, :16])
+    assert not got[16:].any() and not got[:, 16:].any() and not back[0].read(Box((0, 0, 0), (16, 16, 16))).any()