diff --git a/CHANGELOG.md b/CHANGELOG.md
index c0221cf..f45f153 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,12 @@
 
 ## 1.19.0
 
+- 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
+  resolves every dataset version once and rejects two versions sharing a
+  store. A populated foreign directory is never taken over, with or without
+  `force=True` (#93).
 - Add opt-in `verify_files=False` to `inspect_bundle`, `install_bundle` and
   `VersionedDatasetRegistry`'s `inspect`, `local_path`, `is_cached`, `status`,
   `download` and `ensure`: checks ownership, source identity, required file
diff --git a/datacache/bundles.py b/datacache/bundles.py
index 23baf31..f7d7741 100644
--- a/datacache/bundles.py
+++ b/datacache/bundles.py
@@ -215,11 +215,21 @@ 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."""
+    if (path_present(path) and stat.S_ISDIR(path.lstat().st_mode)
+            and not path_present(path / STORE) and any(path.iterdir())):
+        raise FileValidationError(
+            path, 'not a datacache bundle store; a populated directory is never taken over')
+
+
 def _initialize(path):
     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.
             return
         existing_mode = stat.S_IMODE(path.lstat().st_mode)
@@ -286,6 +296,7 @@ def install_bundle(destination, assets, *, force=False, verified=True, verify_fi
     if inspection.status == 'inaccessible':
         raise inspection.error
     if not force and inspection.status == 'invalid':
+        _refuse_foreign(path)
         raise FileValidationError(path, 'invalid bundle; use force=True to explicitly repair') from inspection.error
     if os.name != 'posix':
         raise NotImplementedError('Bundle installation requires a POSIX local filesystem')
@@ -368,17 +379,28 @@ class VersionedDatasetRegistry:
     Each dataset is {default_version, versions: {version: {asset: metadata}}}.
     The hitlist shape {filename, urls: {version: url}, default_version} is also
     accepted with verified=False. cache_root is a path; cache_dir optionally
-    accepts hitlist's zero-argument root callable. Construction never writes.
+    accepts hitlist's zero-argument root callable. Alternatively, store_path is
+    a two-argument (name, version) callback selecting an exact managed store.
+    Construction never writes or invokes either callback.
     """
 
-    def __init__(self, datasets, *, cache_root=None, cache_dir=None, verified=True):
-        if (cache_root is None) == (cache_dir is None):
-            raise ValueError('provide exactly one of cache_root or cache_dir')
+    def __init__(self, datasets, *, cache_root=None, cache_dir=None, store_path=None, verified=True):
+        if sum(value is not None for value in (cache_root, cache_dir, store_path)) != 1:
+            raise ValueError('provide exactly one of cache_root, cache_dir or store_path')
         if not isinstance(verified, bool):
             raise ValueError('verified must be a boolean')
-        self._root = cache_dir if cache_dir is not None else lambda: cache_root
-        if not callable(self._root):
+        if cache_dir is not None and not callable(cache_dir):
             raise ValueError('cache_dir must be callable')
+        if store_path is not None and not callable(store_path):
+            raise ValueError('store_path must be callable')
+        if store_path is None:
+            # A root callable is consulted on every lookup, as hitlist expects.
+            root = cache_dir if cache_dir is not None else lambda: cache_root
+            self._store_path = lambda name, version: Path(root()) / name / version
+        else:
+            self._store_path = store_path
+        # Custom store paths, resolved and checked once on first use.
+        self._custom_stores = None if store_path is None else {}
         self.verified = verified
         self._datasets = {}
         for name, spec in datasets.items():
@@ -406,10 +428,33 @@ def resolve_version(self, name, version=None):
     def bundle_path(self, name, version=None):
         """Resolve the store path without checking it or creating directories."""
         version = self.resolve_version(name, version)
-        parent = Path(self._root()) / name
-        if path_present(parent):
-            _directory(parent)
-        return parent / version
+        if self._custom_stores is None:
+            path = self._store_path(name, version)
+            # <root>/<name> is DataCache's own directory: never follow a link there.
+            if path_present(path.parent):
+                _directory(path.parent)
+            return path
+        # A custom store's parent belongs to the application and may be a link
+        # (e.g. to another disk); the store itself is still never one.
+        if not self._custom_stores:
+            self._custom_stores = self._resolve_custom_stores()
+        return self._custom_stores[(name, version)]
+
+    def _resolve_custom_stores(self):
+        """Every (name, version)'s store_path result: one store per version."""
+        stores, owners = {}, {}
+        for name in sorted(self._datasets):
+            for version in sorted(self._datasets[name]['versions']):
+                path = self._store_path(name, version)
+                if not isinstance(path, (str, os.PathLike)) or not os.fspath(path):
+                    raise ValueError('store_path(%r, %r) returned %r, not a path' % (name, version, path))
+                key = os.path.normpath(os.path.abspath(path))
+                if key in owners:
+                    raise ValueError('store_path gives %s for both %s %s and %s %s; '
+                                     'each version needs its own store' % ((key,) + owners[key] + (name, version)))
+                owners[key] = (name, version)
+                stores[(name, version)] = Path(path)
+        return stores
 
     def inspect(self, name, version=None, *, verify_files=True):
         version = self.resolve_version(name, version)
diff --git a/docs/api.md b/docs/api.md
index d7f1e02..62a778c 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -1289,12 +1289,16 @@ generation. Paths remain usable across refreshes until explicitly removed.
 ## VersionedDatasetRegistry
 
 ```text
-VersionedDatasetRegistry(datasets, *, cache_root=None, cache_dir=None, verified=True)
+VersionedDatasetRegistry(datasets, *, cache_root=None, cache_dir=None, store_path=None, verified=True)
 ```
 
-Select exactly one root path or a zero-argument `cache_dir` callable. Each dataset
+Select exactly one root path, zero-argument `cache_dir` callable, or
+`store_path(name, version)` callback selecting the exact managed store path; it is called once per dataset version on first lookup, and each version needs its own store.
+Root strategies keep `<root>/<name>/<version>`; the exact-path callback owns
+the consumer layout but not the internal generation layout. Each dataset
 specifies a `default_version` and `versions`, mapping concrete versions to asset
-mappings. Construction validates metadata and performs no writes or networking.
+mappings. Construction validates metadata and performs no writes or networking,
+and does not invoke callbacks. Path callbacks should be pure path computations.
 The [bundle guide](bundles.md) includes an example and downstream migration notes.
 
 | Method | Result |
diff --git a/docs/bundles.md b/docs/bundles.md
index ed0b8e4..d91b2b8 100644
--- a/docs/bundles.md
+++ b/docs/bundles.md
@@ -73,6 +73,42 @@ ownership marker, even with `force=True`. Store generated indices and other deri
 outputs under a separate application-owned directory. Reinstalling sources does
 not visit those outputs.
 
+## Consumer-selected store paths
+
+Select exactly one of `cache_root`, a zero-argument `cache_dir` root callable,
+or an exact `store_path(name, version)` callback. Root strategies retain
+`<root>/<name>/<version>`; the callback can keep an application's own layout,
+for example a version directory per release:
+
+```python
+registry = VersionedDatasetRegistry(
+    datasets,
+    store_path=lambda name, version: application_root / version / "sources" / name,
+)
+store = registry.bundle_path("reference")
+```
+
+The callback receives a validated concrete version, including when the caller
+omits it and selects the pinned default. Construction never calls it. The first
+lookup calls it once for every dataset version, without filesystem inspection
+or mutation, and the paths are reused afterwards, so callbacks should only
+compute paths. Each version needs its own store: a callback that gives two
+versions the same path raises `ValueError`, as does one that returns something
+other than a path. Inspection, installation, recovery and refresh all use the
+chosen store, while DataCache still owns its immutable generations.
+
+The store's parent belongs to the application and may be a link, for example to
+another disk; the store itself is never a link. Installation creates missing
+parent directories and keeps its lock file and temporary staging directories
+in the parent, so give the stores a parent of their own rather than a
+directory the application lists or cleans.
+
+Do not point this callback at a populated legacy data/index directory: even
+`force=True` cannot adopt a foreign directory, and installation raises
+`FileValidationError` saying so. Keep those old caches readable
+and deliberately install into a new managed source subdirectory instead. Model,
+biological naming and migration policy remain the application's responsibility.
+
 ## Status and trust
 
 `BundleInspection.status` is `available`, `missing`, `invalid`, `inaccessible`, or
diff --git a/tests/test_bundle_store_paths.py b/tests/test_bundle_store_paths.py
new file mode 100644
index 0000000..3eaa5e1
--- /dev/null
+++ b/tests/test_bundle_store_paths.py
@@ -0,0 +1,188 @@
+"""Consumer-selected paths use the ordinary owned bundle transaction."""
+
+from hashlib import sha256
+import os
+from pathlib import Path
+
+import pytest
+
+from datacache import FileValidationError, VersionedDatasetRegistry
+from datacache import bundles, download
+
+pytestmark = pytest.mark.skipif(os.name != 'posix', reason='POSIX bundle installation')
+
+
+@pytest.fixture
+def datasets(tmp_path):
+    source = tmp_path / 'source.fa'
+    payload = b'>reference\nACGT\n'
+    source.write_bytes(payload)
+    assets = {'records.fa': dict(url=source.as_uri(), sha256=sha256(payload).hexdigest(), size=len(payload))}
+    return {'reference': dict(default_version='110', versions={'110': assets, '109': assets})}
+
+
+def test_custom_path_resolution_is_pure_and_uses_concrete_versions(tmp_path, datasets, monkeypatch):
+    calls = []
+
+    def store_path(name, version):
+        calls.append((name, version))
+        return str(tmp_path / 'GRCh38' / ('ensembl-' + version) / 'sources' / name)
+
+    registry = VersionedDatasetRegistry(datasets, store_path=store_path)
+    assert calls == []
+
+    def forbidden(*args, **kwargs):
+        raise AssertionError('path resolution inspected or mutated the filesystem')
+
+    monkeypatch.setattr(bundles, 'path_present', forbidden)
+    monkeypatch.setattr(Path, 'mkdir', forbidden)
+    assert registry.bundle_path('reference') == tmp_path / 'GRCh38/ensembl-110/sources/reference'
+    # Every version's store is resolved once, on first use, and then reused.
+    assert sorted(calls) == [('reference', '109'), ('reference', '110')]
+    assert registry.bundle_path('reference', '109') == tmp_path / 'GRCh38/ensembl-109/sources/reference'
+    with pytest.raises(ValueError):
+        registry.bundle_path('reference', 'latest')
+    assert len(calls) == 2
+    assert not (tmp_path / 'GRCh38').exists()
+
+
+def test_custom_layout_install_versions_refresh_and_read_only_reuse(tmp_path, datasets, monkeypatch):
+    root = tmp_path / 'application'
+    index = root / 'GRCh38' / 'ensembl-110' / 'index.sqlite'
+    index.parent.mkdir(parents=True)
+    index.write_bytes(b'application-owned derived index')
+    registry = VersionedDatasetRegistry(
+        datasets, store_path=lambda name, version: root / 'GRCh38' / ('ensembl-' + version) / 'sources' / name)
+    assert registry.inspect('reference').status == 'missing'
+    assert not registry.bundle_path('reference').exists()
+    current = registry.download('reference')
+    older = registry.download('reference', '109')
+    assert current != older
+    assert registry.local_path('reference') == Path(current['records.fa'])
+    assert registry.local_path('reference', '109') == Path(older['records.fa'])
+    refreshed = registry.download('reference', force=True)
+    assert refreshed != current
+    assert Path(current['records.fa']).read_bytes() == Path(refreshed['records.fa']).read_bytes()
+    assert index.read_bytes() == b'application-owned derived index'
+    store = registry.bundle_path('reference')
+    entries = [store, *store.rglob('*')]
+    modes = {path: path.stat().st_mode & 0o777 for path in entries}
+
+    def forbidden(*args, **kwargs):
+        raise AssertionError('offline reuse attempted a write or acquisition')
+
+    monkeypatch.setattr(download, 'fetch_file', forbidden)
+    monkeypatch.setattr(bundles, 'file_lock', forbidden)
+    monkeypatch.setattr(bundles, 'write_json', forbidden)
+    try:
+        for path in entries:
+            path.chmod(0o555 if path.is_dir() else 0o444)
+        assert registry.download('reference') == refreshed
+        assert registry.local_path('reference') == Path(refreshed['records.fa'])
+        assert registry.is_cached('reference')
+        assert registry.inspect('reference').verified
+    finally:
+        for path, mode in modes.items():
+            path.chmod(mode)
+
+
+def test_custom_layout_recovers_completed_generation_without_network(tmp_path, datasets, monkeypatch):
+    registry = VersionedDatasetRegistry(
+        datasets, store_path=lambda name, version: tmp_path / ('sources-' + version) / name)
+    original = bundles.write_json
+
+    def interrupted(path, value, **kwargs):
+        if Path(path).name == bundles.CURRENT:
+            raise KeyboardInterrupt('pointer publication interrupted')
+        return original(path, value, **kwargs)
+
+    monkeypatch.setattr(bundles, 'write_json', interrupted)
+    with pytest.raises(KeyboardInterrupt):
+        registry.download('reference')
+    state = registry.inspect('reference')
+    assert state.status == 'recovery-required'
+    monkeypatch.setattr(bundles, 'write_json', original)
+
+    def forbidden(*args, **kwargs):
+        raise AssertionError('recovery attempted acquisition')
+
+    monkeypatch.setattr(download, 'fetch_file', forbidden)
+    paths = registry.download('reference')
+    assert registry.inspect('reference').verified
+    assert registry.local_path('reference') == Path(paths['records.fa'])
+
+
+@pytest.mark.parametrize('force', [False, True])
+def test_custom_paths_never_take_over_foreign_directories(tmp_path, datasets, force):
+    foreign = tmp_path / 'foreign'
+    foreign.mkdir()
+    precious = foreign / 'index.sqlite'
+    precious.write_bytes(b'keep me')
+    registry = VersionedDatasetRegistry(
+        datasets, store_path=lambda name, version: foreign if version == '110' else tmp_path / version)
+    with pytest.raises(FileValidationError, match='never taken over'):
+        registry.download('reference', force=force)
+    assert list(foreign.iterdir()) == [precious]
+    assert precious.read_bytes() == b'keep me'
+
+
+def test_custom_paths_reject_symlinked_stores(tmp_path, datasets):
+    foreign = tmp_path / 'foreign'
+    foreign.mkdir()
+    store = tmp_path / 'link'
+    store.symlink_to(foreign, target_is_directory=True)
+    registry = VersionedDatasetRegistry(
+        datasets, store_path=lambda name, version: store if version == '110' else tmp_path / version)
+    assert registry.inspect('reference').status == 'invalid'
+    with pytest.raises(FileValidationError):
+        registry.download('reference', force=True)
+    assert list(foreign.iterdir()) == []
+
+
+@pytest.mark.parametrize('options', [{}, {'cache_root': '/cache', 'cache_dir': lambda: '/cache'},
+    {'cache_root': '/cache', 'store_path': lambda name, version: '/store'},
+    {'cache_dir': lambda: '/cache', 'store_path': lambda name, version: '/store'},
+    {'cache_root': '/cache', 'cache_dir': lambda: '/cache', 'store_path': lambda name, version: '/store'}])
+def test_destination_strategies_are_mutually_exclusive(datasets, options):
+    with pytest.raises(ValueError, match='exactly one'):
+        VersionedDatasetRegistry(datasets, **options)
+
+
+@pytest.mark.parametrize('key', ['cache_dir', 'store_path'])
+def test_path_callbacks_must_be_callable(datasets, key):
+    with pytest.raises(ValueError, match=key + ' must be callable'):
+        VersionedDatasetRegistry(datasets, **{key: '/not-a-callback'})
+
+
+def test_existing_root_strategies_remain_compatible(tmp_path, datasets):
+    explicit = VersionedDatasetRegistry(datasets, cache_root=tmp_path / 'root')
+    selected = [tmp_path / 'root']
+    dynamic = VersionedDatasetRegistry(datasets, cache_dir=lambda: selected[0])
+    assert explicit.bundle_path('reference') == dynamic.bundle_path('reference')
+    selected[0] = tmp_path / 'different-root'
+    assert dynamic.bundle_path('reference') == tmp_path / 'different-root/reference/110'
+    assert not selected[0].exists()
+
+
+def test_versions_sharing_a_custom_store_are_rejected(tmp_path, datasets):
+    registry = VersionedDatasetRegistry(datasets, store_path=lambda name, version: tmp_path / name)
+    with pytest.raises(ValueError, match='each version needs its own store'):
+        registry.bundle_path('reference')
+    assert not (tmp_path / 'reference').exists()
+
+
+@pytest.mark.parametrize('result', [None, '', 7, b'/bytes'])
+def test_custom_store_paths_must_be_paths(datasets, result):
+    registry = VersionedDatasetRegistry(datasets, store_path=lambda name, version: result)
+    with pytest.raises(ValueError, match='not a path'):
+        registry.bundle_path('reference')
+
+
+def test_custom_store_parent_may_be_a_link(tmp_path, datasets):
+    disk = tmp_path / 'disk'
+    disk.mkdir()
+    (tmp_path / 'data').symlink_to(disk, target_is_directory=True)
+    registry = VersionedDatasetRegistry(
+        datasets, store_path=lambda name, version: tmp_path / 'data' / version / name)
+    paths = registry.download('reference')
+    assert Path(paths['records.fa']).resolve().is_relative_to(disk)