diff --git a/pychunkedgraph/app/common.py b/pychunkedgraph/app/common.py
index 237e11fc0..9fbed58bd 100644
--- a/pychunkedgraph/app/common.py
+++ b/pychunkedgraph/app/common.py
@@ -82,7 +82,7 @@ def after_request(response):
 
     response.data = compression.gzip_compress(response.data)
     response.headers["Content-Encoding"] = "gzip"
-    response.headers["Vary"] = "Accept-Encoding"
+    response.vary.add("Accept-Encoding")
     response.headers["Content-Length"] = len(response.data)
     return response
 
diff --git a/pychunkedgraph/app/meshing/common.py b/pychunkedgraph/app/meshing/common.py
index 8f1a0c20a..a7949a176 100644
--- a/pychunkedgraph/app/meshing/common.py
+++ b/pychunkedgraph/app/meshing/common.py
@@ -11,8 +11,11 @@
 from pychunkedgraph import __version__
 from pychunkedgraph.app import app_utils
 from pychunkedgraph.graph import chunkedgraph
+from pychunkedgraph.graph import exceptions as cg_exceptions
 from pychunkedgraph.app.meshing import tasks as meshing_tasks
 from pychunkedgraph.meshing import meshgen
+from pychunkedgraph.meshing.mesh_meta import MeshMeta
+from pychunkedgraph.meshing.manifest import v2
 from pychunkedgraph.meshing.manifest import get_highest_child_nodes_with_meshes
 from pychunkedgraph.meshing.manifest import get_children_before_start_layer
 from pychunkedgraph.meshing.manifest import ManifestCache
@@ -68,13 +71,21 @@ def handle_get_manifest(table_id, node_id):
         bounding_box = np.array([b.split("-") for b in bounds.split("_")], dtype=int).T
 
     cg = app_utils.get_cg(table_id)
+    mm = MeshMeta(cg)
+    manifest_version = v2.requested_manifest_version(request.headers.get("Accept"))
+    if manifest_version < 2 and mm.needs_v2:
+        raise cg_exceptions.NotAcceptable(
+            "This dataset serves meshes from an absolute path; upgrade your "
+            "client to one that requests "
+            "'Accept: application/x.cave;manifest_version=2'."
+        )
     verify = request.args.get("verify", False)
     verify = verify in ["True", "true", "1", True]
     return_seg_ids = request.args.get("return_seg_ids", False)
     prepend_seg_ids = request.args.get("prepend_seg_ids", False)
     return_seg_ids = return_seg_ids in ["True", "true", "1", True]
     prepend_seg_ids = prepend_seg_ids in ["True", "true", "1", True]
-    start_layer = cg.meta.custom_data.get("mesh", {}).get("max_layer", 2)
+    start_layer = mm.max_layer
     start_layer = int(request.args.get("start_layer", start_layer))
     if "start_layer" in data:
         start_layer = int(data["start_layer"])
@@ -91,8 +102,12 @@ def handle_get_manifest(table_id, node_id):
         flexible_start_layer,
         bounding_box,
         data,
+        manifest_version,
     )
-    return manifest_response(cg, args)
+    response = make_response(jsonify(manifest_response(cg, args)))
+    response.headers["X-Manifest-Version"] = str(manifest_version)
+    response.headers["Vary"] = "Accept"
+    return response
 
 
 def manifest_response(cg, args):
@@ -107,25 +122,32 @@ def manifest_response(cg, args):
         flexible_start_layer,
         bounding_box,
         data,
+        manifest_version,
     ) = args
-    resp = {}
-    seg_ids = []
     if not verify:
