diff --git a/packages/timenet/src/timenet/_fs.py b/packages/timenet/src/timenet/_fs.py
index 97c7bf03e..27856a475 100644
--- a/packages/timenet/src/timenet/_fs.py
+++ b/packages/timenet/src/timenet/_fs.py
@@ -1,4 +1,4 @@
-"""Replace a directory with a complete staging directory."""
+"""Compare directory paths and replace directories from staging."""
 
 from collections.abc import Generator
 from contextlib import contextmanager
@@ -7,6 +7,43 @@
 import uuid
 
 
+def same_directory(a: Path, b: Path) -> bool:
+    """Check whether two paths name the same directory, as the file system sees them.
+
+    Existing paths match when they identify the same directory on disk, and never match a missing path.
+    Missing paths match when their nearest existing parents match and the names below them are equal as written.
+
+    Args:
+        a: The first path, whether or not it exists.
+        b: The second path, whether or not it exists.
+
+    Returns:
+        ``True`` if the paths match.
+    """
+    if a.exists() or b.exists():
+        # On a volume that ignores case, a case variant of an existing path exists too.
+        return a.exists() and b.exists() and a.samefile(b)
+    return _below_existing_parent(a.resolve()) == _below_existing_parent(b.resolve())
+
+
+def _below_existing_parent(path: Path) -> tuple[int, int, tuple[str, ...]]:
+    """Identify a missing path by its nearest existing parent and the names below that parent.
+
+    Args:
+        path: A missing absolute path without symlinks.
+
+    Returns:
+        The parent's device and inode numbers, followed by the names below it as written.
+    """
+    for depth, parent in enumerate(path.parents, start=1):
+        try:
+            info = parent.stat()
+        except OSError:
+            continue
+        return info.st_dev, info.st_ino, path.parts[len(path.parts) - depth :]
+    return 0, 0, path.parts  # nothing exists, not even the root
+
+
 def publish_directory(staging: Path, final: Path) -> None:
     """Rename ``staging`` to ``final``, replacing any existing directory.
 
diff --git a/packages/timenet/src/timenet/cache.py b/packages/timenet/src/timenet/cache.py
index f8c6e5e5a..590697da3 100644
--- a/packages/timenet/src/timenet/cache.py
+++ b/packages/timenet/src/timenet/cache.py
@@ -4,6 +4,11 @@
 download cache. All three sit under the home directory.
 """
 
+# TODO(adrian): Remove temporary glue which maps the "download storage" of this module onto the
+# storage directory, settings().storage_dir, until a later change replaces this module and its
+# view of the storage directory. download() copies into the local registry now, so the storage directory
+# holds only the copies that remote registries cache.
+
 from dataclasses import dataclass
 import os
 from pathlib import Path
@@ -43,12 +48,14 @@ def human_bytes(n: int) -> str:
 
 
 def cached_datasets() -> list[CachedDataset]:
-    """List every dataset version in the local registry and download storage, with sizes.
+    """List dataset versions and their sizes in the local registry and storage directory.
 
     Returns:
-        The cached datasets, sorted by (location, id, version).
+        The cached versions sorted by location, dataset ID, and version.
     """
     config = settings()
+    # TODO(adrian): Remove temporary glue which maps the download storage onto the storage
+    # directory until a later change replaces this module, as the note at the top says.
     found = _scan(config.registry_path, "registry") + _scan(config.storage_dir, "storage")
     return sorted(found, key=lambda dataset: (dataset.location, dataset.dataset_id, dataset.version))
 
@@ -64,20 +71,19 @@ def raw_cache_size() -> int:
 
 
 def clear_cache(*, include_registry: bool) -> tuple[int, list[Path]]:
-    """Delete cached data and return the bytes freed and the directories removed.
+    """Delete the storage directory and raw download cache.
 
-    This always clears the download storage and the raw download cache. The caller can fetch both again.
-    It removes the local registry (locally built datasets) only when the caller sets ``include_registry``.
+    Read or removal failures raise ``OSError``.
 
     Args:
-        include_registry: Also remove the local registry.
+        include_registry: Remove the local registry too.
 
     Returns:
-        A ``(bytes_freed, removed_dirs)`` pair. Every directory in ``removed_dirs`` is gone when this
-        returns. This raises an ``OSError`` if it cannot fully remove a tree. This stops the caller from
-        reporting space it never freed.
+        A ``(bytes_freed, removed_dirs)`` pair with the removed directory paths.
     """
     config = settings()
+    # TODO(adrian): Remove temporary glue which maps the download storage onto the storage
+    # directory until a later change replaces this module, as the note at the top says.
     targets = [config.storage_dir, config.cache_dir]
     if include_registry:
         targets.append(config.registry_path)
diff --git a/packages/timenet/src/timenet/cli/app.py b/packages/timenet/src/timenet/cli/app.py
index 4b27f4610..0134b4b01 100644
--- a/packages/timenet/src/timenet/cli/app.py
+++ b/packages/timenet/src/timenet/cli/app.py
@@ -1,21 +1,21 @@
 """The consumer command-line interface, the command-line mirror of the :class:`~timenet.client.TimeNet` SDK."""
 
 from collections.abc import Callable, Iterable
-from pathlib import Path
-import sys
+from functools import partial
 from typing import TypeVar
 
 from rich.console import Console
-from rich.progress import BarColumn, DownloadColumn, Progress, TextColumn, TimeRemainingColumn, TransferSpeedColumn
 from rich.table import Table
 import typer
 
 from timenet.cache import CachedDataset, cached_datasets, clear_cache, human_bytes, raw_cache_size
 from timenet.cli.runner import run_cli
-from timenet.cli.ui import console as ui
+from timenet.cli.ui import console as ui, progress_renderer
 from timenet.client import TimeNet, default_sources
+from timenet.progress import ProgressCallback, ProgressEvent, StepStarted
 from timenet.types import DatasetMetadata, Domain, License, TaskType
 from timenet.types.access import access_requirements, effective_access
+from timenet.types.references import split_ref
 from timenet.types.tasks import TASKS
 
 
@@ -28,6 +28,15 @@
     "-r",
     help="Registry URL or path that this command reads (else $TIMENET_REGISTRY, then timenet://).",
 )
+# TODO(adrian): Remove temporary glue which maps --registry onto the list after the local registry, until the CLI
+# takes the flags of the defaults.
+_download_registry_option = typer.Option(
+    None,
+    "--registry",
+    "-r",
+    help="Registry URL or path to put after the local registry in the list of sources. The release resolves "
+    "through the whole list.",
+)
 
 T = TypeVar("T")
 
@@ -184,49 +193,41 @@ def info(dataset_id: str, registry: str | None = _registry_option) -> None:
 
 
 @app.command()
-def download(dataset_id: str, registry: str | None = _registry_option) -> None:
-    """Download a dataset version and print its local directory to stdout.
+def download(
+    dataset_id: str,
+    registry: str | None = _download_registry_option,
+    force: bool = typer.Option(
+        False, "--force", help="Copy the release again, even if the local registry holds a complete copy."
+    ),
+) -> None:
+    """Copy a release and its payload sources into the local registry.
 
-    Files go to ``$TIMENET_STORAGE`` or ``<TIMENET_HOME>/storage``.
+    Print the release directory to standard output.
     """
-    client = _client(registry)
-    manifest = client.get(dataset_id)
-    files = list(manifest.files.all_files())
-    total = sum(part.size for part in files)
-    ui.status("⬇️", f"Downloading '{dataset_id}' ({len(files)} files, {human_bytes(total)})…")
-    target = _download(client, dataset_id, total)
-    ui.success(f"Downloaded '{dataset_id}' ({target.name})")
+    sources = default_sources(remote_registry=registry)
+    started: list[StepStarted] = []
+    with progress_renderer() as show:
+        target = TimeNet(sources=sources).download(
+            dataset_id, force=force, progress=partial(_keep_steps, started, show)
+        )
+    if not started:
+        # The local registry already held the release and its payload sources.
+        ui.success(f"Nothing to do: '{split_ref(dataset_id)[0]}' {target.name} is already in {sources[0]}")
     typer.echo(str(target))
 
 
-def _download(client: TimeNet, dataset_id: str, total: int) -> Path:
-    """Download a version, rendering a live progress bar on stderr when the terminal supports it.
+def _keep_steps(started: list[StepStarted], show: ProgressCallback | None, event: ProgressEvent) -> None:
+    """Keep each step event in ``started``, and pass every event on to ``show``.
 
     Args:
-        client: The client to download through.
-        dataset_id: The dataset id, optionally followed by ``@version``.
-        total: The total file size in bytes for the progress bar.
-
-    Returns:
-        The local version directory.
+        started: Receives the step events.
+        show: The callback that shows the events, or ``None`` for a quiet console.
+        event: The progress event.
     """
-    if ui.quiet or not sys.stderr.isatty():
-        return client.download(dataset_id)
-    progress = Progress(
-        TextColumn("[bold blue]{task.description}"),
-        BarColumn(bar_width=None),
-        "[progress.percentage]{task.percentage:>3.0f}%",
-        "•",
-        DownloadColumn(),
-        "•",
-        TransferSpeedColumn(),
-        "•",
-        TimeRemainingColumn(),
-        console=ui.rich,
-    )
-    with progress:
-        task = progress.add_task("downloading", total=total)
-        return client.download(dataset_id, progress_cb=lambda n: progress.advance(task, n))
+    if isinstance(event, StepStarted):
+        started.append(event)
+    if show is not None:
+        show(event)
 
 
 cache_app = typer.Typer(help="Inspect and clear the local cache.", no_args_is_help=True)
@@ -270,14 +271,17 @@ def _cache_table(datasets: list[CachedDataset]) -> Table:
 
 @cache_app.command("clear")
 def cache_clear(
-    all_: bool = typer.Option(False, "--all", help="Also remove the local registry (built datasets)."),
+    all_: bool = typer.Option(False, "--all", help="Also remove the default local registry (builds and downloads)."),
     yes: bool = typer.Option(False, "--yes", "-y", help="Skip the confirmation prompt."),
 ) -> None:
-    """Delete cached data. Clears downloads by default. The ``--all`` flag also removes built datasets."""
+    """Delete the storage directory and raw cache.
+
+    Use ``--all`` to also remove the default local registry.
+    """
     if not yes:
-        what = "downloaded datasets and the raw cache"
+        what = "the storage directory and the raw cache"
         if all_:
-            what += " and the local registry (built datasets)"
+            what += " and the local registry (builds and downloads)"
         typer.confirm(f"Remove {what}?", abort=True)
     freed, removed = clear_cache(include_registry=all_)
     typer.echo(f"Freed {human_bytes(freed)} from {len(removed)} location(s).")
diff --git a/packages/timenet/src/timenet/cli/ui.py b/packages/timenet/src/timenet/cli/ui.py
index f2c8b0ffa..b4ee73d54 100644
--- a/packages/timenet/src/timenet/cli/ui.py
+++ b/packages/timenet/src/timenet/cli/ui.py
@@ -125,7 +125,7 @@ def _download_bars() -> Progress:
 
 
 def _bar_reporter(progress: Progress) -> Callable[[BytesMoved], None]:
-    """Create a callback that displays a progress bar for each transfer URL.
+    """Create a callback that displays a progress bar for each transfer.
 
     Args:
         progress: The display that holds the bars.
@@ -136,10 +136,10 @@ def _bar_reporter(progress: Progress) -> Callable[[BytesMoved], None]:
     tasks: dict[str, TaskID] = {}
 
     def report(event: BytesMoved) -> None:
-        task_id = tasks.get(event.url)
+        task_id = tasks.get(event.subject)
         if task_id is None:
-            task_id = progress.add_task("download", name=event.url.rsplit("/", 1)[-1], total=event.total)
-            tasks[event.url] = task_id
+            task_id = progress.add_task("download", name=event.subject.rsplit("/", 1)[-1], total=event.total)
+            tasks[event.subject] = task_id
         progress.update(task_id, completed=event.done, total=event.total)
 
     return report
@@ -161,10 +161,10 @@ def _text_download_reporter() -> Callable[[BytesMoved], None]:
     def report(event: BytesMoved) -> None:
         # ~20% steps when the size is known, else one line every 50 MB.
         step = event.done * 5 // event.total if event.total else event.done // _UNKNOWN_TOTAL_STEP_BYTES
-        if last.get(event.url) == step:
+        if last.get(event.subject) == step:
             return
-        last[event.url] = step
-        name = event.url.rsplit("/", 1)[-1]
+        last[event.subject] = step
+        name = event.subject.rsplit("/", 1)[-1]
         if event.total:
             console.status("⬇️", f"{name} {event.done * 100 // event.total}% ({human_bytes(event.total)})")
         else:
diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py
index 0a2a49918..c67edadd1 100644
--- a/packages/timenet/src/timenet/client.py
+++ b/packages/timenet/src/timenet/client.py
@@ -1,4 +1,8 @@
-"""Browse and load datasets with the :class:`TimeNet` client."""
+"""Find, read, and download datasets with :class:`TimeNet`.
+
+Clients search their sources in order. Builds and downloads go into the first local registry that is
+not read-only.
+"""
 
 # The public API has a method named ``list``. Deferred annotations keep the type hint
 # ``list[str]`` resolved to the builtin type, not to the method.
