diff --git a/changes/4192.feature.md b/changes/4192.feature.md
new file mode 100644
index 0000000000..11779267b2
--- /dev/null
+++ b/changes/4192.feature.md
@@ -0,0 +1,9 @@
+Added core support for URL pipelines (https://github.com/jbms/url-pipeline):
+`|`-chained URLs that address zarr data through nested storage layers, e.g.
+`s3://bucket/data.zip|zip:|zarr3:`. This PR adds the parser, the single-method
+`zarr.abc.url_pipeline.URLPipelineAdapter` interface, and the
+`zarr.url_adapters` entry-point group through which third-party packages
+(e.g. Icechunk) register adapters for their own schemes. Adapters for a scheme
+are loaded lazily and individually; URLs without a `|` separator (and without
+a registered root scheme) are handled exactly as before. Builtin adapters
+(`zip:`, `zarr2:`/`zarr3:`) follow in separate pull requests.
diff --git a/docs/api/zarr/abc/url_pipeline.md b/docs/api/zarr/abc/url_pipeline.md
new file mode 100644
index 0000000000..2606d4d5d4
--- /dev/null
+++ b/docs/api/zarr/abc/url_pipeline.md
@@ -0,0 +1,5 @@
+---
+title: url_pipeline
+---
+
+::: zarr.abc.url_pipeline
diff --git a/mkdocs.yml b/mkdocs.yml
index 87aaf23430..1f683dd3bc 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -40,6 +40,7 @@ nav:
         - '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.abc.metadata</code>': api/zarr/abc/metadata.md
         - '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.abc.numcodec</code>': api/zarr/abc/numcodec.md
         - '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.abc.store</code>': api/zarr/abc/store.md
+        - '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.abc.url_pipeline</code>': api/zarr/abc/url_pipeline.md
       - '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.api</code>':
         - api/zarr/api/index.md
         - '<code class="doc-symbol doc-symbol-toc doc-symbol-module"></code> <code>zarr.api.asynchronous</code>': api/zarr/api/asynchronous.md
diff --git a/src/zarr/abc/url_pipeline.py b/src/zarr/abc/url_pipeline.py
new file mode 100644
index 0000000000..eb4069e3ad
--- /dev/null
+++ b/src/zarr/abc/url_pipeline.py
@@ -0,0 +1,185 @@
+"""
+Abstract base class and data model for URL pipeline adapters.
+
+A URL pipeline is a `|`-separated chain of sub-URLs, read outer-to-inner,
+as specified by https://github.com/jbms/url-pipeline. The first sub-URL (the
+*root*) locates a resource using a conventional URL, and each subsequent
+sub-URL names an *adapter* that reinterprets everything to its left:
+
+    s3://bucket/data.zip|zip:path/inside|zarr3:
+
+Third-party packages provide adapters by subclassing
+[`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter] and
+registering the class under the `zarr.url_adapters` entry-point group,
+using the URL scheme as the entry-point name.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
+    from zarr.abc.store import Store
+    from zarr.core.common import AccessModeLiteral
+
+__all__ = [
+    "AdapterResolution",
+    "PipelineContext",
+    "PipelineSegment",
+    "URLPipelineAdapter",
+]
+
+
+@dataclass(frozen=True)
+class PipelineSegment:
+    """
+    One `|`-delimited sub-URL of a URL pipeline.
+
+    Attributes
+    ----------
+    scheme : str
+        The lowercased URL scheme. Empty string only for a schemeless root
+        (a bare local path).
+    body : str
+        The text after `scheme:` and before any `?`. Interpretation is
+        scheme-defined; it is **not** URL-normalized, so case-significant
+        content (e.g. icechunk snapshot IDs) is preserved.
+    query : str | None
+        The raw query string after `?`, or None. Interpretation is
+        scheme-defined.
+    raw : str
+        The exact original sub-URL text, preserved for lossless
+        reconstruction of the pipeline.
+    """
+
+    scheme: str
+    body: str
+    query: str | None
+    raw: str
+
+    def __str__(self) -> str:
+        return self.raw
+
+
+@dataclass(frozen=True)
+class AdapterResolution:
+    """
+    The result of resolving a URL pipeline (or a prefix of one).
+
+    Attributes
+    ----------
+    store : Store
+        The resolved store.
+    path : str
+        Residual path *within* the store that the pipeline addresses
+        (e.g. `"path/to/node"` for `...|icechunk://tag.v1/path/to/node`).
+        Empty string when the pipeline addresses the store root.
+    """
+
+    store: Store
+    path: str = ""
+
+
+@dataclass(frozen=True)
+class PipelineContext:
+    """
+    Context handed to a [`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter]
+    describing the pipeline to the left of its segment.
+
+    Attributes
+    ----------
+    preceding : tuple[PipelineSegment, ...]
+        The parsed sub-URLs to the left of the adapter's segment, outer to
+        inner. Empty when the adapter's segment is the pipeline root.
+    mode : AccessModeLiteral | None
+        The access mode requested by the caller (e.g. `zarr.open(mode=...)`),
+        or None when unspecified. Adapters for read-only resources should
+        raise for unambiguous write modes (`"w"`, `"w-"`, `"r+"`) and
+        open read-only otherwise. `"a"` (the `zarr.open` default) means
+        open-or-create: read-only adapters serve the "open" half, and any
+        subsequent write fails at the store level.
+    read_only : bool
+        True when the caller requires a read-only store (`mode == "r"`).
+        Adapters must construct their store read-only when this is set;
+        when it is False, they may construct a writable store if the
+        underlying resource supports writing.
+    storage_options : dict[str, Any] | None
+        Options passed by the caller. By convention these configure the
+        *root* sub-URL (e.g. fsspec options), but adapters may consume
+        adapter-specific keys.
+    """
+
+    preceding: tuple[PipelineSegment, ...]
+    mode: AccessModeLiteral | None
+    read_only: bool
+    storage_options: dict[str, Any] | None
+    _resolver: Callable[[tuple[PipelineSegment, ...]], Awaitable[AdapterResolution]] = field(
+        repr=False
+    )
+
+    @property
+    def preceding_url(self) -> str:
+        """
+        The pipeline to the left of this segment, reconstructed exactly.
+
+        An adapter that consumes this string instead of calling
+        [`resolve_preceding`][zarr.abc.url_pipeline.PipelineContext.resolve_preceding]
+        takes ownership of the *entire* preceding pipeline: it must
+        validate every preceding segment itself and raise
+        [`URLPipelineError`][zarr.errors.URLPipelineError] for segments it
+        does not understand, so that no segment is ever silently ignored.
+        """
+        return "|".join(segment.raw for segment in self.preceding)
+
+    async def resolve_preceding(self) -> AdapterResolution:
+        """
+        Resolve the preceding pipeline into a store.
+
+        This is the entry point for *wrapper* adapters (e.g. `zip:`) that
+        operate on the resource produced by the segments to their left. It
+        composes with any preceding adapters, because each segment is
+        resolved by its own adapter. Adapters backed by their own I/O
+        machinery (e.g. `icechunk:`) may instead consume
+        [`preceding_url`][zarr.abc.url_pipeline.PipelineContext.preceding_url]
+        and never materialize the intermediate store — subject to the
+        ownership contract documented there.
+        """
+        return await self._resolver(self.preceding)
+
+
+class URLPipelineAdapter(ABC):
+    """
+    Handler for one URL pipeline scheme.
+
+    Subclasses implement a single classmethod,
+    [`open_pipeline_segment`][zarr.abc.url_pipeline.URLPipelineAdapter.open_pipeline_segment],
+    and are registered under the `zarr.url_adapters` entry-point group with
+    the URL scheme as the entry-point name:
+
+        [project.entry-points."zarr.url_adapters"]
+        myscheme = "mypackage.zarr_adapter:MyAdapter"
+
+    An adapter is used in two positions:
+
+    - as an *adapter segment*: `s3://bucket/repo|icechunk://tag.v1` — the
+      context carries the preceding sub-URLs;
+    - as a *root scheme*: `gh://org/repo` — `context.preceding` is empty.
+    """
+
+    @classmethod
+    @abstractmethod
+    async def open_pipeline_segment(
+        cls, segment: PipelineSegment, context: PipelineContext
+    ) -> AdapterResolution:
+        """
+        Resolve `segment` (in the context of the pipeline to its left)
+        into a store and an optional residual path within that store.
+
+        The returned store must already be open and must honor
+        `context.read_only`.
+        """
+        ...
diff --git a/src/zarr/errors.py b/src/zarr/errors.py
index 781bebe534..de04a5bdaa 100644
--- a/src/zarr/errors.py
+++ b/src/zarr/errors.py
@@ -12,6 +12,7 @@
     "MetadataValidationError",
     "NegativeStepError",
     "NodeTypeValidationError",
+    "URLPipelineError",
     "UnstableSpecificationWarning",
     "VindexInvalidSelectionError",
     "ZarrDeprecationWarning",
@@ -100,6 +101,12 @@ class UnknownCodecError(BaseZarrError):
     """
 
 
+class URLPipelineError(BaseZarrError):
+    """
+    Raised when a URL pipeline cannot be parsed or resolved.
+    """
+
+
 class NodeTypeValidationError(MetadataValidationError):
     """
     Specialized exception when the node_type of the metadata document is incorrect.
diff --git a/src/zarr/registry.py b/src/zarr/registry.py
index 48f60fabd7..67c72b6fa5 100644
--- a/src/zarr/registry.py
+++ b/src/zarr/registry.py
@@ -7,7 +7,7 @@
 
 from zarr.core.config import BadConfigError, config
 from zarr.core.dtype import data_type_registry
-from zarr.errors import ZarrUserWarning
+from zarr.errors import URLPipelineError, ZarrUserWarning
 
 if TYPE_CHECKING:
     from importlib.metadata import EntryPoint
@@ -21,6 +21,7 @@
         CodecPipeline,
     )
     from zarr.abc.numcodec import Numcodec
+    from zarr.abc.url_pipeline import URLPipelineAdapter
     from zarr.core.buffer import Buffer, NDBuffer
     from zarr.core.chunk_key_encodings import ChunkKeyEncoding
     from zarr.core.common import JSON
@@ -32,11 +33,14 @@
     "get_codec_class",
     "get_ndbuffer_class",
     "get_pipeline_class",
+    "get_url_adapter",
+    "list_url_adapter_schemes",
     "register_buffer",
     "register_chunk_key_encoding",
     "register_codec",
     "register_ndbuffer",
     "register_pipeline",
+    "register_url_adapter",
 ]
 
 
@@ -62,6 +66,7 @@ def register(self, cls: type[T], qualname: str | None = None) -> None:
 _buffer_registry: Registry[Buffer] = Registry()
 _ndbuffer_registry: Registry[NDBuffer] = Registry()
 _chunk_key_encoding_registry: Registry[ChunkKeyEncoding] = Registry()
+_url_adapter_registry: Registry[URLPipelineAdapter] = Registry()
 
 """
 The registry module is responsible for managing implementations of codecs,
@@ -108,6 +113,8 @@ def _collect_entrypoints() -> list[Registry[Any]]:
         entry_points.select(group="zarr", name="chunk_key_encoding")
     )
 
+    _url_adapter_registry.lazy_load_list.extend(entry_points.select(group="zarr.url_adapters"))
+
     _pipeline_registry.lazy_load_list.extend(entry_points.select(group="zarr.codec_pipeline"))
     _pipeline_registry.lazy_load_list.extend(
         entry_points.select(group="zarr", name="codec_pipeline")
@@ -124,6 +131,7 @@ def _collect_entrypoints() -> list[Registry[Any]]:
         _buffer_registry,
         _ndbuffer_registry,
         _chunk_key_encoding_registry,
+        _url_adapter_registry,
     ]
 
 
@@ -303,6 +311,51 @@ def get_chunk_key_encoding_class(key: str) -> type[ChunkKeyEncoding]:
     return _chunk_key_encoding_registry[key]
 
 
+def register_url_adapter(scheme: str, cls: type[URLPipelineAdapter]) -> None:
+    """
+    Register a [`URLPipelineAdapter`][zarr.abc.url_pipeline.URLPipelineAdapter]
+    class for a URL scheme.
+    """
+    _url_adapter_registry.register(cls, scheme.lower())
+
+
+def list_url_adapter_schemes() -> set[str]:
+    """
+    The set of URL schemes with a registered URL pipeline adapter.
+
+    Includes adapters advertised via not-yet-loaded `zarr.url_adapters`
+    entry points; consulting this does not import any adapter code.
+    """
+    return set(_url_adapter_registry) | {e.name for e in _url_adapter_registry.lazy_load_list}
+
+
+def get_url_adapter(scheme: str) -> type[URLPipelineAdapter]:
+    """
+    Get the URL pipeline adapter class registered for `scheme`.
+
+    Loads pending `zarr.url_adapters` entry points for this scheme only, so
+    resolving one scheme never imports other providers' packages.
+    """
+    key = scheme.lower()
+    if key not in _url_adapter_registry:
+        remaining = []
+        for entry_point in _url_adapter_registry.lazy_load_list:
+            if entry_point.name == key:
+                _url_adapter_registry.register(entry_point.load(), qualname=key)
+            else:
+                remaining.append(entry_point)
+        _url_adapter_registry.lazy_load_list[:] = remaining
+    try:
+        return _url_adapter_registry[key]
+    except KeyError:
+        registered = sorted(list_url_adapter_schemes())
+        raise URLPipelineError(
+            f"no URL pipeline adapter is registered for scheme {scheme!r}. "
+            f"Registered schemes: {registered}. Adapters are provided by "
+            "packages via the 'zarr.url_adapters' entry-point group."
+        ) from None
+
+
 _collect_entrypoints()
 
 
diff --git a/src/zarr/storage/_common.py b/src/zarr/storage/_common.py
index 7e9c035c69..b8ba5824d5 100644
--- a/src/zarr/storage/_common.py
+++ b/src/zarr/storage/_common.py
@@ -21,9 +21,15 @@
     AccessModeLiteral,
     ZarrFormat,
 )
