diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/backend.py b/packages/timenet-connectors/src/timenet_connectors/builder/backend.py
index 02b1251a7..aad953d4b 100644
--- a/packages/timenet-connectors/src/timenet_connectors/builder/backend.py
+++ b/packages/timenet-connectors/src/timenet_connectors/builder/backend.py
@@ -1,14 +1,23 @@
 """The build backend the SDK reaches through the ``timenet.builders`` entry point."""
 
 from collections.abc import Mapping
+from dataclasses import dataclass
 from pathlib import Path
 
+from timenet.builders import BuildOptions
+from timenet.config import settings
+from timenet.engine import run_pipeline
+from timenet.errors import TimeNetDatasetNotFoundError, TimeNetVersionUnavailableError
 from timenet.manifest import PayloadSource
-from timenet.registry import BaseRegistry
+from timenet.progress import BytesMoved, ProgressCallback, ShardWritten
+from timenet.registry import BaseRegistry, LocalRegistry
 from timenet.types import DatasetMetadata
 from timenet.types.references import DatasetRef
+from timenet.writer.progress import ProgressStage, WriteProgressEvent
 from timenet_connectors.builder.cli import build_one
-from timenet_connectors.discovery import connector_dir, has_connector, known_ids
+from timenet_connectors.builder.env import run_isolated_build
+from timenet_connectors.discovery import connector_dir, has_connector, known_ids, resolve
+from timenet_connectors.download import DownloadProgress, progress_sink
 
 
 class ConnectorBuilder:
@@ -101,3 +110,104 @@ def build(  # ruff: ignore[too-many-arguments, no-self-use] (protocol)
                 pinned_by=pinned_by,
             )
         )
+
+    def build_prepared(
+        self,
+        release: DatasetRef,
+        store: LocalRegistry,
+        *,
+        force: bool,
+        pinned_by: Mapping[DatasetRef, PayloadSource],
+        options: BuildOptions,
+    ) -> None:
+        """Build a release in a local registry.
+
+        The registry must already contain all parent releases and their payload sources.
+        Set ``TIMENET_ISOLATION=off`` to build in this interpreter.
+
+        Args:
+            release: The exact dataset ID and version to build.
+            store: The local registry to build into.
+            force: Rebuild even if ``store`` already contains the release.
+            pinned_by: Releases that depend on this release, with file pins that the build must preserve.
+            options: Cache and output settings, with a progress callback for builds in this interpreter.
+
+        Raises:
+            TimeNetDatasetNotFoundError: If no connector exists for the dataset.
+            TimeNetVersionUnavailableError: If the requested version differs from the connector's card.
+            TimeNetBuildError: If the isolated build fails.
+        """  # ruff: ignore[docstring-extraneous-exception] - TimeNetBuildError is raised by run_isolated_build
+        card = self.card(release.dataset_id)
+        if card is None:
+            raise TimeNetDatasetNotFoundError(f"no connector builds {release.dataset_id!r}")
+        if card.dataset_version != release.version:
+            raise TimeNetVersionUnavailableError(release, card.dataset_version)
+        if settings().isolation == "on":
+            # The parent reads only the card: the connector and its dependencies exist only in the child.
+            run_isolated_build(
+                release,
+                store.root,
+                force=force,
+                keep_cache=options.keep_cache,
+                quiet=options.quiet,
+                pinned_by=pinned_by,
+            )
+            return
+        _build_in_process(release, store.root, force=force, pinned_by=pinned_by, options=options)
+
+
+def _build_in_process(
+    release: DatasetRef,
+    root: Path,
+    *,
+    force: bool,
+    pinned_by: Mapping[DatasetRef, PayloadSource],
+    options: BuildOptions,
+) -> None:
+    """Build a release in the current interpreter.
+
+    Args:
+        release: The dataset ID and version declared in the connector's card.
+        root: The local registry directory to build into.
+        force: Rebuild even if ``root`` already contains the release.
+        pinned_by: Releases that depend on this release, with file pins that the build must preserve.
+        options: Cache settings and a progress callback.
+    """
+    connector = resolve(release.dataset_id)()
+    relay = None if options.progress is None else _ProgressRelay(release, options.progress)
+    # A connector reports its downloads through the ambient progress sink. Without a sink, a
+    # downloader skips the work that it does only to report progress.
+    with progress_sink(None if relay is None else relay.download):
+        run_pipeline(
+            connector,
+            root,
+            progress_cb=None if relay is None else relay.write,
+            force=force,
+            keep_cache=options.keep_cache,
+            pinned_by=pinned_by,
+        )
+
+
+@dataclass(frozen=True)
+class _ProgressRelay:
+    """Convert download and write updates into build progress events."""
+
+    release: DatasetRef
+    report: ProgressCallback
+
+    def download(self, event: DownloadProgress) -> None:
+        """Report downloaded bytes as a :class:`~timenet.progress.BytesMoved` event.
+
+        Args:
+            event: The download progress update.
+        """
+        self.report(BytesMoved(event.url, event.downloaded, event.total))
+
+    def write(self, event: WriteProgressEvent) -> None:
+        """Report a completed shard as a :class:`~timenet.progress.ShardWritten` event.
+
+        Args:
+            event: The writer progress update.
+        """
+        if event.stage is ProgressStage.SHARD_FINALIZED:
+            self.report(ShardWritten(self.release, event.completed))
diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py
index f026c56f6..ce1e73d0a 100644
--- a/packages/timenet-connectors/src/timenet_connectors/builder/cli.py
+++ b/packages/timenet-connectors/src/timenet_connectors/builder/cli.py
@@ -10,23 +10,12 @@
 from contextlib import contextmanager, nullcontext
 import json
 from pathlib import Path
-import sys
-
-from rich.progress import (
-    BarColumn,
-    DownloadColumn,
-    Progress,
-    TaskID,
-    TextColumn,
-    TimeRemainingColumn,
-    TransferSpeedColumn,
-)
+
 import typer
 
-from timenet.builders import BuilderBackend, declared, find_card, installed_builder
-from timenet.cache import human_bytes
+from timenet.builders import BuilderBackend, BuildOptions, declared, find_card, installed_builder
 from timenet.cli.runner import run_cli
-from timenet.cli.ui import console
+from timenet.cli.ui import console, progress_renderer
 from timenet.config import settings
 from timenet.coordinator.plan import Step, prepare_parents
 from timenet.engine import publish_pipeline, run_pipeline
@@ -38,6 +27,7 @@
     TimeNetRegistryError,
 )
 from timenet.manifest import PayloadSource
+from timenet.progress import BytesMoved
 from timenet.registry import (
     BaseRegistry,
     LocalRegistry,
@@ -50,7 +40,7 @@
 from timenet.writer.progress import ProgressStage, WriteProgressEvent
 from timenet_connectors.builder.env import run_isolated
 from timenet_connectors.discovery import known_ids, resolve
-from timenet_connectors.download import DownloadProgress, ProgressCallback, progress_sink
+from timenet_connectors.download import DownloadProgress, progress_sink
 
 
 app = typer.Typer(help="Build TimeNet datasets from connectors.", no_args_is_help=True)
@@ -186,6 +176,31 @@ def build(
     typer.echo(result)
 
 
+@app.command("build-prepared-release", hidden=True)
+def build_prepared_release_command(
+    release: str,
+    store: Path = typer.Option(..., "--store", help="The local registry to build into."),
+    force: bool = typer.Option(False, "--force", help="Build the release again, even if the local registry holds it."),
+    keep_cache: bool = typer.Option(False, "--keep-cache", help="Keep the raw download cache after building."),
+    pinned_by: Path | None = typer.Option(
+        None, "--pinned-by", help="A JSON file of the releases that read files of this release, with their pins."
+    ),
+) -> None:
+    """Build a release in a local registry.
+
+    The registry must already contain all parent releases and their payload sources.
+    Status and progress go to stderr, and nothing goes to stdout.
+    """
+    # backend.py imports this module, so a module-level import here would be circular.
+    from timenet_connectors.builder.backend import ConnectorBuilder  # ruff: ignore[import-outside-top-level]
+
+    reference = DatasetRef.parse(release)
+    pins = {} if pinned_by is None else _read_pins(pinned_by)
+    with progress_renderer() as progress:
+        options = BuildOptions(keep_cache=keep_cache, progress=progress, quiet=console.quiet)
+        ConnectorBuilder().build_prepared(reference, LocalRegistry(store), force=force, pinned_by=pins, options=options)
+
+
 def build_one(  # ruff: ignore[too-many-arguments]
     dataset_id: str,
     output: str | Path,
@@ -414,92 +429,19 @@ def _report_progress(event: WriteProgressEvent) -> None:
 
 
 @contextmanager
-def _download_progress() -> Generator[Progress | None]:
-    """Install the download-progress sink for a build and render it.
-
-    In an interactive terminal, downloads show as live progress bars, one row per file, updating in
-    parallel. When output goes through a pipe or you set ``--quiet``, this falls back to throttled
-    text lines (or silence), so redirected logs stay readable.
-
-    Yields:
-        The live progress display when rendering bars, else ``None``. This installs the sink for
-        the duration of the block.
-    """
-    if console.quiet:
-        with progress_sink(None):
-            yield None
-        return
-    if not sys.stderr.isatty():
-        with progress_sink(_text_download_reporter()):
-            yield None
-        return
-    progress = Progress(
-        TextColumn("[bold blue]{task.fields[name]}", justify="right"),
-        BarColumn(bar_width=None),
-        "[progress.percentage]{task.percentage:>3.1f}%",
-        "•",
-        DownloadColumn(),
-        "•",
-        TransferSpeedColumn(),
-        "•",
-        TimeRemainingColumn(),
-        console=console.rich,
-    )
-    with progress, progress_sink(_bar_reporter(progress)):
-        yield progress
-
-
-def _bar_reporter(progress: Progress) -> ProgressCallback:
-    """Build a download-progress sink that maps each URL to a live progress-bar task.
-
-    The first event for a URL adds a task (keyed by URL so concurrent downloads stay on their own
-    row). Later events update its position.
-
-    Args:
-        progress: The display to add and update per-file tasks on.
-
-    Returns:
-        A callback for :func:`~timenet_connectors.download.progress_sink`.
-    """
-    tasks: dict[str, TaskID] = {}
-
-    def report(event: DownloadProgress) -> None:
-        task_id = tasks.get(event.url)
-        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
-        progress.update(task_id, completed=event.downloaded, total=event.total)
-
-    return report
-
-
-_UNKNOWN_TOTAL_STEP_BYTES = 50 * 1024 * 1024  # progress cadence when the download size is unknown
-
-
-def _text_download_reporter() -> ProgressCallback:
-    """Build a download-progress sink that prints throttled status lines to the console.
-
-    Reports each file at roughly 20% steps, or every 50 MB when the total size is unknown. Keys each
-    report per URL so concurrent downloads do not interleave into a flood of lines.
+def _download_progress() -> Generator[None]:
+    """Display download progress during the ``with`` block."""
+    with progress_renderer() as render:
+        if render is None:
+            with progress_sink(None):
+                yield
+            return
 
-    Returns:
-        A callback for :func:`~timenet_connectors.download.progress_sink`.
-    """
-    last: dict[str, int] = {}
+        def report(event: DownloadProgress) -> None:
+            render(BytesMoved(event.url, event.downloaded, event.total))
 
-    def report(event: DownloadProgress) -> None:
-        # ~20% steps when the size is known, else one line every 50 MB.
-        step = event.downloaded * 5 // event.total if event.total else event.downloaded // _UNKNOWN_TOTAL_STEP_BYTES
-        if last.get(event.url) == step:
-            return
-        last[event.url] = step
-        name = event.url.rsplit("/", 1)[-1]
-        if event.total:
-            console.status("⬇️", f"{name} {event.downloaded * 100 // event.total}% ({human_bytes(event.total)})")
-        else:
-            console.status("⬇️", f"{name} {human_bytes(event.downloaded)}")
-
-    return report
+        with progress_sink(report):
+            yield
 
 
 def main() -> None:
diff --git a/packages/timenet-connectors/src/timenet_connectors/builder/env.py b/packages/timenet-connectors/src/timenet_connectors/builder/env.py
index 23c1e7390..e18936337 100644
--- a/packages/timenet-connectors/src/timenet_connectors/builder/env.py
+++ b/packages/timenet-connectors/src/timenet_connectors/builder/env.py
@@ -2,7 +2,7 @@
 
 The parent resolves the connector's ``requirements.txt`` without importing the connector (see
 :func:`timenet_connectors.discovery.requirements_for`). It layers that over a base pinned to the
-parent's own ``timenet`` and ``timenet-connectors``, then runs ``timenet-build build`` in that
+parent's own ``timenet`` and ``timenet-connectors``, then runs ``timenet-build`` in that
 environment. uv owns resolution and caching. A repeat build with an unchanged requirement set is a
 cache hit.
 """
@@ -108,31 +108,108 @@ def run_isolated(  # ruff: ignore[too-many-arguments]
     quiet: bool = False,
     pinned_by: Mapping[DatasetRef, PayloadSource] | None = None,
 ) -> str:
-    """Build a dataset in a separate environment with its connector's dependencies.
+    """Build a dataset in a separate environment.
 
-    The build streams status and errors to stderr.
+    The build forwards stderr as it runs and includes its last lines in build errors.
 
     Args:
-        dataset_id: The dataset id.
-        out: The output directory or a ``timenet://``, ``http(s)://``, or ``s3://`` registry URL.
+        dataset_id: The dataset ID.
+        out: A local directory or a ``timenet://``, ``http(s)://``, or ``s3://`` URL.
         force: Rebuild an existing version.
         keep_cache: Keep the raw download cache after building.
-        quiet: Suppress the build's status output.
-        pinned_by: Releases that read this dataset's files, with file pins the build must match.
+        quiet: Hide the build's status output.
+        pinned_by: File pins to preserve, grouped by dependent releases, or ``None``.
 
     Returns:
-        The version directory for a local output, or the published version for a remote output.
+        The local version directory or the published version for a remote target.
 
     Raises:
         TimeNetBuildError: If the build fails or prints no result.
-        LookupError: If no connector exists for the id.
+        LookupError: If no connector exists for the dataset ID.
+    """  # ruff: ignore[docstring-extraneous-exception] (raised by _run_in_env, not directly here)
+    stdout = _run_in_env(
+        dataset_id,
+        ["build", dataset_id, "--out", str(out)],
+        force=force,
+        keep_cache=keep_cache,
+        quiet=quiet,
+        pinned_by=pinned_by,
+        subject=repr(dataset_id),
+    )
+    lines = [line.strip() for line in stdout.splitlines() if line.strip()]
+    if not lines:
+        raise TimeNetBuildError(f"building {dataset_id!r} printed no version")
+    return lines[-1]
+
+
+def run_isolated_build(  # ruff: ignore[too-many-arguments]
+    release: DatasetRef,
+    store_root: Path,
+    *,
+    force: bool,
+    keep_cache: bool,
+    quiet: bool,
+    pinned_by: Mapping[DatasetRef, PayloadSource],
+) -> None:
+    """Build a specific release into a local registry in its connector's environment.
+
+    The registry must already contain all parent releases and their payload sources.
+    Build errors include the last lines of stderr.
+
+    Args:
+        release: The exact dataset ID and version to build.
+        store_root: The local registry directory to build into.
+        force: Rebuild even if the registry already contains the release.
+        keep_cache: Keep the raw download cache after building.
+        quiet: Hide the build's status output.
+        pinned_by: Releases that depend on this release, with file pins that the build must preserve.
+
+    Raises:
+        TimeNetBuildError: If the build fails.
+        LookupError: If no connector exists for the dataset ID.
+    """  # ruff: ignore[docstring-extraneous-exception] (raised by _run_in_env, not directly here)
+    _run_in_env(
+        release.dataset_id,
+        ["build-prepared-release", str(release), "--store", str(store_root)],
+        force=force,
+        keep_cache=keep_cache,
+        quiet=quiet,
+        pinned_by=pinned_by,
+        subject=str(release),
+    )
+
+
+def _run_in_env(  # ruff: ignore[too-many-arguments]
+    dataset_id: str,
+    subcommand: Sequence[str],
+    *,
+    force: bool,
+    keep_cache: bool,
+    quiet: bool,
+    pinned_by: Mapping[DatasetRef, PayloadSource] | None,
+    subject: str,
+) -> str:
+    """Run a ``timenet-build`` subcommand in a connector's environment.
+
+    Args:
+        dataset_id: The dataset ID that selects the connector's environment.
+        subcommand: The subcommand and its arguments, before the build flags.
+        force: Add ``--force``.
+        keep_cache: Add ``--keep-cache``.
+        quiet: Add ``--quiet``.
+        pinned_by: The file pins to pass to the build, or ``None``.
+        subject: The name to include in build errors.
+
+    Returns:
+        The command's stdout.
+
+    Raises:
+        TimeNetBuildError: If the command fails.
+        LookupError: If no connector exists for the dataset ID.
     """  # ruff: ignore[docstring-extraneous-exception] (raised by env_spec, not directly here)
-    # --quiet belongs to timenet-build, not to build, so it goes before the subcommand.
-    argv = ["timenet-build", *(["--quiet"] if quiet else []), "build", dataset_id, "--out", str(out)]
-    if force:
-        argv.append("--force")
-    if keep_cache:
-        argv.append("--keep-cache")
+    # --quiet belongs to timenet-build, not to the subcommand, so it goes first.
+    argv = ["timenet-build", *(["--quiet"] if quiet else []), *subcommand]
+    argv += [*(["--force"] if force else []), *(["--keep-cache"] if keep_cache else [])]
     spec = env_spec(dataset_id)
     # TIMENET_ISOLATION=off is the recursion guard: the child is this same CLI.
     child_env = {**os.environ, "TIMENET_ISOLATION": "off"}
@@ -145,12 +222,9 @@ def run_isolated(  # ruff: ignore[too-many-arguments]
         tail = "\n".join(stderr.splitlines()[-15:]).strip()
         detail = f"\n{tail}" if tail else " The child produced no error output."
         raise TimeNetBuildError(
-            f"building {dataset_id!r} failed with exit code {returncode}. Command: {shlex.join(command)}.{detail}"
+            f"building {subject} failed with exit code {returncode}. Command: {shlex.join(command)}.{detail}"
         )
-    lines = [line.strip() for line in stdout.splitlines() if line.strip()]
-    if not lines:
-        raise TimeNetBuildError(f"building {dataset_id!r} printed no version")
-    return lines[-1]
+    return stdout
 
 
 @contextmanager
diff --git a/packages/timenet-connectors/src/timenet_connectors/download/__init__.py b/packages/timenet-connectors/src/timenet_connectors/download/__init__.py
index 33b6431ed..cecc74d03 100644
--- a/packages/timenet-connectors/src/timenet_connectors/download/__init__.py
+++ b/packages/timenet-connectors/src/timenet_connectors/download/__init__.py
@@ -7,13 +7,13 @@
 """
 
 from timenet_connectors.download.fetch import Artifact, download_files, ensure_archive, find_dir_containing
-from timenet_connectors.download.progress import DownloadProgress, ProgressCallback, progress_sink
+from timenet_connectors.download.progress import DownloadCallback, DownloadProgress, progress_sink
 
 
 __all__ = [
     "Artifact",
+    "DownloadCallback",
     "DownloadProgress",
-    "ProgressCallback",
     "download_files",
     "ensure_archive",
     "find_dir_containing",
diff --git a/packages/timenet-connectors/src/timenet_connectors/download/progress.py b/packages/timenet-connectors/src/timenet_connectors/download/progress.py
index a2798b541..df3a9e2bd 100644
--- a/packages/timenet-connectors/src/timenet_connectors/download/progress.py
+++ b/packages/timenet-connectors/src/timenet_connectors/download/progress.py
@@ -23,13 +23,14 @@ class DownloadProgress:
     total: int | None
 
 
-ProgressCallback = Callable[[DownloadProgress], None]
+DownloadCallback = Callable[[DownloadProgress], None]
+"""A callback that receives download progress events."""
 
-_sink: ContextVar[ProgressCallback | None] = ContextVar("download_progress_sink", default=None)
+_sink: ContextVar[DownloadCallback | None] = ContextVar("download_progress_sink", default=None)
 
 
 @contextmanager
-def progress_sink(callback: ProgressCallback | None) -> Generator[None]:
+def progress_sink(callback: DownloadCallback | None) -> Generator[None]:
     """Install ``callback`` as the download-progress sink for the duration of the block.
 
     Args:
@@ -56,7 +57,7 @@ def report_progress(event: DownloadProgress) -> None:
         sink(event)
 
 
-def current_sink() -> ProgressCallback | None:
+def current_sink() -> DownloadCallback | None:
     """Return the installed download-progress sink, or ``None`` if none is installed.
 
     A downloader captures this on its own thread before handing a callback to a library that reports
diff --git a/packages/timenet-connectors/src/timenet_connectors/download/s3.py b/packages/timenet-connectors/src/timenet_connectors/download/s3.py
index c23601a41..51e35892a 100644
--- a/packages/timenet-connectors/src/timenet_connectors/download/s3.py
+++ b/packages/timenet-connectors/src/timenet_connectors/download/s3.py
@@ -15,7 +15,7 @@
 from urllib.parse import urlparse
 
 from timenet.errors import TimeFFormatError, TimeFValidationError, TimeNetBuildError
-from timenet_connectors.download.progress import DownloadProgress, ProgressCallback, current_sink
+from timenet_connectors.download.progress import DownloadCallback, DownloadProgress, current_sink
 
 
 _LOG = logging.getLogger(__name__)
@@ -71,8 +71,8 @@ def download_s3_object(s3_url: str, dest: Path, *, sha256: str | None = None) ->
         sha256: An optional SHA-256 hash in hexadecimal that the downloaded file must match.
 
     Raises:
-        TimeFValidationError: If the URL uses a scheme other than ``s3://`` or lacks a bucket or key.
-        TimeFFormatError: If the downloaded file does not match the supplied SHA-256 hash.
+        TimeFValidationError: If the URL is not an ``s3://`` URL with a bucket and key.
+        TimeFFormatError: If the downloaded file does not match ``sha256``.
     """
     parsed = urlparse(s3_url)
     bucket, key = parsed.netloc, parsed.path.lstrip("/")
@@ -90,7 +90,7 @@ def download_s3_object(s3_url: str, dest: Path, *, sha256: str | None = None) ->
             total = client.head_object(Bucket=bucket, Key=key)["ContentLength"]
             transferred = 0
 
-            def _on_bytes(count: int, report: ProgressCallback = sink) -> None:
+            def _on_bytes(count: int, report: DownloadCallback = sink) -> None:
                 nonlocal transferred
                 transferred += count
                 report(DownloadProgress(s3_url, transferred, total))
diff --git a/packages/timenet-connectors/tests/test_build_backend.py b/packages/timenet-connectors/tests/test_build_backend.py
index e17d31ab8..29e13c5b0 100644
--- a/packages/timenet-connectors/tests/test_build_backend.py
+++ b/packages/timenet-connectors/tests/test_build_backend.py
@@ -4,24 +4,44 @@
 
 import pytest
 
-from timenet.builders import installed_builder
-from timenet.errors import ConnectorCardWarning, TimeNetInvalidCardError
+from timenet.builders import BuildOptions, installed_builder
+from timenet.errors import (
+    ConnectorCardWarning,
+    TimeFValidationError,
+    TimeNetInvalidCardError,
+    TimeNetVersionUnavailableError,
+)
 from timenet.manifest import Manifest, PayloadSource
 from timenet.manifest.files import FilePart
+from timenet.progress import BytesMoved, ShardWritten
 from timenet.reader import TimeFReader
-from timenet.registry import open_local_version
+from timenet.registry import LocalRegistry, open_local_version
 from timenet.types.references import DatasetRef
 from timenet.values_backends import ValuesBackend
+from timenet.writer.progress import ProgressStage, WriteProgressEvent
 from timenet_connectors import datasets as datasets_module
 from timenet_connectors.builder import backend as backend_module, cli as cli_module
 from timenet_connectors.builder.backend import ConnectorBuilder
 from timenet_connectors.discovery import known_ids
+from timenet_connectors.download.progress import DownloadProgress, current_sink, progress_sink, report_progress
+
+
+HELLO = DatasetRef.parse("timenet/hello-world@1.0.0")
 
 
 def _unexpected_isolation(*args, **kwargs):
     raise AssertionError(f"run_isolated should not have run: {args} {kwargs}")
 
 
+def _unexpected_pipeline(*args, **kwargs):
+    raise AssertionError(f"run_pipeline should not have run: {args} {kwargs}")
+
+
+def _pins():
+    pinned = FilePart(path="control.duckdb", checksum="sha256:" + "0" * 64, size=1)
+    return {DatasetRef.parse("test/reader@1.0.0"): PayloadSource(release=HELLO, files=(pinned,))}
+
+
 def test_knows_accepts_a_known_id():
     assert ConnectorBuilder().knows("timenet/hello-world")
 
@@ -140,3 +160,90 @@ def test_build_in_process_round_trips_the_connector_default_zarr_backend(tmp_pat
         dataset = reader.read()
         assert len(dataset.records) == 3
         assert len(dataset.records[0].signals[0].to_numpy()) == 16
+
+
+# Only BuildOptions.quiet silences the child. An SDK build without a progress callback still prints.
+@pytest.mark.parametrize("quiet", [False, True])
+def test_build_prepared_runs_an_isolated_child_with_the_release_and_the_pins(tmp_path, monkeypatch, quiet):
+    monkeypatch.delenv("TIMENET_ISOLATION", raising=False)
+    monkeypatch.setattr(backend_module, "run_pipeline", _unexpected_pipeline)
+    calls = []
+    monkeypatch.setattr(
+        backend_module, "run_isolated_build", lambda release, root, **options: calls.append((release, root, options))
+    )
+
+    ConnectorBuilder().build_prepared(
+        HELLO,
+        LocalRegistry(tmp_path),
+        force=True,
+        pinned_by=_pins(),
+        options=BuildOptions(keep_cache=True, quiet=quiet),
+    )
+
+    assert calls == [(HELLO, tmp_path, {"force": True, "keep_cache": True, "quiet": quiet, "pinned_by": _pins()})]
+
+
+def test_build_prepared_rejects_another_version(tmp_path, monkeypatch):
+    monkeypatch.setenv("TIMENET_ISOLATION", "on")
+    monkeypatch.setattr(backend_module, "run_isolated_build", _unexpected_isolation)
+    monkeypatch.setattr(backend_module, "run_pipeline", _unexpected_pipeline)
+    release = DatasetRef.parse("timenet/hello-world@9.9.9")
+
+    with pytest.raises(TimeNetVersionUnavailableError) as caught:
+        ConnectorBuilder().build_prepared(
+            release, LocalRegistry(tmp_path), force=False, pinned_by={}, options=BuildOptions()
+        )
+
+    # It is a TimeFValidationError too, so a handler of the old error still catches it.
+    assert isinstance(caught.value, TimeFValidationError)
+    assert (caught.value.release, str(caught.value.declared)) == (release, "1.0.0")
+
+
+def test_build_prepared_reports_downloads_and_shards(tmp_path, monkeypatch):
+    class Connector:
+        pass
+
+    monkeypatch.setenv("TIMENET_ISOLATION", "off")
+    monkeypatch.setattr(backend_module, "resolve", lambda dataset_id: Connector)
+    received = {}
+
+    def pipeline(connector, root, *, progress_cb, **options):
+        received.update(root=root, **options)
+        report_progress(DownloadProgress("https://host/records.zip", 5, 10))
+        progress_cb(WriteProgressEvent(ProgressStage.TIME_SERIES, 3, None))
+        progress_cb(WriteProgressEvent(ProgressStage.SHARD_FINALIZED, 1, None))
+        return root / "timenet/hello-world/1.0.0"
+
+    monkeypatch.setattr(backend_module, "run_pipeline", pipeline)
+    events = []
+
+    ConnectorBuilder().build_prepared(
+        HELLO,
+        LocalRegistry(tmp_path),
+        force=True,
+        pinned_by=_pins(),
+        options=BuildOptions(keep_cache=True, progress=events.append),
+    )
+
+    assert events == [BytesMoved("https://host/records.zip", 5, 10), ShardWritten(HELLO, 1)]
+    expected = {"root": tmp_path, "force": True, "keep_cache": True, "pinned_by": _pins()}
+    assert received == expected
+
+
+def test_build_prepared_without_progress_installs_no_sink(tmp_path, monkeypatch):
+    monkeypatch.setenv("TIMENET_ISOLATION", "off")
+    seen = {}
+
+    def pipeline(connector, root, *, progress_cb, **options):
+        seen.update(progress_cb=progress_cb, sink=current_sink())
+        return root / "timenet/hello-world/1.0.0"
+
+    monkeypatch.setattr(backend_module, "run_pipeline", pipeline)
+
+    # An outer sink must not reach the build either: the build installs its own, empty one.
+    with progress_sink(print):
+        ConnectorBuilder().build_prepared(
+            HELLO, LocalRegistry(tmp_path), force=False, pinned_by={}, options=BuildOptions(progress=None)
+        )
+
+    assert seen == {"progress_cb": None, "sink": None}
diff --git a/packages/timenet-connectors/tests/test_build_cli.py b/packages/timenet-connectors/tests/test_build_cli.py
index 93e4b8fc1..15cfb60d2 100644
--- a/packages/timenet-connectors/tests/test_build_cli.py
+++ b/packages/timenet-connectors/tests/test_build_cli.py
@@ -1,3 +1,4 @@
+import json
 import os
 from pathlib import Path
 import subprocess
@@ -6,11 +7,18 @@
 import pytest
 from typer.testing import CliRunner
 
-from timenet.errors import TimeNetVersionUnavailableError
-from timenet.manifest import Manifest
+from timenet.connectors import BaseConnector
+from timenet.errors import TimeNetDatasetNotFoundError, TimeNetVersionUnavailableError
+from timenet.manifest import Manifest, PayloadSource
+from timenet.manifest.files import FilePart
 from timenet.reader import TimeFReader
 from timenet.registry import open_local_version
-from timenet_connectors.builder import cli as cli_module
+from timenet.testing import make_dataset
+from timenet.types import DatasetMetadata, Version
+from timenet.types.reference_keys import ParentAlias
+from timenet.types.references import DatasetRef
+from timenet_connectors.builder import backend as backend_module, cli as cli_module
+from timenet_connectors.builder.backend import ConnectorBuilder
 from timenet_connectors.builder.cli import app
 
 
@@ -183,3 +191,98 @@ def test_isolated_build_prints_the_version_dir_and_narrates_once(tmp_path):
     # The isolated child is the sole narrator: each status line appears exactly once.
     assert result.stderr.count("🔧 Building 'timenet/hello-world'…") == 1
     assert result.stderr.count("✅ Built 'timenet/hello-world' → 1.0.0") == 1
+
+
+_PARENT = DatasetRef.parse("test/parent@1.0.0")
+
+
+def _demo_card(dataset_id: str, **update) -> DatasetMetadata:
+    """Return the card of the demo dataset under ``dataset_id``, at version 1.0.0."""
+    return make_dataset().metadata.model_copy(
+        update={"dataset_id": dataset_id, "dataset_version": Version(1, 0, 0), **update}
+    )
+
+
+class _Parent(BaseConnector[str]):
+    """Builds the demo dataset as test/parent."""
+
+    def metadata(self):
+        return _demo_card(_PARENT.dataset_id)
+
+    def download(self, cache_dir):
+        return []
+
+    def convert(self, raw_refs):
+        dataset = make_dataset()
+        dataset._metadata = self.metadata()
+        return dataset
+
+
+class _Child(_Parent):
+    """Builds the demo dataset as test/child, which declares test/parent as its parent."""
+
+    def metadata(self):
+        return _demo_card("test/child", parents={ParentAlias("parent"): _PARENT})
+
+    def compose(self, raw_refs, context):
+        return self.convert(raw_refs)
+
+
+def test_the_build_prepared_release_command_does_not_plan(monkeypatch, tmp_path):
+    # The installed connectors build test/child and its parent test/parent, in this process.
+    connectors = {"test/child": _Child, _PARENT.dataset_id: _Parent}
+    monkeypatch.setattr(ConnectorBuilder, "dataset_ids", lambda self: list(connectors))
+    monkeypatch.setattr(ConnectorBuilder, "card", lambda self, dataset_id: connectors[dataset_id]().metadata())
+    monkeypatch.setattr(backend_module, "resolve", connectors.__getitem__)
+    monkeypatch.setenv("TIMENET_ISOLATION", "off")
+    monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home"))
+    monkeypatch.setattr(cli_module.console, "quiet", False)
+
+    # The store lacks the parent. A planner would build it first. The command must fail on the
+    # missing parent instead.
+    result = runner.invoke(app, ["build-prepared-release", "test/child@1.0.0", "--store", str(tmp_path / "store")])
+
+    assert result.exit_code == 1
+    assert isinstance(result.exception, TimeNetDatasetNotFoundError), result.exception
+    assert str(result.exception).startswith(f"complete release {_PARENT} is missing")
+    assert not (tmp_path / "store" / _PARENT.dataset_id).exists()
+
+
+def test_the_build_prepared_release_command_builds_a_release_without_parents(monkeypatch, tmp_path):
+    monkeypatch.setenv("TIMENET_ISOLATION", "off")
+    monkeypatch.setenv("TIMENET_HOME", str(tmp_path / "home"))
+    monkeypatch.setattr(cli_module.console, "quiet", False)
+
+    result = runner.invoke(
+        app, ["build-prepared-release", "timenet/hello-world@1.0.0", "--store", str(tmp_path / "store")]
+    )
+
+    assert result.exit_code == 0, result.output
+    assert (tmp_path / "store/timenet/hello-world/1.0.0/manifest.json").is_file()
+    assert result.stdout == ""
+
+
+@pytest.mark.parametrize("quiet", [False, True])
+def test_the_build_prepared_release_command_passes_the_pins_and_the_options(monkeypatch, tmp_path, quiet):
+    release = DatasetRef.parse("timenet/hello-world@1.0.0")
+    pinned = PayloadSource(release=release, files=(FilePart(path="a.bin", checksum="sha256:" + "0" * 64, size=1),))
+    pins_file = tmp_path / "pins.json"
+    pins_file.write_text(json.dumps({"test/reader@1.0.0": pinned.model_dump(mode="json")}), encoding="utf-8")
+    calls = []
+    monkeypatch.setattr(
+        ConnectorBuilder,
+        "build_prepared",
+        lambda self, release, store, **kwargs: calls.append((release, store, kwargs)),
+    )
+    monkeypatch.setattr(cli_module.console, "quiet", False)
+    options = ["--store", str(tmp_path), "--force", "--keep-cache", "--pinned-by", str(pins_file)]
+
+    result = runner.invoke(app, [*(["--quiet"] if quiet else []), "build-prepared-release", str(release), *options])
+
+    assert result.exit_code == 0, result.output
+    ((built, store, kwargs),) = calls
+    assert (built, store.root, kwargs["force"]) == (release, tmp_path, True)
+    assert kwargs["pinned_by"] == {DatasetRef.parse("test/reader@1.0.0"): pinned}
+    assert kwargs["options"].keep_cache is True
+    assert (kwargs["options"].progress is None) is quiet
+    assert kwargs["options"].quiet is quiet
diff --git a/packages/timenet-connectors/tests/test_build_env.py b/packages/timenet-connectors/tests/test_build_env.py
index 014ed0d69..5c0061440 100644
--- a/packages/timenet-connectors/tests/test_build_env.py
+++ b/packages/timenet-connectors/tests/test_build_env.py
@@ -7,8 +7,11 @@
 import pytest
 
 from timenet.errors import TimeNetBuildError
+from timenet.manifest import PayloadSource
+from timenet.manifest.files import FilePart
+from timenet.types.references import DatasetRef
 from timenet_connectors.builder import env as env_module
-from timenet_connectors.builder.env import EnvSpec, env_spec, run_isolated, uv_command
+from timenet_connectors.builder.env import EnvSpec, env_spec, run_isolated, run_isolated_build, uv_command
 
 
 def _spec(requirements=None, base=("--with", "timenet==0.1.0")):
@@ -228,3 +231,27 @@ def _missing():
 
     with pytest.raises(TimeNetBuildError, match="uv"):
         uv_command(_spec(), ["timenet-build"])
+
+
+def test_run_isolated_build_runs_build_release_with_the_store_and_the_pins(monkeypatch, tmp_path):
+    release = DatasetRef.parse("timenet/hello-world@1.0.0")
+    pinned = PayloadSource(release=release, files=(FilePart(path="a.bin", checksum="sha256:" + "0" * 64, size=1),))
+    captured = {}
+
+    def run(command, env):
+        # The pins file exists only while the child runs.
+        pins = json.loads(Path(command[command.index("--pinned-by") + 1]).read_text(encoding="utf-8"))
+        captured.update(argv=command[command.index("timenet-build") :], env=env, pins=pins)
+        return "", "", 0
+
+    monkeypatch.setattr(env_module, "_run_build", run)
+    pins = {DatasetRef.parse("test/reader@1.0.0"): pinned}
+
+    run_isolated_build(release, tmp_path / "store", force=True, keep_cache=True, quiet=True, pinned_by=pins)
+
+    argv = captured["argv"]
+    assert argv[:4] == ["timenet-build", "--quiet", "build-prepared-release", "timenet/hello-world@1.0.0"]
+    assert argv[argv.index("--store") + 1] == str(tmp_path / "store")
+    assert {"--force", "--keep-cache"} <= set(argv)
+    assert captured["pins"] == {"test/reader@1.0.0": pinned.model_dump(mode="json")}
+    assert captured["env"]["TIMENET_ISOLATION"] == "off"
diff --git a/packages/timenet-connectors/tests/test_build_progress.py b/packages/timenet-connectors/tests/test_build_progress.py
index 8441afa4f..5df32d217 100644
--- a/packages/timenet-connectors/tests/test_build_progress.py
+++ b/packages/timenet-connectors/tests/test_build_progress.py
@@ -1,54 +1,20 @@
 """Tests for the timenet-build download-progress rendering."""
 
+import io
 import sys
 
-from rich.progress import Progress
+from rich.console import Console
 
 from timenet.cli.ui import console
-from timenet_connectors.builder.cli import _bar_reporter, _download_progress
+from timenet_connectors.builder.cli import _download_progress
 from timenet_connectors.download.progress import DownloadProgress, report_progress
 
 
-def test_bar_reporter_tracks_one_task_per_url():
-    progress = Progress()
-    report = _bar_reporter(progress)
-    report(DownloadProgress("https://host/records.zip", 10, 100))
-    report(DownloadProgress("https://host/records.zip", 60, 100))  # same URL -> same row, advanced
-    report(DownloadProgress("https://host/labels.csv", 5, 20))  # new URL -> its own row
-
-    tasks = {task.fields["name"]: task for task in progress.tasks}
-    assert set(tasks) == {"records.zip", "labels.csv"}
-    assert (tasks["records.zip"].completed, tasks["records.zip"].total) == (60, 100)
-    assert (tasks["labels.csv"].completed, tasks["labels.csv"].total) == (5, 20)
-
-
-def test_bar_reporter_handles_unknown_total():
-    progress = Progress()
-    _bar_reporter(progress)(DownloadProgress("https://host/stream.bin", 4096, None))
-    (task,) = progress.tasks
-    assert task.total is None  # indeterminate bar, still tracked
-    assert task.completed == 4096
-
-
-def test_download_progress_is_silent_when_quiet(monkeypatch):
-    monkeypatch.setattr(console, "quiet", True)
-    with _download_progress() as progress:
-        assert progress is None  # no display installed
-
-
-def test_download_progress_falls_back_to_text_off_a_tty(monkeypatch):
+def test_download_progress_shows_the_downloads_of_a_connector(monkeypatch):
+    captured = io.StringIO()
+    monkeypatch.setattr(console, "rich", Console(file=captured, width=200))
     monkeypatch.setattr(console, "quiet", False)
     monkeypatch.setattr(sys.stderr, "isatty", lambda: False, raising=False)
-    with _download_progress() as progress:
-        assert progress is None  # throttled text lines, no bars
-
-
-def test_download_progress_renders_bars_on_a_tty(monkeypatch):
-    monkeypatch.setattr(console, "quiet", False)
-    monkeypatch.setattr(sys.stderr, "isatty", lambda: True, raising=False)
-    with _download_progress() as progress:
-        assert isinstance(progress, Progress)
+    with _download_progress():
         report_progress(DownloadProgress("https://host/a.zip", 3, 9))  # routed through the ambient sink
-        (task,) = progress.tasks
-        assert task.fields["name"] == "a.zip"
-        assert task.completed == 3
+    assert captured.getvalue().splitlines() == ["⬇️ a.zip 33% (9 B)"]
diff --git a/packages/timenet/src/timenet/cli/ui.py b/packages/timenet/src/timenet/cli/ui.py
index b6f73bb6c..f2c8b0ffa 100644
--- a/packages/timenet/src/timenet/cli/ui.py
+++ b/packages/timenet/src/timenet/cli/ui.py
@@ -9,7 +9,24 @@
 This avoids corruption of the display area.
 """
 
+from collections.abc import Callable, Generator
+from contextlib import contextmanager
+from functools import partial
+import sys
+
 from rich.console import Console
+from rich.progress import (
+    BarColumn,
+    DownloadColumn,
+    Progress,
+    TaskID,
+    TextColumn,
+    TimeRemainingColumn,
+    TransferSpeedColumn,
+)
+
+from timenet.cache import human_bytes
+from timenet.progress import BytesMoved, ProgressCallback, ProgressEvent, ShardWritten, StepStarted
 
 
 class _Console:
@@ -42,3 +59,115 @@ def _emit(self, emoji: str, message: str, *, style: str | None = None, force: bo
 
 
 console = _Console()
+
+
+@contextmanager
+def progress_renderer() -> Generator[ProgressCallback | None]:
+    """Show progress events on the console for the block.
+
+    Transfers appear as bars in a terminal or periodic status lines elsewhere.
+    Steps and completed shards appear as status lines.
+
+    Yields:
+        The callback for the events, or ``None`` if the console is quiet.
+    """
+    if console.quiet:
+        yield None
+        return
+    if not sys.stderr.isatty():
+        yield partial(_render, _text_download_reporter())
+        return
+    with _download_bars() as progress:
+        yield partial(_render, _bar_reporter(progress))
+
+
+_STEP_LINES = {"copy": ("📥", "Copying", "Copied"), "build": ("🔧", "Building", "Built")}
+"""The start emoji, start verb, and completion verb for each action."""
+
+
+def _render(report_download: Callable[[BytesMoved], None], event: ProgressEvent) -> None:
+    """Show one progress event on the console.
+
+    Args:
+        report_download: The callback that shows transfer progress.
+        event: The event to show.
+    """
+    if isinstance(event, BytesMoved):
+        report_download(event)
+    elif isinstance(event, ShardWritten):
+        console.status("💾", f"wrote shard {event.count}")
+    else:
+        emoji, doing, done = _STEP_LINES[event.action]
+        if isinstance(event, StepStarted):
+            console.status(emoji, f"{doing} '{event.release.dataset_id}'…")
+        else:
+            console.success(f"{done} '{event.release.dataset_id}' → {event.release.version}")
+
+
+def _download_bars() -> Progress:
+    """Create a live display of download progress bars.
+
+    Returns:
+        The progress display to activate with a ``with`` block.
+    """
+    return Progress(
+        TextColumn("[bold blue]{task.fields[name]}", justify="right"),
+        BarColumn(bar_width=None),
+        "[progress.percentage]{task.percentage:>3.1f}%",
+        "•",
+        DownloadColumn(),
+        "•",
+        TransferSpeedColumn(),
+        "•",
+        TimeRemainingColumn(),
+        console=console.rich,
+    )
+
+
+def _bar_reporter(progress: Progress) -> Callable[[BytesMoved], None]:
+    """Create a callback that displays a progress bar for each transfer URL.
+
+    Args:
+        progress: The display that holds the bars.
+
+    Returns:
+        A callback that adds or updates a bar for each event.
+    """
+    tasks: dict[str, TaskID] = {}
+
+    def report(event: BytesMoved) -> None:
+        task_id = tasks.get(event.url)
+        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
+        progress.update(task_id, completed=event.done, total=event.total)
+
+    return report
+
+
+_UNKNOWN_TOTAL_STEP_BYTES = 50 * 1024 * 1024  # progress cadence when the download size is unknown
+
+
+def _text_download_reporter() -> Callable[[BytesMoved], None]:
+    """Build a callback that shows throttled status lines for each transfer.
+
+    It shows each transfer at about every 20%, or every 50 MB when the size is unknown.
+
+    Returns:
+        The callback.
+    """
+    last: dict[str, int] = {}
+
+    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:
+            return
+        last[event.url] = step
+        name = event.url.rsplit("/", 1)[-1]
+        if event.total:
+            console.status("⬇️", f"{name} {event.done * 100 // event.total}% ({human_bytes(event.total)})")
+        else:
+            console.status("⬇️", f"{name} {human_bytes(event.done)}")
+
+    return report
diff --git a/packages/timenet/src/timenet/progress.py b/packages/timenet/src/timenet/progress.py
index 04a399754..87fb8a9b6 100644
--- a/packages/timenet/src/timenet/progress.py
+++ b/packages/timenet/src/timenet/progress.py
@@ -37,8 +37,36 @@ class StepFinished:
     action: Literal["copy", "build"]
 
 
-ProgressEvent = StepStarted | StepFinished
-"""An event that marks the start or successful end of a copy or build."""
+@dataclass(frozen=True)
+class BytesMoved:
+    """The byte count of 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.
+    """
+
+    url: str
+    done: int
+    total: int | None
+
+
+@dataclass(frozen=True)
+class ShardWritten:
+    """The number of completed shards in a release build.
+
+    Attributes:
+        release: The release being built.
+        count: The number of shards written so far.
+    """
+
+    release: DatasetRef
+    count: int
+
+
+ProgressEvent = StepStarted | StepFinished | BytesMoved | ShardWritten
+"""A step, transfer, or shard progress event."""
 
 ProgressCallback = Callable[[ProgressEvent], None]
 """A callback that receives a progress event and returns nothing."""
diff --git a/packages/timenet/tests/test_ui.py b/packages/timenet/tests/test_ui.py
index 937e2b1d4..706e9e287 100644
--- a/packages/timenet/tests/test_ui.py
+++ b/packages/timenet/tests/test_ui.py
@@ -1,6 +1,17 @@
+import io
+import sys
+
 import pytest
+from rich.console import Console
+from rich.progress import Progress
+
+from timenet.cli import ui
+from timenet.cli.ui import _bar_reporter, _Console, progress_renderer
+from timenet.progress import BytesMoved, ShardWritten, StepFinished, StepStarted
+from timenet.types.references import DatasetRef
 
-from timenet.cli.ui import _Console
+
+HELLO = DatasetRef.parse("timenet/hello-world@1.0.0")
 
 
 @pytest.fixture
@@ -8,6 +19,29 @@ def console():
     return _Console()
 
 
+@pytest.fixture
+def lines(monkeypatch):
+    """Capture the status lines of the shared console, which is not quiet."""
+    captured = io.StringIO()
+    monkeypatch.setattr(ui.console, "rich", Console(file=captured, width=200))
+    monkeypatch.setattr(ui.console, "quiet", False)
+    return captured
+
+
+@pytest.fixture
+def bars(monkeypatch):
+    """Record each live display of download bars that the renderer opens."""
+    opened = []
+    make = ui._download_bars
+
+    def record():
+        opened.append(make())
+        return opened[-1]
+
+    monkeypatch.setattr(ui, "_download_bars", record)
+    return opened
+
+
 def test_status_and_success_go_to_stderr(console, capsys):
     console.status("📥", "downloading")
     console.success("built hello_world")
@@ -28,3 +62,69 @@ def test_quiet_suppresses_status_but_not_errors(console, capsys):
     assert "done" not in captured.err
     assert "heads up" in captured.err  # warnings/errors ignore quiet
     assert "boom" in captured.err
+
+
+def test_bar_reporter_tracks_one_task_per_url():
+    progress = Progress()
+    report = _bar_reporter(progress)
+    report(BytesMoved("https://host/records.zip", 10, 100))
+    report(BytesMoved("https://host/records.zip", 60, 100))  # same URL -> same row, advanced
+    report(BytesMoved("https://host/labels.csv", 5, 20))  # new URL -> its own row
+
+    tasks = {task.fields["name"]: task for task in progress.tasks}
+    assert set(tasks) == {"records.zip", "labels.csv"}
+    assert (tasks["records.zip"].completed, tasks["records.zip"].total) == (60, 100)
+    assert (tasks["labels.csv"].completed, tasks["labels.csv"].total) == (5, 20)
+
+
+def test_bar_reporter_handles_unknown_total():
+    progress = Progress()
+    _bar_reporter(progress)(BytesMoved("https://host/stream.bin", 4096, None))
+    (task,) = progress.tasks
+    assert task.total is None  # indeterminate bar, still tracked
+    assert task.completed == 4096
+
+
+def test_the_renderer_is_silent_when_quiet(monkeypatch, bars):
+    monkeypatch.setattr(ui.console, "quiet", True)
+    with progress_renderer() as progress:
+        assert progress is None
+    assert bars == []
+
+
+def test_the_renderer_falls_back_to_text_off_a_tty(monkeypatch, lines, bars):
+    monkeypatch.setattr(sys.stderr, "isatty", lambda: False, raising=False)
+    with progress_renderer() as progress:
+        assert progress is not None
+        progress(BytesMoved("https://host/a.zip", 3, 9))
+        progress(ShardWritten(HELLO, 2))
+    assert lines.getvalue().splitlines() == ["⬇️ a.zip 33% (9 B)", "💾 wrote shard 2"]
+    assert bars == []
+
+
+def test_the_renderer_shows_downloads_as_bars_on_a_tty(monkeypatch, bars):
+    monkeypatch.setattr(ui.console, "quiet", False)
+    monkeypatch.setattr(sys.stderr, "isatty", lambda: True, raising=False)
+    with progress_renderer() as progress:
+        assert progress is not None
+        progress(BytesMoved("https://host/a.zip", 3, 9))
+        ((task,),) = [display.tasks for display in bars]
+        assert (task.fields["name"], task.completed) == ("a.zip", 3)
+
+
+def test_the_renderer_shows_copy_and_build_steps(monkeypatch, lines, bars):
+    monkeypatch.setattr(sys.stderr, "isatty", lambda: True, raising=False)
+    with progress_renderer() as progress:
+        assert progress is not None
+        for action in ("copy", "build"):
+            progress(StepStarted(HELLO, action))
+            progress(StepFinished(HELLO, action))
+
+    # The live display ends with an empty line.
+    assert [line for line in lines.getvalue().splitlines() if line] == [
+        "📥 Copying 'timenet/hello-world'…",
+        "✅ Copied 'timenet/hello-world' → 1.0.0",
+        "🔧 Building 'timenet/hello-world'…",
+        "✅ Built 'timenet/hello-world' → 1.0.0",
+    ]
+    assert len(bars) == 1