@@ -7,25 +11,41 @@
 from collections.abc import Sequence
 import os
 from pathlib import Path
+import shlex
 from typing import TYPE_CHECKING, TypeAlias, TypeVar
 
+from timenet.builders import BuildOptions
 from timenet.config import settings
-from timenet.coordinator.read import read_release
+from timenet.coordinator.build import run_plan
+from timenet.coordinator.plan import Lookups, plan_download
+from timenet.coordinator.read import raise_not_found, read_release
 from timenet.dataset import TimeFDataset
 from timenet.errors import (
     TimeFValidationError,
     TimeNetAccessError,
     TimeNetDatasetNotFoundError,
+    TimeNetError,
+    TimeNetFileError,
     TimeNetMissingParentError,
+    TimeNetVersionUnavailableError,
 )
 from timenet.manifest import Manifest
+from timenet.progress import ProgressCallback
 from timenet.reader import LoadPlan, TimeFReader
 from timenet.registry import BaseRegistry, LocalRegistry, open_registry
 from timenet.registry._paths import release_dir
-from timenet.registry.base import ProgressCallback
-from timenet.sources import InstalledConnectors, ReleaseSource
+from timenet.registry.local import is_same_registry
+from timenet.sources import (
+    Found,
+    FoundInConnector,
+    FoundInRegistry,
+    InstalledConnectors,
+    NotFound,
+    NotServed,
+    ReleaseSource,
+    resolve_release,
+)
 from timenet.types import DatasetMetadata, Domain, License, Task
-from timenet.types.access import require_open_access
 from timenet.types.references import DatasetRef, split_ref
 
 
@@ -221,35 +241,66 @@ def download(
         version: str | None = None,
         *,
         force: bool = False,
-        progress_cb: ProgressCallback | None = None,
+        progress: ProgressCallback | None = None,
     ) -> Path:
-        """Download a dataset version and the data needed to read it.
+        """Copy a dataset release and its payload sources into the local registry.
 
-        The client uses the first registry other than the local registry, or the local registry if it is the only one.
-        Files go to ``$TIMENET_STORAGE`` or ``<TIMENET_HOME>/storage``.
+        The release resolves as in :meth:`open`, without a build. The release and each payload source that the
+        local registry lacks or holds damaged come from the first registry other than the local registry that
+        provides them.
 
         Args:
-            dataset_id: The dataset ID.
-            version: The version string, or ``None`` for the latest.
-            force: Download again, even if an up-to-date copy already exists.
-            progress_cb: A callback that receives each file's byte count after download.
+            dataset_id: The dataset ID, optionally followed by ``@version``.
+            version: An exact version, ``"latest"``, or ``None``, as in :meth:`open`.
+            force: Copy the release again, even if the local registry holds a complete copy.
+            progress: Receives step events and byte counts for each copied release.
 
         Returns:
-            The local ``<storage>/<dataset_id>/<version>/`` directory.
-        """
-        # TODO(adrian): Remove temporary glue which maps download() onto a copy from one registry
-        # into the storage directory, as before the list of sources, until a download into the
-        # local registry replaces it.
+            The release directory in the local registry.
+
+        Raises:
+            TimeFValidationError: If the sources lack a writable local registry, or both arguments specify a version.
+            TimeNetAccessError: If a required release is only listed or a remote release is not open.
+            TimeNetDatasetNotFoundError: If no registry other than the local registry provides the release.
+            TimeNetVersionUnavailableError: If the version is missing and its connector builds another version.
+            TimeNetMissingParentError: If no registry provides a missing payload source.
+            TimeNetVersionConflictError: If the local registry holds a release of the copy with different files.
+            TimeNetFileError: If a copied file is damaged or no registry can replace a damaged local copy.
+            TimeFFormatError: If payload sources have conflicting file records or form a dependency cycle.
+        """  # ruff: ignore[docstring-extraneous-exception] (raised by the plan and the copies)
         dataset_id, version = _resolve_ref(dataset_id, version)
-        registry = self._catalog()
-        self._reject_unhosted_access(dataset_id, version)
-        manifest = registry.get_manifest(dataset_id, version)
-        resolved = str(manifest.metadata.dataset_version)
-        target = release_dir(settings().storage_dir, DatasetRef.parse(f"{dataset_id}@{resolved}"))
-        # The registry owns the fetch: the base implementation stages each file and swaps atomically,
-        # and a remote registry overrides it to resolve and stream every file in parallel.
-        registry.download_version(dataset_id, resolved, target, force=force, manifest=manifest, progress_cb=progress_cb)
-        return target
+        store = self._store
+        if store is None:
+            # Without a local registry, a download has no place to go.
+            raise TimeFValidationError(
+                "download() needs a writable local registry to copy into. "
+                "Add a LocalRegistry without read_only=True to the sources"
+            )
+        lookups = Lookups()
+        answer = lookups.resolve(self._sources, dataset_id, version)
+        if isinstance(answer, NotServed):
+            # A registry lists the release, but no source serves its files.
+            error = answer.access_error(dataset_id)
+            error.hint = _access_hint(dataset_id, self._sources)
+            raise error
+        if isinstance(answer, NotFound):
+            # No source provides the release.
+            try:
+                raise_not_found(dataset_id, version, self._sources)
+            except TimeNetVersionUnavailableError:
+                # An installed connector builds another version, and the error names it.
+                raise
+            except TimeNetDatasetNotFoundError as error:
+                raise TimeNetDatasetNotFoundError(f"{error} Add a registry that holds it to the sources.") from error
+        release = DatasetRef(dataset_id=dataset_id, version=answer.version)
+        try:
+            plan = plan_download(release, store, sources=self._sources, force=force, lookups=lookups)
+        except TimeNetFileError as error:
+            # The local registry holds a damaged copy that no other registry provides.
+            error.hint = self._file_hint(error)
+            raise
+        run_plan(plan, store, options=BuildOptions(progress=progress))
+        return release_dir(store.root, release)
 
     def open(self, dataset_id: str, version: str | None = None) -> TimeFReader:
         """Open a dataset version for lazy reads, where it is.
@@ -295,7 +346,9 @@ def open(self, dataset_id: str, version: str | None = None) -> TimeFReader:
             TimeNetMissingParentError: If no source provides a payload source, or a release that a
                 build needs. The message says which source to add.
             TimeNetFileError: If a release that the call opens in a local registry has a missing
-                file or a file of another size.
+                file or a file of another size. If the damaged copy is in the local registry, the message says
+                how to replace it: with the :meth:`download` that copies an intact release from a
+                registry, else with a new build, if an installed connector builds the release.
             TimeFFormatError: If a payload source differs from the files that the version pins,
                 for example because its build does not reproduce them.
             TimeNetBuildError: If a build runs but fails.
@@ -306,6 +359,9 @@ def open(self, dataset_id: str, version: str | None = None) -> TimeFReader:
         except TimeNetAccessError as error:
             error.hint = _access_hint(dataset_id, self._sources)
             raise
+        except TimeNetFileError as error:
+            error.hint = self._file_hint(error)
+            raise
         except TimeNetMissingParentError as error:
             error.hint = _not_provided_hint(self._sources)
             raise
@@ -348,13 +404,13 @@ def load(self, dataset_id: str, version: str | None = None) -> TimeFDataset:
             return reader.read(load=LoadPlan.ALL_EAGER)
 
     def _catalog(self) -> BaseRegistry:
-        """Select a registry from the configured sources.
+        """Select a registry from the sources.
 
         Returns:
-            The first registry other than the local registry, or the local registry if it is the only one.
+            The first registry other than the local registry, or the local registry if no other registry exists.
 
         Raises:
-            TimeFValidationError: If the sources contain no registry.
+            TimeFValidationError: If no source is a registry.
         """
         # TODO(adrian): Remove temporary glue which maps these methods onto one registry until
         # list() and search() ask every source. This registry gives the results of the client of
@@ -362,30 +418,32 @@ def _catalog(self) -> BaseRegistry:
         registries = [source for source in self._sources if isinstance(source, BaseRegistry)]
         registry = next((registry for registry in registries if registry is not self._store), self._store)
         if registry is None:
-            raise TimeFValidationError("list(), get(), search() and download() need a registry in the sources")
+            raise TimeFValidationError("list(), get() and search() need a registry in the sources")
         return registry
 
-    # TODO(adrian): Remove temporary glue which checks access only for the interim download()
-    # until a later change replaces download().
-    def _reject_unhosted_access(self, dataset_id: str, version: str | None) -> None:
-        """Reject datasets without open access when the registry requires it.
+    def _file_hint(self, error: TimeNetFileError) -> str | None:
+        """Suggest how to replace a damaged release in the local registry.
 
         Args:
-            dataset_id: The dataset ID without a version suffix.
-            version: The resolved version, or ``None`` for the latest.
+            error: The error that names the damaged release.
 
-        Raises:
-            TimeNetAccessError: If the dataset lacks open access and the registry requires it.
+        Returns:
+            A repair hint, or ``None`` if the local registry has no damaged copy or the lookup raises a TimeNet error.
         """
-        registry = self._catalog()
-        if not registry.requires_open_access:
-            return
+        store = self._store
+        if store is None:
+            return None
+        damaged = error.release
         try:
-            manifest = registry.get_manifest(dataset_id, version)
-            require_open_access(manifest.metadata, manifest.lineage)
-        except TimeNetAccessError as error:
-            error.hint = _unhosted_hint(dataset_id)
-            raise
+            held = store.find_manifest(damaged)
+            if held is None or store.damage(damaged, held) is None:
+                # The local registry holds no damaged copy, so the error is about another registry.
+                return None
+            found = _copy_source(self._sources, store, damaged)
+        except TimeNetError:
+            # The hint must not hide the error that it explains, for example if a registry is offline.
+            return None
+        return _repair_hint(damaged, found, store)
 
     def open_torch(
         self, dataset_id: str, version: str | None = None, *, load: LoadPlan = LoadPlan.ALL_EAGER
@@ -445,12 +503,39 @@ def _access_hint(dataset_id: str, sources: Sequence[ReleaseSource]) -> str:
     return f"No source serves its files. Add InstalledConnectors() to the sources, or build it locally with {build}."
 
 
-# TODO(adrian): Remove temporary glue which gives a hint only for the interim download() until a
-# later change replaces download().
-def _unhosted_hint(dataset_id: str) -> str:
-    """Suggest a local build for a dataset whose registry does not serve its files.
+def _copy_source(sources: Sequence[ReleaseSource], store: LocalRegistry, release: DatasetRef) -> Found | None:
+    """Find the answer of a release source outside the local registry.
+
+    Args:
+        sources: The sources to search, in order.
+        store: The local registry to exclude.
+        release: The dataset ID and exact version.
+
+    Returns:
+        The answer of the source that provides or builds the release, or ``None`` if none exists.
+    """
+    others = [source for source in sources if not is_same_registry(source, store)]
+    answer = resolve_release(others, release.dataset_id, str(release.version))
+    return answer if isinstance(answer, Found) else None
+
+
+def _repair_hint(release: DatasetRef, found: Found | None, store: LocalRegistry) -> str:
+    """Suggest how to replace a damaged release in the local registry.
+
+    Args:
+        release: The damaged release.
+        found: The answer of the available source, or ``None`` if none exists.
+        store: The local registry.
 
     Returns:
-        A hint with the build command.
+        A message with a download or build command, or the reason replacement is unavailable.
     """
-    return f"TimeNet does not host its data, so build it locally with `timenet-build build {dataset_id}`."
+    if isinstance(found, FoundInRegistry) and not is_same_registry(found.registry, store):
+        # A registry other than the local registry provides the release.
+        return f"Copy an intact release into the local registry with download('{release}')."
+    if isinstance(found, FoundInConnector):
+        # Only an installed connector provides the release.
+        command = f"timenet-build build {release.dataset_id} --out {shlex.quote(str(store.root))} --force"
+        return f"Build it again with `{command}`."
+    # Only the local registry holds the release.
+    return "No registry other than the local registry provides it, so download() cannot replace it."
diff --git a/packages/timenet/src/timenet/coordinator/build.py b/packages/timenet/src/timenet/coordinator/build.py
index 6ac98f8e4..d13243ef1 100644
--- a/packages/timenet/src/timenet/coordinator/build.py
+++ b/packages/timenet/src/timenet/coordinator/build.py
@@ -83,6 +83,7 @@ def run_plan(plan: Plan, store: LocalRegistry, *, options: BuildOptions | None =
                     step.registry,
                     store,
                     with_payload_sources=step.with_payload_sources,
+                    force=step.force,
                     progress=options.progress,
                 )
             case BuildStep():
diff --git a/packages/timenet/src/timenet/coordinator/plan.py b/packages/timenet/src/timenet/coordinator/plan.py
index 219a5635f..4140ec3cc 100644
--- a/packages/timenet/src/timenet/coordinator/plan.py
+++ b/packages/timenet/src/timenet/coordinator/plan.py
@@ -22,7 +22,7 @@
     find_payload_sources,
     validate_manifest_reference,
 )