-        seg_ids, resp["fragments"] = speculative_manifest_sharded(
+        seg_ids, fragments = speculative_manifest_sharded(
             cg, node_id, start_layer=start_layer, bounding_box=bounding_box
         )
-
     else:
-        seg_ids, resp["fragments"] = get_highest_child_nodes_with_meshes(
+        seg_ids, fragments = get_highest_child_nodes_with_meshes(
             cg,
             np.uint64(node_id),
             start_layer=start_layer,
             bounding_box=bounding_box,
         )
-    if prepend_seg_ids:
-        resp["fragments"] = [f"~{i}:{f}" for i, f in zip(seg_ids, resp["fragments"])]
-    if return_seg_ids:
-        resp["seg_ids"] = seg_ids
+
+    if manifest_version >= 2:
+        mm = MeshMeta(cg)
+        initial, dynamic = v2.to_v2_groups(seg_ids, fragments, prepend_seg_ids)
+        resp = v2.assemble(mm.initial_path, mm.dynamic_path, initial, dynamic)
+        if return_seg_ids:
+            resp["seg_ids"] = seg_ids
+    else:
+        resp = {"fragments": fragments}
+        if prepend_seg_ids:
+            resp["fragments"] = [f"~{i}:{f}" for i, f in zip(seg_ids, fragments)]
+        if return_seg_ids:
+            resp["seg_ids"] = seg_ids
     return _check_post_options(cg, resp, data, seg_ids)
 
 
@@ -180,22 +202,15 @@ def handle_remesh(table_id):
 
 def _remeshing(serialized_cg_info, lvl2_nodes):
     cg = chunkedgraph.ChunkedGraph(**serialized_cg_info)
-    cv_mesh_dir = cg.meta.dataset_info["mesh"]
-    cv_unsharded_mesh_dir = cg.meta.dataset_info["mesh_metadata"]["unsharded_mesh_dir"]
-    cv_unsharded_mesh_path = os.path.join(
-        cg.meta.data_source.WATERSHED, cv_mesh_dir, cv_unsharded_mesh_dir
-    )
-    mesh_data = cg.meta.custom_data["mesh"]
-
-    # TODO: stop_layer and mip should be configurable by dataset
+    mm = MeshMeta(cg)
     meshgen.remeshing(
         cg,
         lvl2_nodes,
-        stop_layer=mesh_data["max_layer"],
-        mip=mesh_data["mip"],
-        max_err=mesh_data["max_error"],
-        cv_sharded_mesh_dir=cv_mesh_dir,
-        cv_unsharded_mesh_path=cv_unsharded_mesh_path,
+        stop_layer=mm.max_layer,
+        mip=mm.mip,
+        max_err=mm.max_error,
+        cv_sharded_mesh_dir=mm.dir,
+        cv_unsharded_mesh_path=mm.dynamic_path,
     )
 
     return Response(status=200)
diff --git a/pychunkedgraph/app/meshing/tasks.py b/pychunkedgraph/app/meshing/tasks.py
index a1f11ca68..373f934bd 100644
--- a/pychunkedgraph/app/meshing/tasks.py
+++ b/pychunkedgraph/app/meshing/tasks.py
@@ -1,27 +1,19 @@
 from pychunkedgraph.app import app_utils
-from pychunkedgraph.meshing import meshgen, meshgen_utils
+from pychunkedgraph.meshing import meshgen
+from pychunkedgraph.meshing.mesh_meta import MeshMeta
 import numpy as np
-import os
 
 
 def remeshing(table_id, lvl2_nodes):
     lvl2_nodes = np.array(lvl2_nodes, dtype=np.uint64)
     cg = app_utils.get_cg(table_id, skip_cache=True)
-
-    cv_mesh_dir = cg.meta.dataset_info["mesh"]
-    cv_unsharded_mesh_dir = cg.meta.dataset_info["mesh_metadata"]["unsharded_mesh_dir"]
-    cv_unsharded_mesh_path = os.path.join(
-        cg.meta.data_source.WATERSHED, cv_mesh_dir, cv_unsharded_mesh_dir
-    )
-    mesh_data = cg.meta.custom_data["mesh"]
-
-    # TODO: stop_layer and mip should be configurable by dataset
+    mm = MeshMeta(cg)
     meshgen.remeshing(
         cg,
         lvl2_nodes,
-        stop_layer=mesh_data["max_layer"],
-        mip=mesh_data["mip"],
-        max_err=mesh_data["max_error"],
-        cv_sharded_mesh_dir=cv_mesh_dir,
-        cv_unsharded_mesh_path=cv_unsharded_mesh_path,
+        stop_layer=mm.max_layer,
+        mip=mm.mip,
+        max_err=mm.max_error,
+        cv_sharded_mesh_dir=mm.dir,
+        cv_unsharded_mesh_path=mm.dynamic_path,
     )
\ No newline at end of file
diff --git a/pychunkedgraph/app/segmentation/common.py b/pychunkedgraph/app/segmentation/common.py
index 3250248f2..67ec6728b 100644
--- a/pychunkedgraph/app/segmentation/common.py
+++ b/pychunkedgraph/app/segmentation/common.py
@@ -14,6 +14,7 @@
 
 from pychunkedgraph import __version__
 from pychunkedgraph.app import app_utils
+from pychunkedgraph.meshing.mesh_meta import MeshMeta
 from pychunkedgraph.graph import (
     attributes,
     cutting,
@@ -111,6 +112,13 @@ def handle_info(table_id):
         combined_info["mesh_dir"] = mesh_dir
     elif combined_info.get("mesh_dir", None) is not None:
         combined_info["mesh_dir"] = "graphene_meshes"
+    # Absolute mesh bucket locations (== the v2 manifest keys) so clients that
+    # bypass the manifest resolve initial/dynamic meshes from the real buckets.
+    mm = MeshMeta(cg)
+    mesh_metadata = dict(combined_info.get("mesh_metadata", {}))
+    mesh_metadata["initial_mesh_path"] = mm.initial_path
+    mesh_metadata["dynamic_mesh_path"] = mm.dynamic_path
+    combined_info["mesh_metadata"] = mesh_metadata
     return jsonify(combined_info)
 
 
diff --git a/pychunkedgraph/graph/basetypes.py b/pychunkedgraph/graph/basetypes.py
new file mode 100644
index 000000000..25195a6c7
--- /dev/null
+++ b/pychunkedgraph/graph/basetypes.py
@@ -0,0 +1,6 @@
+"""Compat shim: ``basetypes`` lives at ``graph/utils/basetypes.py`` on this line
+but at ``graph/basetypes.py`` on the pcgv3 line. Re-export so the shorter import
+path resolves identically on both, keeping meshing code branch-agnostic.
+"""
+
+from .utils.basetypes import *  # noqa: F401,F403
diff --git a/pychunkedgraph/graph/exceptions.py b/pychunkedgraph/graph/exceptions.py
index 45aa57fc7..5ca56d8a2 100644
--- a/pychunkedgraph/graph/exceptions.py
+++ b/pychunkedgraph/graph/exceptions.py
@@ -64,6 +64,11 @@ class Forbidden(ClientError):
     status_code = http_client.FORBIDDEN
 
 
+class NotAcceptable(ClientError):
+    """Exception mapping a ``406 Not Acceptable`` response."""
+    status_code = http_client.NOT_ACCEPTABLE
+
+
 class Conflict(ClientError):
     """Exception mapping a ``409 Conflict`` response."""
     status_code = http_client.CONFLICT
diff --git a/pychunkedgraph/meshing/manifest/cache.py b/pychunkedgraph/meshing/manifest/cache.py
index f38a830c2..0decc3a65 100644
--- a/pychunkedgraph/meshing/manifest/cache.py
+++ b/pychunkedgraph/meshing/manifest/cache.py
@@ -12,6 +12,8 @@
 DOES_NOT_EXIST = "X"
 INITIAL_PATH_PREFIX = "initial_path_prefix"
 
+MANIFEST_TTL_SECONDS = 3 * 24 * 3600
+
 REDIS_HOST = os.environ.get("MANIFEST_CACHE_REDIS_HOST", "localhost")
 REDIS_PORT = os.environ.get("MANIFEST_CACHE_REDIS_PORT", "6379")
 REDIS_PASSWORD = os.environ.get("MANIFEST_CACHE_REDIS_PASSWORD", "")
@@ -74,6 +76,27 @@ def clear_fragments(self, node_ids) -> None:
         keys = [f"{self.namespace}:{n}" for n in node_ids]
         REDIS.delete(*keys)
 
+    def clear_namespace(self, batch_size: int = 1000) -> int:
+        """Delete every key under this graph_id's namespace.
+
+        SCAN-based pattern delete (non-blocking on the redis side).
+        Returns the number of keys deleted.
+        """
+        if REDIS is None:
+            return 0
+
+        pattern = f"{self.namespace}:*"
+        deleted = 0
+        batch = []
+        for key in REDIS.scan_iter(match=pattern, count=batch_size):
+            batch.append(key)
+            if len(batch) >= batch_size:
+                deleted += REDIS.delete(*batch)
+                batch.clear()
+        if batch:
+            deleted += REDIS.delete(*batch)
+        return deleted
+
     def _get_cached_initial_fragments(self, node_ids: List[np.uint64]):
         if REDIS is None:
             return {}, node_ids, []
@@ -140,10 +163,16 @@ def _set_cached_initial_fragments(
         for node_id, fragment_info in fragments_d.items():
             path, offset, size = fragment_info
             key = f"{self.namespace}:{node_id}"
-            pipeline.set(key, f"{path[prefix_idx:]}:{offset}:{size}")
+            pipeline.set(
+                key, f"{path[prefix_idx:]}:{offset}:{size}", ex=MANIFEST_TTL_SECONDS
+            )
 
         for node_id in not_existing:
-            pipeline.set(f"{self.namespace}:{node_id}", DOES_NOT_EXIST)
+            pipeline.set(
+                f"{self.namespace}:{node_id}",
+                DOES_NOT_EXIST,
+                ex=MANIFEST_TTL_SECONDS,
+            )
 
         pipeline.execute()
 
@@ -155,9 +184,15 @@ def _set_cached_dynamic_fragments(
 
         pipeline = REDIS.pipeline()
         for node_id, fragment in fragments_d.items():
-            pipeline.set(f"{self.namespace}:{node_id}", fragment)
+            pipeline.set(
+                f"{self.namespace}:{node_id}", fragment, ex=MANIFEST_TTL_SECONDS
+            )
 
         for node_id in not_existing:
-            pipeline.set(f"{self.namespace}:{node_id}", DOES_NOT_EXIST)
+            pipeline.set(
+                f"{self.namespace}:{node_id}",
+                DOES_NOT_EXIST,
+                ex=MANIFEST_TTL_SECONDS,
+            )
 
         pipeline.execute()
diff --git a/pychunkedgraph/meshing/manifest/sharded.py b/pychunkedgraph/meshing/manifest/sharded.py
index 2576fcb2f..afb41189a 100644
--- a/pychunkedgraph/meshing/manifest/sharded.py
+++ b/pychunkedgraph/meshing/manifest/sharded.py
@@ -8,8 +8,9 @@
 from .utils import get_children_before_start_layer
 from ...graph import ChunkedGraph
 from ...graph.types import empty_1d
-from ...graph.utils.basetypes import NODE_ID
+from ...graph.basetypes import NODE_ID
 from ...graph.chunks import utils as chunk_utils
+from ..mesh_meta import MeshMeta
 
 
 def verified_manifest(
@@ -61,7 +62,7 @@ def speculative_manifest(
     from ..meshgen_utils import get_json_info
 
     if start_layer is None:
-        start_layer = cg.meta.custom_data.get("mesh", {}).get("max_layer", 2)
+        start_layer = MeshMeta(cg).max_layer
 
     start = time()
     bounding_box = chunk_utils.normalize_bounding_box(
@@ -89,7 +90,7 @@ def speculative_manifest(
 
     readers = CloudVolume(  # pylint: disable=no-member
         "graphene://https://localhost/segmentation/table/dummy",
-        mesh_dir=cg.meta.custom_data.get("mesh", {}).get("dir", "graphene_meshes"),
+        mesh_dir=MeshMeta(cg).dir,
         info=get_json_info(cg),
     ).mesh.readers
 
diff --git a/pychunkedgraph/meshing/manifest/utils.py b/pychunkedgraph/meshing/manifest/utils.py
index 67e600653..af5c58f38 100644
--- a/pychunkedgraph/meshing/manifest/utils.py
+++ b/pychunkedgraph/meshing/manifest/utils.py
@@ -14,9 +14,10 @@
 from .cache import ManifestCache
 from ..meshgen_utils import get_mesh_name
 from ..meshgen_utils import get_json_info
+from ..mesh_meta import MeshMeta
 from ...graph import ChunkedGraph
 from ...graph.types import empty_1d
-from ...graph.utils.basetypes import NODE_ID
+from ...graph.basetypes import NODE_ID
 from ...graph.utils import generic as misc_utils
 
 
@@ -40,7 +41,7 @@ def _get_children(cg, node_ids: Sequence[np.uint64], children_cache: Dict):
     if len(node_ids) == 0:
         return empty_1d.copy()
     node_ids = np.array(node_ids, dtype=NODE_ID)
-    mask = np.in1d(node_ids, np.fromiter(children_cache.keys(), dtype=NODE_ID))
+    mask = np.isin(node_ids, np.fromiter(children_cache.keys(), dtype=NODE_ID))
     children_d = cg.get_children(node_ids[~mask])
     children_cache.update(children_d)
 
@@ -105,8 +106,7 @@ def _get_dynamic_meshes(cg, node_ids: Sequence[np.uint64]) -> Tuple[Dict, List]:
     if len(node_ids) == 0:
         return result, not_existing
 
-    mesh_dir = cg.meta.custom_data.get("mesh", {}).get("dir", "graphene_meshes")
-    mesh_path = f"{cg.meta.data_source.WATERSHED}/{mesh_dir}/dynamic"
+    mesh_path = MeshMeta(cg).dynamic_path
     cf = CloudFiles(mesh_path)
     manifest_cache = ManifestCache(cg.graph_id, initial=False)
 
@@ -180,7 +180,7 @@ def segregate_node_ids(cg, node_ids):
     new = created by proofreading edit operations
     """
 
-    initial_ts = cg.meta.custom_data["mesh"]["initial_ts"]
+    initial_ts = MeshMeta(cg).initial_ts
     initial_mesh_dt = np.datetime64(datetime.fromtimestamp(initial_ts))
     node_ids_ts = cg.get_node_timestamps(node_ids)
     initial_mesh_mask = node_ids_ts < initial_mesh_dt
@@ -194,10 +194,17 @@ def get_mesh_paths(
     node_ids: Sequence[np.uint64],
     stop_layer: int = 2,
 ) -> Dict:
+    # Point the shard reader at initial_path: cloud-volume resolves shards at
+    # join(info["data_dir"], info["mesh"], "initial"), so reader_anchor splits
+    # initial_path into that (data_dir, mesh) pair — repoints a migrated bucket.
+    info = get_json_info(cg)
+    data_dir, mesh_dir = MeshMeta(cg).reader_anchor
+    info["data_dir"] = data_dir
+    info["mesh"] = mesh_dir
     shard_readers = CloudVolume(  # pylint: disable=no-member
         "graphene://https://localhost/segmentation/table/dummy",
-        mesh_dir=cg.meta.custom_data.get("mesh", {}).get("dir", "graphene_meshes"),
-        info=get_json_info(cg),
+        mesh_dir=mesh_dir,
+        info=info,
     ).mesh
 
     result = {}
diff --git a/pychunkedgraph/meshing/manifest/v2.py b/pychunkedgraph/meshing/manifest/v2.py
new file mode 100644
index 000000000..430f01c64
--- /dev/null
+++ b/pychunkedgraph/meshing/manifest/v2.py
@@ -0,0 +1,59 @@
+"""v2 graphene manifest: Accept-header negotiation + format assembly.
+
+Branch-agnostic, pure-stdlib. The v2 format keys ``fragments`` by absolute mesh
+bucket path so mesh location travels in the manifest itself; a client advertises
+support via ``Accept: application/x.cave;manifest_version=2``.
+"""
+
+ACCEPT_MEDIA_TYPE = "application/x.cave"
+MANIFEST_VERSION = 2  # highest manifest version this server produces
+
+
+def requested_manifest_version(accept_header, default: int = 1) -> int:
+    """Manifest version the client advertised via
+    ``Accept: application/x.cave;manifest_version=N``.
+
+    Parsed with stdlib — werkzeug's ``MIMEAccept`` mangles media-type params.
+    """
+    if not accept_header:
+        return default
+    for part in accept_header.split(","):
+        params = [p.strip() for p in part.split(";")]
+        if params[0].lower() != ACCEPT_MEDIA_TYPE:
+            continue
+        for param in params[1:]:
+            key, _, value = param.partition("=")
+            if key.strip().lower() == "manifest_version":
+                try:
+                    return int(value.strip())
+                except ValueError:
+                    return default
+    return default
+
+
+def to_v2_groups(node_ids, fragments, prepend_seg_ids):
+    """Group v1 fragments into ``(initial, dynamic)`` by the leading ``~`` marker.
+
+    Each fragment is emitted exactly as the v1 manifest would (seg id prepended
+    when requested); only the initial-vs-dynamic grouping is added. The raw ``~``
+    selects the group, matching v1's ``~<segid>:<fragment>``.
+    """
+    initial, dynamic = [], []
+    for node_id, frag in zip(node_ids, fragments):
+        out = f"~{node_id}:{frag}" if prepend_seg_ids else frag
+        (initial if frag.startswith("~") else dynamic).append(out)
+    return initial, dynamic
+
+
+def assemble(initial_path: str, dynamic_path: str, initial_frags, dynamic_frags) -> dict:
+    """v2 manifest dict: fragment lists grouped by absolute bucket path.
+
+    Empty groups are omitted; each bucket value is a sub-object so per-bucket
+    metadata can be added later without breaking the shape.
+    """
+    fragments = {}
+    if len(initial_frags):
+        fragments[initial_path] = {"fragments": list(initial_frags)}
+    if len(dynamic_frags):
+        fragments[dynamic_path] = {"fragments": list(dynamic_frags)}
+    return {"manifest_version": MANIFEST_VERSION, "fragments": fragments}
diff --git a/pychunkedgraph/meshing/mesh_analysis.py b/pychunkedgraph/meshing/mesh_analysis.py
index 97bb28f5b..8fb63e07b 100644
--- a/pychunkedgraph/meshing/mesh_analysis.py
+++ b/pychunkedgraph/meshing/mesh_analysis.py
@@ -1,7 +1,7 @@
 from cloudvolume import CloudVolume
 import numpy as np
-import os
 from pychunkedgraph.meshing import meshgen, meshgen_utils
+from pychunkedgraph.meshing.mesh_meta import MeshMeta
 
 
 def compute_centroid_by_range(vertices):
@@ -58,15 +58,7 @@ def compute_mesh_centroids_of_l2_ids(cg, l2_ids, flatten=False):
     :param l2_ids: Sequence[np.uint64]
     :return: Union[Dict[np.uint64, np.ndarray], [np.uint64], [np.uint64]]
     """
-    cv_sharded_mesh_dir = cg.meta.dataset_info["mesh"]
-    cv_unsharded_mesh_dir = cg.meta.dataset_info["mesh_metadata"][
-        "unsharded_mesh_dir"
-    ]
-    cv_unsharded_mesh_path = os.path.join(
-        cg.meta.data_source.WATERSHED,
-        cv_sharded_mesh_dir,
-        cv_unsharded_mesh_dir,
-    )
+    cv_sharded_mesh_dir = MeshMeta(cg).dir
     cv = CloudVolume(
         f"graphene://https://localhost/segmentation/table/dummy",
         mesh_dir=cv_sharded_mesh_dir,
diff --git a/pychunkedgraph/meshing/mesh_meta/README.md b/pychunkedgraph/meshing/mesh_meta/README.md
new file mode 100644
index 000000000..6451f5847
--- /dev/null
+++ b/pychunkedgraph/meshing/mesh_meta/README.md
@@ -0,0 +1,69 @@
+# mesh_meta
+
+Single, branch-agnostic home for **graphene mesh metadata**. Every value comes
+from **`custom_data["mesh"]`, the single source of truth**. `MeshMeta(cg)` wraps a
+ChunkedGraph and exposes, as properties, the mesh dir/bucket locations and the
+per-dataset mesh params used by the v2 manifest and by remeshing; multi-resolution
+(LOD) mesh meta will live here too as it lands, so this is the module to grow
+rather than scattering mesh-meta reads across the code.
+
+## Why it exists
+
+Supervoxels (watershed) can move to a cheap, egress-waived bucket while meshes
+stay put. The v2 graphene manifest therefore carries **absolute mesh bucket
+paths** rather than paths composed relative to the watershed `data_dir`, so a
+client fetches meshes straight from wherever they live. This module is the one
+place that decides where meshes live, and what the mesh params are, for a given
+ChunkedGraph.
+
+## Two mesh locations
+
+- **initial** — the sharded meshes produced at ingest. **Shared / graph-independent**:
+  several ChunkedGraphs (e.g. copies) point at the *same* initial meshes, so this
+  location is never namespaced by graph id.
+- **dynamic** — the unsharded meshes produced by proofreading edits.
+  **Per-graph**: each graph keeps its own, so on the pcgv3 line it defaults to a
+  graph-id-derived dir (`dynamic_<graph_id>`) and stays isolated even while
+  sharing initial meshes; on the pcgv2 line it is a plain subdir.
+
+## Configuration
+
+All read from `custom_data["mesh"]`:
+
+- `dir` — the mesh root under the watershed (default `graphene_meshes`).
+- `initial_mesh_dir` — the initial location (default `initial`).
+- `dynamic_mesh_dir` — the dynamic location (default `dynamic`; pcgv3 setup fills
+  `dynamic_<graph_id>`).
+- `max_layer`, `mip`, `max_error` — mesh params (start layer for manifests, and
+  the mip / max-error used by remeshing).
+- `initial_ts` — the ingest timestamp boundary that splits initial (sharded) node
+  ids from proofread (dynamic) ones.
+
+Resolution rule for the two location values: a value containing a cloud scheme
+(`gs://`, `s3://`, …) is treated as an **absolute** bucket and used as-is;
+otherwise it is joined under `<watershed>/<dir>`. This lets a dataset move
+initial and/or dynamic meshes to different buckets without touching anything else.
+
+Branch-agnostic on purpose: this module reads only `custom_data["mesh"]` and the
+watershed path — accessors identical on the pcgv2 and pcgv3 lines — so it
+cherry-picks clean between them. How that config gets *populated* (e.g. pcgv3's
+setup-time `MeshConfig` from the dataset yaml) is branch-specific and lives
+upstream; this module only reads the result.
+
+## API
+
+`MeshMeta(cg)` exposes these properties:
+
+- `initial_path` → absolute dir of the initial (sharded) meshes.
+- `dynamic_path` → absolute dir of the dynamic (unsharded) meshes.
+- `dir` → the sharded mesh dir name under the watershed.
+- `max_layer`, `mip`, `max_error`, `initial_ts` → the mesh params.
+- `needs_v2` → true when either location falls **outside** the watershed dir,
+  i.e. meshes are not co-located with the watershed. Old clients compose paths
+  relative to the watershed, so they cannot reach such meshes: the v2 manifest
+  serves them correctly, and the v1 path must fail loudly rather than return
+  unreachable paths.
+- `reader_anchor` → the `(data_dir, mesh)` pair to hand a CloudVolume mesh reader.
+  The reader appends `initial/` internally, so it must be anchored at the
+  **parent** of the initial dir; this yields shard byte-ranges from the right
+  bucket even for a migrated graph.
diff --git a/pychunkedgraph/meshing/mesh_meta/__init__.py b/pychunkedgraph/meshing/mesh_meta/__init__.py
new file mode 100644
index 000000000..33b6c72fd
--- /dev/null
+++ b/pychunkedgraph/meshing/mesh_meta/__init__.py
@@ -0,0 +1,5 @@
+"""Branch-agnostic graphene mesh metadata. See README.md."""
+
+from .core import MeshMeta
+
+__all__ = ["MeshMeta"]
diff --git a/pychunkedgraph/meshing/mesh_meta/core.py b/pychunkedgraph/meshing/mesh_meta/core.py
new file mode 100644
index 000000000..ae396ace9
--- /dev/null
+++ b/pychunkedgraph/meshing/mesh_meta/core.py
@@ -0,0 +1,71 @@
+"""MeshMeta — mesh dir locations and per-dataset params for a ChunkedGraph.
+
+``custom_data["mesh"]`` is the single source of truth. See README.md.
+"""
+
+import posixpath
+
+_DEFAULT_MESH_DIR = "graphene_meshes"
+
+
+def _resolve(root: str, value: str) -> str:
+    """Absolute ``value`` (a cloud URL) as-is, else joined under ``root``."""
+    return value.rstrip("/") if "://" in value else posixpath.join(root, value)
+
+
+class MeshMeta:
+    """Mesh metadata for a ChunkedGraph, read from ``custom_data["mesh"]``."""
+
+    def __init__(self, cg):
+        self._mesh = cg.meta.custom_data.get("mesh", {})
+        self._watershed = cg.meta.data_source.WATERSHED
+
+    @property
+    def dir(self) -> str:
+        """Sharded mesh dir name under the watershed."""
+        return self._mesh.get("dir", _DEFAULT_MESH_DIR)
+
+    @property
+    def _root(self) -> str:
+        return posixpath.join(self._watershed, self.dir)
+
+    @property
+    def initial_path(self) -> str:
+        """Absolute dir of the shared initial (sharded) meshes."""
+        return _resolve(self._root, self._mesh.get("initial_mesh_dir", "initial"))
+
+    @property
+    def dynamic_path(self) -> str:
+        """Absolute dir of the per-graph dynamic (unsharded) meshes."""
+        return _resolve(self._root, self._mesh.get("dynamic_mesh_dir", "dynamic"))
+
+    @property
+    def needs_v2(self) -> bool:
+        """True when meshes live outside the watershed dir (old clients can't reach)."""
+        root = self._root
+        return not (
+            self.initial_path.startswith(root)
+            and self.dynamic_path.startswith(root)
+        )
+
+    @property
+    def reader_anchor(self):
+        """``(data_dir, mesh)`` so a CloudVolume that appends ``initial/`` lands at
+        ``initial_path`` — points the shard reader at a migrated bucket."""
+        return posixpath.split(posixpath.dirname(self.initial_path))
+
+    @property
+    def max_layer(self) -> int:
+        return self._mesh.get("max_layer", 2)
+
+    @property
+    def mip(self) -> int:
+        return self._mesh["mip"]
+
+    @property
+    def max_error(self):
+        return self._mesh["max_error"]
+
+    @property
+    def initial_ts(self):
+        return self._mesh["initial_ts"]
diff --git a/pychunkedgraph/meshing/meshgen_utils.py b/pychunkedgraph/meshing/meshgen_utils.py
index 711c09322..3865ba419 100644
--- a/pychunkedgraph/meshing/meshgen_utils.py
+++ b/pychunkedgraph/meshing/meshgen_utils.py
@@ -14,6 +14,7 @@
 
 from pychunkedgraph.graph.utils.basetypes import NODE_ID  # noqa
 from ..graph.types import empty_1d
+from .mesh_meta import MeshMeta
 
 
 def str_to_slice(slice_str: str):
@@ -145,7 +146,14 @@ def get_json_info(cg):
     dataset_info = cg.meta.dataset_info
     dummy_app_info = {"app": {"supported_api_versions": [0, 1]}}
     info = {**dataset_info, **dummy_app_info}
-    info["mesh"] = cg.meta.custom_data.get("mesh", {}).get("dir", "graphene_meshes")
+    mm = MeshMeta(cg)
+    info["mesh"] = mm.dir
+    # Absolute initial/dynamic bucket locations (the v2 manifest's bucket keys)
+    # so cloud-volume's manifest-bypass reads resolve to the real buckets.
+    mesh_metadata = dict(info.get("mesh_metadata", {}))
+    mesh_metadata["initial_mesh_path"] = mm.initial_path
+    mesh_metadata["dynamic_mesh_path"] = mm.dynamic_path
+    info["mesh_metadata"] = mesh_metadata
     info_str = dumps(info)
     return loads(info_str)
 
diff --git a/pychunkedgraph/tests/meshing/test_manifest_v2.py b/pychunkedgraph/tests/meshing/test_manifest_v2.py
new file mode 100644
index 000000000..b3337e267
--- /dev/null
+++ b/pychunkedgraph/tests/meshing/test_manifest_v2.py
@@ -0,0 +1,28 @@
+"""Tests for pychunkedgraph.meshing.manifest.v2."""
+
+from pychunkedgraph.meshing.manifest.v2 import assemble, to_v2_groups
+
+
+class TestToV2Groups:
+    def test_prepend_matches_v1_grouped_by_type(self):
+        """With prepend_seg_ids each fragment is v1's ``~<segid>:<frag>``; grouping
+        is by the raw ~ (initial/sharded) vs no-~ (dynamic/whole-file)."""
+        node_ids = [386, 529]
+        frags = ["~5/774017-0.shard:74809813:4532", "529:0:90112-98304"]
+        ini, dyn = to_v2_groups(node_ids, frags, prepend_seg_ids=True)
+        assert ini == ["~386:~5/774017-0.shard:74809813:4532"]
+        assert dyn == ["~529:529:0:90112-98304"]
+
+    def test_no_prepend_keeps_raw(self):
+        ini, dyn = to_v2_groups(
+            [386, 529], ["~a.shard:1:2", "529:0:bbox"], prepend_seg_ids=False
+        )
+        assert ini == ["~a.shard:1:2"]
+        assert dyn == ["529:0:bbox"]
+
+    def test_assemble_groups_under_bucket(self):
+        resp = assemble(
+            "gs://b/initial", "gs://b/dynamic", ["~386:~a.shard:1:2"], ["~529:529:0:y"]
+        )
+        assert resp["fragments"]["gs://b/initial"]["fragments"] == ["~386:~a.shard:1:2"]
+        assert resp["manifest_version"] == 2
diff --git a/pychunkedgraph/tests/meshing/test_mesh_meta.py b/pychunkedgraph/tests/meshing/test_mesh_meta.py
new file mode 100644
index 000000000..46a76c762
--- /dev/null
+++ b/pychunkedgraph/tests/meshing/test_mesh_meta.py
@@ -0,0 +1,35 @@
+"""Tests for pychunkedgraph.meshing.mesh_meta.MeshMeta."""
+
+import posixpath
+from types import SimpleNamespace
+
+from pychunkedgraph.meshing.mesh_meta import MeshMeta
+
+
+def _cg(watershed, mesh):
+    meta = SimpleNamespace(
+        custom_data={"mesh": mesh},
+        data_source=SimpleNamespace(WATERSHED=watershed),
+    )
+    return SimpleNamespace(meta=meta)
+
+
+class TestReaderAnchor:
+    """reader_anchor must recompose to initial_path via cloud-volume's hardcoded
+    ``initial`` subdir, so the verified shard reader reads the real bucket."""
+
+    def test_colocated_anchor_is_watershed_dir(self):
+        mm = MeshMeta(_cg("gs://ws/seg", {"dir": "graphene_meshes"}))
+        assert mm.reader_anchor == ("gs://ws/seg", "graphene_meshes")
+        assert posixpath.join(*mm.reader_anchor, "initial") == mm.initial_path
+        assert mm.needs_v2 is False
+
+    def test_migrated_initial_anchor_recomposes(self):
+        mesh = {
+            "dir": "graphene_meshes",
+            "initial_mesh_dir": "gs://old/graphene_meshes/initial",
+        }
+        mm = MeshMeta(_cg("gs://cheap/seg", mesh))
+        assert mm.initial_path == "gs://old/graphene_meshes/initial"
+        assert posixpath.join(*mm.reader_anchor, "initial") == mm.initial_path
+        assert mm.needs_v2 is True
diff --git a/workers/mesh_worker.py b/workers/mesh_worker.py
index b8f1e0024..788251b45 100644
--- a/workers/mesh_worker.py
+++ b/workers/mesh_worker.py
@@ -3,7 +3,6 @@
 import gc
 import pickle
 import logging
-from os import path
 from os import getenv
 
 import numpy as np
@@ -12,6 +11,7 @@
 from pychunkedgraph.graph import ChunkedGraph
 from pychunkedgraph.graph.utils import basetypes
 from pychunkedgraph.meshing import meshgen
+from pychunkedgraph.meshing.mesh_meta import MeshMeta
 
 
 PCG_CACHE = {}
@@ -41,21 +41,16 @@ def callback(payload):
     )
 
     try:
-        mesh_meta = cg.meta.custom_data["mesh"]
-        mesh_dir = mesh_meta["dir"]
-        layer = mesh_meta["max_layer"]
-        mip = mesh_meta["mip"]
-        err = mesh_meta["max_error"]
-        cv_unsharded_mesh_dir = mesh_meta.get("dynamic_mesh_dir", "dynamic")
+        mm = MeshMeta(cg)
+        layer = mm.max_layer
+        mip = mm.mip
+        err = mm.max_error
+        mesh_dir = mm.dir
+        mesh_path = mm.dynamic_path
     except KeyError:
         logging.warning(f"No metadata found for {cg.graph_id}; ignoring...")
         return
 
-    mesh_path = path.join(
-        cg.meta.data_source.WATERSHED, mesh_dir, cv_unsharded_mesh_dir
-    )
-
-
     logging.log(INFO_HIGH, f"remeshing {l2ids}; graph {table_id} operation {op_id}.")
     meshgen.remeshing(
         cg,