-from zarr.errors import ContainsArrayAndGroupError, ContainsArrayError, ContainsGroupError
+from zarr.errors import (
+    ContainsArrayAndGroupError,
+    ContainsArrayError,
+    ContainsGroupError,
+    URLPipelineError,
+)
 from zarr.storage._local import LocalStore
 from zarr.storage._memory import ManagedMemoryStore, MemoryStore
+from zarr.storage._url_pipeline import is_url_pipeline, resolve_pipeline
 from zarr.storage._utils import _join_paths, normalize_path, parse_store_url
 
 _has_fsspec = importlib.util.find_spec("fsspec")
@@ -348,6 +354,15 @@ async def make_store(
     """
     from zarr.storage._fsspec import FsspecStore  # circular import
 
+    if isinstance(store_like, str) and is_url_pipeline(store_like):
+        result = await resolve_pipeline(store_like, mode=mode, storage_options=storage_options)
+        if result.path:
+            raise URLPipelineError(
+                f"the URL pipeline {store_like!r} resolves to a path inside a store; "
+                "use zarr.open() or make_store_path() instead of make_store()"
+            )
+        return result.store
+
     # Parse URL early so we can reuse the result for both validation and routing
     parsed = parse_store_url(store_like) if isinstance(store_like, str) else None
 
@@ -453,6 +468,17 @@ async def make_store_path(
     """
     path_normalized = normalize_path(path)
 
+    if isinstance(store_like, str) and is_url_pipeline(store_like):
+        result = await resolve_pipeline(store_like, mode=mode, storage_options=storage_options)
+        combined_path = _join_paths([normalize_path(result.path), path_normalized])
+        # mode "a" (the zarr.open default) means open-or-create; when the
+        # pipeline resolved to a read-only resource, honor the "open" half
+        # rather than failing outright. Writes still fail at the store level.
+        open_mode: AccessModeLiteral | None = (
+            "r" if (mode == "a" and result.store.read_only) else mode
+        )
+        return await StorePath.open(result.store, path=combined_path, mode=open_mode)
+
     if isinstance(store_like, StorePath):
         # Already a StorePath
         if storage_options:
diff --git a/src/zarr/storage/_url_pipeline.py b/src/zarr/storage/_url_pipeline.py
new file mode 100644
index 0000000000..f3cb689828
--- /dev/null
+++ b/src/zarr/storage/_url_pipeline.py
@@ -0,0 +1,182 @@
+"""
+Parsing and resolution of URL pipelines (https://github.com/jbms/url-pipeline).
+
+Importing this module is cheap and has no side effects: third-party adapters
+(registered through the `zarr.url_adapters` entry-point group) are loaded
+only when a pipeline URL naming their scheme is actually resolved.
+
+As a zarr-python extension to the specification (which requires the root
+sub-URL of an absolute pipeline to carry a scheme), the root sub-URL may be
+a schemeless local filesystem path, e.g. `data/example.zip|zip:`. Such
+pipelines are not portable to other URL pipeline implementations; portable
+pipelines should spell the root as a `file:` URL.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING, Any
+
+from zarr.abc.url_pipeline import (
+    AdapterResolution,
+    PipelineContext,
+    PipelineSegment,
+)
+from zarr.errors import URLPipelineError
+from zarr.registry import get_url_adapter, list_url_adapter_schemes
+from zarr.storage._utils import parse_store_url
+
+if TYPE_CHECKING:
+    from zarr.core.common import AccessModeLiteral
+
+__all__ = ["is_url_pipeline", "parse_pipeline", "resolve_pipeline"]
+
+# Adapter scheme per RFC 3986 plus "." to permit vendor-prefixed
+# nonstandard schemes (e.g. "earthmover.myscheme").
+_SCHEME_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9+.\-]*$")
+
+
+def _root_scheme(root_sub_url: str) -> str:
+    """
+    Detect the scheme of the root sub-URL.
+
+    Uses `parse_store_url` (which knows Windows drive letters are not
+    schemes), falling back to plain scheme extraction when `urlparse`
+    rejects an exotic authority (e.g. bracketed non-IP text) — the pipeline
+    grammar constrains only the scheme, not the authority.
+    """
+    try:
+        return parse_store_url(root_sub_url).scheme.lower()
+    except ValueError:
+        match = re.match(r"([a-zA-Z][a-zA-Z0-9+.\-]*):", root_sub_url)
+        return match.group(1).lower() if match else ""
+
+
+def _split_query(sub_url: str) -> tuple[str, str | None]:
+    """Split a sub-URL on the first `?`. Fragments are not supported."""
+    if "#" in sub_url:
+        raise URLPipelineError(
+            f"URL pipeline sub-URLs do not support fragments: {sub_url!r}. "
+            "Percent-encode '#' as '%23' if it is part of the path."
+        )
+    body, sep, query = sub_url.partition("?")
+    return body, query if sep else None
+
+
+def parse_pipeline(url: str) -> tuple[PipelineSegment, ...]:
+    """
+    Parse a URL pipeline into its `|`-delimited segments.
+
+    The first segment is the *root* sub-URL; its scheme is detected with the
+    same rules as ordinary store URLs (Windows drive letters are not
+    schemes). Subsequent segments are *adapter* sub-URLs of the form
+    `scheme:body` where the trailing colon is optional when the body is
+    empty (`zip` is equivalent to `zip:`).
+
+    A schemeless root (a bare local path) is accepted as a zarr-python
+    extension to the specification; see the module docstring.
+
+    Segment text is preserved verbatim (no case or percent-encoding
+    normalization) except that schemes are lowercased.
+    """
+    parts = url.split("|")
+    if any(not part for part in parts):
+        raise URLPipelineError(f"URL pipeline contains an empty sub-URL: {url!r}")
+
+    segments: list[PipelineSegment] = []
+    for index, part in enumerate(parts):
+        body_and_scheme, query = _split_query(part)
+        if index == 0:
+            scheme = _root_scheme(body_and_scheme)
+            body = body_and_scheme
+            if scheme and body_and_scheme.lower().startswith(f"{scheme}:"):
+                body = body_and_scheme[len(scheme) + 1 :]
+            segments.append(PipelineSegment(scheme=scheme, body=body, query=query, raw=part))
+        else:
+            scheme, _, body = body_and_scheme.partition(":")
+            scheme = scheme.lower()
+            if not _SCHEME_RE.match(scheme):
+                raise URLPipelineError(
+                    f"invalid adapter scheme {scheme!r} in pipeline segment {part!r}"
+                )
+            segments.append(PipelineSegment(scheme=scheme, body=body, query=query, raw=part))
+    return tuple(segments)
+
+
+def is_url_pipeline(url: str) -> bool:
+    """
+    Whether `url` should be routed through the URL pipeline machinery.
+
+    True when the URL contains a `|` separator, or when its scheme has a
+    registered URL pipeline adapter (a *root adapter* such as `gh:`).
+    The registry check inspects entry-point names only — no adapter code is
+    imported here.
+    """
+    if "|" in url:
+        return True
+    scheme = _root_scheme(url)
+    return bool(scheme) and scheme in list_url_adapter_schemes()
+
+
+async def resolve_pipeline(
+    url: str,
+    *,
+    mode: AccessModeLiteral | None = None,
+    storage_options: dict[str, Any] | None = None,
+) -> AdapterResolution:
+    """
+    Resolve a URL pipeline into a store and a residual path.
+
+    Parameters
+    ----------
+    url : str
+        A URL pipeline, e.g. `"s3://bucket/data.zip|zip:|zarr3:"`.
+    mode : AccessModeLiteral | None
+        The caller's access mode. `"r"` requires adapters to construct
+        read-only stores.
+    storage_options : dict | None
+        Options forwarded to the root sub-URL's store (and visible to
+        adapters via the context).
+    """
+    segments = parse_pipeline(url)
+    if len(segments) == 1 and segments[0].scheme not in list_url_adapter_schemes():
+        raise URLPipelineError(
+            f"{url!r} is not a URL pipeline: it has no '|' separator and no "
+            f"URL pipeline adapter is registered for scheme {segments[0].scheme!r}"
+        )
+    return await _resolve(segments, mode=mode, storage_options=storage_options)
+
+
+async def _resolve(
+    segments: tuple[PipelineSegment, ...],
+    *,
+    mode: AccessModeLiteral | None,
+    storage_options: dict[str, Any] | None,
+) -> AdapterResolution:
+    if len(segments) == 1 and segments[0].scheme not in list_url_adapter_schemes():
+        # Base case: a plain root URL. Delegate to the existing StoreLike
+        # machinery (local paths, memory://, fsspec fallthrough) unchanged.
+        from zarr.storage._common import make_store  # circular import
+
+        store = await make_store(segments[0].raw, mode=mode, storage_options=storage_options)
+        return AdapterResolution(store=store)
+
+    *preceding, last = segments
+    adapter_cls = get_url_adapter(last.scheme)
+
+    async def _resolver(preceding: tuple[PipelineSegment, ...]) -> AdapterResolution:
+        if not preceding:
+            raise URLPipelineError(
+                f"adapter {last.scheme!r} was used as a pipeline root but "
+                "requires a preceding sub-URL"
+            )
+        return await _resolve(preceding, mode=mode, storage_options=storage_options)
+
+    context = PipelineContext(
+        preceding=tuple(preceding),
+        mode=mode,
+        read_only=mode == "r",
+        storage_options=storage_options,
+        _resolver=_resolver,
+    )
+    return await adapter_cls.open_pipeline_segment(last, context)
diff --git a/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt b/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt
index 7eb0eb7c86..c7bb830901 100644
--- a/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt
+++ b/tests/package_with_entrypoint-0.1.dist-info/entry_points.txt
@@ -13,4 +13,6 @@ another_ndbuffer = package_with_entrypoint:TestEntrypointGroup.NDBuffer
 [zarr.codec_pipeline]
 another_pipeline = package_with_entrypoint:TestEntrypointGroup.Pipeline
 [zarr.data_type]
-new_data_type = package_with_entrypoint:TestDataType
\ No newline at end of file
+new_data_type = package_with_entrypoint:TestDataType
+[zarr.url_adapters]
+entrypoint-scheme = package_with_entrypoint:TestEntrypointURLAdapter
diff --git a/tests/package_with_entrypoint/__init__.py b/tests/package_with_entrypoint/__init__.py
index 23afcf1dc2..b128b8cca1 100644
--- a/tests/package_with_entrypoint/__init__.py
+++ b/tests/package_with_entrypoint/__init__.py
@@ -7,6 +7,7 @@
 
 import zarr.core.buffer
 from zarr.abc.codec import ArrayBytesCodec, CodecInput, CodecPipeline
+from zarr.abc.url_pipeline import AdapterResolution, URLPipelineAdapter
 from zarr.codecs import BytesCodec
 from zarr.core.buffer import Buffer, NDBuffer
 from zarr.core.dtype.npy.bool import Bool
@@ -16,6 +17,7 @@
     from collections.abc import Iterable
     from typing import Any, ClassVar, Literal, Self
 
+    from zarr.abc.url_pipeline import PipelineContext, PipelineSegment
     from zarr.core.array_spec import ArraySpec
     from zarr.core.common import ZarrFormat
     from zarr.core.dtype.common import DTypeJSON, DTypeSpec_V2
@@ -100,3 +102,16 @@ def to_json(self, zarr_format: ZarrFormat) -> str | DTypeSpec_V2:  # type: ignor
         if zarr_format == 3:
             return self._zarr_v3_name
         raise ValueError("zarr_format must be 2 or 3")
+
+
+class TestEntrypointURLAdapter(URLPipelineAdapter):
+    """URL pipeline adapter discovered via the zarr.url_adapters entry point."""
+
+    @classmethod
+    async def open_pipeline_segment(
+        cls, segment: PipelineSegment, context: PipelineContext
+    ) -> AdapterResolution:
+        from zarr.storage import MemoryStore
+
+        store = await MemoryStore.open(read_only=False)
+        return AdapterResolution(store=store, path=segment.body)
diff --git a/tests/test_url_pipeline/__init__.py b/tests/test_url_pipeline/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/tests/test_url_pipeline/conftest.py b/tests/test_url_pipeline/conftest.py
new file mode 100644
index 0000000000..13c717e04e
--- /dev/null
+++ b/tests/test_url_pipeline/conftest.py
@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+
+import zarr.registry
+
+if TYPE_CHECKING:
+    from collections.abc import Generator
+
+
+@pytest.fixture
+def clean_url_adapter_registry() -> Generator[None, None, None]:
+    """Snapshot and restore the URL adapter registry around a test."""
+    registry = zarr.registry._url_adapter_registry
+    saved = dict(registry)
+    saved_lazy = list(registry.lazy_load_list)
+    yield
+    registry.clear()
+    registry.update(saved)
+    registry.lazy_load_list[:] = saved_lazy
diff --git a/tests/test_url_pipeline/test_parser.py b/tests/test_url_pipeline/test_parser.py
new file mode 100644
index 0000000000..6c6f76464e
--- /dev/null
+++ b/tests/test_url_pipeline/test_parser.py
@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import pytest
+
+from zarr.errors import URLPipelineError
+from zarr.storage._url_pipeline import parse_pipeline
+
+
+def test_single_root_url() -> None:
+    (segment,) = parse_pipeline("s3://bucket/key")
+    assert segment.scheme == "s3"
+    assert segment.body == "//bucket/key"
+    assert segment.query is None
+    assert segment.raw == "s3://bucket/key"
+    assert str(segment) == "s3://bucket/key"
+
+
+def test_schemeless_root() -> None:
+    (segment,) = parse_pipeline("/local/path")
+    assert segment.scheme == ""
+    assert segment.body == "/local/path"
+    assert segment.raw == "/local/path"
+
+
+def test_adapter_chain() -> None:
+    segments = parse_pipeline("s3://bucket/data.zip|zip:inner/path|zarr3:")
+    assert [s.scheme for s in segments] == ["s3", "zip", "zarr3"]
+    assert segments[1].body == "inner/path"
+    assert segments[2].body == ""
+
+
+def test_trailing_colon_optional() -> None:
+    with_colon = parse_pipeline("file:/tmp/x.zip|zip:")
+    without_colon = parse_pipeline("file:/tmp/x.zip|zip")
+    assert with_colon[1].scheme == without_colon[1].scheme == "zip"
+    assert with_colon[1].body == without_colon[1].body == ""
+
+
+def test_scheme_case_insensitive() -> None:
+    segments = parse_pipeline("FILE:/tmp/x.zip|ZIP:Inner/Path")
+    assert segments[0].scheme == "file"
+    assert segments[1].scheme == "zip"
+    # bodies are case-preserved
+    assert segments[1].body == "Inner/Path"
+
+
+def test_case_preserved_in_raw() -> None:
+    # e.g. icechunk snapshot IDs are case-significant
+    segments = parse_pipeline("file:/tmp/repo|icechunk://ABCDEFGH12345678ABCD/x")
+    assert segments[1].raw == "icechunk://ABCDEFGH12345678ABCD/x"
+    assert segments[1].body == "//ABCDEFGH12345678ABCD/x"
+
+
+def test_vendor_prefixed_scheme() -> None:
+    segments = parse_pipeline("file:/data|vendor-1.custom+adapter:sub/path")
+    assert segments[1].scheme == "vendor-1.custom+adapter"
+    assert segments[1].body == "sub/path"
+
+
+def test_query_strings() -> None:
+    segments = parse_pipeline("https://example.com/d.zip?token=abc|zip:x?opt=1")
+    assert segments[0].query == "token=abc"
+    assert segments[0].body == "//example.com/d.zip"
+    assert segments[1].query == "opt=1"
+    assert segments[1].body == "x"
+
+
+def test_empty_query() -> None:
+    (segment,) = parse_pipeline("https://example.com/d?")
+    assert segment.query == ""
+
+
+def test_windows_drive_path_is_not_a_scheme() -> None:
+    # On Windows parse_store_url treats C:\... as a local path; elsewhere the
+    # single-letter scheme is preserved but must not crash the parser.
+    (segment,) = parse_pipeline(r"C:\data\store")
+    assert segment.raw == r"C:\data\store"
+
+
+@pytest.mark.parametrize("url", ["a||b:", "|zip:", "file:/tmp|", ""])
+def test_empty_sub_url_rejected(url: str) -> None:
+    with pytest.raises(URLPipelineError, match="empty sub-URL"):
+        parse_pipeline(url)
+
+
+def test_fragment_rejected() -> None:
+    with pytest.raises(URLPipelineError, match="fragment"):
+        parse_pipeline("file:/tmp/x.zip|zip:inner#frag")
+
+
+@pytest.mark.parametrize("segment", ["1zip:", "zi p:x", "zip@:x"])
+def test_invalid_adapter_scheme_rejected(segment: str) -> None:
+    with pytest.raises(URLPipelineError, match="invalid adapter scheme"):
+        parse_pipeline(f"file:/tmp/x|{segment}")
+
+
+def test_round_trip() -> None:
+    url = "s3://bucket/a.zip?v=2|zip:b/inner.zip|zip:c|zarr3:"
+    segments = parse_pipeline(url)
+    assert "|".join(s.raw for s in segments) == url
diff --git a/tests/test_url_pipeline/test_resolver.py b/tests/test_url_pipeline/test_resolver.py
new file mode 100644
index 0000000000..193b1c81ce
--- /dev/null
+++ b/tests/test_url_pipeline/test_resolver.py
@@ -0,0 +1,265 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, ClassVar
+
+import pytest
+
+import zarr
+import zarr.registry
+from zarr.abc.store import Store
+from zarr.abc.url_pipeline import (
+    AdapterResolution,
+    PipelineContext,
+    PipelineSegment,
+    URLPipelineAdapter,
+)
+from zarr.errors import URLPipelineError
+from zarr.registry import (
+    get_url_adapter,
+    list_url_adapter_schemes,
+    register_url_adapter,
+)
+from zarr.storage import MemoryStore, WrapperStore
+from zarr.storage._common import make_store, make_store_path
+from zarr.storage._url_pipeline import is_url_pipeline, resolve_pipeline
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+pytestmark = pytest.mark.usefixtures("clean_url_adapter_registry")
+
+
+class TracingStore(WrapperStore[Store]):
+    """Wrapper that records the context it was created from."""
+
+    context: PipelineContext
+    segment: PipelineSegment
+
+
+class WrapperAdapter(URLPipelineAdapter):
+    """A wrapper-style adapter: resolves the preceding pipeline into a store."""
+
+    @classmethod
+    async def open_pipeline_segment(
+        cls, segment: PipelineSegment, context: PipelineContext
+    ) -> AdapterResolution:
+        preceding = await context.resolve_preceding()
+        store = TracingStore(preceding.store)
+        store.context = context
+        store.segment = segment
+        return AdapterResolution(store=store, path=segment.body)
+
+
+class NativeAdapter(URLPipelineAdapter):
+    """A native-style adapter: consumes the preceding URL as a string."""
+
+    seen_urls: ClassVar[list[str]] = []
+
+    @classmethod
+    async def open_pipeline_segment(
+        cls, segment: PipelineSegment, context: PipelineContext
+    ) -> AdapterResolution:
+        cls.seen_urls.append(context.preceding_url)
+        store = await MemoryStore.open(read_only=context.read_only)
+        return AdapterResolution(store=store, path=segment.body)
+
+
+class RootAdapter(URLPipelineAdapter):
+    """A root-scheme adapter (no preceding segments), like al://."""
+
+    @classmethod
+    async def open_pipeline_segment(
+        cls, segment: PipelineSegment, context: PipelineContext
+    ) -> AdapterResolution:
+        assert context.preceding == ()
+        if context.mode in ("w", "w-", "r+"):
+            raise ValueError("read-only scheme")
+        store = await MemoryStore.open(read_only=True)
+        return AdapterResolution(store=store, path=segment.body.lstrip("/"))
+
+
+async def _store_of(resolution: AdapterResolution) -> Store:
+    return resolution.store
+
+
+class TestRegistry:
+    def test_register_and_get(self) -> None:
+        register_url_adapter("demo", WrapperAdapter)
+        assert get_url_adapter("demo") is WrapperAdapter
+        assert get_url_adapter("DEMO") is WrapperAdapter
+        assert "demo" in list_url_adapter_schemes()
+
+    def test_unknown_scheme(self) -> None:
+        with pytest.raises(URLPipelineError, match="no URL pipeline adapter is registered"):
+            get_url_adapter("nonexistent-scheme")
+
+    @pytest.mark.usefixtures("set_path")
+    def test_entrypoint_discovery(self) -> None:
+        assert "entrypoint-scheme" in list_url_adapter_schemes()
+        cls = get_url_adapter("entrypoint-scheme")
+        assert cls.__name__ == "TestEntrypointURLAdapter"
+
+    @pytest.mark.usefixtures("set_path")
+    async def test_entrypoint_end_to_end(self) -> None:
+        result = await resolve_pipeline("memory://src|entrypoint-scheme:sub/path")
+        assert result.path == "sub/path"
+
+    @pytest.mark.usefixtures("set_path")
+    def test_loading_one_scheme_leaves_others_pending(self) -> None:
+        # resolving one scheme must not import other providers' entry points
+        registry = zarr.registry._url_adapter_registry
+        assert any(e.name == "entrypoint-scheme" for e in registry.lazy_load_list)
+        with pytest.raises(URLPipelineError, match="no URL pipeline adapter"):
+            get_url_adapter("some-other-scheme")
+        assert any(e.name == "entrypoint-scheme" for e in registry.lazy_load_list)
+        assert get_url_adapter("entrypoint-scheme").__name__ == "TestEntrypointURLAdapter"
+
+
+class TestIsURLPipeline:
+    def test_pipe_routes(self) -> None:
+        assert is_url_pipeline("memory://x|demo:")
+
+    def test_registered_root_scheme_routes(self) -> None:
+        register_url_adapter("rooty", RootAdapter)
+        assert is_url_pipeline("rooty://org/repo")
+
+    @pytest.mark.parametrize("url", ["s3://bucket/key", "/local/path", "memory://x", "C:.zarr"])
+    def test_plain_urls_do_not_route(self, url: str) -> None:
+        assert not is_url_pipeline(url)
+
+    def test_exotic_authority_root_scheme(self) -> None:
+        # The spec's own root-URL example: urlparse rejects the bracketed
+        # non-IP authority, so scheme detection must use the same fallback
+        # as the parser rather than raising before routing.
+        url = "vendor1-2.custom-1+proto.ext://[authority]/path?query/part?x"
+        assert not is_url_pipeline(url)
+        register_url_adapter("vendor1-2.custom-1+proto.ext", RootAdapter)
+        assert is_url_pipeline(url)
+
+
+class TestResolve:
+    async def test_wrapper_adapter_chain(self, tmp_path: Path) -> None:
+        register_url_adapter("wrap", WrapperAdapter)
+        result = await resolve_pipeline(f"{tmp_path}|wrap:inner/path")
+        assert isinstance(result.store, TracingStore)
+        assert result.path == "inner/path"
+        assert result.store.context.preceding_url == str(tmp_path)
+
+    async def test_native_adapter_gets_preceding_url(self) -> None:
+        register_url_adapter("native", NativeAdapter)
+        NativeAdapter.seen_urls.clear()
+        await resolve_pipeline("s3://bucket/repo|native:")
+        assert NativeAdapter.seen_urls == ["s3://bucket/repo"]
+
+    async def test_multi_segment_preceding_url(self) -> None:
+        register_url_adapter("wrap", WrapperAdapter)
+        register_url_adapter("native", NativeAdapter)
+        NativeAdapter.seen_urls.clear()
+        await resolve_pipeline("memory://base|wrap:a|native:x")
+        assert NativeAdapter.seen_urls == ["memory://base|wrap:a"]
+
+    async def test_root_adapter(self) -> None:
+        register_url_adapter("rooty", RootAdapter)
+        result = await resolve_pipeline("rooty://org/repo")
+        assert result.path == "org/repo"
+        assert result.store.read_only
+
+    async def test_root_adapter_composes_with_chain(self) -> None:
+        register_url_adapter("rooty", RootAdapter)
+        register_url_adapter("native", NativeAdapter)
+        NativeAdapter.seen_urls.clear()
+        await resolve_pipeline("rooty://org/repo|native:x")
+        assert NativeAdapter.seen_urls == ["rooty://org/repo"]
+
+    async def test_read_only_flag(self, tmp_path: Path) -> None:
+        register_url_adapter("wrap", WrapperAdapter)
+        result = await resolve_pipeline(f"{tmp_path}|wrap:", mode="r")
+        assert isinstance(result.store, TracingStore)
+        assert result.store.context.read_only
+        assert result.store.context.mode == "r"
+        result = await resolve_pipeline(f"{tmp_path}|wrap:")
+        assert isinstance(result.store, TracingStore)
+        assert not result.store.context.read_only
+        assert result.store.context.mode is None
+
+    async def test_storage_options_visible_to_adapter(self) -> None:
+        register_url_adapter("native", NativeAdapter)
+
+        class OptionsProbe(NativeAdapter):
+            seen_options: dict[str, object] | None = None
+
+            @classmethod
+            async def open_pipeline_segment(
+                cls, segment: PipelineSegment, context: PipelineContext
+            ) -> AdapterResolution:
+                cls.seen_options = context.storage_options
+                return await super().open_pipeline_segment(segment, context)
+
+        register_url_adapter("probe", OptionsProbe)
+        opts = {"anon": True}
+        await resolve_pipeline("s3://bucket/x|probe:", storage_options=opts)
+        assert OptionsProbe.seen_options == opts
+
+    async def test_wrapper_at_root_position_raises(self) -> None:
+        # a wrapper adapter used as the pipeline root has nothing to wrap
+        register_url_adapter("wrap", WrapperAdapter)
+        with pytest.raises(URLPipelineError, match="requires a preceding sub-URL"):
+            await resolve_pipeline("wrap:whatever")
+
+    async def test_not_a_pipeline_raises(self) -> None:
+        with pytest.raises(URLPipelineError, match="is not a URL pipeline"):
+            await resolve_pipeline("s3://bucket/plain")
+
+    async def test_unknown_adapter_scheme_raises(self) -> None:
+        with pytest.raises(URLPipelineError, match="no URL pipeline adapter is registered"):
+            await resolve_pipeline("memory://base|no-such-adapter:")
+
+
+class TestMakeStoreIntegration:
+    async def test_make_store_path_combines_paths(self, tmp_path: Path) -> None:
+        register_url_adapter("wrap", WrapperAdapter)
+        store_path = await make_store_path(f"{tmp_path}|wrap:residual", path="user/sub")
+        assert store_path.path == "residual/user/sub"
+
+    async def test_make_store_rejects_residual_path(self, tmp_path: Path) -> None:
+        register_url_adapter("wrap", WrapperAdapter)
+        with pytest.raises(URLPipelineError, match="resolves to a path inside a store"):
+            await make_store(f"{tmp_path}|wrap:residual")
+
+    async def test_make_store_no_residual_path(self, tmp_path: Path) -> None:
+        register_url_adapter("wrap", WrapperAdapter)
+        store = await make_store(f"{tmp_path}|wrap:")
+        assert isinstance(store, TracingStore)
+
+    async def test_zarr_open_end_to_end(self, tmp_path: Path) -> None:
+        register_url_adapter("wrap", WrapperAdapter)
+        group = zarr.open_group(f"{tmp_path}|wrap:", mode="w")
+        array = group.create_array("x", shape=(4,), dtype="i4")
+        array[:] = [1, 2, 3, 4]
+        assert zarr.open_array(f"{tmp_path}|wrap:x")[2] == 3
+
+    async def test_pipeline_store_read_only_mode(self) -> None:
+        register_url_adapter("native", NativeAdapter)
+        store_path = await make_store_path("memory://base|native:", mode="r")
+        assert store_path.read_only
+
+    async def test_mode_a_downgrades_to_read_only_open(self) -> None:
+        # mode "a" (the zarr.open default) is open-or-create: a pipeline
+        # that resolves to a read-only store serves the "open" half instead
+        # of failing outright.
+        register_url_adapter("rooty", RootAdapter)
+        store_path = await make_store_path("rooty://org/repo", mode="a")
+        assert store_path.read_only
+
+    async def test_explicit_write_mode_reaches_adapter(self) -> None:
+        register_url_adapter("rooty", RootAdapter)
+        with pytest.raises(ValueError, match="read-only scheme"):
+            await make_store_path("rooty://org/repo", mode="w")
+
+    async def test_storage_options_forwarded_to_root(self) -> None:
+        # storage_options reach the root sub-URL via resolve_preceding ->
+        # make_store. A local-path root does not accept storage_options, so
+        # forwarding them must raise the same TypeError as a non-pipeline open.
+        register_url_adapter("wrap", WrapperAdapter)
+        with pytest.raises(TypeError, match="'storage_options' was provided but unused"):
+            await make_store("/tmp/some/path|wrap:", storage_options={"anon": True})
diff --git a/tests/test_url_pipeline/test_spec_examples.py b/tests/test_url_pipeline/test_spec_examples.py
new file mode 100644
index 0000000000..5538074bea
--- /dev/null
+++ b/tests/test_url_pipeline/test_spec_examples.py
@@ -0,0 +1,258 @@
+"""
+Conformance tests against the URL pipeline specification.
+
+Every entry in `SPEC_EXAMPLES` is an example URL from the specification
+repository (https://github.com/jbms/url-pipeline @ 1a01ce6), extracted
+with the same rule as the spec's own linter (`scripts/lint.py`): backticked
+examples on bullet lines, skipping spans without a `:` or `|`. The spec
+validates each example (and an uppercased-scheme variant) against its ABNF
+grammar, so this corpus is grammar-valid by construction.
+
+The parser must accept every example — including schemes zarr-python has no
+adapter for, since parsing is independent of adapter availability — split it
+into the expected sub-URL schemes, and preserve the text losslessly.
+
+To regenerate after a spec update: extract examples per the rule above and
+recompute the expected scheme tuples with `parse_pipeline`, reviewing any
+changes against the spec diff.
+"""
+
+from __future__ import annotations
+
+import re
+
+import pytest
+
+from zarr.storage._url_pipeline import parse_pipeline
+
+SPEC_EXAMPLES: list[tuple[str, tuple[str, ...]]] = [
+    # README.md
+    ("s3://bucket/path/to/archive.zip|zip:path/within/zip.zarr/|zarr3:", ("s3", "zip", "zarr3")),
+    (
+        "file:///tmp/dataset.ocdbt/|ocdbt://2025-01-01T01:23:45.678Z/path/within/database",
+        ("file", "ocdbt"),
+    ),
+    (
+        "s3+https://example.com/path/to/database.icechunk/|icechunk://tag.v5/path/to/node/|zarr3:",
+        ("s3+https", "icechunk", "zarr3"),
+    ),
+    (
+        "vendor1-2.custom-1+proto.ext://[authority]/path?query/part?x",
+        ("vendor1-2.custom-1+proto.ext",),
+    ),
+    (
+        "http://example.com/file|vendor1-2.custom+adapter:/path/within/adapter",
+        ("http", "vendor1-2.custom+adapter"),
+    ),
+    ("a.b:?", ("a.b",)),
+    ("http://somehost/downloads/somefile.zip|zip:", ("http", "zip")),
+    ("http://example.com/archive.jar|zip:path/to/file.txt", ("http", "zip")),
+    ("https://host/archive.zip|zip:path/in/outer.zip|zip:path/in/inner", ("https", "zip", "zip")),
+    ("file:///path/to/archive.zip|zip:path/within/archive", ("file", "zip")),
+    # avif.md
+    ("file:/path/to/image.avif|avif:", ("file", "avif")),
+    ("file:/path/to/image.avif|avif", ("file", "avif")),
+    # bmp.md
+    ("file:/path/to/image.bmp|bmp:", ("file", "bmp")),
+    ("file:/path/to/image.bmp|bmp", ("file", "bmp")),
+    # byte-range.md
+    ("file:/path/to/data|byte-range:1000-2000", ("file", "byte-range")),
+    ("file:/path/to/data|byte-range:0-1", ("file", "byte-range")),
+    # file.md
+    ("file:/", ("file",)),
+    ("file:/a", ("file",)),
+    ("file:/path/to/file.txt", ("file",)),
+    ("file://localhost/path/to/file.txt", ("file",)),
+    ("file://LOCALHOST/path/to/file.txt", ("file",)),
+    ("file:///path/to/file.txt", ("file",)),
+    ("file://somehost/sharename/path/to/file.txt", ("file",)),
+    # gs.md
+    ("gs://bucket", ("gs",)),
+    ("gs://bucket/", ("gs",)),
+    ("gs://bucket/path/within/bucket", ("gs",)),
+    ("gs://aaa", ("gs",)),
+    (
+        "gs://label1-is-sixty-two-characters-long-xxxxxxxxxxxxxxxxxxxxxxxxxx.label2-is-sixty-two-characters-long-yyyyyyyyyyyyyyyyyyyyyyyyyy.label3-is-sixty-two-characters-long-zzzzzzzzzzzzzzzzzzzzzzzzzz.label4-is-thirty-one-characters",
+        ("gs",),
+    ),
+    # gzip.md
+    ("file:/path/to/data.gz|gzip:", ("file", "gzip")),
+    ("file:/path/to/data.gz|gzip", ("file", "gzip")),
+    # hdf5.md
+    ("s3://bucket/path/to/file.h5|hdf5:/path/to/dataset", ("s3", "hdf5")),
+    ("s3://bucket/path/to/file.h5|hdf5:a", ("s3", "hdf5")),
+    ("file:///path/to/file.h5|hdf5:", ("file", "hdf5")),
+    ("file:///path/to/file.h5|hdf5", ("file", "hdf5")),
+    # http.md
+    ("https://example.com/path/to/resource%20name", ("https",)),
+    ("https://example.com/path/to/resource?query=value", ("https",)),
+    ("https://example.com/path/to/?query=value", ("https",)),
+    ("https://example.com/path/with:colon", ("https",)),
+    ("https://server.example.com:1234/path/to/array", ("https",)),
+    ("http://a", ("http",)),
+    ("http://example.com", ("http",)),
+    ("http://local%68ost", ("http",)),
+    ("http://example.com:", ("http",)),
+    ("http://192.168.10.1:1234", ("http",)),
+    ("http://[::1]", ("http",)),
+    ("http://[::ffff:127.0.0.1]", ("http",)),
+    ("http://example.com/", ("http",)),
+    # icechunk.md
+    ("file:///path/to/repo.zarr.icechunk/|icechunk:", ("file", "icechunk")),
+    ("file:///path/to/repo.zarr.icechunk/|icechunk", ("file", "icechunk")),
+    ("file:///path/to/repo.zarr.icechunk/|icechunk://branch.main/", ("file", "icechunk")),
+    ("file:///path/to/repo.zarr.icechunk/|icechunk:path/to/node/", ("file", "icechunk")),
+    ("file:///path/to/repo.zarr.icechunk/|icechunk:a", ("file", "icechunk")),
+    ("file:///path/to/repo.zarr.icechunk/|icechunk:/path/to/node/", ("file", "icechunk")),
+    ("file:///path/to/repo.zarr.icechunk/|icechunk:path/to/node/zarr.json", ("file", "icechunk")),
+    ("file:///path/to/repo.zarr.icechunk/|icechunk:path/to/node/c/0/0/1", ("file", "icechunk")),
+    (
+        "file:///path/to/repo.zarr.icechunk/|icechunk://branch.mybranch/path/to/node/",
+        ("file", "icechunk"),
+    ),
+    ("file:///path/to/repo.zarr.icechunk/|icechunk://branch.a", ("file", "icechunk")),
+    (
+        "file:///path/to/repo.zarr.icechunk/|icechunk://tag.mytag/path/to/node/",
+        ("file", "icechunk"),
+    ),
+    ("file:///path/to/repo.zarr.icechunk/|icechunk://tag.a", ("file", "icechunk")),
+    (
+        "file:///path/to/repo.zarr.icechunk/|icechunk://FWWFQGAW742XMX0F5MF0/path/to/node/",
+        ("file", "icechunk"),
+    ),
+    (
+        "file:///path/to/repo.zarr.icechunk/|icechunk://ABCDEFGHJKMNPQRSTVWX/path/to/node/",
+        ("file", "icechunk"),
+    ),
+    (
+        "file:///path/to/repo.zarr.icechunk/|icechunk:|zarr3:path/to/array/",
+        ("file", "icechunk", "zarr3"),
+    ),
+    (
+        "file:///path/to/repo.zarr.icechunk/|icechunk://branch.other/|zarr3:path/to/array/",
+        ("file", "icechunk", "zarr3"),
+    ),
+    (
+        "file:///path/to/repo.zarr.icechunk/|icechunk://tag.v5/|zarr3:path/to/array/",
+        ("file", "icechunk", "zarr3"),
+    ),
+    (
+        "file:///path/to/repo.zarr.icechunk/|icechunk://4N0217AZA4VNPYD0HR0G/|zarr3:path/to/array/",
+        ("file", "icechunk", "zarr3"),
+    ),
+    # jpeg.md
+    ("file:/path/to/image.jpeg|jpeg:", ("file", "jpeg")),
+    ("file:/path/to/image.jpeg|jpeg", ("file", "jpeg")),
+    # json.md
+    ("file:/path/to/data.json|json:", ("file", "json")),
+    ("file:/path/to/data.json|json", ("file", "json")),
+    ("file:/path/to/data.json|json:/path/to/node", ("file", "json")),
+    ("file:/path/to/data.json|json:/abc~0def", ("file", "json")),
+    ("file:/path/to/data.json|json:/abc~1def", ("file", "json")),
+    ("file:/path/to/data.json|json:/", ("file", "json")),
+    # memory.md
+    ("memory:", ("memory",)),
+    ("memory:/", ("memory",)),
+    ("memory://", ("memory",)),
+    ("memory:path/to/resource", ("memory",)),
+    ("memory:a", ("memory",)),
+    ("memory:another@path+with,lots&of(special;characters)*_!-$'", ("memory",)),
+    ("memory:/path/to/resource", ("memory",)),
+    ("memory://path/to/resource", ("memory",)),
+    # n5.md
+    ("file:///tmp/data.n5/|n5:path/to/array", ("file", "n5")),
+    ("file:///tmp/data.n5/|n5:a", ("file", "n5")),
+    ("file:///tmp/data.n5/|n5:/path/to/array", ("file", "n5")),
+    ("file:///tmp/data.n5/|n5:", ("file", "n5")),
+    ("file:///tmp/data.n5/|n5", ("file", "n5")),
+    # neuroglancer-precomputed.md
+    (
+        "file:///tmp/dataset.precomputed/|neuroglancer-precomputed:",
+        ("file", "neuroglancer-precomputed"),
+    ),
+    (
+        "file:///tmp/dataset.precomputed/|neuroglancer-precomputed",
+        ("file", "neuroglancer-precomputed"),
+    ),
+    # ocdbt.md
+    ("file:///path/to/repo.ocdbt/|ocdbt:", ("file", "ocdbt")),
+    ("file:///path/to/repo.ocdbt/|ocdbt", ("file", "ocdbt")),
+    ("file:///path/to/repo.ocdbt/|ocdbt:path/within/repo/", ("file", "ocdbt")),
+    ("file:///path/to/repo.ocdbt/|ocdbt:path/within/repo", ("file", "ocdbt")),
+    ("file:///path/to/repo.ocdbt/|ocdbt:a", ("file", "ocdbt")),
+    ("file:///path/to/repo.ocdbt/|ocdbt:/path/within/repo", ("file", "ocdbt")),
+    ("file:///path/to/repo.ocdbt/|ocdbt://v123/", ("file", "ocdbt")),
+    ("file:///path/to/repo.ocdbt/|ocdbt://v1", ("file", "ocdbt")),
+    ("file:///path/to/repo.ocdbt/|ocdbt://v123/path/within/repo", ("file", "ocdbt")),
+    (
+        "file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45.678Z/path/within/database",
+        ("file", "ocdbt"),
+    ),
+    ("file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45Z", ("file", "ocdbt")),
+    ("file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45.1Z", ("file", "ocdbt")),
+    ("file:///path/to/repo.ocdbt/|ocdbt://2025-01-01T01:23:45.123456789Z", ("file", "ocdbt")),
+    # png.md
+    ("file:/path/to/image.png|png:", ("file", "png")),
+    ("file:/path/to/image.png|png", ("file", "png")),
+    # s3+http.md
+    ("s3+https://endpoint/path/within/bucket", ("s3+https",)),
+    ("s3+https://endpoint/bucket/path/within/bucket", ("s3+https",)),
+    ("s3+https://mybucket.s3.amazonaws.com/path/to/file", ("s3+https",)),
+    ("s3+https://s3.amazonaws.com/mybucket/path/to/file", ("s3+https",)),
+    ("s3+http://example.com", ("s3+http",)),
+    ("s3+http://example.com/", ("s3+http",)),
+    # s3.md
+    ("s3://bucket", ("s3",)),
+    ("s3://bucket/", ("s3",)),
+    ("s3://bucket/path/within/bucket", ("s3",)),
+    ("s3://aaa", ("s3",)),
+    (
+        "s3://aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+        ("s3",),
+    ),
+    # tiff.md
+    ("file:/path/to/image.tiff|tiff:", ("file", "tiff")),
+    ("file:/path/to/image.tiff|tiff", ("file", "tiff")),
+    # webp.md
+    ("file:/path/to/image.webp|webp:", ("file", "webp")),
+    ("file:/path/to/image.webp|webp", ("file", "webp")),
+    # zarr.md
+    ("file:///path/to/node.zarr/|zarr3:", ("file", "zarr3")),
+    ("file:///path/to/node.zarr/|zarr3", ("file", "zarr3")),
+    ("file:///path/to/node.zarr/|zarr2:", ("file", "zarr2")),
+    ("file:///path/to/node.zarr/|zarr2", ("file", "zarr2")),
+    ("file:///path/to/node.zarr/|zarr:", ("file", "zarr")),
+    ("file:///path/to/node.zarr/|zarr", ("file", "zarr")),
+    # zip.md
+    ("file:/path/to/archive.zip|zip:path/to/file.txt", ("file", "zip")),
+    ("file:/path/to/archive.zip|zip", ("file", "zip")),
+    ("file:/path/to/archive.zip|zip:/path/to/file.txt", ("file", "zip")),
+    ("file:/path/to/outer.zip|zip:path/to/inner.zip|zip:path/to/file.txt", ("file", "zip", "zip")),
+    # zstd.md
+    ("file:/path/to/data.zstd|zstd:", ("file", "zstd")),
+    ("file:/path/to/data.zstd|zstd", ("file", "zstd")),
+]
+
+
+def _uppercase_schemes(example: str) -> str:
+    """Uppercase every sub-URL scheme, as the spec's linter does."""
+
+    def repl(match: re.Match[str]) -> str:
+        return match.group(1) + match.group(2).upper()
+
+    return re.sub(r"((?:^|\|)\s*)([a-zA-Z][a-zA-Z0-9+.-]*)", repl, example)
+
+
+@pytest.mark.parametrize(("url", "expected_schemes"), SPEC_EXAMPLES, ids=lambda v: str(v))
+def test_spec_example_parses(url: str, expected_schemes: tuple[str, ...]) -> None:
+    segments = parse_pipeline(url)
+    assert tuple(segment.scheme for segment in segments) == expected_schemes
+    # lossless round-trip of the original text
+    assert "|".join(segment.raw for segment in segments) == url
+
+
+@pytest.mark.parametrize(("url", "expected_schemes"), SPEC_EXAMPLES, ids=lambda v: str(v))
+def test_spec_example_schemes_case_insensitive(url: str, expected_schemes: tuple[str, ...]) -> None:
+    upper = _uppercase_schemes(url)
+    segments = parse_pipeline(upper)
+    assert tuple(segment.scheme for segment in segments) == expected_schemes