-from timenet.registry.local import LocalRegistry
+from timenet.registry.local import LocalRegistry, is_same_registry
 from timenet.registry.version import DatasetVersion
 from timenet.sources import (
     Found,
@@ -64,6 +64,8 @@ class CopyStep:
     """The releases with steps that must finish before this step runs."""
     with_payload_sources: bool = True
     """Also copy the payload sources that the manifest of the release lists."""
+    force: bool = False
+    """Copy the release again, also if the local registry holds it."""
     action: ClassVar[Literal["copy"]] = "copy"
     """The kind of step, for display."""
 
@@ -204,6 +206,8 @@ class _Graph:
     """The sources in preference order. A read reads the output like any other source."""
     open_only: bool
     """True if each release that the walk visits must declare ``access: open``."""
+    builds: bool = True
+    """Whether the plan allows builds from installed connectors."""
     releases: dict[DatasetRef, _Release] = field(default_factory=dict)
     """The nodes, in the order in which the walk found them."""
     pins: dict[DatasetRef, dict[DatasetRef, PayloadSource]] = field(default_factory=dict)
@@ -217,9 +221,9 @@ class _Graph:
     def build_sources(self) -> list[ReleaseSource]:
         """The sources without the output, which is never a source of its own copies and builds."""
         output = self.output
-        return [source for source in self.sources if output is None or not _is_same_registry(source, output)]
+        return [source for source in self.sources if output is None or not is_same_registry(source, output)]
 
-    def visit(
+    def visit(  # ruff: ignore[too-many-arguments]
         self,
         reference: DatasetRef,
         needed_by: DatasetRef | None,
@@ -227,6 +231,7 @@ def visit(
         *,
         needed_in_local_registry: bool,
         answer: ReleaseLookup | None = None,
+        force: bool = False,
     ) -> None:
         """Add a release and its dependencies to the graph.
 
@@ -234,11 +239,12 @@ def visit(
         copy in the output instead of a read.
 
         Args:
-            reference: The exact release to add.
-            needed_by: The release that needs ``reference``, or ``None`` for the requested release.
-            pinned: File requirements from ``needed_by``, or ``None`` for a parent dataset.
+            reference: The release to add.
+            needed_by: The release that requires ``reference``, or ``None`` for the requested release.
+            pinned: The file records that ``needed_by`` requires, or ``None`` for a parent dataset.
             needed_in_local_registry: True if the plan needs a complete copy in the output.
             answer: The answer of the sources for ``reference``, or ``None`` to ask them.
+            force: Copy the release again, even if the output holds a complete copy.
         """
         role: Role = "dataset" if pinned is None else "payload-source"
         if pinned is not None and needed_by is not None:
@@ -250,7 +256,7 @@ def visit(
             if (role, needed_in_local_registry) == (found.role, found.needed_in_local_registry):
                 return
         if needed_in_local_registry:
-            self.local(reference, needed_by, role)
+            self.local(reference, needed_by, role, force=force)
         else:
             self.read(reference, needed_by, role, answer)
 
@@ -302,16 +308,19 @@ def read(
         else:
             raise self.missing(reference, needed_by, role, self.sources)
 
-    def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role) -> None:
+    def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role, *, force: bool = False) -> None:
         """Plan a complete copy of a release in the output, from the output itself, a copy or a build.
 
         Args:
             reference: The exact release.
             needed_by: The release that needs ``reference``, or ``None`` for the requested release.
             role: Whether the release is a dataset or a payload source.
+            force: Copy the release again, even if the output holds a complete copy.
 
         Raises:
             TimeFValidationError: If the plan has no output.
+            TimeNetDatasetNotFoundError: If the plan builds nothing and only installed connectors provide the release,
+                no source provides the requested release, or ``force`` is set and only the output holds it.
             TimeNetFileError: If the output holds a damaged copy that no other registry provides.
             TimeNetAccessError: If a source lists the release but no source serves it.
             TimeNetMissingParentError: If no source provides the release.
@@ -320,7 +329,7 @@ def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role)
         if output is None:
             raise TimeFValidationError(f"{reference} must be in a local registry, but the read has none")
         held, damage = self.in_output(reference)
-        if held is not None:
+        if held is not None and not force:
             if self.open_only:
                 require_open_access(held.metadata, held.lineage)
             sources = list(held.payload_sources.values()) if role == "dataset" else []
@@ -330,11 +339,20 @@ def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role)
             return
         answer = self.lookups.resolve(self.build_sources, reference.dataset_id, str(reference.version))
         if isinstance(answer, FoundInRegistry):
-            self.copy(reference, role, answer.registry)
+            self.copy(reference, role, answer.registry, force=force)
         elif damage is not None:
             # A copy from another registry repairs a damaged copy. Nothing else replaces it.
             raise damage
+        elif held is not None:
+            # With force, the copy needs a registry other than the output, and none provides the release.
+            raise TimeNetDatasetNotFoundError(
+                f"No registry other than the local registry provides {reference}, so download(force=True) cannot copy "
+                "it again."
+            )
         elif isinstance(answer, FoundInConnector):
+            if not self.builds:
+                # A download builds nothing.
+                raise TimeNetDatasetNotFoundError(_not_built(reference))
             self.build(reference, role, answer.builder)
         elif isinstance(answer, NotServed):
             raise answer.access_error(reference.dataset_id)
@@ -342,7 +360,7 @@ def local(self, reference: DatasetRef, needed_by: DatasetRef | None, role: Role)
             # No source provides the release, or only the output does.
             raise self.missing(reference, needed_by, role, (output, *self.build_sources))
 
-    def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> None:
+    def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry, *, force: bool = False) -> None:
         """Add a copy step and plan the payload sources that the copy does not bring.
 
         A dataset copy brings each payload source that ``registry`` serves. Each other payload source that the output
@@ -352,9 +370,10 @@ def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> Non
             reference: The exact release to copy.
             role: Whether the release is a dataset or a payload source.
             registry: The registry to copy from.
+            force: Copy the release again, even if the output holds a complete copy.
 
         Raises:
-            TimeNetDatasetNotFoundError: If the release is absent from the registry.
+            TimeNetDatasetNotFoundError: If the registry lacks the release.
             TimeFFormatError: If a payload source that the copy brings or reuses fails a file pin.
         """  # ruff: ignore[docstring-extraneous-exception] - raised by find_payload_sources
         manifest = self.lookups.manifest(registry, reference)
@@ -374,7 +393,7 @@ def copy(self, reference: DatasetRef, role: Role, registry: BaseRegistry) -> Non
         if self.open_only:
             for found in (*sources.fetched.values(), *sources.held.values()):
                 require_open_access(found.metadata, found.lineage)
-        step = CopyStep(reference, registry, with_payload_sources=role == "dataset")
+        step = CopyStep(reference, registry, with_payload_sources=role == "dataset", force=force)
         # The node holds its final dependencies before the walk visits them. So a release of the walk
         # that depends on it closes a cycle and does not look it up again.
         self.releases[reference] = _Release(role, True, step, tuple(sources.missing))
@@ -574,19 +593,19 @@ def plan_steps(
     force_build: bool,
     open_access_only: bool,
 ) -> tuple[DatasetRef, list[Step]]:
-    """Plan a dataset build and all required releases.
+    """Plan the steps to build a dataset and its dependencies.
 
     A damaged copy of the requested release in the output gets a forced build.
 
     Args:
         builder: The installed connector that builds the dataset.
-        output: The registry to write built releases to.
+        output: The destination registry.
         parents_from: The sources to search for missing releases, in order.
-        force_build: Rebuild the requested release even if the output holds it.
+        force_build: Rebuild the requested release, even if the output holds it.
         open_access_only: Require open access for every release.
 
     Returns:
-        The requested release and ordered steps, ending with its build if needed.
+        The requested release and its steps, with its own build last if needed.
     """
     card = builder.card
     if open_access_only:
@@ -616,19 +635,56 @@ def plan_steps(
     return root, graph.create_ordered_steps()
 
 
-def _is_same_registry(source: ReleaseSource, output: LocalRegistry) -> bool:
-    """Check whether a source is the output registry.
+def plan_download(
+    release: DatasetRef,
+    store: LocalRegistry,
+    *,
+    sources: Sequence[ReleaseSource],
+    force: bool = False,
+    lookups: Lookups | None = None,
+) -> Plan:
+    """Plan the copies that put a release and its payload sources into the local registry.
+
+    Each release that the local registry lacks comes from the first registry other than the local registry
+    that provides it. The plan builds nothing.
 
     Args:
-        source: The source to compare.
-        output: The destination registry.
+        release: The exact release.
+        store: The local registry to copy into.
+        sources: The sources to search in order, excluding registries at the directory of ``store``.
+        force: Copy the release again, even if the local registry holds a complete copy.
+        lookups: The answers and manifests that resolved ``release``, or ``None`` to ask the sources.
+
+    Returns:
+        The copy steps, in the order in which they run.
+
+    Raises:
+        TimeNetDatasetNotFoundError: If a needed release requires a build or no registry can supply a forced copy.
+        TimeNetFileError: If the local registry holds a damaged copy that no other registry provides.
+        TimeNetAccessError: If a source lists a release that the local registry lacks but no source serves it.
+        TimeNetMissingParentError: If no registry provides a payload source that the local registry lacks.
+        TimeFFormatError: If a payload source differs from the files that a release pins.
+    """  # ruff: ignore[docstring-extraneous-exception] - raised by the walk and as values
+    others = [source for source in sources if not is_same_registry(source, store)]
+    graph = _Graph(store, others, open_only=False, builds=False, lookups=lookups or Lookups())
+    graph.visit(release, None, needed_in_local_registry=True, force=force)
+    graph.check()
+    return Plan(tuple(graph.create_ordered_steps()))
+
+
+def _not_built(release: DatasetRef) -> str:
+    """Describe a release that only installed connectors provide, in a plan that builds nothing.
+
+    Args:
+        release: The release.
 
     Returns:
-        ``True`` for the same object or local registries with the same resolved directory.
+        The error message.
     """
-    if source is output:
-        return True
-    return isinstance(source, LocalRegistry) and source.root.resolve() == output.root.resolve()
+    return (
+        f"{release} is not built yet: only an installed connector provides it. open() or load() builds it into "
+        "the local registry."
+    )
 
 
 def describe_searched_sources(sources: Iterable[ReleaseSource], reference: DatasetRef) -> str:
diff --git a/packages/timenet/src/timenet/coordinator/read.py b/packages/timenet/src/timenet/coordinator/read.py
index a1a771529..560dcd0fd 100644
--- a/packages/timenet/src/timenet/coordinator/read.py
+++ b/packages/timenet/src/timenet/coordinator/read.py
@@ -5,7 +5,7 @@
 """
 
 from collections.abc import Sequence
-from typing import cast
+from typing import NoReturn, cast
 
 from timenet.builders import declared, find_card
 from timenet.coordinator.build import run_plan
@@ -86,6 +86,31 @@ def read_release(
     )
 
 
+def raise_not_found(dataset_id: str, version: str | None, sources: Sequence[ReleaseSource]) -> NoReturn:
+    """Raise the error for a release that no source provides.
+
+    For an exact version, an installed connector that builds another version gives a more exact error.
+
+    Args:
+        dataset_id: The dataset ID.
+        version: An exact version, ``"latest"``, or ``None`` or ``""`` for no version.
+        sources: The sources searched for the release.
+
+    Raises:
+        TimeNetVersionUnavailableError: If the installed connector declares a different version.
+        TimeNetInvalidCardError: If the card of the connector cannot be read.
+        TimeNetDatasetNotFoundError: Otherwise, with a message that names the searched sources.
+    """  # ruff: ignore[docstring-extraneous-exception] - raised by declared
+    requested = None if is_latest(version) else DatasetRef.parse(f"{dataset_id}@{version}")
+    connectors = any(isinstance(source, InstalledConnectors) for source in sources)
+    if requested is not None and connectors and find_card(dataset_id) is not None:
+        # The installed connector builds another version.
+        declared(dataset_id, requested.version)
+    names = ", ".join(str(source) for source in sources)
+    wanted = dataset_id if requested is None else str(requested)
+    raise TimeNetDatasetNotFoundError(f"{wanted} is in none of the sources: {names}.")
+
+
 def _walk(
     dataset_id: str, version: str | None, *, sources: Sequence[ReleaseSource], store: LocalRegistry | None
 ) -> tuple[_Graph, DatasetRef]:
@@ -96,23 +121,16 @@ def _walk(
 
     Raises:
         TimeNetDatasetNotFoundError: If no source provides the release.
-    """
+    """  # ruff: ignore[docstring-extraneous-exception] - raised by raise_not_found
     graph = _Graph(store, sources, open_only=False)
     answer = graph.lookups.resolve(sources, dataset_id, version)
     if isinstance(answer, NotFound):
-        requested = None if is_latest(version) else DatasetRef.parse(f"{dataset_id}@{version}")
-        if requested is not None:
+        if not is_latest(version):
             # Nothing but a registry replaces a damaged copy in the local registry.
-            damage = graph.in_output(requested)[1]
+            damage = graph.in_output(DatasetRef.parse(f"{dataset_id}@{version}"))[1]
             if damage is not None:
                 raise damage
-            connectors = any(isinstance(source, InstalledConnectors) for source in sources)
-            if connectors and find_card(dataset_id) is not None:
-                # The installed connector builds another version.
-                declared(dataset_id, requested.version)
-        names = ", ".join(str(source) for source in sources)
-        wanted = dataset_id if requested is None else str(requested)
-        raise TimeNetDatasetNotFoundError(f"{wanted} is in none of the sources: {names}.")
+        raise_not_found(dataset_id, version, sources)
     release = DatasetRef(dataset_id=dataset_id, version=answer.version)
     # The walk takes the answer, so that a read without an exact version does not ask again for the exact version.
     graph.visit(release, None, needed_in_local_registry=False, answer=answer)
diff --git a/packages/timenet/src/timenet/dataset/edit.py b/packages/timenet/src/timenet/dataset/edit.py
index 2275fc5e2..c51fba7eb 100644
--- a/packages/timenet/src/timenet/dataset/edit.py
+++ b/packages/timenet/src/timenet/dataset/edit.py
@@ -509,7 +509,7 @@ def edit_version(
 
     The derived version reuses the base's payload files, so it is written into the base's registry
     as ``<root>/<id>/<dataset_version>/``, where readers find the base. To derive a version in
-    another registry, copy the base there first, for example with ``download_version``, then edit
+    another registry, copy the base there first, for example with ``copy_release()``, then edit
     the copy.
 
     The derived version keeps the base's parent cards, lineage, and import map, without the
diff --git a/packages/timenet/src/timenet/progress.py b/packages/timenet/src/timenet/progress.py
index 87fb8a9b6..d05ce2713 100644
--- a/packages/timenet/src/timenet/progress.py
+++ b/packages/timenet/src/timenet/progress.py
@@ -39,15 +39,15 @@ class StepFinished:
 
 @dataclass(frozen=True)
 class BytesMoved:
-    """The byte count of a transfer in progress.
+    """The byte count for a transfer in progress.
 
     Attributes:
-        url: The source or destination URL.
-        done: The number of bytes moved so far.
-        total: The transfer size in bytes, or ``None`` if unknown.
+        subject: What moves: a source or target URL, or a copied release ID such as ``"org/name@1.0.0"``.
+        done: The bytes moved so far.
+        total: The total byte count, or ``None`` if unknown.
     """
 
-    url: str
+    subject: str
     done: int
     total: int | None
 
diff --git a/packages/timenet/src/timenet/registry/_copy.py b/packages/timenet/src/timenet/registry/_copy.py
index f29131ebb..15fc05b99 100644
--- a/packages/timenet/src/timenet/registry/_copy.py
+++ b/packages/timenet/src/timenet/registry/_copy.py
@@ -4,12 +4,13 @@
 Manifests must match for existing releases, except for their build environments.
 """
 
+from collections.abc import Callable
 from pathlib import Path
 import tempfile
 
-from timenet.errors import TimeNetMissingParentError, TimeNetVersionConflictError
+from timenet.errors import TimeFValidationError, TimeNetMissingParentError, TimeNetVersionConflictError
 from timenet.manifest import Manifest, same_release
-from timenet.progress import ProgressCallback, StepFinished, StepStarted
+from timenet.progress import BytesMoved, ProgressCallback, StepFinished, StepStarted
 from timenet.registry._paths import release_dir
 from timenet.registry.base import BaseRegistry
 from timenet.registry.dependencies import find_payload_sources, validate_manifest_reference
@@ -19,39 +20,49 @@
 from timenet.types.references import DatasetRef
 
 
-def copy_release(
+def copy_release(  # ruff: ignore[too-many-arguments]
     reference: DatasetRef,
     source: BaseRegistry,
     dest: WritableRegistry,
     *,
     with_payload_sources: bool = True,
+    force: bool = False,
     progress: ProgressCallback | None = None,
 ) -> None:
-    """Copy a release and its required payload sources to another registry.
+    """Copy a release and its required payload sources to a registry.
 
-    The copy reuses complete releases in ``dest`` and fetches damaged local copies again.
-    It checks access, file records and pins before copying payload sources and then the requested release.
+    Payload sources go first, a complete copy in ``dest`` is reused, and a damaged local copy is fetched again.
+    All checks run before the first copy, so a failed check copies nothing.
 
     Args:
         reference: The exact release to copy.
         source: The registry to copy from.
         dest: The registry to copy to.
         with_payload_sources: Also copy the payload sources that the manifest of the release lists.
-        progress: Callback for ``StepStarted`` and ``StepFinished`` events for each copied release.
+        force: Fetch ``reference`` again, even if ``dest`` has a complete local copy.
+        progress: Receives step events for each copied release and byte counts when ``dest`` is local.
 
     Raises:
         TimeNetRegistryError: If ``dest`` is read-only.
-        TimeNetVersionConflictError: If ``dest`` holds other files or another card for a release being copied.
-        TimeNetAccessError: If a release fails an open access requirement of ``dest`` or ``source``.
-        TimeNetDatasetNotFoundError: If ``source`` does not hold the release.
-        TimeNetMissingParentError: If ``source`` lacks a payload source and ``dest`` holds no complete copy of it.
-        TimeFFormatError: If a manifest names a different release or a pin fails.
+        TimeFValidationError: If ``force`` is set and ``dest`` is not local.
+        TimeNetVersionConflictError: If ``dest`` holds a release of the copy with other files or another card.
+        TimeNetAccessError: If a release is not open and ``dest`` or ``source`` requires open access.
+        TimeNetDatasetNotFoundError: If ``source`` lacks the release.
+        TimeNetMissingParentError: If neither registry holds a complete payload source.
+        TimeFFormatError: If a manifest names the wrong release or a pin fails.
     """  # ruff: ignore[docstring-extraneous-exception] - raised by the helpers
     dest.check_writable(f"a copy of {reference}")
+    if force and not isinstance(dest, LocalRegistry):
+        raise TimeFValidationError(f"force=True needs a local dest: {dest} never writes a published version again")
     manifests = _manifests(reference, source, dest, with_payload_sources=with_payload_sources)
     # Check every release before anything is copied, so that a refusal leaves dest unchanged. A payload
-    # source that only dest holds is complete there, so it is never pending.
-    pending = [ref for ref, manifest in manifests.items() if _needs_copy(reference, ref, manifest, source, dest)]
+    # source that only dest holds is complete there, so it is never pending. _needs_copy() runs first,
+    # so that other files raise also with force.
+    pending = [
+        ref
+        for ref, manifest in manifests.items()
+        if _needs_copy(reference, ref, manifest, source, dest) or (force and ref == reference)
+    ]
     _check_access(manifests, set(pending), source, dest)
     if not pending:
         return
@@ -61,26 +72,38 @@ def copy_release(
         for ref in pending:
             if progress is not None:
                 progress(StepStarted(ref, "copy"))
-            _copy_one(ref, manifests[ref], source, dest, Path(staging))
+            again = force and ref == reference
+            _copy_one(ref, manifests[ref], source, dest, Path(staging), force=again, progress=progress)
             if progress is not None:
                 progress(StepFinished(ref, "copy"))
 
 
-def _copy_one(
-    reference: DatasetRef, manifest: Manifest, source: BaseRegistry, dest: WritableRegistry, staging: Path
+def _copy_one(  # ruff: ignore[too-many-arguments]
+    reference: DatasetRef,
+    manifest: Manifest,
+    source: BaseRegistry,
+    dest: WritableRegistry,
+    staging: Path,
+    *,
+    force: bool,
+    progress: ProgressCallback | None,
 ) -> None:
-    """Copy one release without its payload sources to ``dest``, or repair its local copy.
+    """Copy or repair one release without its payload sources in the destination registry.
 
     Args:
         reference: The release to copy.
-        manifest: The release's manifest from ``source``.
+        manifest: The release manifest from ``source``.
         source: The registry to copy from.
         dest: The registry to copy to.
-        staging: Temporary download root when both registries are remote.
+        staging: A temporary directory for downloads from a remote source.
+        force: Fetch the release again, even if ``dest`` has a complete local copy.
+        progress: Receives byte progress events when ``dest`` is local.
     """
     # A local registry is a folder that the copy reads or writes directly.
     if isinstance(dest, LocalRegistry):
-        source.download_release(reference, release_dir(dest.local_root, reference), manifest=manifest)
+        report = None if progress is None else _bytes_of(reference, manifest, progress)
+        directory = release_dir(dest.local_root, reference)
+        source.download_release(reference, directory, force=force, manifest=manifest, progress_cb=report)
         return
     if isinstance(source, LocalRegistry):
         directory = release_dir(source.root, reference)
@@ -93,6 +116,28 @@ def _copy_one(
     dest.put_release(directory)
 
 
+def _bytes_of(reference: DatasetRef, manifest: Manifest, progress: ProgressCallback) -> Callable[[int], None]:
+    """Create a callback that reports cumulative bytes for a release.
+
+    Args:
+        reference: The release to report.
+        manifest: The manifest whose file sizes define the total.
+        progress: The callback that receives ``BytesMoved`` events.
+
+    Returns:
+        A callback that accepts each file or chunk's byte count and reports the bytes moved so far.
+    """
+    total = sum(part.size for part in manifest.files.all_files())
+    done = 0
+
+    def report(count: int) -> None:
+        nonlocal done
+        done += count
+        progress(BytesMoved(str(reference), done, total))
+
+    return report
+
+
 def _check_access(
     manifests: dict[DatasetRef, Manifest], fetched: set[DatasetRef], source: BaseRegistry, dest: WritableRegistry
 ) -> None:
diff --git a/packages/timenet/src/timenet/registry/base.py b/packages/timenet/src/timenet/registry/base.py
index 81e7be8d4..d6bd2c131 100644
--- a/packages/timenet/src/timenet/registry/base.py
+++ b/packages/timenet/src/timenet/registry/base.py
@@ -13,12 +13,12 @@
 from typing import BinaryIO, TypeVar
 
 from timenet._fs import staged_directory
-from timenet.errors import TimeFValidationError, TimeNetDatasetNotFoundError, TimeNetFileError, TimeNetRegistryError
+from timenet.errors import TimeNetDatasetNotFoundError, TimeNetFileError, TimeNetRegistryError
 from timenet.format.checksums import file_checksum
 from timenet.format.constants import MANIFEST_FILE
 from timenet.manifest import FilePart, Manifest
-from timenet.registry._paths import safe_version_path, split_release_dir
-from timenet.registry.dependencies import prepare_local_copy, validate_manifest_reference
+from timenet.registry._paths import safe_version_path
+from timenet.registry.dependencies import validate_manifest_reference
 from timenet.registry.shared import complete_copy
 from timenet.registry.version import DatasetVersion
 from timenet.sources import FoundInRegistry, NotFound, NotServed, ReleaseLookup
@@ -29,7 +29,7 @@
 
 T = TypeVar("T")
 
-# Reports download progress: called with each file (or chunk) of bytes as it lands. A ``download_version``
+# Reports download progress: called with each file (or chunk) of bytes as it lands. A ``download_release``
 # implementation calls it so a caller can render a progress bar; ``None`` downloads silently.
 ProgressCallback = Callable[[int], None]
 
@@ -176,47 +176,6 @@ def resolve(self, dataset_id: str, version: Version | None) -> ReleaseLookup:
             return FoundInRegistry(held, self, manifest)
         return NotServed(held, access, access_requirements(manifest.metadata, manifest.lineage), manifest)
 
-    def download_version(  # ruff: ignore[too-many-arguments]
-        self,
-        dataset_id: str,
-        version: str,
-        dest_dir: str | Path,
-        *,
-        force: bool = False,
-        manifest: Manifest | None = None,
-        progress_cb: ProgressCallback | None = None,
-    ) -> None:
-        """Download complete local copies of a release and of each of its payload sources.
-
-        Only the release's own payload sources are downloaded: it lists every release whose files it
-        reads. Each is checked against the files the release pins.
-
-        Args:
-            dataset_id: The dataset ID.
-            version: The exact release version.
-            dest_dir: The local ``<registry>/<org>/<name>/<version>`` directory.
-            force: Download the requested release again.
-            manifest: An already-loaded manifest for the requested release.
-            progress_cb: Report transferred bytes.
-
-        Raises:
-            TimeFValidationError: If the destination does not identify the requested release.
-        """
-        reference = DatasetRef.parse(f"{dataset_id}@{version}")
-        root, destination_ref = split_release_dir(Path(dest_dir))
-        if destination_ref != reference:
-            raise TimeFValidationError(
-                f"download destination {dest_dir} identifies {destination_ref}, expected {reference}"
-            )
-        prepare_local_copy(
-            root,
-            reference,
-            fetch_from=self,
-            manifest=manifest,
-            force=force,
-            progress_cb=progress_cb,
-        )
-
     def download_release(
         self,
         reference: DatasetRef,
diff --git a/packages/timenet/src/timenet/registry/local.py b/packages/timenet/src/timenet/registry/local.py
index 22dc22664..c28acec6d 100644
--- a/packages/timenet/src/timenet/registry/local.py
+++ b/packages/timenet/src/timenet/registry/local.py
@@ -7,16 +7,17 @@
 
 import pyarrow.fs as pafs
 
-from timenet._fs import staged_directory
+from timenet._fs import same_directory, staged_directory
 from timenet.dataset import TimeFDataset
 from timenet.errors import TimeFFormatError, TimeFValidationError, TimeNetDatasetNotFoundError, TimeNetFileError
 from timenet.format.constants import MANIFEST_FILE
 from timenet.manifest import Manifest, PayloadSource, read_manifest
 from timenet.registry._paths import release_dir
+from timenet.registry.base import BaseRegistry
 from timenet.registry.dependencies import check_file_sizes, validate_manifest_reference
 from timenet.registry.version import DatasetVersion
 from timenet.registry.writable import WritableRegistry
-from timenet.sources import FoundInRegistry, NotFound, ReleaseLookup
+from timenet.sources import FoundInRegistry, NotFound, ReleaseLookup, ReleaseSource
 from timenet.types import DatasetMetadata, Version
 from timenet.types.metadata_fields import validate_dataset_id
 from timenet.types.references import DatasetRef, is_latest
@@ -325,6 +326,25 @@ def _latest_version(self, dataset_id: str) -> str | None:
         return max(versions, key=Version.parse)
 
 
+def is_same_registry(source: ReleaseSource, registry: BaseRegistry) -> bool:
+    """Check whether a source matches a registry.
+
+    Args:
+        source: The source to compare.
+        registry: The registry to compare against.
+
+    Returns:
+        ``True`` for the same object or local registries over the same directory.
+    """
+    if source is registry:
+        return True
+    return (
+        isinstance(source, LocalRegistry)
+        and isinstance(registry, LocalRegistry)
+        and same_directory(source.root, registry.root)
+    )
+
+
 def _is_version(name: str) -> bool:
     """Return whether a directory name is a parseable version.
 
diff --git a/packages/timenet/tests/coordinator/test_build.py b/packages/timenet/tests/coordinator/test_build.py
index c957ffd8c..fc866963a 100644
--- a/packages/timenet/tests/coordinator/test_build.py
+++ b/packages/timenet/tests/coordinator/test_build.py
@@ -8,7 +8,7 @@
 from timenet.coordinator.plan import BuildStep, CopyStep, Plan
 from timenet.dataset import TimeFDataset
 from timenet.errors import TimeNetAccessError, TimeNetRegistryError, TimeNetVersionUnavailableError
-from timenet.progress import StepFinished, StepStarted
+from timenet.progress import BytesMoved, StepFinished, StepStarted
 from timenet.reader import TimeFReader
 from timenet.registry import LocalRegistry, open_local_version
 from timenet.registry._paths import release_dir
@@ -144,7 +144,8 @@ def test_a_build_copies_a_missing_parent_and_reports_each_step_once(tmp_path, st
         CopyStep(PARENT_REF, origin),
         _build(CHILD_REF, depends_on=(PARENT_REF,)),
     )
-    assert events == [
+    # A copy also sends the bytes that it moves.
+    assert [event for event in events if not isinstance(event, BytesMoved)] == [
         StepStarted(PARENT_REF, "copy"),
         StepFinished(PARENT_REF, "copy"),
         StepStarted(CHILD_REF, "build"),
diff --git a/packages/timenet/tests/reader/test_shared_storage.py b/packages/timenet/tests/reader/test_shared_storage.py
index 0d9a7b49b..c09ca7c1e 100644
--- a/packages/timenet/tests/reader/test_shared_storage.py
+++ b/packages/timenet/tests/reader/test_shared_storage.py
@@ -13,6 +13,7 @@
 from timenet.manifest import Manifest
 from timenet.reader import LoadPlan, TimeFReader
 from timenet.registry import LocalRegistry, open_local_version
+from timenet.registry.dependencies import prepare_local_copy
 from timenet.testing import make_dataset
 from timenet.types import (
     Annotation,
@@ -599,7 +600,7 @@ def test_a_download_fetches_only_the_direct_payload_sources(tmp_path):
         b = reader.read()
         store(registry_root, _derive(b, "test/c", [record for record in b.records if _name(record) == "b"]))
     cache = tmp_path / "cache"
-    LocalRegistry(registry_root).download_version("test/c", "1.0.0", cache / "test/c/1.0.0")
+    prepare_local_copy(cache, DatasetRef.parse("test/c@1.0.0"), fetch_from=LocalRegistry(registry_root))
     assert sorted(path.parent.relative_to(cache).as_posix() for path in cache.rglob("manifest.json")) == [
         "test/b/1.0.0",
         "test/c/1.0.0",
diff --git a/packages/timenet/tests/registry/test_copy.py b/packages/timenet/tests/registry/test_copy.py
index 43514741f..bb52830e4 100644
--- a/packages/timenet/tests/registry/test_copy.py
+++ b/packages/timenet/tests/registry/test_copy.py
@@ -17,11 +17,12 @@
 from timenet.dataset import Record, Signal, Source, TimeFDataset
 from timenet.errors import (
     TimeFFormatError,
+    TimeFValidationError,
     TimeNetAccessError,
     TimeNetMissingParentError,
     TimeNetVersionConflictError,
 )
-from timenet.progress import ProgressEvent, StepFinished, StepStarted
+from timenet.progress import BytesMoved, ProgressEvent, StepFinished, StepStarted
 from timenet.reader import TimeFReader
 from timenet.registry import LocalRegistry, RemoteRegistry, WritableRegistry, copy_release, open_local_version
 from timenet.registry.shared import complete_copy
@@ -149,6 +150,11 @@ def _remote(tmp_path: Path, name: str, *, token: str | None = None) -> tuple[Rem
     return registry, root
 
 
+def _steps(events: list[ProgressEvent]) -> list[ProgressEvent]:
+    """Return the step events, without the byte events of each copy."""
+    return [event for event in events if not isinstance(event, BytesMoved)]
+
+
 @pytest.fixture
 def source(tmp_path) -> LocalRegistry:
     root = tmp_path / "source"
@@ -264,7 +270,7 @@ def test_copy_release_reads_a_payload_source_that_only_dest_holds(source, dest):
     copy_release(child_ref, _without_parent(source), dest, progress=events.append)
     assert dest.get_manifest(*child) == source.get_manifest(*child)
     assert dest.get_manifest(*parent) == before
-    assert events == [StepStarted(child_ref, "copy"), StepFinished(child_ref, "copy")]
+    assert _steps(events) == [StepStarted(child_ref, "copy"), StepFinished(child_ref, "copy")]
 
 
 def test_copy_release_refuses_a_payload_source_that_only_dest_holds_with_other_files(source, dest):
@@ -321,17 +327,42 @@ def test_copy_release_repairs_a_damaged_local_copy(source, tmp_path):
     events: list[ProgressEvent] = []
     copy_release(child_ref, source, dest, progress=events.append)
     assert complete_copy(child_dir, manifest)
-    assert events == [StepStarted(child_ref, "copy"), StepFinished(child_ref, "copy")]
+    assert _steps(events) == [StepStarted(child_ref, "copy"), StepFinished(child_ref, "copy")]
+
+
+def test_copy_release_with_force_copies_a_complete_local_copy_again(source, tmp_path):
+    dest = LocalRegistry(tmp_path / "dest")
+    copy_release(child_ref, source, dest)
+    child_dir = dest.root / child_ref.dataset_id / str(child_ref.version)
+    parent_dir = dest.root / parent_ref.dataset_id / str(parent_ref.version)
+    for directory in (child_dir, parent_dir):
+        (directory / "stale.txt").write_text("left over")
+    events: list[ProgressEvent] = []
+    copy_release(child_ref, source, dest, force=True, progress=events.append)
+    # The requested release is copied again as a whole. Its complete payload source stays.
+    assert not (child_dir / "stale.txt").exists()
+    assert (parent_dir / "stale.txt").exists()
+    assert complete_copy(child_dir, dest.get_manifest(*child))
+    assert _steps(events) == [StepStarted(child_ref, "copy"), StepFinished(child_ref, "copy")]
+    # force=True does not replace a copy with other files.
+    other = LocalRegistry(tmp_path / "other")
+    other.put_dataset(_other_build(child_ref.dataset_id))
+    with pytest.raises(TimeNetVersionConflictError, match="new version"):
+        copy_release(child_ref, source, other, force=True)
+
+
+def test_copy_release_with_force_refuses_a_registry_that_is_not_local(source, tmp_path):
+    dest, _ = _remote(tmp_path, "dest", token="tok_rw")
+    with pytest.raises(TimeFValidationError, match=r"^force=True needs a local dest"):
+        copy_release(child_ref, source, dest, force=True)
+    assert not dest.exists(*child)
+    assert not dest.exists(*parent)
 
 
 def test_copy_release_reports_each_copied_release(source, tmp_path):
     events: list[ProgressEvent] = []
     copy_release(child_ref, source, LocalRegistry(tmp_path / "dest"), progress=events.append)
-    assert [(e.action, e.release) for e in events if isinstance(e, StepStarted)] == [
-        ("copy", parent_ref),
-        ("copy", child_ref),
-    ]
-    assert events == [
+    assert _steps(events) == [
         StepStarted(parent_ref, "copy"),
         StepFinished(parent_ref, "copy"),
         StepStarted(child_ref, "copy"),
@@ -339,6 +370,29 @@ def test_copy_release_reports_each_copied_release(source, tmp_path):
     ]
 
 
+@pytest.mark.parametrize("kind", ["local", "remote"])
+def test_copy_release_reports_the_bytes_of_each_release_that_it_copies_to_a_local_registry(kind, tmp_path):
+    if kind == "local":
+        _write_parent_and_child(tmp_path / "source")
+        origin: LocalRegistry | RemoteRegistry = LocalRegistry(tmp_path / "source")
+    else:
+        origin, root = _remote(tmp_path, "source")
+        _write_parent_and_child(root)
+    events: list[ProgressEvent] = []
+    copy_release(child_ref, origin, LocalRegistry(tmp_path / "dest"), progress=events.append)
+    for release in (parent_ref, child_ref):
+        # The bytes of a release arrive between the start and the end of its step.
+        start, end = events.index(StepStarted(release, "copy")), events.index(StepFinished(release, "copy"))
+        moved = events[start + 1 : end]
+        manifest = origin.get_manifest(release.dataset_id, str(release.version))
+        total = sum(part.size for part in manifest.files.all_files())
+        done = [event.done for event in moved if isinstance(event, BytesMoved)]
+        assert done
+        assert moved == [BytesMoved(str(release), count, total) for count in done]
+        assert done == sorted(done)
+        assert done[-1] == total
+
+
 def test_copy_release_copies_only_the_payload_sources_that_the_release_lists(chain_source, dest):
     # The grandchild reads the middle release only. The source lacks the grandparent, which the middle reads.
     shutil.rmtree(chain_source.root / grandparent_ref.dataset_id)
diff --git a/packages/timenet/tests/registry/test_remote.py b/packages/timenet/tests/registry/test_remote.py
index 59c182c16..6a4337c35 100644
--- a/packages/timenet/tests/registry/test_remote.py
+++ b/packages/timenet/tests/registry/test_remote.py
@@ -22,7 +22,7 @@
 from timenet.format.checksums import file_checksum
 from timenet.manifest import Manifest
 from timenet.reader import LoadPlan, TimeFReader
-from timenet.registry import LocalRegistry, RemoteRegistry, open_local_version
+from timenet.registry import LocalRegistry, RemoteRegistry, copy_release, open_local_version
 from timenet.sources import FoundInRegistry, NotServed
 from timenet.testing import assert_datasets_equal, make_dataset
 from timenet.types import Access, DatasetSchema, Domain, License, Version
@@ -196,13 +196,26 @@ def test_token_required_when_configured(version_dir, tmp_path):
 def test_full_download_writes_identical_bytes(version_dir, tmp_path):
     registry, _ = _remote(version_dir, tmp_path)
     directory, manifest = version_dir
-    dest = tmp_path / "out" / manifest.metadata.dataset_id / str(manifest.metadata.dataset_version)
-    registry.download_version(manifest.metadata.dataset_id, str(manifest.metadata.dataset_version), dest)
+    release = DatasetRef(dataset_id=manifest.metadata.dataset_id, version=manifest.metadata.dataset_version)
+    dest = tmp_path / "out" / release.dataset_id / str(release.version)
+    registry.download_release(release, dest)
     for part in manifest.files.all_parts():
         assert (dest / part).read_bytes() == (directory / part).read_bytes()
     assert (dest / "manifest.json").exists()
 
 
+def test_a_download_into_the_store_reads_the_remote_manifest_for_the_plan_and_for_the_copy(version_dir, tmp_path):
+    registry, requests = _remote(version_dir, tmp_path)
+    directory, manifest = version_dir
+    store = LocalRegistry(tmp_path / "store")
+
+    path = TimeNet(sources=[store, registry]).download(manifest.metadata.dataset_id)
+
+    assert (path / "manifest.json").read_bytes() == (directory / "manifest.json").read_bytes()
+    # The plan reads the manifest once, and the copy reads it again before it writes.
+    assert sum(request.url.path.endswith("/manifest") for request in requests) == 2
+
+
 def test_full_download_waits_out_a_rate_limited_presign(version_dir, tmp_path, monkeypatch, caplog):
     directory, manifest = version_dir
     transport, _ = build_fake(directory)
@@ -217,9 +230,10 @@ def handler(request):
     sleep = AsyncMock()
     monkeypatch.setattr("timenet.registry.remote._http.asyncio.sleep", sleep)
     registry = RemoteRegistry("http://api.local", transport=httpx.MockTransport(handler), cache_dir=tmp_path / "cache")
-    dest = tmp_path / "out" / manifest.metadata.dataset_id / str(manifest.metadata.dataset_version)
+    release = DatasetRef(dataset_id=manifest.metadata.dataset_id, version=manifest.metadata.dataset_version)
+    dest = tmp_path / "out" / release.dataset_id / str(release.version)
 
-    registry.download_version(manifest.metadata.dataset_id, str(manifest.metadata.dataset_version), dest)
+    registry.download_release(release, dest)
 
     assert limited
     assert sleep.await_args_list == [call(1.0)] * len(limited)
@@ -239,10 +253,11 @@ def handler(request):
 
     monkeypatch.setattr("timenet.registry.remote._http.asyncio.sleep", AsyncMock())
     registry = RemoteRegistry("http://api.local", transport=httpx.MockTransport(handler), cache_dir=tmp_path / "cache")
-    dest = tmp_path / "out" / manifest.metadata.dataset_id / str(manifest.metadata.dataset_version)
+    release = DatasetRef(dataset_id=manifest.metadata.dataset_id, version=manifest.metadata.dataset_version)
+    dest = tmp_path / "out" / release.dataset_id / str(release.version)
 
     with pytest.raises(TimeNetRegistryError, match="429"):
-        registry.download_version(manifest.metadata.dataset_id, str(manifest.metadata.dataset_version), dest)
+        registry.download_release(release, dest)
     assert not dest.exists()
 
 
@@ -257,7 +272,7 @@ def test_a_corrupted_remote_download_names_the_file_and_its_checksums(version_di
     dest = tmp_path / "out" / release.dataset_id / str(release.version)
 
     with pytest.raises(TimeNetFileError) as caught:
-        registry.download_version(release.dataset_id, str(release.version), dest)
+        registry.download_release(release, dest)
 
     error = caught.value
     # A caller that handles format errors still catches it.
@@ -279,7 +294,7 @@ def test_a_remote_download_whose_size_differs_from_its_record_names_both_sizes(v
     dest = tmp_path / "out" / release.dataset_id / str(release.version)
 
     with pytest.raises(TimeNetFileError) as caught:
-        registry.download_version(release.dataset_id, str(release.version), dest)
+        registry.download_release(release, dest)
 
     error = caught.value
     size = record["size"]
@@ -485,8 +500,8 @@ def handler(request):
 
 
 def test_a_registry_url_in_the_sources_caches_under_the_storage_directory(tmp_path, monkeypatch):
-    # A remote registry that the sources name by URL caches under the storage directory, where
-    # download() copies to, so download() and load() share one cache-first directory.
+    # A remote registry that the sources name by URL reuses the complete copies under the storage
+    # directory.
     monkeypatch.setenv("TIMENET_STORAGE", str(tmp_path / "store"))
     (registry,) = TimeNet(sources=["https://registry.example.test"])._sources
     assert isinstance(registry, RemoteRegistry)
@@ -545,7 +560,7 @@ def handler(request):
             assert any(parent_id + "/1.0.0/download/" in path for path in requests)
             assert not any(parent_id + "/1.0.0/download/control.duckdb" in path for path in requests)
             assert not (cache / parent_id / "1.0.0").exists()
-        remote.download_version("test/child", "1.0.0", cache / "test/child/1.0.0")
+        copy_release(DatasetRef.parse("test/child@1.0.0"), remote, LocalRegistry(cache))
     with TimeNet(sources=[remote]).open("test/child", "1.0.0") as reader:
         assert reader.get(Signal, signal_id).to_arrow().equals(expected)
     assert not any("test/build-only" in path for path in requests)
@@ -555,7 +570,9 @@ def handler(request):
         assert reader.get(Signal, signal_id).to_arrow().equals(expected)
     # The parent is already complete: preparing it again requires no network requests.
     prepared_requests = list(requests)
-    remote.download_version(parent_id, "1.0.0", cache / parent_id / "1.0.0")
+    parent = DatasetRef.parse(f"{parent_id}@1.0.0")
+    manifest = LocalRegistry(cache).get_manifest(parent_id, "1.0.0")
+    remote.download_release(parent, cache / parent_id / "1.0.0", manifest=manifest)
     assert (cache / parent_id / "1.0.0/control.duckdb").is_file()
     assert requests == prepared_requests
 
@@ -578,13 +595,15 @@ def _gated_release(origin: Path, page: str) -> tuple[str, DatasetRef]:
     return dataset.metadata.dataset_id, release
 
 
-def test_a_read_checks_access_before_it_reads_from_a_remote_registry(tmp_path):
+@pytest.mark.parametrize("entry_point", ["open", "download"])
+def test_a_read_or_a_download_checks_access_before_it_reads_from_a_remote_registry(tmp_path, entry_point):
     dataset_id, release = _gated_release(tmp_path / "origin", "ancestor")
     transport, requests = build_fake(next((tmp_path / "origin").rglob("manifest.json")).parent)
     remote = RemoteRegistry("http://api.local", transport=transport, cache_dir=tmp_path / "cache")
+    client = TimeNet(sources=[LocalRegistry(tmp_path / "local"), remote])
 
     with pytest.raises(TimeNetAccessError) as caught:
-        TimeNet(sources=[LocalRegistry(tmp_path / "local"), remote]).open(dataset_id)
+        getattr(client, entry_point)(dataset_id)
 
     assert caught.value.requirements == (AccessRequirement(release, Access.CREDENTIALED, "https://example.org/dua"),)
     assert caught.value.hint == (
diff --git a/packages/timenet/tests/registry/test_s3.py b/packages/timenet/tests/registry/test_s3.py
index 4797ab5bc..a63e3a3c3 100644
--- a/packages/timenet/tests/registry/test_s3.py
+++ b/packages/timenet/tests/registry/test_s3.py
@@ -99,13 +99,14 @@ def test_objects_live_under_a_datasets_prefix(s3_root, tmp_path):
     assert all(key.startswith("registry/datasets/") for key in keys)
 
 
-def test_download_then_load_serves_from_cache(s3_root, tmp_path, monkeypatch):
+def test_download_then_load_serves_from_cache(s3_root, tmp_path):
     registry = S3Registry(s3_root, cache_dir=tmp_path / "cache")
     dataset_id = make_dataset().metadata.dataset_id
     registry.put_dataset(make_dataset())
-    monkeypatch.setenv("TIMENET_STORAGE", str(tmp_path / "cache"))
-    client = TimeNet(sources=[registry])
-    path = client.download(dataset_id)  # boto3 GetObject per file, into the cache
+    # The store is the cache directory of the registry, so the registry reuses each download.
+    client = TimeNet(sources=[LocalRegistry(tmp_path / "cache"), registry])
+    path = client.download(dataset_id)  # boto3 GetObject per file, into the store
+    assert path == tmp_path / "cache" / dataset_id / "1.0.0"
     assert (path / "manifest.json").exists()
     handle = registry.open_release(DatasetRef.parse(f"{dataset_id}@1.0.0"))
     assert isinstance(handle.filesystem, pafs.LocalFileSystem)  # cache-first: no S3 reads
diff --git a/packages/timenet/tests/registry/test_version.py b/packages/timenet/tests/registry/test_version.py
index d0e5c3b4b..2bfcbc81b 100644
--- a/packages/timenet/tests/registry/test_version.py
+++ b/packages/timenet/tests/registry/test_version.py
@@ -169,14 +169,6 @@ def test_preparation_needs_only_the_direct_payload_sources(registry_root, tmp_pa
     assert prepared.resolve_payload_source(SourceKey(str(parent_ref))).payload_sources == {}
 
 
-@pytest.mark.parametrize("suffix", ["different/ecg/2.0.0", "demo/ecg/3.0.0", "demo/ecg/latest"])
-def test_download_rejects_a_destination_for_another_release(registry_root, tmp_path, suffix):
-    destination = tmp_path / "downloads" / suffix
-    with pytest.raises(TimeFValidationError, match=r"expected demo/ecg@2\.0\.0|release directory"):
-        LocalRegistry(registry_root).download_version("demo/ecg", "2.0.0", destination)
-    assert not destination.exists()
-
-
 @pytest.mark.parametrize("damage", [None, "missing", "truncated"])
 def test_a_complete_copy_has_every_file_with_its_recorded_size(registry_root, damage):
     version = open_local_version(_version_dir(registry_root))
diff --git a/packages/timenet/tests/test_cli.py b/packages/timenet/tests/test_cli.py
index 8e82e141a..991fcbede 100644
--- a/packages/timenet/tests/test_cli.py
+++ b/packages/timenet/tests/test_cli.py
@@ -4,16 +4,18 @@
 
 import click
 import pytest
+from rich.console import Console
 import typer
 from typer.testing import CliRunner
 
+from timenet import builders
 from timenet.cache import cached_datasets, raw_cache_size
 from timenet.cli.app import _enum_list, app
 from timenet.dataset import TimeFDataset
 from timenet.errors import TimeNetDatasetNotFoundError
 from timenet.reader import TimeFReader
 from timenet.registry import open_local_version
-from timenet.testing import make_dataset
+from timenet.testing import install_backends, make_dataset
 from timenet.types import Access, Domain
 from timenet.types.reference_keys import ParentAlias
 from timenet.types.references import DatasetRef
@@ -151,12 +153,44 @@ def test_info_lists_the_access_page_of_each_release_that_sets_one(tmp_path):
     assert cells[start + 3][0] == "domains"
 
 
-def test_download(registry_root, tmp_path, monkeypatch):
+def test_download_command_copies_into_the_store_and_then_has_nothing_to_do(registry_root, tmp_path, monkeypatch):
+    # The default sources end with the installed connectors. Keep the real ones out of the resolution.
+    monkeypatch.setattr(builders, "entry_points", lambda group: [])
+    # The last line names a long path, so the console must not wrap it.
+    monkeypatch.setattr(_cli_module.ui, "rich", Console(stderr=True, width=1000))
     monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home"))
-    monkeypatch.setenv("TIMENET_STORAGE", str(tmp_path / "store"))
-    result = runner.invoke(app, ["download", "timenet/hello-world", "--registry", str(registry_root)])
-    assert result.exit_code == 0, result.output
-    assert (tmp_path / "store" / "timenet/hello-world" / "1.0.0" / "manifest.json").exists()
+    monkeypatch.setenv("TIMENET_REGISTRY", str(tmp_path / "store"))
+    command = ["download", "timenet/hello-world", "--registry", str(registry_root)]
+    target = tmp_path / "store" / "timenet/hello-world" / "1.0.0"
+
+    copied = runner.invoke(app, command)
+    again = runner.invoke(app, command)
+
+    assert copied.exit_code == 0, copied.output
+    assert copied.stdout == f"{target}\n"  # only the directory, for a script to capture
+    assert (target / "manifest.json").exists()
+    successes = [line for line in click.unstyle(copied.stderr).splitlines() if line.startswith("✅")]
+    assert successes == ["✅ Copied 'timenet/hello-world' → 1.0.0"]
+    assert again.exit_code == 0, again.output
+    assert again.stdout == f"{target}\n"
+    stored = tmp_path / "store"
+    assert click.unstyle(again.stderr).splitlines() == [
+        f"✅ Nothing to do: 'timenet/hello-world' 1.0.0 is already in local registry {stored}"
+    ]
+
+
+def test_download_command_with_force_copies_a_held_release_again(registry_root, tmp_path, monkeypatch):
+    install_backends(monkeypatch)
+    monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home"))
+    monkeypatch.setenv("TIMENET_REGISTRY", str(tmp_path / "store"))
+    command = ["download", "timenet/hello-world", "--registry", str(registry_root)]
+    assert runner.invoke(app, command).exit_code == 0
+
+    forced = runner.invoke(app, [*command, "--force"])
+
+    assert forced.exit_code == 0, forced.output
+    successes = [line for line in click.unstyle(forced.stderr).splitlines() if line.startswith("✅")]
+    assert successes == ["✅ Copied 'timenet/hello-world' → 1.0.0"]
 
 
 def test_search_rejects_unknown_filter_value(registry_root):
diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py
index 36ba34bd7..7f696b85d 100644
--- a/packages/timenet/tests/test_client.py
+++ b/packages/timenet/tests/test_client.py
@@ -1,3 +1,5 @@
+from pathlib import Path
+import re
 import shutil
 from types import SimpleNamespace
 
@@ -15,17 +17,20 @@
     TimeFValidationError,
     TimeNetAccessError,
     TimeNetDatasetNotFoundError,
+    TimeNetFileError,
     TimeNetMissingParentError,
     TimeNetRegistryError,
+    TimeNetVersionConflictError,
     TimeNetVersionUnavailableError,
 )
 from timenet.format.stored_values import StoredText
 from timenet.manifest.files import FilePart
+from timenet.progress import BytesMoved, ProgressEvent, StepFinished, StepStarted
 from timenet.reader import Loading, LoadPlan, TimeFReader
 from timenet.registry import TIMENET_REGISTRY_URL, LocalRegistry, RemoteRegistry, open_local_version
 from timenet.sources import InstalledConnectors
 from timenet.testing import assert_datasets_equal, install_backends, make_dataset
-from timenet.types import Access, DatasetMetadata, License, Version
+from timenet.types import Access, Annotation, DatasetMetadata, License, Version
 from timenet.types.access import AccessRequirement
 from timenet.types.parent_dataset_card import ParentDatasetCard
 from timenet.types.reference_keys import ParentAlias
@@ -105,7 +110,7 @@ def compose(self, raw_refs, context):
         return child
 
 
-def test_resampled_child_reads_and_downloads_without_its_build_only_parent(tmp_path, monkeypatch):
+def test_resampled_child_reads_and_downloads_without_its_build_only_parent(tmp_path):
     origin = LocalRegistry(tmp_path / "origin")
     origin.put_dataset(make_dataset())
     metadata = _Resampled().metadata()
@@ -124,10 +129,9 @@ def test_resampled_child_reads_and_downloads_without_its_build_only_parent(tmp_p
     with pytest.raises(TimeNetDatasetNotFoundError, match=reference.dataset_id):
         open_parents(metadata, origin)
 
-    cache = tmp_path / "download"
-    monkeypatch.setenv("TIMENET_STORAGE", str(cache))
-    directory = TimeNet(sources=[origin]).download(metadata.dataset_id)
-    assert not (cache / reference.dataset_id).exists()
+    store = LocalRegistry(tmp_path / "store")
+    directory = TimeNet(sources=[store, origin]).download(metadata.dataset_id)
+    assert not (store.root / reference.dataset_id).exists()
     shutil.rmtree(origin.root)
     version = open_local_version(directory)
     assert version.manifest.parent_cards == manifest.parent_cards
@@ -136,6 +140,41 @@ def test_resampled_child_reads_and_downloads_without_its_build_only_parent(tmp_p
         assert reader.read().records[0].signals[0].to_arrow().equals(expected)
 
 
+def test_a_damaged_release_outside_the_store_names_no_download(tmp_path):
+    origin, _ = _child_of_the_demo(tmp_path / "origin")
+    mirror = LocalRegistry(tmp_path / "mirror")
+    shutil.copytree(origin.root, mirror.root)
+    (mirror.root / "test/child/1.0.0/control.duckdb").write_bytes(b"cut short")
+    store = LocalRegistry(tmp_path / "store")
+
+    # download() replaces only a copy in the store, so it cannot repair the mirror.
+    with pytest.raises(TimeNetFileError) as caught:
+        TimeNet(sources=[store, mirror, origin]).open("test/child")
+    assert (caught.value.release, caught.value.hint) == (DatasetRef.parse("test/child@1.0.0"), None)
+
+
+def test_a_damaged_build_in_the_store_names_the_build_that_replaces_it(tmp_path, monkeypatch):
+    calls = []
+    install_backends(monkeypatch, _Builder(calls))
+    store = LocalRegistry(tmp_path / "store")
+    # A registry in the list that lacks the release, as the hosted registry lacks a local build.
+    empty = RemoteRegistry(
+        "http://api.local",
+        transport=httpx.MockTransport(lambda request: httpx.Response(404)),
+        cache_dir=tmp_path / "cache",
+    )
+    client = TimeNet(sources=[store, empty, InstalledConnectors()])
+    client.load("timenet/hello-world@1.0.0")  # the build puts the release into the store
+    (store.root / "timenet/hello-world/1.0.0/control.duckdb").write_bytes(b"cut short")
+
+    # No registry other than the local registry provides the release, so download() cannot replace it.
+    hint = f"Build it again with `timenet-build build timenet/hello-world --out {store.root} --force`."
+    with pytest.raises(TimeNetFileError) as caught:
+        client.open("timenet/hello-world@1.0.0")
+    assert caught.value.hint == hint
+    assert len(calls) == 1  # nothing builds over the damaged copy
+
+
 @pytest.fixture(autouse=True)
 def _clean_env(monkeypatch):
     for var in ("TIMENET_HOME", "TIMENET_STORAGE", "TIMENET_CACHE", "TIMENET_REGISTRY"):
@@ -151,13 +190,6 @@ def registry_root(tmp_path):
     return tmp_path / "reg"
 
 
-@pytest.fixture
-def storage(tmp_path, monkeypatch):
-    """The directory that ``download()`` copies into."""
-    monkeypatch.setenv("TIMENET_STORAGE", str(tmp_path / "storage"))
-    return tmp_path / "storage"
-
-
 def _entry(source):
     """Describe one source of a list: a local registry by its root, another source by its text."""
     if isinstance(source, LocalRegistry):
@@ -189,6 +221,7 @@ def test_the_default_list_follows_timenet_registry(tmp_path, monkeypatch, regist
 
 
 def test_timenet_registry_env_selects_a_local_registry(tmp_path, monkeypatch):
+    monkeypatch.setattr(builders_module, "entry_points", lambda group: [])
     monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home"))
     monkeypatch.setenv("TIMENET_REGISTRY", str(tmp_path / "home" / "registry"))
     dataset = make_dataset()
@@ -201,7 +234,7 @@ def test_timenet_registry_env_selects_a_local_registry(tmp_path, monkeypatch):
         "installed connectors",
     ]
     assert {m.dataset_id for m in client.list()} == {"timenet/hello-world"}
-    assert client.download("timenet/hello-world") == tmp_path / "home" / "storage" / "timenet/hello-world" / "1.0.0"
+    assert client.download("timenet/hello-world") == tmp_path / "home" / "registry" / "timenet/hello-world" / "1.0.0"
 
 
 def test_a_client_takes_an_ordered_list_of_sources(tmp_path):
@@ -290,52 +323,279 @@ def test_get_list_and_search_ask_the_first_registry_that_is_not_the_store(regist
     assert [metadata.dataset_id for metadata in alone.list()] == ["test/other"]
 
 
-def test_download_copies_into_storage(registry_root, storage):
-    client = TimeNet(sources=[str(registry_root)])
-    version_dir = client.download("timenet/hello-world")
-    assert version_dir == storage / "timenet/hello-world" / "1.0.0"
-    assert (version_dir / "manifest.json").exists()
-    assert list(version_dir.glob("time_series/part-*.parquet"))
+def test_download_needs_a_local_registry_that_is_not_read_only(registry_root):
+    with pytest.raises(TimeFValidationError) as caught:
+        TimeNet(sources=[LocalRegistry(registry_root, read_only=True)]).download("timenet/hello-world")
+    assert str(caught.value) == (
+        "download() needs a writable local registry to copy into. "
+        "Add a LocalRegistry without read_only=True to the sources"
+    )
 
 
-def test_download_reports_progress(registry_root, storage):
-    client = TimeNet(sources=[str(registry_root)])
-    reported: list[int] = []
-    client.download("timenet/hello-world", progress_cb=reported.append)
-    expected = sum(part.size for part in client.get("timenet/hello-world").files.all_files())
-    assert reported  # the callback fired
-    assert sum(reported) == expected  # summed to the version's total byte size
+def test_download_of_a_release_that_only_a_connector_provides_builds_nothing(tmp_path, monkeypatch):
+    calls = []
+    install_backends(monkeypatch, _Builder(calls))
+    store = LocalRegistry(tmp_path / "store")
 
+    with pytest.raises(TimeNetDatasetNotFoundError) as caught:
+        TimeNet(sources=[store, InstalledConnectors()]).download("timenet/hello-world")
 
-def test_manifest_file_rejects_path_traversal():
-    with pytest.raises(ValidationError, match="stay within the dataset root"):
-        FilePart(path="../../escape.txt", checksum="sha256:" + "0" * 64, size=0)
+    assert str(caught.value) == (
+        "timenet/hello-world@1.0.0 is not built yet: only an installed connector provides it. "
+        "open() or load() builds it into the local registry."
+    )
+    assert calls == []
+    assert not store.root.exists()
 
 
-def test_download_is_idempotent(registry_root, storage):
-    client = TimeNet(sources=[str(registry_root)])
+def test_download_of_latest_fails_when_a_connector_declares_a_newer_version(registry_root, tmp_path, monkeypatch):
+    calls = []
+    monkeypatch.setattr(_Builder, "VERSION", "2.0.0")
+    install_backends(monkeypatch, _Builder(calls))
+    store = LocalRegistry(tmp_path / "store")
+    client = TimeNet(sources=[store, str(registry_root), InstalledConnectors()])
+
+    # "Latest" is the version of the connector, which no registry holds.
+    with pytest.raises(TimeNetDatasetNotFoundError, match=r"^timenet/hello-world@2\.0\.0 is not built yet: "):
+        client.download("timenet/hello-world@latest")
+    assert calls == []
+    # Without a version, the download takes the registry, the first source that serves the dataset.
+    assert client.download("timenet/hello-world") == store.root / "timenet/hello-world/1.0.0"
+
+
+def test_download_of_a_release_that_no_source_provides_names_each_source(tmp_path, monkeypatch):
+    monkeypatch.setattr(builders_module, "entry_points", lambda group: [])
+    sources = [LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "empty"), InstalledConnectors()]
+
+    with pytest.raises(TimeNetDatasetNotFoundError) as caught:
+        TimeNet(sources=sources).download("x/y")
+
+    # download() never builds, so the hint names only a registry.
+    names = ", ".join(str(source) for source in sources)
+    assert str(caught.value) == f"x/y is in none of the sources: {names}. Add a registry that holds it to the sources."
+
+
+def test_download_of_a_version_that_the_connector_does_not_build_names_the_built_version(tmp_path, monkeypatch):
+    calls = []
+    install_backends(monkeypatch, _Builder(calls))
+    client = TimeNet(sources=[LocalRegistry(tmp_path / "store"), InstalledConnectors()])
+
+    with pytest.raises(TimeNetVersionUnavailableError, match=r"builds version 1\.0\.0, not the requested 9\.9\.9"):
+        client.download("timenet/hello-world@9.9.9")
+    assert calls == []
+
+
+def test_download_copies_each_release_from_the_first_registry_that_provides_it(tmp_path):
+    origin, reference = _child_of_the_demo(tmp_path / "origin")
+    # The first registry holds only the child, and the second only the payload source that it reads.
+    shutil.copytree(origin.root / "test/child", tmp_path / "first/test/child")
+    shutil.copytree(origin.root / reference.dataset_id, tmp_path / "second" / reference.dataset_id)
+    store = LocalRegistry(tmp_path / "store")
+
+    path = TimeNet(sources=[store, tmp_path / "first", tmp_path / "second"]).download("test/child")
+
+    assert path == store.root / "test/child/1.0.0"
+    with TimeNet(sources=[store]).open("test/child") as reader:
+        reader.verify()
+
+
+@pytest.mark.parametrize("case", ["force", "damaged"])
+def test_download_refuses_a_store_copy_with_other_files(registry_root, tmp_path, case):
+    store = LocalRegistry(tmp_path / "store")
+    other = make_dataset()
+    other.records[0].annotate(Annotation(key="build", value="another"))
+    store.put_dataset(other)
+    directory = store.root / "timenet/hello-world/1.0.0"
+    client = TimeNet(sources=[store, str(registry_root)])
+    if case == "force":
+        # A complete copy in the store returns at once, so only force=True compares its files.
+        assert client.download("timenet/hello-world") == directory
+    else:
+        (directory / "control.duckdb").write_bytes(b"cut short")
+    before = (directory / "manifest.json").read_text()
+
+    with pytest.raises(TimeNetVersionConflictError, match="new version"):
+        client.download("timenet/hello-world", force=case == "force")
+
+    assert (directory / "manifest.json").read_text() == before
+
+
+def test_download_names_the_store_when_no_other_registry_provides_its_copy(tmp_path, monkeypatch):
+    calls = []
+    install_backends(monkeypatch, _Builder(calls))
+    store = LocalRegistry(tmp_path / "store")
+    store.put_dataset(make_dataset())
+    client = TimeNet(sources=[store, InstalledConnectors()])
+
+    # The store holds the release, so the error is about its copy, not about a build.
+    message = (
+        r"^No registry other than the local registry provides timenet/hello-world@1\.0\.0, so download\(force=True\)"
+    )
+    with pytest.raises(TimeNetDatasetNotFoundError, match=message):
+        client.download("timenet/hello-world", force=True)
+    assert calls == []
+
+
+@pytest.mark.parametrize("case", ["force", "damaged"])
+def test_download_counts_a_registry_over_the_directory_of_the_store_as_the_store(tmp_path, monkeypatch, case):
+    monkeypatch.setattr(builders_module, "entry_points", lambda group: [])
+    root = tmp_path / "reg"
+    LocalRegistry(root).put_dataset(make_dataset())
+    directory = root / "timenet/hello-world/1.0.0"
+    if case == "damaged":
+        (directory / "control.duckdb").write_bytes(b"cut short")
+    before = {path: path.read_bytes() for path in directory.rglob("*") if path.is_file()}
+    # As for `TIMENET_REGISTRY=./reg timenet download timenet/hello-world --registry ./reg`: the
+    # registry after the store reads the directory of the store, so it cannot replace its copy.
+    monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home"))
+    monkeypatch.setenv("TIMENET_REGISTRY", str(root))
+    client = TimeNet(sources=default_sources(remote_registry=root))
+
+    if case == "force":
+        with pytest.raises(TimeNetDatasetNotFoundError) as caught:
+            client.download("timenet/hello-world", force=True)
+        assert str(caught.value) == (
+            "No registry other than the local registry provides timenet/hello-world@1.0.0, so download(force=True) "
+            "cannot copy it again."
+        )
+    else:
+        with pytest.raises(TimeNetFileError) as caught:
+            client.download("timenet/hello-world")
+        assert (
+            caught.value.hint
+            == "No registry other than the local registry provides it, so download() cannot replace it."
+        )
+    assert {path: path.read_bytes() for path in directory.rglob("*") if path.is_file()} == before
+
+
+def test_download_counts_a_symlink_to_the_store_directory_as_the_store(tmp_path):
+    store = LocalRegistry(tmp_path / "store")
+    store.put_dataset(make_dataset())
+    (tmp_path / "link").symlink_to(store.root, target_is_directory=True)
+    client = TimeNet(sources=[store, LocalRegistry(tmp_path / "link")])
+
+    # The second registry reads the directory of the store, so it cannot copy the release again.
+    with pytest.raises(TimeNetDatasetNotFoundError, match=r"^No registry other than the local registry provides "):
+        client.download("timenet/hello-world", force=True)
+
+
+def test_download_completes_the_payload_sources_of_a_complete_store_copy(tmp_path):
+    origin, reference = _child_of_the_demo(tmp_path / "origin")
+    store = LocalRegistry(tmp_path / "store")
+    client = TimeNet(sources=[store, origin])
+    client.download("test/child")
+    (store.root / reference.dataset_id / str(reference.version) / "control.duckdb").write_bytes(b"cut short")
+
+    # The child is complete in the store, but the payload source that it reads is not.
+    assert client.download("test/child") == store.root / "test/child/1.0.0"
+    with TimeFReader(store.open_release(reference)) as reader:
+        reader.verify()
+
+
+def _store_reading(registry: LocalRegistry, release: DatasetRef, source: DatasetRef, record: int) -> None:
+    """Store ``release``, which keeps one Record of ``source`` and adds one Record of its own values."""
+    dataset = TimeFDataset(
+        metadata=make_dataset().metadata.model_copy(
+            update={"dataset_id": release.dataset_id, "parents": {ParentAlias("source"): source}}
+        )
+    )
+    with TimeFReader(open_local_version(registry.root / source.dataset_id / str(source.version))) as reader:
+        dataset.import_from(reader, alias=ParentAlias("source"), records=[record])
+    own = make_dataset().records[0].signals[0]
+    signal = Signal.from_values(own.to_numpy(), spec=own.spec, name="own", time_axis=RegularAxis.from_rate_hz(16))
+    dataset.add_record(record=Record(sources=(Source(name="Monitor", signals=(signal,)),)))
+    registry.put_dataset(dataset)
+
+
+def test_download_copies_a_payload_source_without_the_payload_sources_of_that_source(tmp_path):
+    # The store holds READER, which reads the own Record of SOURCE. SOURCE reads DEMO. Origin holds
+    # SOURCE without DEMO, which READER does not need.
+    work = LocalRegistry(tmp_path / "work")
+    work.put_dataset(make_dataset())
+    demo = DatasetRef(dataset_id=make_dataset().metadata.dataset_id, version=make_dataset().metadata.dataset_version)
+    source, reader = DatasetRef.parse("test/source@1.0.0"), DatasetRef.parse("test/reader@1.0.0")
+    _store_reading(work, source, demo, make_dataset().records[0].id)
+    _store_reading(work, reader, source, 2)
+    assert [pinned.release for pinned in work.get_manifest(reader.dataset_id).payload_sources.values()] == [source]
+    store = LocalRegistry(tmp_path / "store")
+    origin = LocalRegistry(tmp_path / "origin")
+    for registry, release in ((store, reader), (origin, source)):
+        path = Path(release.dataset_id, str(release.version))
+        shutil.copytree(work.root / path, registry.root / path)
+
+    assert TimeNet(sources=[store, origin]).download(reader.dataset_id) == store.root / "test/reader/1.0.0"
+    assert store.exists(source.dataset_id, str(source.version))
+    assert not store.exists(demo.dataset_id, str(demo.version))
+
+
+def test_download_repairs_a_damaged_store_copy(tmp_path):
+    origin, reference = _child_of_the_demo(tmp_path / "origin")
+    store = LocalRegistry(tmp_path / "store")
+    client = TimeNet(sources=[store, origin.root])
+    client.download("test/child")
+    (store.root / reference.dataset_id / str(reference.version) / "control.duckdb").write_bytes(b"cut short")
+
+    # The read stops at the damaged payload source, and names the download that repairs it.
+    hint = f"Copy an intact release into the local registry with download('{reference}')."
+    with pytest.raises(TimeNetFileError, match=re.escape(hint) + "$"):
+        client.open("test/child")
+    client.download(str(reference))
+
+    with client.open("test/child") as reader:
+        reader.verify()
+
+
+def test_download_returns_a_complete_store_copy_at_once(registry_root, tmp_path):
+    client = TimeNet(sources=[LocalRegistry(tmp_path / "store"), str(registry_root)])
     first = client.download("timenet/hello-world")
-    second = client.download("timenet/hello-world")
-    assert first == second
+    # The store copy is complete, so the next download needs no registry.
+    shutil.rmtree(registry_root)
+    events: list[ProgressEvent] = []
+    assert client.download("timenet/hello-world", progress=events.append) == first
+    assert events == []  # nothing is copied again
 
 
-def test_force_redownload_replaces_atomically(registry_root, storage):
-    client = TimeNet(sources=[str(registry_root)])
+def test_download_with_force_copies_again(registry_root, tmp_path):
+    client = TimeNet(sources=[LocalRegistry(tmp_path / "store"), str(registry_root)])
     first = client.download("timenet/hello-world")
-    (first / "stale.txt").write_text("left over")  # a marker that a clean replace should remove
+    (first / "stale.txt").write_text("left over")  # a marker that a clean replace removes
 
     again = client.download("timenet/hello-world", force=True)
     assert again == first
     assert (again / "manifest.json").exists()
-    assert not (again / "stale.txt").exists()  # replaced wholesale, not written in place
-    assert not list(again.parent.glob("*.tmp-*"))  # staging dir cleaned up
+    assert not (again / "stale.txt").exists()  # replaced as a whole, not written in place
+    assert not list(again.parent.glob("*.tmp-*"))  # the staging directory is gone
 
 
-def test_download_leaves_sibling_staging_dirs_untouched(registry_root, storage):
-    client = TimeNet(sources=[str(registry_root)])
+def test_download_reports_its_steps_and_bytes(tmp_path):
+    origin, reference = _child_of_the_demo(tmp_path / "origin")
+    child = DatasetRef.parse("test/child@1.0.0")
+    events: list[ProgressEvent] = []
+    TimeNet(sources=[LocalRegistry(tmp_path / "store"), origin.root]).download("test/child", progress=events.append)
+
+    assert [event for event in events if not isinstance(event, BytesMoved)] == [
+        StepStarted(reference, "copy"),
+        StepFinished(reference, "copy"),
+        StepStarted(child, "copy"),
+        StepFinished(child, "copy"),
+    ]
+    for release in (reference, child):
+        done = [event.done for event in events if isinstance(event, BytesMoved) and event.subject == str(release)]
+        manifest = origin.get_manifest(release.dataset_id, str(release.version))
+        assert done[-1] == sum(part.size for part in manifest.files.all_files())
+
+
+def test_manifest_file_rejects_path_traversal():
+    with pytest.raises(ValidationError, match="stay within the dataset root"):
+        FilePart(path="../../escape.txt", checksum="sha256:" + "0" * 64, size=0)
+
+
+def test_download_leaves_sibling_staging_dirs_untouched(registry_root, tmp_path):
+    store = LocalRegistry(tmp_path / "store")
+    client = TimeNet(sources=[store, str(registry_root)])
     # A concurrent download of the same version has a live <version>.tmp-* dir. download() must not
     # delete a sibling staging dir: sweeping siblings would corrupt that other download mid-write.
-    sibling = storage / "timenet/hello-world" / "1.0.0.tmp-deadbeef"
+    sibling = store.root / "timenet/hello-world" / "1.0.0.tmp-deadbeef"
     sibling.mkdir(parents=True)
     (sibling / "inflight.parquet").write_text("partial")
 
@@ -391,9 +651,9 @@ def counting_close(reader):
     assert_datasets_equal(expected, restored)
 
 
-def test_load_reads_in_place_without_downloading(registry_root, tmp_path, storage, monkeypatch):
+def test_load_reads_in_place_without_downloading(registry_root, tmp_path, monkeypatch):
     # load() reads the release where it is, not download-then-read, so it must not write a copy into
-    # the store or into local storage the way download() does.
+    # the store the way download() does.
     calls = []
     install_backends(monkeypatch, _Builder(calls))
     store = tmp_path / "store"
@@ -401,7 +661,6 @@ def test_load_reads_in_place_without_downloading(registry_root, tmp_path, storag
     client.load("timenet/hello-world")
     assert calls == []
     assert not store.exists()  # nothing copied into the store
-    assert not (storage / "timenet/hello-world").exists()  # nothing fetched into the download cache
 
 
 def test_open_torch(registry_root, tmp_path, monkeypatch):
@@ -489,11 +748,6 @@ def recording_open(client, *args, **kwargs):
     assert str(reader.metadata.dataset_version) == "1.0.0"
 
 
-def test_version_ref_download_pins(versioned_registry, storage):
-    client = TimeNet(sources=[str(versioned_registry)])
-    assert client.download("timenet/hello-world@1.0.0") == storage / "timenet/hello-world" / "1.0.0"
-
-
 def test_version_ref_missing_pin_raises(versioned_registry):
     client = TimeNet(sources=[str(versioned_registry)])
     with pytest.raises(TimeNetDatasetNotFoundError):
@@ -590,9 +844,9 @@ def get_manifest(self, dataset_id, version=None):
 
 
 def test_load_rejects_a_credentialed_dataset_from_a_hosted_registry(tmp_path):
-    client = TimeNet(
-        sources=[_hosted(tmp_path, _card("org/gated", Access.CREDENTIALED, "https://physionet.example/dua"))]
-    )
+    hosted = _hosted(tmp_path, _card("org/gated", Access.CREDENTIALED, "https://physionet.example/dua"))
+    store = LocalRegistry(tmp_path / "store")
+    client = TimeNet(sources=[store, hosted])
     message = (
         "org/gated@1.0.0 is credentialed. Get access at https://physionet.example/dua for org/gated@1.0.0. "
         "No source serves its files. Add InstalledConnectors() to the sources, or build it locally with "
@@ -601,6 +855,7 @@ def test_load_rejects_a_credentialed_dataset_from_a_hosted_registry(tmp_path):
     with pytest.raises(TimeNetAccessError) as caught:
         client.load("org/gated")
     assert str(caught.value) == message
+    assert not store.root.exists()
 
 
 def test_the_access_hint_says_that_no_installed_connector_builds_the_release(tmp_path, monkeypatch):
diff --git a/packages/timenet/tests/test_fs.py b/packages/timenet/tests/test_fs.py
index 6fcd240b8..c381cdb9b 100644
--- a/packages/timenet/tests/test_fs.py
+++ b/packages/timenet/tests/test_fs.py
@@ -2,7 +2,24 @@
 
 import pytest
 
-from timenet._fs import publish_directory
+from timenet._fs import publish_directory, same_directory
+
+
+def test_a_symlink_and_its_target_are_the_same_directory(tmp_path):
+    (tmp_path / "store").mkdir()
+    (tmp_path / "link").symlink_to(tmp_path / "store", target_is_directory=True)
+    assert same_directory(tmp_path / "link", tmp_path / "store")
+
+
+def test_two_missing_paths_that_differ_only_in_case_are_different_directories(tmp_path):
+    assert not same_directory(tmp_path / "store", tmp_path / "Store")
+
+
+def test_a_missing_path_is_never_the_same_directory_as_an_existing_one(tmp_path):
+    (tmp_path / "store").mkdir()
+    if (tmp_path / "Store").exists():
+        pytest.skip("the temporary folder ignores case, so the case variant exists too")
+    assert not same_directory(tmp_path / "Store", tmp_path / "store")
 
 
 def _directories(tmp_path: Path) -> tuple[Path, Path]: