diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index 9f3ad89..f77ce6d 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -51,6 +51,7 @@ jobs:
         run: |
           python -m examples.basic_usage
           python -m examples.versioned_datasets
+          python -m examples.reference_dna_materialization
       - name: Publish coverage to Coveralls
         # Coverage upload is best-effort: parallel matrix jobs race to post to
         # the same Coveralls build, and the loser gets a "build already closed"
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f45f153..2e775ac 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,18 @@
 
 ## 1.19.0
 
+- Add `materialize` and offline `inspect_materialization` for caller-built
+  multi-file artifacts with complete raw-input/transform/output dependency
+  receipts. Preserve completed inputs and HTTP resume state across failures,
+  publish immutable output generations atomically, and discard owned inputs
+  only after successful publication unless retention is requested. Local
+  sources stay caller-owned; scientific validation stays in the builder (#90).
+  `force=True` always rebuilds, copies local sources again while the original
+  exists, and repairs corrupt private input receipts.
+- Recover the newest complete generation when a bundle, archive or
+  materialization pointer is lost. Generation names are random, so recovery
+  previously chose an arbitrary matching generation and could roll back to
+  older content; it now orders by the creation time each receipt records.
 - Add a `store_path(name, version)` callback to `VersionedDatasetRegistry` for
   exact consumer-chosen managed-store locations. It is mutually exclusive with
   root configuration and does not run at construction; the first lookup
diff --git a/README.md b/README.md
index 180c5a0..686e5d0 100644
--- a/README.md
+++ b/README.md
@@ -19,8 +19,9 @@ python -m pip install datacache
 python -m pip install "datacache[html]"
 ```
 
-tqdm is installed by default; progress displays are opt-in with
-`show_progress=True`. Normal use does not import tqdm or configure your
+tqdm is installed by default; existing download APIs enable displays with
+`show_progress=True`, while `materialize` enables them by default. Quiet use
+does not import tqdm or configure your
 application's logging. The old `datacache[progress]` install syntax remains
 supported as a compatibility alias.
 
@@ -61,6 +62,8 @@ guides; their old links land on the guide that replaced them.
 - [Progress and logging](https://github.com/openvax/datacache/blob/master/docs/progress.md)
 - <a id="versioned-datasets-and-large-downloads"></a>[Pinned bundles](https://github.com/openvax/datacache/blob/master/docs/bundles.md), [archive trees](https://github.com/openvax/datacache/blob/master/docs/archives.md) and the [legacy file registry](https://github.com/openvax/datacache/blob/master/docs/file_registry.md)
 - <a id="choose-the-right-api"></a><a id="guarantees-and-limits"></a>[Interface selection and guarantees](https://github.com/openvax/datacache/blob/master/docs/reference/choosing.md)
+- [Derived-artifact materialization](https://github.com/openvax/datacache/blob/master/docs/materialization.md): caller-owned
+  builders, atomic output generations, dependency receipts and resumable inputs.
 - [Complete API reference](https://github.com/openvax/datacache/blob/master/docs/api.md)
 - [Downstream integration](https://github.com/openvax/datacache/blob/master/docs/integration.md): contracts for consuming libraries.
 - [Release notes](https://github.com/openvax/datacache/blob/master/CHANGELOG.md) and [release procedure](https://github.com/openvax/datacache/blob/master/RELEASING.md).
diff --git a/datacache/__init__.py b/datacache/__init__.py
index 8dda36e..b967737 100644
--- a/datacache/__init__.py
+++ b/datacache/__init__.py
@@ -38,6 +38,7 @@
 from .archives import (
     ArchiveInspection, VersionedArchiveRegistry, inspect_archive, install_archive)
 from .file_registry import VersionedFileRegistry
+from .materialization import MaterializationInspection, inspect_materialization, materialize
 from .version import __version__
 
 __all__ = [
@@ -46,6 +47,9 @@
     'fetch_bytes',
     'discard_partial',
     'BundleInspection',
+    'MaterializationInspection',
+    'inspect_materialization',
+    'materialize',
     'ArchiveInspection',
     'VersionedArchiveRegistry',
     'VersionedDatasetRegistry',
diff --git a/datacache/archives.py b/datacache/archives.py
index fa383ca..506bff5 100644
--- a/datacache/archives.py
+++ b/datacache/archives.py
@@ -26,7 +26,7 @@
 from filelock import FileLock
 
 from ._filesystem import open_regular, path_present, read_json, write_json
-from .bundles import _relative_name
+from .bundles import _newest_generations_first, _relative_name
 from .inspection import FileInspection
 from .integrity import FileValidationError, _validate_expectations
 from .provenance import redact_url
@@ -419,7 +419,8 @@ def _initialize(path):
 
 
 def _recover(path, definition):
-    for entry in sorted((path / "generations").iterdir(), reverse=True):
+    for name in _newest_generations_first(path, MANIFEST, "fetched_at"):
+        entry = path / "generations" / name
         try:
             candidate = _inspect_generation(path, entry.name, definition)
         except (OSError, ValueError, KeyError, TypeError, RecursionError):
diff --git a/datacache/bundles.py b/datacache/bundles.py
index f7d7741..7183ff4 100644
--- a/datacache/bundles.py
+++ b/datacache/bundles.py
@@ -88,10 +88,10 @@ def _directory(path):
         raise FileValidationError(path, 'expected a directory, not a link')
 
 
-def _store(path):
+def _store(path, *, marker=STORE):
     _directory(path)
-    if read_json(path / STORE) != {'format': FORMAT}:
-        raise FileValidationError(path, 'unrecognized bundle store')
+    if read_json(path / marker) != {'format': FORMAT}:
+        raise FileValidationError(path, 'unrecognized generation store')
     _directory(path / 'generations')
 
 
@@ -215,29 +215,29 @@ def _paths(inspection):
     return {name: value.path for name, value in inspection.files.items()}
 
 
-def _refuse_foreign(path):
-    """Raise if path is a populated directory that isn't a bundle store, which
-    neither installation nor force=True ever takes over."""
+def _refuse_foreign(path, *, marker=STORE):
+    """Raise if path is a populated directory that isn't a generation store,
+    which neither installation nor force=True ever takes over."""
     if (path_present(path) and stat.S_ISDIR(path.lstat().st_mode)
-            and not path_present(path / STORE) and any(path.iterdir())):
+            and not path_present(path / marker) and any(path.iterdir())):
         raise FileValidationError(
-            path, 'not a datacache bundle store; a populated directory is never taken over')
+            path, 'not a datacache generation store; a populated directory is never taken over')
 
 
-def _initialize(path):
+def _initialize(path, *, marker=STORE):
     existing_mode = None
     if path_present(path):
         _directory(path)
         if any(path.iterdir()):
-            _refuse_foreign(path)
-            _store(path)  # Never take over an arbitrary nonempty directory.
+            _refuse_foreign(path, marker=marker)
+            _store(path, marker=marker)  # Never adopt a foreign directory.
             return
         existing_mode = stat.S_IMODE(path.lstat().st_mode)
     # Publish a complete store skeleton at once: concurrent first-time readers
     # see absence or a recognized store, never a half-written ownership marker.
     staging = Path(tempfile.mkdtemp(prefix='.datacache-store-', dir=path.parent))
     try:
-        write_json(staging / STORE, {'format': FORMAT}, mode=_file_mode(path.parent))
+        write_json(staging / marker, {'format': FORMAT}, mode=_file_mode(path.parent))
         (staging / 'generations').mkdir()
         os.chmod(staging, existing_mode if existing_mode is not None else
                  stat.S_IMODE((staging / 'generations').stat().st_mode))
@@ -252,9 +252,28 @@ def _file_mode(path):
     return _normal_creation_mode(path)
 
 
+def _newest_generations_first(store, manifest, time_key):
+    """Generation names, newest first by the UTC time their receipt records
+    under time_key, else by directory modification time. Names are random
+    UUIDs, so their own order says nothing about age."""
+    def created(entry):
+        try:
+            recorded = datetime.fromisoformat(read_json(entry / manifest, limit=None)[time_key])
+            if recorded.tzinfo is None:
+                recorded = recorded.replace(tzinfo=timezone.utc)
+            return recorded.timestamp()
+        except (OSError, ValueError, KeyError, TypeError, RecursionError, OverflowError):
+            try:
+                return entry.lstat().st_mtime
+            except OSError:
+                return float('-inf')
+    entries = list((store / 'generations').iterdir())
+    return [entry.name for entry in sorted(entries, key=lambda e: (created(e), e.name), reverse=True)]
+
+
 def _recover(path, assets):
-    generations = path / 'generations'
-    for entry in sorted(generations.iterdir(), reverse=True):
+    for name in _newest_generations_first(path, MANIFEST, 'fetched_at'):
+        entry = path / 'generations' / name
         try:
             candidate = _inspect_generation(path, entry.name, assets)
         except (OSError, ValueError, KeyError, TypeError, RecursionError):
diff --git a/datacache/materialization.py b/datacache/materialization.py
new file mode 100644
index 0000000..77db025
--- /dev/null
+++ b/datacache/materialization.py
@@ -0,0 +1,551 @@
+"""Transactional caller-built artifacts with durable raw-input dependencies."""
+
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+import hashlib
+import json
+import logging
+import os
+from pathlib import Path
+import shutil
+import stat
+from urllib.parse import urlsplit
+from urllib.request import url2pathname
+from uuid import uuid4
+
+from ._filesystem import file_lock, open_regular, path_present, read_json, write_json
+from .bundles import (
+    _directory, _file_mode, _generation, _initialize, _newest_generations_first, _relative_name, _store,
+)
+from .inspection import FileInspection
+from .integrity import FileValidationError, _validate_expectations
+from .progress import Progress
+from . import provenance
+
+STORE = '.datacache-materialization.json'
+MANIFEST = '.datacache-manifest.json'
+CURRENT = 'current.json'
+INPUTS = '.datacache-inputs.json'
+FORMAT = 1
+CHUNK_SIZE = 2 ** 20
+
+logger = logging.getLogger(__name__)
+
+
+def _json_copy(value):
+    # Reject non-JSON identities and NaNs, and detach from caller mutation.
+    return json.loads(json.dumps(value, sort_keys=True, allow_nan=False))
+
+
+def _fingerprint(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True, allow_nan=False).encode()).hexdigest()
+
+
+def _inventory(mapping):
+    if not isinstance(mapping, dict) or not mapping:
+        raise ValueError('sources and outputs must be nonempty mappings')
+    for name in mapping:
+        _relative_name(name)
+    folded = {name.casefold() for name in mapping}
+    parents = {'/'.join(name.split('/')[:i]).casefold()
+               for name in mapping for i in range(1, len(name.split('/')))}
+    if len(folded) != len(mapping) or folded & parents:
+        raise ValueError('paths collide as files/directories or ignoring case')
+    return mapping
+
+
+def _expectations(spec):
+    if not isinstance(spec, dict) or set(spec) - {'sha256', 'size'}:
+        raise ValueError('output expectations must contain only sha256 and size')
+    digest, size = spec.get('sha256'), spec.get('size')
+    _validate_expectations(digest, size)
+    return dict(sha256=digest.lower() if digest else None, size=size)
+
+
+def _transform(value):
+    if (not isinstance(value, dict) or set(value) - {'version', 'options'} or
+            not isinstance(value.get('version'), str) or not value['version']):
+        raise ValueError('transform requires a nonempty version and optional JSON options')
+    return dict(version=value['version'], options=_json_copy(value.get('options', {})))
+
+
+def _definition(sources, transform, outputs):
+    normalized, acquisition = {}, {}
+    for name, spec in _inventory(sources).items():
+        if not isinstance(spec, dict) or set(spec) - {'url', 'path', 'sha256', 'size'}:
+            raise ValueError('source metadata supports url or path, sha256 and size')
+        if ('url' in spec) == ('path' in spec):
+            raise ValueError('each source requires exactly one of url or path')
+        expected = _expectations({key: spec[key] for key in ('sha256', 'size') if key in spec})
+        if 'path' in spec:
+            source = Path(spec['path']).absolute()
+            origin = source.as_uri()
+            acquisition[name] = ('path', str(source))
+        else:
+            origin = spec['url']
+            if not isinstance(origin, str) or not origin:
+                raise ValueError('source URL must be a nonempty string')
+            parts = urlsplit(origin)
+            if parts.scheme not in ('http', 'https', 'ftp', 'file'):
+                raise ValueError('source URL must use HTTP(S), FTP or file')
+            if parts.scheme == 'file':
+                if parts.netloc not in ('', 'localhost') or parts.query or parts.fragment:
+                    raise ValueError('local file URLs cannot have remote hosts, queries or fragments')
+                acquisition[name] = ('path', str(Path(url2pathname(parts.path)).absolute()))
+            else:
+                acquisition[name] = ('url', origin)
+        normalized[name] = dict(expected, identity=_fingerprint(origin), origin=provenance.redact_url(origin))
+    sidecars = {Path(provenance.sidecar_path(name)).as_posix().casefold() for name in normalized}
+    occupied = {name.casefold() for name in normalized}
+    parents = {'/'.join(name.split('/')[:i]).casefold() for name in normalized
+               for i in range(1, len(name.split('/')))}
+    if sidecars & (occupied | parents):
+        raise ValueError('source paths collide with automatic provenance sidecars')
+    output_specs = {name: _expectations(spec) for name, spec in _inventory(outputs).items()}
+    return dict(sources=normalized, transform=_transform(transform), outputs=output_specs), acquisition
+
+
+def _recorded_definition(value):
+    if not isinstance(value, dict) or set(value) != {'sources', 'transform', 'outputs'}:
+        raise ValueError('invalid materialization dependency definition')
+    for spec in _inventory(value['sources']).values():
+        if not isinstance(spec, dict) or set(spec) != {'origin', 'identity', 'sha256', 'size'}:
+            raise ValueError('invalid source dependency definition')
+        _expectations({key: spec[key] for key in ('sha256', 'size')})
+        _validate_expectations(spec['identity'], None)
+        if spec['identity'] is None or not isinstance(spec['origin'], str):
+            raise ValueError('source identity and origin are required')
+    for spec in _inventory(value['outputs']).values():
+        _expectations(spec)
+    if _transform(value['transform']) != value['transform']:
+        raise ValueError('invalid transform definition')
+    return value
+
+
+def _check_observed(path, observed, expected):
+    if not isinstance(observed, dict) or set(observed) != {'sha256', 'size'}:
+        raise ValueError('invalid observed file metadata')
+    _validate_expectations(observed['sha256'], observed['size'])
+    if observed['sha256'] is None or observed['size'] is None:
+        raise ValueError('observed hashes and sizes are required')
+    for key in ('sha256', 'size'):
+        if expected[key] is not None and observed[key] != expected[key]:
+            raise FileValidationError(path, '%s disagrees with dependency expectations' % key)
+
+
+def _observe(path, expected=None, *, verify_files=True, show_progress=False):
+    with os.fdopen(open_regular(path), 'rb') as handle:
+        info = os.fstat(handle.fileno())
+        if verify_files:
+            digest, size = hashlib.sha256(), 0
+            with Progress(show_progress, 'Verifying', info.st_size) as progress:
+                for chunk in iter(lambda: handle.read(CHUNK_SIZE), b''):
+                    digest.update(chunk)
+                    size += len(chunk)
+                    progress(size, info.st_size)
+            if size != info.st_size or os.fstat(handle.fileno()).st_mtime_ns != info.st_mtime_ns:
+                raise FileValidationError(path, 'file changed during verification')
+            observed = dict(sha256=digest.hexdigest(), size=size)
+            if expected is not None:
+                _check_observed(path, observed, expected)
+        else:
+            observed = dict(sha256=None, size=info.st_size)
+            if expected is not None and expected['size'] is not None and info.st_size != expected['size']:
+                raise FileValidationError(path, 'file size disagrees with receipt')
+    return observed, info
+
+
+def _source_records(value, definition):
+    if not isinstance(value, dict) or set(value) != set(definition):
+        raise ValueError('source receipt inventory disagrees with definition')
+    for name, record in value.items():
+        if not isinstance(record, dict):
+            raise ValueError('invalid source receipt')
+        observed = {key: record.get(key) for key in ('sha256', 'size')}
+        _check_observed(name, observed, definition[name])
+        trusted = definition[name]['sha256'] is not None
+        if (record.get('identity') != definition[name]['identity'] or
+                record.get('origin') != definition[name]['origin'] or
+                record.get('verified') is not trusted or
+                not isinstance(record.get('fetched_at'), str)):
+            raise ValueError('source receipt identity or trust disagrees with definition')
+        transport = record.get('transport')
+        if transport is not None:
+            from .resume import _valid_validator
+            if not _valid_validator(transport):
+                raise ValueError('invalid validated transport metadata')
+    return value
+
+
+def _tree(root, recorded, trusted=None, *, verify_files=True, show_progress=False, allow_manifest=True):
+    """Check the exact inventory without following directories or special files."""
+    files, found, stack = {}, set(), [(Path(root), '')]
+    parents = {'/'.join(name.split('/')[:i]) for name in recorded
+               for i in range(1, len(name.split('/')))}
+    while stack:
+        directory, prefix = stack.pop()
+        _directory(directory)
+        with os.scandir(directory) as entries:
+            for entry in entries:
+                name = prefix + entry.name
+                if allow_manifest and not prefix and name == MANIFEST:
+                    continue
+                _relative_name(name)
+                if entry.is_dir(follow_symlinks=False):
+                    if name not in parents:
+                        raise FileValidationError(entry.path, 'undeclared output directory')
+                    stack.append((Path(entry.path), name + '/'))
+                    continue
+                if name not in recorded:
+                    raise FileValidationError(entry.path, 'undeclared output file')
+                observed, info = _observe(entry.path, recorded[name], verify_files=verify_files,
+                                          show_progress=show_progress)
+                found.add(name)
+                files[name] = FileInspection(entry.path, 'available',
+                    verified=bool(verify_files and trusted and trusted[name]['sha256']),
+                    size=info.st_size, mtime=info.st_mtime)
+    if found != set(recorded):
+        raise FileValidationError(root, 'missing declared outputs: %s' % sorted(set(recorded) - found))
+    return files
+
+
+@dataclass(frozen=True)
+class MaterializationInspection:
+    """One read-only snapshot, including source provenance and transform identity.
+
+    verified refers only to outputs matched against caller-trusted hashes now.
+    Source records distinguish acquisition-time trusted hashes from observation.
+    """
+    path: str
+    status: str
+    verified: bool = False
+    generation: object = None
+    files: dict = field(default_factory=dict)
+    sources: dict = field(default_factory=dict)
+    transform: object = None
+    error: object = None
+
+
+def _inspect_tree(store, directory, expected=None, *, verify_files=True, show_progress=False):
+    receipt = read_json(directory / MANIFEST, limit=None)
+    if not isinstance(receipt, dict) or receipt.get('format') != FORMAT:
+        raise ValueError('unrecognized materialization receipt')
+    definition = _recorded_definition(receipt.get('definition'))
+    if expected is not None and definition != expected:
+        raise FileValidationError(directory, 'dependency identity changed; use force=True to refresh')
+    sources = _source_records(receipt.get('sources'), definition['sources'])
+    outputs = receipt.get('outputs')
+    if not isinstance(outputs, dict) or set(outputs) != set(definition['outputs']):
+        raise ValueError('output receipt inventory disagrees with definition')
+    for name, observed in outputs.items():
+        _check_observed(name, observed, definition['outputs'][name])
+    trusted = expected['outputs'] if expected is not None else None
+    files = _tree(directory, outputs, trusted, verify_files=verify_files, show_progress=show_progress)
+    verified = verify_files and trusted is not None and all(spec['sha256'] for spec in trusted.values())
+    return MaterializationInspection(str(store), 'available', bool(verified), str(directory),
+                                      files, sources, definition['transform'])
+
+
+def inspect_materialization(destination, sources=None, *, transform=None, outputs=None, verify_files=True):
+    """Inspect offline without writes, locks, source access or automatic recovery.
+
+    Supply all of sources/transform/outputs to check dependency identity; omit
+    all three for receipt-only consistency checks. Metadata-only inspection
+    checks regular files and sizes, never claims verification of payload hashes.
+    """
+    if not isinstance(verify_files, bool):
+        raise ValueError('verify_files must be a boolean')
+    supplied = sum(value is not None for value in (sources, transform, outputs))
+    if supplied not in (0, 3):
+        raise ValueError('provide sources, transform and outputs together, or omit all three')
+    expected = _definition(sources, transform, outputs)[0] if supplied else None
+    return _inspect_materialization(destination, expected, verify_files=verify_files)
+
+
+def _inspect_materialization(destination, expected, *, verify_files=True):
+    path = Path(destination)
+    try:
+        if not path_present(path):
+            return MaterializationInspection(str(path), 'missing')
+        try:
+            _store(path, marker=STORE)
+        except FileNotFoundError:
+            _directory(path)
+            if not any(path.iterdir()):
+                return MaterializationInspection(str(path), 'missing')
+            _store(path, marker=STORE)
+        try:
+            pointer = read_json(path / CURRENT)
+        except FileNotFoundError:
+            status = 'recovery-required' if any((path / 'generations').iterdir()) else 'missing'
+            return MaterializationInspection(str(path), status)
+        return _inspect_tree(path, _generation(path, pointer['generation']), expected, verify_files=verify_files)
+    except PermissionError as error:
+        return MaterializationInspection(str(path), 'inaccessible', error=error)
+    except (OSError, ValueError, KeyError, TypeError, RecursionError) as error:
+        return MaterializationInspection(str(path), 'invalid', error=error)
+
+
+def _private(path):
+    if not path_present(path):
+        path.mkdir(mode=0o700)
+    info = path.lstat()
+    if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or
+            stat.S_IMODE(info.st_mode) & 0o077):
+        raise FileValidationError(path, 'working directory must be private and owner-owned')
+
+
+def _input_directory(store, definition):
+    return store / ('.inputs-%s-%d' % (_fingerprint(definition['sources']), os.getuid()))
+
+
+def _cleanup_inputs(store, definition, retain_sources):
+    directory = _input_directory(store, definition)
+    if not retain_sources and path_present(directory):
+        _private(directory)
+        shutil.rmtree(directory)
+
+
+def _acquire_inputs(directory, definition, acquisition, options, *, force=False):
+    from .download import fetch_file
+    _private(directory)
+    receipt_path = directory / INPUTS
+    try:
+        records = read_json(receipt_path, limit=None) if path_present(receipt_path) else {}
+        if not isinstance(records, dict) or set(records) - set(definition):
+            raise FileValidationError(receipt_path, 'invalid private input receipt')
+    except (ValueError, RecursionError) as error:
+        if not force:
+            raise FileValidationError(
+                receipt_path, 'invalid private input receipt; use force=True to repair') from error
+        # Unrecorded inputs are checked or acquired again below.
+        records = {}
+        write_json(receipt_path, records)
+    paths = {}
+    for name, spec in definition.items():
+        target = directory / name
+        for parent in reversed(target.relative_to(directory).parents):
+            current = directory / parent
+            if not path_present(current):
+                current.mkdir(mode=0o700)
+            _directory(current)
+        kind, source = acquisition[name]
+        if force and kind == 'path' and name in records and path_present(Path(source)):
+            # The caller may have corrected a local source in place without
+            # changing its path, so a forced refresh copies it again while the
+            # original exists; once it's gone, the retained copy still serves.
+            del records[name]
+            write_json(receipt_path, records)
+            if path_present(target):
+                os.close(open_regular(target))  # Never remove a planted link/special file.
+                target.unlink()
+            provenance.remove(target)
+        if name in records:
+            try:
+                _source_records({name: records[name]}, {name: spec})
+                _observe(target, {key: records[name][key] for key in ('sha256', 'size')})
+            except (FileNotFoundError, ValueError) as error:
+                if not force:
+                    raise FileValidationError(
+                        target, 'invalid private input; use force=True to repair') from error
+                if path_present(target):
+                    fd = open_regular(target)  # Never repair a planted link/special file.
+                    os.close(fd)
+                    target.unlink()
+                provenance.remove(target)
+                del records[name]
+                write_json(receipt_path, records)
+            else:
+                paths[name] = str(target)
+                continue
+        if path_present(target):
+            # Only installer-created, private completed bytes can exist here.
+            # Validate their type before fetch_file's ordinary cache-hit path.
+            fd = open_regular(target)
+            os.close(fd)
+            # A hash-pinned acquisition is independently reusable. A bare
+            # size-only/unverified file has no durable completion receipt yet.
+            if spec['sha256'] is None:
+                target.unlink()
+            else:
+                try:
+                    _observe(target, spec)
+                except FileValidationError:
+                    if not force:
+                        raise FileValidationError(target, 'invalid private input; use force=True to repair')
+                    target.unlink()
+                    provenance.remove(target)
+        copied = None
+        if kind == 'url':
+            fetch_file(source, destination=target, raw=True, record_provenance=True,
+                       expected_sha256=spec['sha256'], expected_size=spec['size'], **options)
+        elif not path_present(target):
+            temporary = directory / ('.copy-' + uuid4().hex)
+            try:
+                with os.fdopen(open_regular(source), 'rb') as handle, temporary.open('xb') as output:
+                    digest = hashlib.sha256()
+                    with Progress(options['show_progress'], 'Copying', os.fstat(handle.fileno()).st_size) as progress:
+                        size = 0
+                        for chunk in iter(lambda: handle.read(options.get('chunk_size', CHUNK_SIZE)), b''):
+                            output.write(chunk)
+                            digest.update(chunk)  # Hash while copying: one read.
+                            size += len(chunk)
+                            progress(size, progress.total)
+                            if options.get('progress_callback') is not None:
+                                options['progress_callback'](size, progress.total)
+                    output.flush()
+                    os.fsync(output.fileno())
+                copied = dict(sha256=digest.hexdigest(), size=size)
+                _check_observed(temporary, copied, spec)
+                os.replace(temporary, target)
+            finally:
+                temporary.unlink(missing_ok=True)
+        if copied is not None:
+            observed, info = copied, os.stat(target)
+        else:
+            observed, info = _observe(target, spec, show_progress=options['show_progress'])
+        origin_record = provenance.read(target, info) or {}
+        record = dict(observed, origin=spec['origin'], identity=spec['identity'],
+                      verified=spec['sha256'] is not None,
+                      fetched_at=origin_record.get('fetched_at', datetime.now(timezone.utc).isoformat()))
+        if origin_record.get('transport') is not None:
+            record['transport'] = origin_record['transport']
+        records[name] = record
+        write_json(receipt_path, records)
+        paths[name] = str(target)
+    return paths, records
+
+
+def materialize(destination, sources, *, transform, outputs, builder, force=False,
+                retain_sources=False, verify_files=True, download_options=None):
+    """Build all declared outputs and their receipt as one immutable generation.
+
+    builder(source_paths, output_paths) must create/close exactly the declared
+    outputs and leave inputs unchanged. Source definitions contain url or path
+    plus optional raw sha256/size; outputs map names to optional sha256/size.
+    transform is {version: nonempty string, options: opaque JSON value}.
+    Changed dependencies and corrupt generations require explicit force=True.
+    Force rebuilds outputs while reusing any complete matching private inputs.
+
+    Inputs survive failures, including without HTTP resume. resume=True enables
+    existing raw HTTP resume rules for remote sources; local inputs are copied.
+    Successful publication removes owned inputs unless retain_sources=True.
+    Caller-owned inputs and old output generations are never deleted. Cache
+    hits are offline and read-only; verify_files=False skips payload hashing
+    only on reuse. New publication and explicit recovery always check bytes.
+    Download/copy/verification progress is on by default; builder progress is
+    caller-owned. Installation requires POSIX locks and atomic local renames.
+    """
+    if not all(isinstance(value, bool) for value in (force, retain_sources, verify_files)):
+        raise ValueError('force, retain_sources and verify_files must be booleans')
+    if not callable(builder):
+        raise ValueError('builder must be callable')
+    definition, acquisition = _definition(sources, transform, outputs)
+    options = dict(download_options or {})
+    allowed = {'timeout', 'chunk_size', 'progress_callback', 'show_progress',
+               'max_retries', 'retry_backoff', 'retry_max_delay', 'resume'}
+    if set(options) - allowed:
+        raise ValueError('unsupported materialization download options')
+    options.setdefault('show_progress', True)
+    if not isinstance(options['show_progress'], bool) or not isinstance(options.get('resume', False), bool):
+        raise ValueError('show_progress and resume must be booleans')
+    from .retries import validate_retry_options
+    from .download import DEFAULT_CHUNK_SIZE, DEFAULT_MAX_RETRIES, DEFAULT_RETRY_BACKOFF, DEFAULT_RETRY_MAX_DELAY
+    validate_retry_options(options.get('max_retries', DEFAULT_MAX_RETRIES),
+                           options.get('retry_backoff', DEFAULT_RETRY_BACKOFF),
+                           options.get('retry_max_delay', DEFAULT_RETRY_MAX_DELAY))
+    chunk_size = options.get('chunk_size', DEFAULT_CHUNK_SIZE)
+    if not isinstance(chunk_size, int) or isinstance(chunk_size, bool) or chunk_size <= 0:
+        raise ValueError('chunk_size must be a positive integer')
+    if options.get('progress_callback') is not None and not callable(options['progress_callback']):
+        raise ValueError('progress_callback must be callable')
+    if options.get('resume'):
+        from .resume import validate_resume
+        for name, (kind, source) in acquisition.items():
+            if kind == 'url':
+                validate_resume(source, definition['sources'][name]['sha256'], definition['sources'][name]['size'])
+    path = Path(destination).absolute()
+
+    def inspect_current():
+        # A forced refresh rebuilds whatever is there; it only needs to know
+        # whether the store is readable, not to hash the old outputs.
+        return _inspect_materialization(path, definition, verify_files=verify_files and not force)
+
+    def check_hit(inspection):
+        if inspection.status == 'inaccessible':
+            raise inspection.error
+        if not force and inspection.status == 'invalid':
+            raise FileValidationError(path, 'invalid materialization; use force=True to refresh or repair') from inspection.error
+        return not force and inspection.status == 'available'
+
+    inspection = inspect_current()
+    if check_hit(inspection):
+        return {name: value.path for name, value in inspection.files.items()}
+    if os.name != 'posix':
+        raise NotImplementedError('Materialization requires a POSIX local filesystem')
+    path.parent.mkdir(parents=True, exist_ok=True)
+    key = hashlib.sha256(os.fsencode(path.name)).hexdigest()[:32]
+    with file_lock(path.parent / ('.datacache-materialization-lock-' + key)):
+        _initialize(path, marker=STORE)
+        inspection = inspect_current()
+        if check_hit(inspection):
+            return {name: value.path for name, value in inspection.files.items()}
+        if not force and inspection.status in ('missing', 'recovery-required'):
+            for name in _newest_generations_first(path, MANIFEST, 'created_at'):
+                try:
+                    candidate = _inspect_tree(path, _generation(path, name), definition)
+                except (OSError, ValueError, KeyError, TypeError, RecursionError):
+                    continue
+                write_json(path / CURRENT, {'generation': name}, mode=_file_mode(path))
+                _cleanup_inputs(path, definition, retain_sources)
+                return {name: value.path for name, value in candidate.files.items()}
+        inputs = _input_directory(path, definition)
+        source_paths, records = _acquire_inputs(inputs, definition['sources'], acquisition, options, force=force)
+        working = path / ('.staging-%s-%d' % (_fingerprint(definition), os.getuid()))
+        _private(working)
+        # Only this user's installer-owned failed outputs are discarded.
+        for entry in working.iterdir():
+            if entry.is_dir() and not entry.is_symlink():
+                shutil.rmtree(entry)
+            else:
+                entry.unlink()
+        staged = working / 'files'
+        staged.mkdir()
+        output_paths = {name: str(staged / name) for name in definition['outputs']}
+        for target in output_paths.values():
+            Path(target).parent.mkdir(parents=True, exist_ok=True)
+        try:
+            builder(dict(source_paths), dict(output_paths))
+            # Builder callbacks are trusted code, not a sandbox; enforce their
+            # read-only-input contract before creating a dependency receipt.
+            for name, source in source_paths.items():
+                _observe(source, {key: records[name][key] for key in ('sha256', 'size')})
+            observed = {}
+            # Reject extras/links/specials before hashing any declared output.
+            _tree(staged, definition['outputs'], verify_files=False, allow_manifest=False)
+            mode = _file_mode(staged)
+            for name, target in output_paths.items():
+                observed[name] = _observe(target, definition['outputs'][name],
+                                          show_progress=options['show_progress'])[0]
+                os.chmod(target, mode)
+            receipt = dict(format=FORMAT, created_at=datetime.now(timezone.utc).isoformat(),
+                           definition=definition, sources=records, outputs=observed)
+            write_json(staged / MANIFEST, receipt, mode=mode)
+            # The outputs were just hashed into the receipt; check its structure,
+            # inventory and sizes without hashing them again.
+            _inspect_tree(path, staged, definition, verify_files=False)
+            generation = uuid4().hex
+            final = path / 'generations' / generation
+            os.replace(staged, final)
+            # Remap only after validating the receipt in its private staging.
+            candidate = _inspect_tree(path, final, definition, verify_files=False)
+            write_json(path / CURRENT, {'generation': generation}, mode=_file_mode(path))
+            _cleanup_inputs(path, definition, retain_sources)
+            return {name: value.path for name, value in candidate.files.items()}
+        finally:
+            # Cleanup never replaces the builder's error or fails a publication.
+            try:
+                _private(working)  # Only ever remove our own private directory.
+                shutil.rmtree(working)
+            except (OSError, ValueError) as error:
+                logger.warning("Could not remove the staging directory %s: %s", working, error)
diff --git a/datacache/provenance.py b/datacache/provenance.py
index 34fa4d1..69858a2 100644
--- a/datacache/provenance.py
+++ b/datacache/provenance.py
@@ -60,22 +60,27 @@ def redact_url(url):
     return urlunsplit((parts.scheme, host, parts.path, "", ""))
 
 
-def describe(download_url, info, sha256=None):
+def describe(download_url, info, sha256=None, *, transport=None):
     """Provenance, as JSON text, for the bytes that stat info describes.
 
     info must come from the exact bytes being published (the staged file), not
     from whatever is at the destination afterwards, which a concurrent writer
     may have replaced. sha256 is recorded only when those bytes were verified
     against a trusted expectation: a digest of what a server sent proves nothing.
+    transport optionally records the validator accepted by resumable transport;
+    it is evidence of representation consistency, not a trusted content hash.
     """
-    return json.dumps({
+    record = {
         "format": FORMAT,
         "url": redact_url(download_url),
         "fetched_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
         "size": info.st_size,
         "mtime_ns": info.st_mtime_ns,
         "sha256": sha256.lower() if sha256 else None,
-    }, sort_keys=True)
+    }
+    if transport is not None:
+        record['transport'] = transport
+    return json.dumps(record, sort_keys=True)
 
 
 def remove(path):
diff --git a/datacache/resume.py b/datacache/resume.py
index 13239f4..2636f76 100644
--- a/datacache/resume.py
+++ b/datacache/resume.py
@@ -61,6 +61,17 @@ def discard_partial(destination):
             (directory / name).unlink(missing_ok=True)
 
 
+def _valid_validator(validator):
+    """Whether a recorded transport validator is a usable If-Range value:
+    {header: ETag (strong) or Last-Modified, value: one header line}."""
+    return (
+        isinstance(validator, dict) and set(validator) == {'header', 'value'}
+        and validator['header'] in ('ETag', 'Last-Modified')
+        and isinstance(validator['value'], str)
+        and '\r' not in validator['value'] and '\n' not in validator['value']
+        and (validator['header'] != 'ETag' or _strong_etag(validator['value'])))
+
+
 def _strong_etag(value):
     # RFC 9110 entity-tag syntax, without the weak W/ prefix. Embedded quotes,
     # whitespace, and control characters cannot authorize an If-Range request.
@@ -107,11 +118,7 @@ def download_resumable(download_url, destination, *, expected_sha256, expected_s
             # Unlink, never truncate an unknown path planted in the state dir.
             partial.unlink(missing_ok=True)
         validator = metadata.get('validator')
-        valid_validator = (
-                isinstance(validator, dict) and validator.get('header') in ('ETag', 'Last-Modified')
-                and isinstance(validator.get('value'), str)
-                and '\r' not in validator['value'] and '\n' not in validator['value']
-                and (validator['header'] != 'ETag' or _strong_etag(validator['value'])))
+        valid_validator = _valid_validator(validator)
         strong_validator = (valid_validator and validator['header'] == 'ETag')
         if ((validator is not None and not valid_validator)
                 or (expected_sha256 is None and not strong_validator)):
@@ -244,7 +251,8 @@ def report():
                 staged = handle.name
                 with partial.open('rb') as source:
                     download.copyfileobj(source, handle)
-            record = (provenance.describe(download_url, os.stat(staged), expected_sha256)
+            record = (provenance.describe(download_url, os.stat(staged), expected_sha256,
+                                          transport=metadata.get('validator'))
                       if record_provenance else None)
             download._publish_file(staged, destination, record)
         finally:
diff --git a/docs/api.md b/docs/api.md
index 62a778c..b908a00 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -17,6 +17,7 @@ explain the longer workflows and compatibility guarantees.
 | Downloading | [fetch_file](#fetch_file), [fetch_bytes](#fetch_bytes), [fetch_csv_dataframe](#fetch_csv_dataframe), [fetch_and_transform](#fetch_and_transform) |
 | Complete archive trees | [install_archive](#install_archive), [inspect_archive](#inspect_archive), [ArchiveInspection](#archiveinspection), [VersionedArchiveRegistry](#versionedarchiveregistry) |
 | Versioned file bundles | [install_bundle](#install_bundle), [inspect_bundle](#inspect_bundle), [BundleInspection](#bundleinspection), [VersionedDatasetRegistry](#versioneddatasetregistry), [VersionedFileRegistry](#versionedfileregistry) |
+| Derived artifacts | [materialize](#materialize), [inspect_materialization](#inspect_materialization), [MaterializationInspection](#materializationinspection) |
 | Paths and presence | [expected_path](#expected_path), [file_exists](#file_exists), [build_local_filename](#build_local_filename), [get_data_dir](#get_data_dir), [get_cache_root](#get_cache_root), [resolve_path](#resolve_path), [build_path](#build_path), [ensure_dir](#ensure_dir), [clear_cache](#clear_cache) |
 | Integrity and permissions | [validate_file](#validate_file), [inspect_file](#inspect_file), [inspect_files](#inspect_files), [make_file_readable](#make_file_readable) |
 | Results and exceptions | [FileInspection](#fileinspection), [CacheInspection](#cacheinspection), [FileValidationError](#filevalidationerror) |
@@ -1312,6 +1313,52 @@ The [bundle guide](bundles.md) includes an example and downstream migration note
 | `is_cached(name, version=None, *, verify_files=True)` | Whether inspection reports `available`; `verify_files=False` skips hashing. |
 | `status(*, verify_files=True)` | One row per pinned default: name, version, description, available_versions and inspection. `verify_files=False` skips hashing. |
 
+## materialize
+
+```text
+materialize(destination, sources, *, transform, outputs, builder, force=False,
+            retain_sources=False, verify_files=True, download_options=None)
+```
+
+Install caller-built outputs and their dependency receipt as one immutable
+generation. Sources map relative names to `{url: ...}` or `{path: ...}`, plus
+optional raw-byte `sha256`/`size`; outputs map names to optional output-byte
+`sha256`/`size`. Transform is `{version: nonempty_string, options: JSON_value}`.
+The builder receives private `(source_paths, output_paths)` mappings, must leave
+inputs unchanged and create/close exactly the declared regular output files.
+Returns name-to-string snapshot paths; builder return values are ignored.
+
+Dependency mismatch/corruption requires explicit `force=True`. Completed inputs
+survive failures; successful publication removes owned inputs unless retention
+is requested. Caller-owned local inputs and old generations are never deleted.
+Cache hits are offline/read-only. Fast reuse is opt-in; new publication/recovery
+always hash files. Download/copy/verification progress defaults on, configurable
+through download options. Installation initially requires POSIX local storage.
+See the [materialization guide](materialization.md) for receipts, resume, callback
+contracts, retention, recovery and peak/retained disk use.
+
+## inspect_materialization
+
+```text
+inspect_materialization(destination, sources=None, *, transform=None, outputs=None,
+                        verify_files=True)
+```
+
+Return `MaterializationInspection` offline without writes, locks, original
+input access or repair. Supply all of sources/transform/outputs for requested
+dependency checks, or omit all three for receipt-only consistency checks.
+`verify_files=False` validates inventory, types and sizes without payload reads
+and never claims trusted verification.
+
+## MaterializationInspection
+
+A read-only snapshot with `path`, `status`, `verified`, `generation`, `files`,
+`sources`, `transform` and `error`. Status is available, missing, invalid,
+inaccessible or recovery-required. `verified` means all output files matched
+caller-trusted hashes now. Source records separately report raw observed hashes,
+sizes, redacted origins, identity, acquisition time, acquisition-time trusted
+verification and available validated transport metadata.
+
 ## VersionedFileRegistry
 
 `VersionedFileRegistry(datasets, *, cache_dir, error_cls=RuntimeError)`
diff --git a/docs/data.md b/docs/data.md
index 26d6c91..552280e 100644
--- a/docs/data.md
+++ b/docs/data.md
@@ -150,6 +150,11 @@ rebuild a matching database; also increment `version` when its data changes.
 
 ## Custom single-file transformations
 
+For versioned dependencies, multiple outputs, resumable compressed inputs or
+automatic post-publication source cleanup, use the
+[derived-artifact materialization API](materialization.md). The existing helper
+below retains its compatible single-file behavior.
+
 ```python
 from pathlib import Path
 from datacache import fetch_and_transform
diff --git a/docs/design/materialization.md b/docs/design/materialization.md
new file mode 100644
index 0000000..9686f7d
--- /dev/null
+++ b/docs/design/materialization.md
@@ -0,0 +1,36 @@
+# Derived-artifact transaction (#90)
+
+Add `materialize` and `inspect_materialization`, leaving `fetch_and_transform`
+unchanged. A caller chooses the exact store, raw source definitions, a JSON
+transform identity (`version` plus opaque `options`), required output inventory,
+and `builder(source_paths, output_paths)`. Biological parsing stays in the
+builder. DataCache validates regular files, hashes bytes in bounded memory,
+records dependency receipts and publishes one immutable generation atomically.
+
+Use existing filesystem, transport, resume, progress and generation-store
+primitives. Materialization stores have a distinct ownership marker: neither
+bundles nor arbitrary legacy directories can be overwritten even with force.
+Source and transform identity mismatches raise until explicit `force=True`.
+Input expectations describe raw (including compressed) bytes; output
+expectations describe builder-produced bytes. Observed digests are not trust.
+
+Writers serialize per artifact. Inputs live in owner-only, source-keyed working
+directories, survive download/build/publication interruptions, and can be
+reused after a transform-only change. Builders receive only private staged
+paths and must close all handles, leave inputs unchanged, and create exactly
+the declared regular outputs. Outputs and the receipt publish together through
+one generation rename and one current-pointer replacement. Failed refreshes
+never modify the previous pointer or old generation. Missing pointers can be
+recovered explicitly from fully verified matching local generations.
+
+Default successful publication removes only the transaction's private owned
+inputs; retention is opt-in. Cache hits stay read-only and never clean up.
+Local inputs are copied and never modified/deleted. Old output generations are
+retained. Installation initially has the same POSIX-local-filesystem constraint
+as resumable bundle installation; cross-platform bundle support is separate.
+
+Test tiny gzip and two-output builds, actual HTTP range resumption, killed
+builders, completed-input retry without network, identity mismatch, source and
+output corruption, unsafe paths/files, failure at both publication boundaries,
+read-only offline reuse, per-artifact serialization and unrelated writer
+independence. Document disk use including transport's temporary resume copy.
diff --git a/docs/materialization.md b/docs/materialization.md
new file mode 100644
index 0000000..9fddf10
--- /dev/null
+++ b/docs/materialization.md
@@ -0,0 +1,181 @@
+# Derived-artifact materialization
+
+Use `materialize` when an application must download raw inputs, build one or
+more derived files, validate them, and publish them together. For example,
+PyEnsembl can keep its FASTA parsing and biological validation in a builder
+while DataCache owns reference-DNA download, retry, receipts and publication.
+Existing `fetch_and_transform` keeps its compatible single-file behavior.
+
+```python
+from datacache import materialize, inspect_materialization
+
+sources = {
+    "dna.fa.gz": {
+        "url": compressed_dna_url,
+        "sha256": compressed_dna_sha256,
+        "size": compressed_dna_size,
+    },
+}
+outputs = {"dna.fa": {}, "index.json": {}}
+transform = {"version": "dna-parser-1", "options": {"assembly": "GRCh38"}}
+
+def build(source_paths, output_paths):
+    # These are private staged paths, not the caller's source or public output.
+    # Stream gzip into output_paths["dna.fa"], validate sequence/contig contents,
+    # and create output_paths["index.json"]. Scientific policy stays here.
+    parse_and_validate_dna(source_paths["dna.fa.gz"], output_paths)
+
+paths = materialize(
+    artifact_store, sources, transform=transform, outputs=outputs, builder=build,
+    download_options={"resume": True, "timeout": 60},
+)
+dna_path = paths["dna.fa"]
+index_path = paths["index.json"]
+state = inspect_materialization(
+    artifact_store, sources, transform=transform, outputs=outputs,
+)
+```
+
+The [offline reference-DNA example](https://github.com/openvax/datacache/blob/master/examples/reference_dna_materialization.py)
+is runnable with `python -m examples.reference_dna_materialization` and uses
+tiny gzip data without any biological-library dependency.
+
+## Definitions and builder contract
+
+Choose an exact managed store path, independently of the Python package
+version. Sources are a nonempty mapping from relative names to either `url` or
+`path`, plus optional `sha256` and `size`. Local paths and `file://` URLs are
+copied into private working storage; their original bytes and permissions are
+never modified, and caller-owned inputs are never deleted. Links, hard-linked
+files, FIFOs and other nonregular local inputs are rejected without blocking.
+
+Source expectations describe the **raw input bytes**, including compression.
+Outputs are a nonempty mapping from relative names to optional `sha256` and
+`size`, describing **builder-produced bytes**. Empty `{}` expectations are
+supported; observation and a local consistency receipt do not authenticate
+these bytes. An expected size of zero supports explicitly empty inputs/outputs.
+Names cannot escape the store, collide by case or as files/directories, or
+occupy reserved metadata names. Sources also cannot collide with their
+automatic provenance-sidecar paths.
+
+`transform` contains a nonempty opaque `version` string and optional JSON
+`options`. Bump this identity when parsing, validation, index format or other
+scientific decisions change. DataCache does not infer code changes or decide
+biological compatibility. Do not put secrets in options: they are recorded.
+
+`builder(source_paths, output_paths)` receives name-to-string mappings. Output
+paths are initially absent, with required parent directories already present.
+The builder must create exactly those outputs, close all handles, and leave
+inputs unchanged before returning. Only regular, single-link files and the
+declared output parent directories are accepted. Return values are ignored;
+`materialize` returns name-to-absolute-path mappings belonging to the published generation.
+The callback is trusted application code, not a sandbox. Builder progress,
+streamed decompression and biological validation are caller-owned.
+
+## Identity, receipts and trust
+
+Each generation records the complete source/transform/output dependency
+definition and observed sizes/SHA-256 hashes. Source receipts also record
+redacted origins, acquisition time, acquisition-time trusted-hash verification,
+and validated transport metadata when available. Resumable HTTP acquisitions
+record the accepted strong ETag or Last-Modified validator. These validators
+establish transport consistency, not scientific correctness or authenticity.
+Ordinary transfers without validated transport metadata do not invent it.
+
+Full source references are represented by SHA-256 identity fingerprints.
+Display origins omit URL credentials, query text and fragments, but retain the
+path; avoid secret-bearing paths. Changing the full URL/path (including signed
+query text), integrity expectations, transform identity or output inventory
+makes an existing generation `invalid` for that request. Refresh is explicit:
+`materialize(..., force=True)` builds a replacement. This API does not claim
+different mirror URLs are the same dependency; caller-declared source identity
+policy is separate. No remote `latest` lookup occurs.
+
+`inspect_materialization` reports `available`, `missing`, `invalid`,
+`inaccessible` or `recovery-required`, with generation paths, output inspections,
+source receipts, transform identity and an error cause. Supply all three of
+`sources`, `transform`, and `outputs` to check the requested identity, or omit
+all three for receipt-only consistency checks. Inspection never accesses the
+original inputs, performs network requests, writes, locks or repairs.
+
+Full inspection and cache-hit checks hash outputs by default. `verified=True`
+means all outputs matched caller-supplied trusted hashes **now**; source receipt
+verification is explicitly historical. Without trusted output hashes, even
+fully consistent results have `verified=False`. `verify_files=False` checks
+ownership, dependency identity, exact inventory, types, readability and sizes,
+but cannot detect same-size corruption and always reports unverified outputs.
+This fast flag applies only to reuse. New output publication and explicit
+recovery always hash the bytes.
+
+## Interruption, publication and retention
+
+Writers serialize per artifact; unrelated artifacts have independent locks.
+Completed inputs persist in owner-only source-keyed directories even without
+HTTP resume. `download_options={"resume": True}` also preserves partial raw
+HTTP(S) downloads, using existing hash/size or size/strong-ETag resume rules.
+Remote resumable inputs require `size`; local inputs are simply copied.
+The supported transport options are timeout, chunk size, progress callback,
+progress display, retry count/backoff/delay and resume. Download, copy and hash
+progress is enabled by default; set `show_progress=False` in download options
+for noninteractive use. Valid cache hits remain quiet.
+
+Download, gzip, validation and builder failures leave the previously selected
+output generation untouched. Failed builder outputs are discarded; completed
+inputs survive for retry without another download. A killed builder can leave
+private partial outputs, which the next transaction with the same definition
+discards. Changing only the transform can reuse retained matching inputs.
+`force=True` always rebuilds outputs: it never selects an older generation,
+even when `current.json` is missing, and it doesn't hash the outputs it replaces.
+It reuses remote inputs that still match, copies a local source again while
+the original exists (so an in-place correction is picked up; once the original
+is gone, a retained copy still serves), and repairs corrupted private inputs or
+input receipts. It never adopts unsafe paths or foreign directories.
+
+All declared outputs and `.datacache-manifest.json` are staged and checked as
+one tree. DataCache renames that tree into `generations/<id>` and atomically
+replaces `current.json`. It does not present independent fixed-file renames as
+a multi-file transaction. Readers see the complete old or complete new tree;
+returned snapshot paths continue to exist across later refreshes. Do not add
+indexes or edit files in a published generation; declare them as outputs or
+place independently mutable files outside the managed store.
+
+If the first pointer publication is interrupted, explicit `materialize`
+without `force` can fully check and select a matching complete local
+generation without fetching inputs or calling the builder. It tries the newest
+generation first, by the creation time its receipt records. Inspection alone
+does not recover anything. A failure to remove the private staging directory is
+logged; it never replaces a builder's error or fails a published build.
+
+After successful pointer publication/recovery, owned inputs for that source
+definition are removed by default. `retain_sources=True` keeps them privately
+for future builds. Cache hits never perform cleanup: a crash between pointer
+publication and cleanup may leave inputs for an explicit later refresh. Inputs
+for other source definitions/users and old output generations are never pruned
+implicitly. Abandoned input/staging variants require deliberate operator
+cleanup; general inventory/pruning is separate work. Retention is installation
+policy, not output identity, so toggling it on a cache hit does not mutate disk.
+
+The distinct `.datacache-materialization.json` ownership marker prevents
+adoption of populated legacy directories or bundle stores, even with force.
+Empty precreated stores are allowed and keep their directory modes. As with
+current resumable bundles, installation requires a POSIX local filesystem with
+`flock` and atomic sibling renames. Read-only existing hits require no lock.
+This is atomic visibility, not a guarantee against filesystem/power-loss faults.
+
+## Disk use
+
+Let `C` be total completed raw inputs, `D` the new output generation, and `O`
+all old retained output generations. During a build, storage is approximately
+`O + C + D`, plus scratch files created by the builder. Generation publication
+is a rename and does not duplicate `D`. After successful default cleanup,
+retained storage is `O + D`; opt-in source retention uses `O + C + D`.
+
+The existing resumable transport copies a completed private partial into file
+publication staging before removing the partial. That phase can need an extra
+compressed-input-sized copy: peak is approximately
+`O + max(C + largest_input, C + D + builder_scratch)`, excluding any previously
+abandoned input/staging variants. For one 1 GB compressed / 3 GB decompressed
+human-DNA input, a first build needs roughly 4 GB plus builder scratch, then
+retains roughly 3 GB (or 4 GB with source retention). A refresh retaining an old
+3 GB output needs roughly 7 GB peak and retains 6 GB afterward. Caller-owned
+local originals occupy additional storage outside this managed store.
diff --git a/docs/reference/choosing.md b/docs/reference/choosing.md
index 41f2d62..7844b18 100644
--- a/docs/reference/choosing.md
+++ b/docs/reference/choosing.md
@@ -8,6 +8,7 @@ lists every public signature, default, return value, exception, and example.
 | Task | API | Result |
 | --- | --- | --- |
 | Install or reuse a versioned dataset | `VersionedDatasetRegistry`, `install_bundle(...)` | Mapping of asset names to snapshot paths |
+| Build derived files from versioned dependencies | `materialize(...)` | Mapping of output names to snapshot paths |
 | Install versioned archive trees | `VersionedArchiveRegistry`, `install_archive(...)` | Immutable extracted-generation `Path` |
 | Reuse an established fixed-path versioned file cache | `VersionedFileRegistry` | One Path and a legacy-compatible root receipt |
 | Inspect a complete dataset generation | `inspect_bundle(...)` | `BundleInspection` |
diff --git a/examples/reference_dna_materialization.py b/examples/reference_dna_materialization.py
new file mode 100644
index 0000000..03ff9d4
--- /dev/null
+++ b/examples/reference_dna_materialization.py
@@ -0,0 +1,55 @@
+"""Tiny offline reference-DNA transaction; scientific work stays in the builder."""
+
+import gzip
+from hashlib import sha256
+import json
+from pathlib import Path
+import shutil
+from tempfile import TemporaryDirectory
+
+from datacache import inspect_materialization, materialize
+
+
+def build_reference(source_paths, output_paths):
+    with gzip.open(source_paths['dna.fa.gz'], 'rb') as source, open(output_paths['dna.fa'], 'wb') as output:
+        shutil.copyfileobj(source, output, length=2 ** 20)
+    # A toy caller-side biological check, not a DataCache FASTA parser.
+    with open(output_paths['dna.fa'], 'rb') as fasta:
+        if fasta.readline() != b'>chr1\n':
+            raise ValueError('unexpected contig header')
+        length = sum(len(line.strip()) for line in fasta)
+    Path(output_paths['index.json']).write_text(json.dumps({'length': length}))
+
+
+def main():
+    with TemporaryDirectory() as temporary:
+        root = Path(temporary)
+        dna = b'>chr1\nACGTACGT\n'
+        compressed = gzip.compress(dna, mtime=0)
+        original = root / 'upstream.fa.gz'
+        original.write_bytes(compressed)
+        sources = {'dna.fa.gz': {'path': original, 'sha256': sha256(compressed).hexdigest(),
+                                'size': len(compressed)}}
+        outputs = {'dna.fa': {'sha256': sha256(dna).hexdigest(), 'size': len(dna)}, 'index.json': {}}
+        transform = {'version': 'toy-fasta-1', 'options': {'contig': 'chr1'}}
+        store = root / 'reference' / 'release-110'
+        paths = materialize(store, sources, transform=transform, outputs=outputs, builder=build_reference,
+                            download_options={'show_progress': False})
+        assert Path(paths['dna.fa']).read_bytes() == dna
+        assert json.loads(Path(paths['index.json']).read_text()) == {'length': 8}
+        assert not list(store.glob('.inputs-*'))
+        original.unlink()
+        # Neither the upstream source nor a builder is needed on an offline hit.
+        def forbidden(*args):
+            raise AssertionError('offline reuse attempted to rebuild')
+        assert materialize(store, sources, transform=transform, outputs=outputs,
+                           builder=forbidden, verify_files=False) == paths
+        state = inspect_materialization(store, sources, transform=transform, outputs=outputs)
+        assert state.status == 'available'
+        assert not state.verified  # Index has observed metadata, not a trusted hash.
+        assert state.sources['dna.fa.gz']['verified']
+        print('Reference DNA: atomic pair, compressed provenance, input cleanup and offline reuse pass')
+
+
+if __name__ == '__main__':
+    main()
diff --git a/mkdocs.yml b/mkdocs.yml
index fde9f1c..8a7495a 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -39,6 +39,7 @@ nav:
 - Versioned datasets:
   - Install bundles: bundles.md
   - Install archive trees: archives.md
+  - Build derived files: materialization.md
   - Use the legacy file registry: file_registry.md
   - Integrate a consuming library: integration.md
 - Reference:
diff --git a/tests/test_bundles.py b/tests/test_bundles.py
index 449ca24..46e60f6 100644
--- a/tests/test_bundles.py
+++ b/tests/test_bundles.py
@@ -6,6 +6,7 @@
 import os
 from pathlib import Path
 import threading
+import time
 
 import pytest
 
@@ -519,3 +520,25 @@ def pause_after_missing_marker(path, *args, **kwargs):
             resume_inspection.set()
         assert future.result() == winner
     assert len(list((dest / 'generations').iterdir())) == 1
+
+
+def test_recovery_picks_the_newest_generation_not_the_highest_name(tmp_path, assets):
+    dest = tmp_path / 'bundle'
+    older = Path(next(iter(install_bundle(dest, assets).values()))).parent
+    newer = Path(next(iter(install_bundle(dest, assets, force=True).values()))).parent
+    # Generation names are random: make the older one's name sort highest.
+    older.rename(dest / 'generations' / ('f' * 32))
+    newer.rename(dest / 'generations' / ('0' * 32))
+    (dest / bundles.CURRENT).unlink()
+    recovered = install_bundle(dest, assets)
+    generations = {Path(path).relative_to(dest / 'generations').parts[0] for path in recovered.values()}
+    assert generations == {'0' * 32}
+
+
+def test_generation_age_falls_back_to_modification_time(tmp_path):
+    store = tmp_path / 'store'
+    for name, age in [('a' * 32, 100), ('b' * 32, 10)]:
+        directory = store / 'generations' / name
+        directory.mkdir(parents=True)
+        os.utime(directory, (time.time() - age, time.time() - age))
+    assert bundles._newest_generations_first(store, bundles.MANIFEST, 'fetched_at') == ['b' * 32, 'a' * 32]
diff --git a/tests/test_materialization.py b/tests/test_materialization.py
new file mode 100644
index 0000000..20bed5b
--- /dev/null
+++ b/tests/test_materialization.py
@@ -0,0 +1,878 @@
+"""Caller-built generations, private retained inputs, and raw gzip dependencies."""
+
+from concurrent.futures import ThreadPoolExecutor
+import gzip
+from hashlib import sha256
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import socket
+import subprocess
+import sys
+import threading
+import time
+
+import pytest
+
+from datacache import FileValidationError, inspect_bundle, inspect_materialization, materialize
+from datacache import download, materialization as module
+from datacache.resume import _state_directory
+
+pytestmark = pytest.mark.skipif(os.name != 'posix', reason='POSIX materialization locks')
+DATA = b'>chr1\nACGTACGT\n'
+COMPRESSED = gzip.compress(DATA, mtime=0)
+TRANSFORM = {'version': 'fasta-1', 'options': {'validate': True}}
+
+
+def metadata(data):
+    return {'sha256': sha256(data).hexdigest(), 'size': len(data)}
+
+
+@pytest.fixture
+def sources(tmp_path):
+    source = tmp_path / 'caller-owned.fa.gz'
+    source.write_bytes(COMPRESSED)
+    return {'dna.fa.gz': dict(path=source, **metadata(COMPRESSED))}
+
+
+@pytest.fixture
+def outputs():
+    return {'dna.fa': metadata(DATA), 'index/info.json': metadata(b'{"length":8}\n')}
+
+
+def build(sources, outputs):
+    assert all(Path(path).parent.stat().st_mode & 0o077 == 0 for path in sources.values())
+    with gzip.open(sources['dna.fa.gz'], 'rb') as source, open(outputs['dna.fa'], 'wb') as target:
+        while chunk := source.read(4):
+            target.write(chunk)
+    Path(outputs['index/info.json']).write_bytes(b'{"length":8}\n')
+
+
+def install(store, sources, outputs, **options):
+    defaults = dict(transform=TRANSFORM, outputs=outputs, builder=build,
+                    download_options={'show_progress': False})
+    defaults.update(options)
+    return materialize(store, sources, **defaults)
+
+
+def inspected(store, sources, outputs, **options):
+    return inspect_materialization(store, sources, transform=TRANSFORM, outputs=outputs, **options)
+
+
+def input_dir(store):
+    return next(store.glob('.inputs-*'))
+
+
+def test_transaction_receipt_cleanup_and_immutable_paths(tmp_path, sources, outputs):
+    store = tmp_path / 'artifacts' / 'dna-110'
+    paths = install(store, sources, outputs)
+    assert Path(paths['dna.fa']).read_bytes() == DATA
+    assert Path(paths['index/info.json']).read_bytes() == b'{"length":8}\n'
+    state = inspected(store, sources, outputs)
+    assert state.status == 'available' and state.verified
+    assert all(file.verified for file in state.files.values())
+    assert state.transform == TRANSFORM
+    record = state.sources['dna.fa.gz']
+    assert record['verified'] is True
+    assert {key: record[key] for key in ('sha256', 'size')} == metadata(COMPRESSED)
+    receipt = json.loads((Path(state.generation) / module.MANIFEST).read_text())
+    assert receipt['definition']['transform'] == TRANSFORM
+    assert set(receipt['outputs']) == set(outputs)
+    assert Path(sources['dna.fa.gz']['path']).read_bytes() == COMPRESSED
+    assert not list(store.glob('.inputs-*'))
+    assert not list(store.glob('.staging-*'))
+    assert inspect_bundle(store).status == 'invalid'  # Distinct store ownership.
+    assert not inspect_materialization(store).verified
+    new_paths = install(store, sources, outputs, force=True)
+    assert paths != new_paths
+    assert all(Path(path).exists() for path in paths.values())
+
+
+def forbidden(*args, **kwargs):
+    raise AssertionError('offline cache hit tried to write, build, or acquire')
+
+
+def test_readonly_offline_hit_never_touches_sources(tmp_path, sources, outputs, monkeypatch):
+    store = tmp_path / 'dna'
+    paths = install(store, sources, outputs)
+    Path(sources['dna.fa.gz']['path']).unlink()
+    changed_modes = []
+    for path in [*store.rglob('*'), store]:
+        changed_modes.append((path, path.stat().st_mode & 0o777))
+        os.chmod(path, 0o555 if path.is_dir() else 0o444)
+    try:
+        monkeypatch.setattr(download, 'fetch_file', forbidden)
+        monkeypatch.setattr(module, 'file_lock', forbidden)
+        monkeypatch.setattr(module, 'write_json', forbidden)
+        monkeypatch.setattr(Path, 'mkdir', forbidden)
+        assert install(store, sources, outputs, builder=forbidden) == paths
+        assert inspected(store, sources, outputs).verified
+        fast = inspected(store, sources, outputs, verify_files=False)
+        assert fast.status == 'available' and not fast.verified
+        assert not any(file.verified for file in fast.files.values())
+    finally:
+        for path, mode in reversed(changed_modes):
+            os.chmod(path, mode)
+
+
+@pytest.mark.parametrize('failure', ['builder', 'keyboard', 'last-output', 'validation', 'rename', 'pointer'])
+def test_failed_refresh_preserves_old_and_keeps_complete_input(tmp_path, sources, outputs, monkeypatch, failure):
+    store = tmp_path / 'dna'
+    old_paths = install(store, sources, outputs)
+    old_pointer = (store / module.CURRENT).read_bytes()
+
+    def failing_build(inputs, targets):
+        build(inputs, targets)
+        assert (store / module.CURRENT).read_bytes() == old_pointer
+        if failure == 'builder':
+            raise RuntimeError('biological validation failed')
+        if failure == 'keyboard':
+            raise KeyboardInterrupt
+        if failure == 'last-output':
+            Path(targets['index/info.json']).unlink()
+        if failure == 'validation':
+            Path(targets['index/info.json']).write_bytes(b'wrong')
+
+    original_replace, original_write = module.os.replace, module.write_json
+    if failure == 'rename':
+        def fail_replace(source, target):
+            if Path(target).parent.name == 'generations':
+                raise OSError('generation rename failed')
+            return original_replace(source, target)
+        monkeypatch.setattr(module.os, 'replace', fail_replace)
+    if failure == 'pointer':
+        def fail_write(path, value, **options):
+            if Path(path).name == module.CURRENT:
+                raise OSError('pointer failed')
+            return original_write(path, value, **options)
+        monkeypatch.setattr(module, 'write_json', fail_write)
+    with pytest.raises((RuntimeError, KeyboardInterrupt, FileValidationError, OSError)):
+        install(store, sources, outputs, builder=failing_build, force=True)
+    assert (store / module.CURRENT).read_bytes() == old_pointer
+    assert inspected(store, sources, outputs).verified
+    assert all(Path(path).exists() for path in old_paths.values())
+    assert (input_dir(store) / 'dna.fa.gz').read_bytes() == COMPRESSED
+    assert not list(store.glob('.staging-*'))
+    Path(sources['dna.fa.gz']['path']).unlink()
+    monkeypatch.setattr(module.os, 'replace', original_replace)
+    monkeypatch.setattr(module, 'write_json', original_write)
+    assert install(store, sources, outputs, force=True) != old_paths
+    assert not list(store.glob('.inputs-*'))
+
+
+def test_missing_pointer_recovers_locally_and_only_then_discards_inputs(tmp_path, sources, outputs, monkeypatch):
+    store = tmp_path / 'dna'
+    original = module.write_json
+    def interrupted(path, value, **options):
+        if Path(path).name == module.CURRENT:
+            raise KeyboardInterrupt
+        return original(path, value, **options)
+    monkeypatch.setattr(module, 'write_json', interrupted)
+    with pytest.raises(KeyboardInterrupt):
+        install(store, sources, outputs)
+    assert inspected(store, sources, outputs).status == 'recovery-required'
+    assert (input_dir(store) / 'dna.fa.gz').exists()
+    Path(sources['dna.fa.gz']['path']).unlink()
+    monkeypatch.setattr(module, 'write_json', original)
+    monkeypatch.setattr(download, 'fetch_file', forbidden)
+    paths = install(store, sources, outputs, builder=forbidden)
+    assert Path(paths['dna.fa']).read_bytes() == DATA
+    assert inspected(store, sources, outputs).verified
+    assert not list(store.glob('.inputs-*'))
+
+
+def test_recovery_always_hashes_outputs_even_with_fast_hit_flag(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    paths = install(store, sources, outputs, retain_sources=True)
+    (store / module.CURRENT).unlink()
+    Path(paths['dna.fa']).write_bytes(b'x' * len(DATA))
+    repaired = install(store, sources, outputs, verify_files=False)
+    assert repaired != paths
+    assert Path(repaired['dna.fa']).read_bytes() == DATA
+
+
+@pytest.mark.parametrize('change', ['source', 'transform-version', 'transform-options', 'output-inventory'])
+def test_dependency_mismatch_is_explicit_and_does_not_touch_old(tmp_path, sources, outputs, change):
+    store = tmp_path / 'dna'
+    paths = install(store, sources, outputs, retain_sources=True)
+    new_sources = {name: dict(spec) for name, spec in sources.items()}
+    options = {}
+    new_outputs = outputs
+    if change == 'source':
+        replacement = tmp_path / 'another.fa.gz'
+        replacement.write_bytes(COMPRESSED)
+        new_sources['dna.fa.gz']['path'] = replacement
+    elif change == 'transform-version':
+        options['transform'] = {'version': 'fasta-2', 'options': TRANSFORM['options']}
+    elif change == 'transform-options':
+        options['transform'] = {'version': TRANSFORM['version'], 'options': {'validate': False}}
+    else:
+        new_outputs = {'dna.fa': outputs['dna.fa']}
+        options['builder'] = lambda inputs, targets: Path(targets['dna.fa']).write_bytes(DATA)
+    with pytest.raises(FileValidationError, match='force=True'):
+        install(store, new_sources, new_outputs, **options)
+    assert inspected(store, sources, outputs).verified
+    assert install(store, new_sources, new_outputs, force=True, **options) != paths
+
+
+def test_transform_change_can_reuse_retained_inputs_after_upstream_removed(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    install(store, sources, outputs, retain_sources=True)
+    owned = input_dir(store)
+    Path(sources['dna.fa.gz']['path']).unlink()
+    paths = install(store, sources, outputs, force=True, transform={'version': 'fasta-2'})
+    assert Path(paths['dna.fa']).read_bytes() == DATA
+    assert not owned.exists()
+
+
+def test_untrusted_observation_is_not_trusted_verification(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    untrusted_sources = {'dna.fa.gz': {'path': sources['dna.fa.gz']['path']}}
+    untrusted_outputs = {name: {} for name in outputs}
+    paths = install(store, untrusted_sources, untrusted_outputs)
+    state = inspected(store, untrusted_sources, untrusted_outputs)
+    assert state.status == 'available' and not state.verified
+    assert state.sources['dna.fa.gz']['verified'] is False
+    assert state.sources['dna.fa.gz']['sha256'] == sha256(COMPRESSED).hexdigest()
+    Path(paths['dna.fa']).write_bytes(b'x' * len(DATA))
+    assert inspected(store, untrusted_sources, untrusted_outputs, verify_files=False).status == 'available'
+    assert inspected(store, untrusted_sources, untrusted_outputs).status == 'invalid'
+    with pytest.raises(FileValidationError):
+        install(store, untrusted_sources, untrusted_outputs)
+    assert install(store, untrusted_sources, untrusted_outputs, force=True) != paths
+
+
+@pytest.mark.parametrize('kind', ['symlink', 'hardlink', 'fifo', 'extra', 'extra-directory', 'input-mutation'])
+def test_unsafe_or_undeclared_builder_results_are_never_published(tmp_path, sources, outputs, kind):
+    store = tmp_path / 'dna'
+    def unsafe(inputs, targets):
+        build(inputs, targets)
+        target = Path(targets['dna.fa'])
+        if kind in ('symlink', 'hardlink', 'fifo'):
+            target.unlink()
+            if kind == 'symlink':
+                target.symlink_to(sources['dna.fa.gz']['path'])
+            elif kind == 'hardlink':
+                os.link(inputs['dna.fa.gz'], target)
+            else:
+                os.mkfifo(target)
+        elif kind == 'extra':
+            (target.parent / 'undeclared').write_text('extra')
+        elif kind == 'extra-directory':
+            (target.parent / 'unused').mkdir()
+        else:
+            Path(inputs['dna.fa.gz']).write_bytes(b'changed')
+    with pytest.raises(FileValidationError):
+        install(store, sources, outputs, builder=unsafe)
+    assert inspected(store, sources, outputs).status == 'missing'
+    assert list((store / 'generations').iterdir()) == []
+    assert Path(sources['dna.fa.gz']['path']).read_bytes() == COMPRESSED
+
+
+@pytest.mark.parametrize('kind', ['symlink', 'hardlink', 'fifo', 'directory'])
+def test_nonregular_local_sources_rejected_without_blocking(tmp_path, sources, outputs, kind):
+    target = tmp_path / 'unsafe'
+    if kind == 'symlink':
+        target.symlink_to(sources['dna.fa.gz']['path'])
+    elif kind == 'hardlink':
+        os.link(sources['dna.fa.gz']['path'], target)
+    elif kind == 'fifo':
+        os.mkfifo(target)
+    else:
+        target.mkdir()
+    sources['dna.fa.gz']['path'] = target
+    with pytest.raises(FileValidationError):
+        install(tmp_path / 'dna', sources, outputs)
+
+
+@pytest.mark.parametrize('force', [False, True])
+def test_foreign_directory_never_adopted(tmp_path, sources, outputs, force):
+    store = tmp_path / 'foreign'
+    store.mkdir()
+    (store / 'precious').write_text('keep')
+    with pytest.raises((FileValidationError, OSError)):
+        install(store, sources, outputs, force=force)
+    assert (store / 'precious').read_text() == 'keep'
+
+
+def test_complete_gzip_but_failed_decompression_reuses_input(tmp_path, sources, outputs):
+    invalid_gzip = b'not-gzip'
+    source = sources['dna.fa.gz']['path']
+    source.write_bytes(invalid_gzip)
+    sources['dna.fa.gz'].update(metadata(invalid_gzip))
+    store = tmp_path / 'dna'
+    with pytest.raises(gzip.BadGzipFile):
+        install(store, sources, outputs)
+    source.unlink()
+    owned = input_dir(store) / 'dna.fa.gz'
+    assert owned.read_bytes() == invalid_gzip
+    calls = []
+    def failure(inputs, targets):
+        calls.append(Path(inputs['dna.fa.gz']).read_bytes())
+        raise RuntimeError('retry reached builder with complete input')
+    with pytest.raises(RuntimeError, match='retry reached'):
+        install(store, sources, outputs, builder=failure)
+    assert calls == [invalid_gzip] and owned.exists()
+
+
+def test_same_artifact_serializes_but_unrelated_builders_can_run(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    entered, release = threading.Event(), threading.Event()
+    calls = []
+    def paused(inputs, targets):
+        calls.append(1)
+        entered.set()
+        assert release.wait(8)
+        build(inputs, targets)
+    with ThreadPoolExecutor(max_workers=3) as pool:
+        first = pool.submit(install, store, sources, outputs, builder=paused)
+        assert entered.wait(8)
+        second = pool.submit(install, store, sources, outputs, builder=paused)
+        unrelated = pool.submit(install, tmp_path / 'other', sources, outputs)
+        try:
+            assert unrelated.result(timeout=8)
+            assert not second.done()
+        finally:
+            release.set()
+        assert first.result(timeout=8) == second.result(timeout=8)
+    assert calls == [1]
+
+
+def test_reader_sees_whole_old_pair_until_pointer_publication(tmp_path, sources, outputs, monkeypatch):
+    store = tmp_path / 'dna'
+    old = install(store, sources, outputs)
+    entered, release = threading.Event(), threading.Event()
+    original = module.write_json
+    def paused(path, value, **options):
+        if Path(path).name == module.CURRENT:
+            entered.set()
+            assert release.wait(8)
+        return original(path, value, **options)
+    monkeypatch.setattr(module, 'write_json', paused)
+    with ThreadPoolExecutor() as pool:
+        future = pool.submit(install, store, sources, outputs, force=True)
+        assert entered.wait(8)
+        try:
+            state = inspected(store, sources, outputs)
+            assert state.verified
+            assert {name: file.path for name, file in state.files.items()} == old
+            assert (input_dir(store) / 'dna.fa.gz').exists()
+        finally:
+            release.set()
+        new = future.result(timeout=8)
+    assert new != old
+    assert Path(new['index/info.json']).parents[1] == Path(new['dna.fa']).parent
+
+
+def test_large_dependency_receipts_are_readable(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    transform = {'version': '1', 'options': 'x' * (1024 * 1024 + 1)}
+    paths = install(store, sources, outputs, transform=transform)
+    receipt = Path(paths['dna.fa']).parent / module.MANIFEST
+    assert receipt.stat().st_size > 1024 * 1024
+    assert inspect_materialization(store, sources, transform=transform, outputs=outputs).verified
+
+
+@pytest.fixture
+def server():
+    requests = []
+    actions = []
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self):
+            requests.append(dict(self.headers))
+            action = actions.pop(0) if actions else {}
+            offset = int(self.headers.get('Range', 'bytes=0-')[6:-1])
+            body = COMPRESSED[offset:]
+            self.send_response(206 if offset else 200)
+            self.send_header('Content-Length', str(len(body)))
+            self.send_header('ETag', '"dna-110"')
+            if offset:
+                self.send_header('Content-Range', 'bytes %d-%d/%d' % (offset, len(COMPRESSED) - 1, len(COMPRESSED)))
+            self.end_headers()
+            try:
+                self.wfile.write(body[:action.get('drop', len(body))])
+                self.wfile.flush()
+                if 'drop' in action:
+                    self.connection.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+        def log_message(self, *args):
+            pass
+    srv = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+    thread = threading.Thread(target=srv.serve_forever, daemon=True)
+    thread.start()
+    yield 'http://127.0.0.1:%d/dna.gz?token=secret' % srv.server_port, requests, actions
+    srv.shutdown()
+    srv.server_close()
+    thread.join()
+
+
+@pytest.mark.parametrize('trusted', [True, False])
+def test_resume_gzip_and_record_validated_transport(server, tmp_path, outputs, trusted):
+    url, requests, actions = server
+    actions.append({'drop': 12})
+    sources = {'dna.fa.gz': dict(url=url, size=len(COMPRESSED))}
+    if trusted:
+        sources['dna.fa.gz']['sha256'] = sha256(COMPRESSED).hexdigest()
+    store = tmp_path / 'dna'
+    paths = install(store, sources, outputs, download_options={'resume': True, 'chunk_size': 4,
+                    'retry_backoff': 0, 'show_progress': False})
+    assert Path(paths['dna.fa']).read_bytes() == DATA
+    assert requests[1]['Range'] == 'bytes=12-'
+    assert requests[1]['If-Range'] == '"dna-110"'
+    state = inspected(store, sources, outputs)
+    record = state.sources['dna.fa.gz']
+    assert record['verified'] is trusted
+    assert record['transport'] == {'header': 'ETag', 'value': '"dna-110"'}
+    assert record['origin'] == url.split('?')[0]
+    assert 'secret' not in (Path(state.generation) / module.MANIFEST).read_text()
+    assert not list(store.glob('.inputs-*'))
+
+
+def test_interrupted_download_and_builder_retry_without_network(server, tmp_path, outputs, monkeypatch):
+    url, requests, actions = server
+    sources = {'dna.fa.gz': dict(url=url, **metadata(COMPRESSED))}
+    store = tmp_path / 'dna'
+    def interrupt(done, total):
+        if done >= 8:
+            raise KeyboardInterrupt
+    options = {'resume': True, 'chunk_size': 4, 'progress_callback': interrupt, 'show_progress': False}
+    with pytest.raises(KeyboardInterrupt):
+        install(store, sources, outputs, download_options=options)
+    owned = input_dir(store) / 'dna.fa.gz'
+    assert (_state_directory(owned) / 'partial').read_bytes() == COMPRESSED[:8]
+    assert inspected(store, sources, outputs).status == 'missing'
+    del options['progress_callback']
+    def interrupted_build(inputs, targets):
+        with gzip.open(inputs['dna.fa.gz'], 'rb') as source, open(targets['dna.fa'], 'wb') as target:
+            target.write(source.read(4))
+            raise KeyboardInterrupt
+    with pytest.raises(KeyboardInterrupt):
+        install(store, sources, outputs, builder=interrupted_build, download_options=options)
+    assert requests[1]['Range'] == 'bytes=8-'
+    assert owned.read_bytes() == COMPRESSED
+    assert not (_state_directory(owned) / 'partial').exists()
+    monkeypatch.setattr(download, 'fetch_file', forbidden)
+    assert install(store, sources, outputs, download_options=options)
+    assert len(requests) == 2
+    assert not owned.exists()
+
+
+def test_killed_builder_retries_with_completed_local_input(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    ready = tmp_path / 'builder-ready'
+    code = '''
+import gzip, json, sys, time
+from pathlib import Path
+from datacache import materialize
+store, source, ready, outputs = sys.argv[1:]
+def builder(inputs, targets):
+    with gzip.open(inputs['dna.fa.gz'], 'rb') as compressed:
+        Path(targets['dna.fa']).write_bytes(compressed.read(4))
+        Path(ready).write_text('ready')
+        time.sleep(30)
+materialize(store, {'dna.fa.gz': {'path': source}}, transform={'version':'fasta-1','options':{'validate':True}},
+            outputs=json.loads(outputs), builder=builder, download_options={'show_progress':False})
+'''
+    process = subprocess.Popen([sys.executable, '-c', code, str(store), str(sources['dna.fa.gz']['path']),
+                                str(ready), json.dumps(outputs)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    try:
+        deadline = time.monotonic() + 8
+        while not ready.exists() and process.poll() is None and time.monotonic() < deadline:
+            time.sleep(0.02)
+        assert ready.exists(), process.communicate(timeout=2)
+        process.kill()
+        process.communicate(timeout=5)
+        assert list(store.glob('.staging-*'))
+        Path(sources['dna.fa.gz']['path']).unlink()
+        untrusted = {'dna.fa.gz': {'path': sources['dna.fa.gz']['path']}}
+        paths = install(store, untrusted, outputs)
+        assert Path(paths['dna.fa']).read_bytes() == DATA
+        assert not list(store.glob('.staging-*'))
+        assert not list(store.glob('.inputs-*'))
+    finally:
+        if process.poll() is None:
+            process.kill()
+            process.communicate(timeout=5)
+
+
+@pytest.mark.parametrize('options', [
+    {'force': 1}, {'retain_sources': 'yes'}, {'verify_files': None}, {'builder': None},
+    {'transform': {'version': ''}}, {'transform': {'version': '1', 'options': float('nan')}},
+    {'outputs': {'../escape': {}}}, {'outputs': {'A': {}, 'a/b': {}}},
+    {'download_options': {'raw': False}}, {'download_options': {'show_progress': 1}},
+    {'download_options': {'resume': 'yes'}},
+    {'download_options': {'chunk_size': 0}}, {'download_options': {'progress_callback': 'yes'}},
+    {'download_options': {'max_retries': -1}},
+])
+def test_invalid_options_fail_before_creating_store(tmp_path, sources, outputs, options):
+    store = tmp_path / 'missing' / 'dna'
+    with pytest.raises((ValueError, TypeError)):
+        install(store, sources, outputs, **options)
+    assert not store.parent.exists()
+
+
+def test_fast_reuse_does_not_hash_payloads(tmp_path, sources, outputs, monkeypatch):
+    store = tmp_path / 'dna'
+    paths = install(store, sources, outputs)
+    original = module._observe
+    def no_hash(*args, **options):
+        assert options.get('verify_files') is False
+        return original(*args, **options)
+    monkeypatch.setattr(module, '_observe', no_hash)
+    assert install(store, sources, outputs, verify_files=False, builder=forbidden) == paths
+
+
+def test_input_hash_failure_and_bad_final_output_never_publish(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    sources['dna.fa.gz']['sha256'] = '0' * 64
+    with pytest.raises(FileValidationError):
+        install(store, sources, outputs, verify_files=False)
+    assert not list((store / 'generations').iterdir())
+    sources['dna.fa.gz'].update(metadata(COMPRESSED))
+    outputs['dna.fa']['sha256'] = '0' * 64
+    with pytest.raises(FileValidationError):
+        install(store, sources, outputs, verify_files=False)
+    assert not list((store / 'generations').iterdir())
+
+
+def test_source_file_url_is_private_copy_and_not_deleted(tmp_path, sources, outputs):
+    source = sources['dna.fa.gz']['path']
+    sources['dna.fa.gz'] = dict(url=source.as_uri(), **metadata(COMPRESSED))
+    assert install(tmp_path / 'dna', sources, outputs)
+    assert source.read_bytes() == COMPRESSED
+
+
+def test_private_input_corruption_requires_explicit_repair(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    def interrupted(inputs, targets):
+        raise KeyboardInterrupt
+    with pytest.raises(KeyboardInterrupt):
+        install(store, sources, outputs, builder=interrupted)
+    owned = input_dir(store) / 'dna.fa.gz'
+    owned.write_bytes(b'x' * len(COMPRESSED))
+    with pytest.raises(FileValidationError, match='force=True'):
+        install(store, sources, outputs)
+    assert install(store, sources, outputs, force=True)
+    assert not owned.exists()
+
+
+def test_bare_complete_input_without_receipt_can_be_explicitly_repaired(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    def interrupted(inputs, targets):
+        raise KeyboardInterrupt
+    with pytest.raises(KeyboardInterrupt):
+        install(store, sources, outputs, builder=interrupted)
+    directory = input_dir(store)
+    (directory / module.INPUTS).unlink()
+    (directory / 'dna.fa.gz').write_bytes(b'x' * len(COMPRESSED))
+    with pytest.raises(FileValidationError, match='force=True'):
+        install(store, sources, outputs)
+    assert install(store, sources, outputs, force=True)
+
+
+def test_untrusted_bare_input_requires_new_acquisition(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    sources = {'dna.fa.gz': {'path': sources['dna.fa.gz']['path']}}
+    def interrupted(inputs, targets):
+        raise KeyboardInterrupt
+    with pytest.raises(KeyboardInterrupt):
+        install(store, sources, outputs, builder=interrupted)
+    directory = input_dir(store)
+    (directory / module.INPUTS).unlink()
+    (directory / 'dna.fa.gz').write_bytes(b'unverified unknown')
+    assert install(store, sources, outputs)
+
+
+def test_inspection_requires_complete_definition_or_none(tmp_path, sources, outputs):
+    with pytest.raises(ValueError):
+        inspect_materialization(tmp_path / 'dna', sources, outputs=outputs)
+    with pytest.raises(ValueError):
+        inspect_materialization(tmp_path / 'dna', verify_files='yes')
+
+
+def test_precreated_empty_store_preserves_mode(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    store.mkdir(mode=0o750)
+    assert inspected(store, sources, outputs).status == 'missing'
+    install(store, sources, outputs)
+    assert store.stat().st_mode & 0o777 == 0o750
+
+
+def test_retention_toggle_on_hit_does_not_cleanup(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    paths = install(store, sources, outputs, retain_sources=True)
+    owned = input_dir(store) / 'dna.fa.gz'
+    assert install(store, sources, outputs) == paths
+    assert owned.exists()
+    assert install(store, sources, outputs, force=True) != paths
+    assert not owned.exists()
+
+
+@pytest.mark.parametrize('kind', ['symlink', 'fifo', 'shared-directory'])
+def test_planted_private_state_is_not_followed_or_adopted(tmp_path, sources, outputs, kind):
+    store = tmp_path / 'dna'
+    def interrupted(inputs, targets):
+        raise KeyboardInterrupt
+    with pytest.raises(KeyboardInterrupt):
+        install(store, sources, outputs, builder=interrupted)
+    owned_directory = input_dir(store)
+    owned = owned_directory / 'dna.fa.gz'
+    if kind == 'shared-directory':
+        os.chmod(owned_directory, 0o755)
+    else:
+        owned.unlink()
+        if kind == 'symlink':
+            owned.symlink_to(sources['dna.fa.gz']['path'])
+        else:
+            os.mkfifo(owned)
+    with pytest.raises(FileValidationError):
+        install(store, sources, outputs, force=True)
+    assert Path(sources['dna.fa.gz']['path']).read_bytes() == COMPRESSED
+
+
+@pytest.mark.parametrize('malformation', [
+    'missing-output', 'extra-output', 'source-size', 'source-identity', 'source-trust',
+    'source-inventory', 'transform', 'output-digest', 'transport', 'definition',
+])
+def test_malformed_manifest_is_invalid_even_for_fast_inspection(tmp_path, sources, outputs, malformation):
+    store = tmp_path / 'dna'
+    paths = install(store, sources, outputs)
+    manifest = Path(paths['dna.fa']).parent / module.MANIFEST
+    receipt = json.loads(manifest.read_text())
+    if malformation == 'missing-output':
+        del receipt['outputs']['dna.fa']
+    elif malformation == 'extra-output':
+        receipt['outputs']['extra'] = metadata(b'')
+    elif malformation == 'source-size':
+        receipt['sources']['dna.fa.gz']['size'] += 1
+    elif malformation == 'source-identity':
+        receipt['sources']['dna.fa.gz']['identity'] = '0' * 64
+    elif malformation == 'source-trust':
+        receipt['sources']['dna.fa.gz']['verified'] = False
+    elif malformation == 'source-inventory':
+        receipt['sources'] = {}
+    elif malformation == 'transform':
+        receipt['definition']['transform'] = {'version': ''}
+    elif malformation == 'output-digest':
+        receipt['outputs']['dna.fa']['sha256'] = None
+    elif malformation == 'transport':
+        receipt['sources']['dna.fa.gz']['transport'] = {'header': 'ETag', 'value': 'W/"weak"'}
+    else:
+        receipt['definition'] = None
+    manifest.write_text(json.dumps(receipt))
+    assert inspected(store, sources, outputs, verify_files=False).status == 'invalid'
+    assert inspect_materialization(store, verify_files=False).status == 'invalid'
+
+
+def test_inspection_permission_error_is_inaccessible(tmp_path, sources, outputs, monkeypatch):
+    store = tmp_path / 'dna'
+    install(store, sources, outputs)
+    def permission_denied(*args, **kwargs):
+        raise PermissionError('cannot read receipt')
+    monkeypatch.setattr(module, 'read_json', permission_denied)
+    assert inspected(store, sources, outputs).status == 'inaccessible'
+    with pytest.raises(PermissionError):
+        install(store, sources, outputs)
+
+
+@pytest.mark.parametrize('bad_sources', [
+    {}, {'a': {}}, {'a': {'url': 'https://example.test', 'path': 'local'}},
+    {'a': {'url': ''}}, {'a': {'url': 'unsupported://example.test'}},
+    {'a': {'url': 'file://remote/file'}}, {'a': {'url': 'file:///file?query'}},
+    {'a': {'url': 'https://example.test', 'decompress': True}},
+    {'a': {'path': 'local'}, '.a.datacache.json': {'path': 'local'}},
+    {'a': {'path': 'local'}, '.a.datacache.json/extra': {'path': 'local'}},
+])
+def test_invalid_source_definitions_fail_without_writes(tmp_path, outputs, bad_sources):
+    store = tmp_path / 'missing' / 'dna'
+    with pytest.raises(ValueError):
+        install(store, bad_sources, outputs)
+    assert not store.parent.exists()
+
+
+def test_multiple_inputs_keep_completed_first_source_on_later_failure(tmp_path, sources, outputs, monkeypatch):
+    store = tmp_path / 'dna'
+    original = Path(sources['dna.fa.gz']['path'])
+    sources['auxiliary'] = {'path': tmp_path / 'missing'}
+    with pytest.raises(FileNotFoundError):
+        install(store, sources, outputs)
+    completed = input_dir(store) / 'dna.fa.gz'
+    assert completed.read_bytes() == COMPRESSED
+    original.unlink()
+    Path(sources['auxiliary']['path']).write_bytes(b'auxiliary')
+    paths = install(store, sources, outputs)
+    assert Path(paths['dna.fa']).read_bytes() == DATA
+    assert set(inspected(store, sources, outputs).sources) == {'dna.fa.gz', 'auxiliary'}
+
+
+def test_explicitly_empty_input_and_output(tmp_path):
+    source = tmp_path / 'empty'
+    source.touch()
+    sources = {'empty': dict(path=source, **metadata(b''))}
+    outputs = {'empty-result': metadata(b'')}
+    def empty(inputs, targets):
+        Path(targets['empty-result']).touch()
+    store = tmp_path / 'artifact'
+    materialize(store, sources, outputs=outputs, transform={'version': '1'}, builder=empty,
+                download_options={'show_progress': False})
+    assert inspect_materialization(store, sources, transform={'version': '1'}, outputs=outputs).verified
+    assert source.exists()
+
+
+def test_progress_defaults_on_and_existing_hits_stay_quiet(tmp_path, sources, outputs, monkeypatch):
+    enabled = []
+    original = module.Progress
+    def progress(enabled_flag, *args, **options):
+        enabled.append(enabled_flag)
+        return original(False, *args, **options)
+    monkeypatch.setattr(module, 'Progress', progress)
+    store = tmp_path / 'dna'
+    paths = install(store, sources, outputs, download_options={})
+    assert any(enabled)
+    enabled.clear()
+    assert install(store, sources, outputs, download_options={}) == paths
+    assert not any(enabled)
+
+
+def test_builder_paths_stay_absolute_after_working_directory_change(tmp_path, sources, outputs, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    other = tmp_path / 'other'
+    other.mkdir()
+    def changing_directory(inputs, targets):
+        assert all(Path(path).is_absolute() for path in [*inputs.values(), *targets.values()])
+        monkeypatch.chdir(other)
+        build(inputs, targets)
+    paths = install(Path('relative') / 'dna', sources, outputs, builder=changing_directory)
+    assert Path(paths['dna.fa']).read_bytes() == DATA
+
+
+def test_local_copy_honors_chunk_size_and_progress_callback(tmp_path, sources, outputs):
+    events = []
+    install(tmp_path / 'dna', sources, outputs,
+            download_options={'chunk_size': 4, 'progress_callback': lambda *event: events.append(event),
+                              'show_progress': False})
+    assert events == [(count, len(COMPRESSED)) for count in range(4, len(COMPRESSED) + 1, 4)]
+
+
+def test_interrupted_local_copy_never_modifies_original(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    def interrupt(done, total):
+        raise KeyboardInterrupt
+    with pytest.raises(KeyboardInterrupt):
+        install(store, sources, outputs, download_options={'chunk_size': 4, 'progress_callback': interrupt,
+                                                         'show_progress': False})
+    assert Path(sources['dna.fa.gz']['path']).read_bytes() == COMPRESSED
+    assert not (input_dir(store) / 'dna.fa.gz').exists()
+    assert not list(input_dir(store).glob('.copy-*'))
+    assert install(store, sources, outputs)
+
+
+# Review fixes
+
+
+def unpinned(tmp_path, data):
+    source = tmp_path / 'local.fa.gz'
+    source.write_bytes(gzip.compress(data, mtime=0))
+    return source, {'dna.fa.gz': dict(path=source)}
+
+
+def unpinned_outputs():
+    return {'dna.fa': {}, 'index/info.json': {}}
+
+
+def test_recovery_picks_the_newest_generation_not_the_highest_name(tmp_path):
+    source, sources = unpinned(tmp_path, DATA)
+    store = tmp_path / 'dna'
+    older = Path(install(store, sources, unpinned_outputs())['dna.fa']).parent
+    source.write_bytes(gzip.compress(b'>chr1\nTTTTTTTT\n', mtime=0))
+    newer = Path(install(store, sources, unpinned_outputs(), force=True)['dna.fa']).parent
+    # Names are random: make the older generation's name sort highest.
+    older.rename(store / 'generations' / ('f' * 32))
+    newer.rename(store / 'generations' / ('0' * 32))
+    (store / module.CURRENT).unlink()
+    recovered = install(store, sources, unpinned_outputs())
+    assert Path(recovered['dna.fa']).parent.name == '0' * 32
+    assert Path(recovered['dna.fa']).read_bytes() == b'>chr1\nTTTTTTTT\n'
+
+
+def test_forced_refresh_copies_an_edited_local_source_again(tmp_path):
+    source, sources = unpinned(tmp_path, DATA)
+    store = tmp_path / 'dna'
+    install(store, sources, unpinned_outputs(), retain_sources=True)
+    source.write_bytes(gzip.compress(b'>chr1\nTTTTTTTT\n', mtime=0))
+    paths = install(store, sources, unpinned_outputs(), force=True)
+    assert Path(paths['dna.fa']).read_bytes() == b'>chr1\nTTTTTTTT\n'
+    state = inspect_materialization(store)
+    assert state.sources['dna.fa.gz']['sha256'] == sha256(source.read_bytes()).hexdigest()
+
+
+def test_force_repairs_a_corrupt_private_input_receipt(tmp_path, sources, outputs):
+    import shutil
+    store = tmp_path / 'dna'
+    install(store, sources, outputs, retain_sources=True)
+    # No generations left, so the next call has to acquire inputs again.
+    (store / module.CURRENT).unlink()
+    for generation in (store / 'generations').iterdir():
+        shutil.rmtree(generation)
+    (input_dir(store) / module.INPUTS).write_text('{')
+    with pytest.raises(FileValidationError, match='use force=True to repair'):
+        install(store, sources, outputs)
+    paths = install(store, sources, outputs, force=True)
+    assert Path(paths['dna.fa']).read_bytes() == DATA
+
+
+def test_force_rebuilds_instead_of_recovering_a_lost_pointer(tmp_path, sources, outputs):
+    store = tmp_path / 'dna'
+    old = install(store, sources, outputs)
+    (store / module.CURRENT).unlink()
+    built = []
+
+    def counting_build(inputs, targets):
+        built.append(True)
+        build(inputs, targets)
+
+    new = install(store, sources, outputs, builder=counting_build, force=True)
+    assert built == [True] and new != old
+    (store / module.CURRENT).unlink()
+    assert install(store, sources, outputs, builder=counting_build) == new  # Plain call recovers.
+    assert built == [True]
+
+
+def test_forced_refresh_does_not_hash_the_old_outputs(tmp_path, sources, outputs, monkeypatch):
+    store = tmp_path / 'dna'
+    install(store, sources, outputs)
+    modes = []
+    original = module._inspect_materialization
+
+    def recording(path, expected, *, verify_files=True):
+        modes.append(verify_files)
+        return original(path, expected, verify_files=verify_files)
+
+    monkeypatch.setattr(module, '_inspect_materialization', recording)
+    install(store, sources, outputs, force=True)
+    assert modes and not any(modes)
+
+
+@pytest.mark.parametrize('builder_fails', [True, False])
+def test_cleanup_failure_never_masks_the_outcome(tmp_path, sources, outputs, monkeypatch, caplog, builder_fails):
+    store = tmp_path / 'dna'
+    original = module.shutil.rmtree
+
+    def failing_rmtree(path, *args, **kwargs):
+        if Path(path).name.startswith('.staging-'):
+            raise OSError('busy')
+        return original(path, *args, **kwargs)
+
+    monkeypatch.setattr(module.shutil, 'rmtree', failing_rmtree)
+
+    def failing_build(inputs, targets):
+        raise RuntimeError('builder failed')
+
+    if builder_fails:
+        with pytest.raises(RuntimeError, match='builder failed'):
+            install(store, sources, outputs, builder=failing_build)
+    else:
+        assert Path(install(store, sources, outputs)['dna.fa']).read_bytes() == DATA
+    assert 'Could not remove the staging directory' in caplog.text