diff --git a/packages/timenet/src/timenet/_warnings.py b/packages/timenet/src/timenet/_warnings.py
new file mode 100644
index 000000000..064f57bed
--- /dev/null
+++ b/packages/timenet/src/timenet/_warnings.py
@@ -0,0 +1,24 @@
+"""Warn at the first caller outside the timenet package."""
+
+import inspect
+import warnings
+
+
+def warn_at_caller(warning: Warning) -> None:
+    """Emit a warning at the first frame of the call stack whose module is outside the ``timenet`` package.
+
+    Args:
+        warning: The warning to emit.
+    """
+    # The walk starts at this function, which is level 1 of the stack and lies in the package.
+    frame = inspect.currentframe()
+    level = 1
+    while frame is not None and _in_timenet(frame.f_globals.get("__name__", "")):
+        frame = frame.f_back
+        level += 1
+    warnings.warn(warning, stacklevel=level)
+
+
+def _in_timenet(module: str) -> bool:
+    """Return whether a module name lies in the ``timenet`` package."""
+    return module == "timenet" or module.startswith("timenet.")
diff --git a/packages/timenet/src/timenet/builders.py b/packages/timenet/src/timenet/builders.py
index 30d766082..8c0ad7de9 100644
--- a/packages/timenet/src/timenet/builders.py
+++ b/packages/timenet/src/timenet/builders.py
@@ -6,9 +6,9 @@
 package plug in the same way the first-party one does.
 """
 
-from collections.abc import Iterable, Mapping
+from collections.abc import Iterable, Iterator, Mapping
 from dataclasses import dataclass
-from importlib.metadata import entry_points
+from importlib.metadata import EntryPoint, entry_points
 from typing import TYPE_CHECKING, Literal, Protocol
 
 from timenet.errors import TimeNetDatasetNotFoundError, TimeNetVersionUnavailableError
@@ -123,6 +123,38 @@ def release(self) -> DatasetRef:
         return DatasetRef(dataset_id=self.card.dataset_id, version=self.card.dataset_version)
 
 
+@dataclass(frozen=True)
+class ConnectorClaim:
+    """A dataset ID and the installed backend that owns it.
+
+    Attributes:
+        dataset_id: The dataset ID.
+        backend: The first backend that lists the ID.
+        entry: The entry point that registers the backend.
+    """
+
+    dataset_id: str
+    backend: BuilderBackend
+    entry: EntryPoint
+
+
+def claims() -> Iterator[ConnectorClaim]:
+    """Yield each dataset ID of the installed backends once, owned by the first backend that lists it.
+
+    The walk loads each backend only when it reaches its entry point.
+
+    Yields:
+        One claim per dataset ID, in the order of the entry points.
+    """
+    seen: set[str] = set()
+    for entry in entry_points(group=ENTRY_POINT_GROUP):
+        backend = entry.load()()
+        for dataset_id in backend.dataset_ids():
+            if dataset_id not in seen:
+                seen.add(dataset_id)
+                yield ConnectorClaim(dataset_id, backend, entry)
+
+
 def installed_builder(dataset_id: str) -> InstalledBuilder | None:
     """Find the first registered backend that lists a dataset, and read its card.
 
@@ -135,12 +167,11 @@ def installed_builder(dataset_id: str) -> InstalledBuilder | None:
     Raises:
         TimeNetInvalidCardError: If the dataset's card cannot be read.
     """  # ruff: ignore[docstring-extraneous-exception] - raised by the backend
-    for entry in entry_points(group=ENTRY_POINT_GROUP):
-        backend = entry.load()()
-        if dataset_id in backend.dataset_ids():
-            card = backend.card(dataset_id)
-            return None if card is None else InstalledBuilder(backend, card)
-    return None
+    claim = next((claim for claim in claims() if claim.dataset_id == dataset_id), None)
+    if claim is None:
+        return None
+    card = claim.backend.card(dataset_id)
+    return None if card is None else InstalledBuilder(claim.backend, card)
 
 
 def declared(dataset_id: str, version: Version | str | None = None) -> InstalledBuilder:
diff --git a/packages/timenet/src/timenet/cli/app.py b/packages/timenet/src/timenet/cli/app.py
index 3dbdd1a3d..4d31eaca9 100644
--- a/packages/timenet/src/timenet/cli/app.py
+++ b/packages/timenet/src/timenet/cli/app.py
@@ -1,10 +1,12 @@
 """The consumer command-line interface, the command-line mirror of the :class:`~timenet.client.TimeNet` SDK."""
 
-from collections.abc import Callable, Iterable, Sequence
+from collections.abc import Callable, Generator, Iterable, Sequence
+from contextlib import contextmanager
 from functools import partial
 from pathlib import Path
 import shlex
 from typing import TypeVar
+import warnings
 
 from rich.console import Console
 from rich.table import Table
@@ -16,7 +18,7 @@
 from timenet.cli.ui import console as ui, progress_renderer
 from timenet.client import TimeNet, default_local_registry, default_sources
 from timenet.config import settings
-from timenet.errors import TimeNetDatasetNotFoundError, TimeNetNotBuiltError
+from timenet.errors import CatalogWarning, ConnectorCardWarning, TimeNetDatasetNotFoundError, TimeNetNotBuiltError
 from timenet.producer import publish as publish_release
 from timenet.progress import ProgressCallback, ProgressEvent, StepStarted
 from timenet.registry import open_registry
@@ -103,16 +105,35 @@ def _metadata_table(rows: list[DatasetMetadata], title: str) -> Table:
     return table
 
 
+@contextmanager
+def _warning_lines() -> Generator[None]:
+    """Show each CatalogWarning and ConnectorCardWarning of the block as a warning line, without a file and a line."""
+    caught: list[warnings.WarningMessage] = []
+    try:
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always", CatalogWarning)
+            warnings.simplefilter("always", ConnectorCardWarning)
+            yield
+    finally:
+        for item in caught:
+            if issubclass(item.category, (CatalogWarning, ConnectorCardWarning)):
+                ui.warn(str(item.message))
+            else:
+                # Any other warning gets its usual output.
+                warnings.warn_explicit(item.message, item.category, item.filename, item.lineno)
+
+
 @app.command("list")
 def list_datasets(
     local_registry: str | None = LOCAL_REGISTRY,
     remote_registry: str | None = REMOTE_REGISTRY,
     auto_build: bool | None = AUTO_BUILD,
 ) -> None:
-    """List the datasets in the registry."""
-    rows = _client(local_registry, remote_registry, auto_build).list()
+    """List each dataset at its highest version across all sources."""
+    with _warning_lines():
+        rows = _client(local_registry, remote_registry, auto_build).list()
     if not rows:
-        typer.echo("No datasets in the registry.")
+        typer.echo("No datasets in the sources.")
         return
     console.print(_metadata_table(rows, "Datasets"))
 
@@ -131,20 +152,21 @@ def search(  # ruff: ignore[too-many-arguments, too-many-positional-arguments]
     remote_registry: str | None = REMOTE_REGISTRY,
     auto_build: bool | None = AUTO_BUILD,
 ) -> None:
-    """Search for datasets that match the filters.
+    """Search datasets across all sources with filters.
 
-    Repeat a filter flag to supply multiple values.
+    Repeat a filter flag to pass several values.
     """
-    results = _client(local_registry, remote_registry, auto_build).search(
-        query=query or None,
-        domain=_enum_list(domain, Domain, "--domain", [d.value for d in Domain]),
-        task=_enum_list(task, lambda value: TASKS[TaskType(value)], "--task", [t.value for t in TaskType]),
-        license=_enum_list(license, License, "--license", [lic.value for lic in License]),
-        time_series_spec=spec or None,
-        dataset_id=dataset_id or None,
-        tag=tag or None,
-        limit=limit,
-    )
+    with _warning_lines():
+        results = _client(local_registry, remote_registry, auto_build).search(
+            query=query or None,
+            domain=_enum_list(domain, Domain, "--domain", [d.value for d in Domain]),
+            task=_enum_list(task, lambda value: TASKS[TaskType(value)], "--task", [t.value for t in TaskType]),
+            license=_enum_list(license, License, "--license", [lic.value for lic in License]),
+            time_series_spec=spec or None,
+            dataset_id=dataset_id or None,
+            tag=tag or None,
+            limit=limit,
+        )
     if not results:
         typer.echo("No datasets match.")
         return
diff --git a/packages/timenet/src/timenet/client.py b/packages/timenet/src/timenet/client.py
index ad257451e..4818af2ab 100644
--- a/packages/timenet/src/timenet/client.py
+++ b/packages/timenet/src/timenet/client.py
@@ -1,7 +1,6 @@
-"""Find, inspect, download, and read datasets with :class:`TimeNet`.
+"""Find, inspect, download, and read TimeF datasets.
 
-A client checks sources in order and builds releases from installed connectors into its first local
-registry that is not read-only.
+:class:`TimeNet` checks sources in order and uses :func:`default_sources` by default.
 """
 
 # The public API has a method named ``list``. Deferred annotations keep the type hint
@@ -15,6 +14,7 @@
 import shlex
 from typing import TYPE_CHECKING, TypeAlias, TypeVar
 
+from timenet._warnings import warn_at_caller
 from timenet.builders import BuildOptions
 from timenet.config import settings
 from timenet.coordinator.build import run_plan
@@ -22,12 +22,14 @@
 from timenet.coordinator.read import raise_not_found, read_release
 from timenet.dataset import TimeFDataset
 from timenet.errors import (
+    CatalogWarning,
     TimeFValidationError,
     TimeNetAccessError,
     TimeNetDatasetNotFoundError,
     TimeNetError,
     TimeNetFileError,
     TimeNetMissingParentError,
+    TimeNetNoCatalogError,
     TimeNetNotBuiltError,
     TimeNetVersionUnavailableError,
 )
@@ -47,7 +49,7 @@
     ReleaseSource,
     resolve_release,
 )
-from timenet.types import DatasetMetadata, Domain, License, Task
+from timenet.types import DatasetFilter, DatasetMetadata, Domain, License, ListedRelease, Task
 from timenet.types.references import DatasetRef, split_ref
 
 
@@ -60,6 +62,7 @@
 # annotations.
 T = TypeVar("T")
 _Metadatas: TypeAlias = list[DatasetMetadata]
+_Listed: TypeAlias = list[ListedRelease]
 _OrList: TypeAlias = T | list[T] | None
 
 
@@ -197,14 +200,48 @@ def __init__(self, sources: Sequence[ReleaseSource | str | Path] | None = None)
             )
 
     def list(self) -> _Metadatas:
-        """Return metadata for every dataset in the selected registry.
+        """List each dataset at its highest version across all sources, including connector cards.
 
-        The client uses the first registry other than the local registry, or the local registry if it is the only one.
+        A listed version can be newer than the one that :meth:`open` reads, which skips a version that a registry
+        does not serve and, without a version, takes the first source that serves the dataset.
+        A source without a catalog is left out with a :class:`~timenet.errors.CatalogWarning`. Source errors stop
+        the call.
 
         Returns:
-            One :class:`~timenet.types.DatasetMetadata` per dataset.
+            One :class:`~timenet.types.DatasetMetadata` per dataset, sorted by dataset ID.
+
+        Raises:
+            TimeNetNoCatalogError: If no source has a catalog.
+            TimeNetInvalidCardError: If an installed backend cannot list its dataset IDs.
+        """  # ruff: ignore[docstring-extraneous-exception] (raised by the backend)
+        return [release.card for release in self._listed()]
+
+    def _listed(self) -> _Listed:
+        """Return each dataset at its highest version across the sources, sorted by dataset ID.
+
+        For equal versions, the earlier source keeps the row. A connector card has no registry. A source without a
+        catalog is left out with a :class:`~timenet.errors.CatalogWarning`.
+
+        Returns:
+            One listed release per dataset.
+
+        Raises:
+            TimeNetNoCatalogError: If no source has a catalog.
         """
-        return self._catalog().list_datasets()
+        if not any(source.has_catalog for source in self._sources):
+            names = ", ".join(str(source) for source in self._sources)
+            raise TimeNetNoCatalogError(f"none of the sources has a catalog: {names}")
+        kept: dict[str, ListedRelease] = {}
+        for source in self._sources:
+            if not source.has_catalog:
+                warn_at_caller(CatalogWarning(f"{source} has no catalog, so the list leaves it out."))
+                continue
+            registry = source if isinstance(source, BaseRegistry) else None
+            for card in source.list_datasets():
+                held = kept.get(card.dataset_id)
+                if held is None or card.dataset_version > held.card.dataset_version:
+                    kept[card.dataset_id] = ListedRelease(card, registry)
+        return [kept[dataset_id] for dataset_id in sorted(kept)]
 
     def get_manifest(self, dataset_id: str, version: str | None = None) -> Manifest:
         """Return the manifest of the release that :meth:`open` reads, without a copy or a build.
@@ -269,24 +306,33 @@ def search(  # ruff: ignore[too-many-arguments]
         tag: _OrList[str] = None,
         limit: int = 100,
     ) -> _Metadatas:
-        """Search the selected registry for datasets that match the filters.
+        """Search the rows of :meth:`list`: each dataset at its highest version across all sources.
 
-        The client uses the first registry other than the local registry, or the local registry if it is the only one.
+        So the search checks only the highest version of each dataset, and an older version that matches is not found.
+        A ``task`` or ``time_series_spec`` filter reads the manifest of a row only if its card matches, and the search
+        stops at ``limit`` matches. Connector cards lack schemas, so they never match these filters.
+        A source without a catalog is left out with a :class:`~timenet.errors.CatalogWarning`. Source errors stop
+        the call.
 
         Args:
-            query: Free text terms for the name, description, and tags.
+            query: Match any term in the name, description, or tags, ignoring case.
             domain: Match any of these domains.
             task: Match any of these task classes in the schema.
             license: Match any of these licenses.
             time_series_spec: Match all of these ``spec_type`` values.
-            dataset_id: Match only these IDs.
+            dataset_id: Match any of these IDs.
             tag: Match all of these tags.
             limit: The maximum number of results.
 
         Returns:
-            The matching dataset metadata.
-        """
-        return self._catalog().search(
+            The matching dataset metadata, sorted by dataset ID.
+
+        Raises:
+            TimeFValidationError: If ``limit`` is negative.
+            TimeNetNoCatalogError: If no source has a catalog.
+            TimeNetInvalidCardError: If an installed backend cannot list its dataset IDs.
+        """  # ruff: ignore[docstring-extraneous-exception] (raised by the backend)
+        wanted = DatasetFilter.of(
             query=query,
             domain=domain,
             task=task,
@@ -296,6 +342,7 @@ def search(  # ruff: ignore[too-many-arguments]
             tag=tag,
             limit=limit,
         )
+        return wanted.select(self._listed())
 
     def download(
         self,
@@ -450,24 +497,6 @@ def load(self, dataset_id: str, version: str | None = None) -> TimeFDataset:
         with self.open(dataset_id, version) as reader:
             return reader.read(load=LoadPlan.ALL_EAGER)
 
-    def _catalog(self) -> BaseRegistry:
-        """Choose the registry to browse.
-
-        Returns:
-            The first registry other than the local registry, or the local registry if it is the only one.
-
-        Raises:
-            TimeFValidationError: If the sources contain no registry.
-        """
-        # TODO(adrian): Remove temporary glue which maps these methods onto one registry until
-        # list() and search() ask every source. This registry gives the results of the client of
-        # one registry, from before the list of sources.
-        registries = [source for source in self._sources if isinstance(source, BaseRegistry)]
-        registry = next((registry for registry in registries if registry is not self._store), self._store)
-        if registry is None:
-            raise TimeFValidationError("list() and search() need a registry in the sources")
-        return registry
-
     @contextmanager
     def _hints(self, dataset_id: str) -> Generator[None]:
         """Add recovery hints to errors raised inside the block.
diff --git a/packages/timenet/src/timenet/errors.py b/packages/timenet/src/timenet/errors.py
index 59677c185..54457fef1 100644
--- a/packages/timenet/src/timenet/errors.py
+++ b/packages/timenet/src/timenet/errors.py
@@ -31,6 +31,10 @@ class TimeNetRegistryError(TimeNetError):
     """TimeNet could not load or reach a registry, or the registry could not serve a request."""
 
 
+class TimeNetNoCatalogError(TimeNetRegistryError):
+    """No source of a dataset listing has a catalog."""
+
+
 class TimeNetDatasetNotFoundError(TimeNetError):
     """A requested dataset id or version is not present in a registry or on disk."""
 
@@ -219,7 +223,11 @@ class TimeNetWarning(UserWarning):
 
 
 class ConnectorCardWarning(TimeNetWarning):
-    """An unreadable or invalid connector card excludes its dataset from listings."""
+    """A dataset listing skips a dataset with an unreadable connector card."""
+
+
+class CatalogWarning(TimeNetWarning):
+    """A dataset listing skips a source without a catalog."""
 
 
 class SpanOutsideWindowWarning(TimeNetWarning):
diff --git a/packages/timenet/src/timenet/registry/base.py b/packages/timenet/src/timenet/registry/base.py
index e87fa33d7..df1da31a6 100644
--- a/packages/timenet/src/timenet/registry/base.py
+++ b/packages/timenet/src/timenet/registry/base.py
@@ -10,7 +10,7 @@
 from collections.abc import Callable
 from pathlib import Path
 import shutil
-from typing import BinaryIO, TypeVar
+from typing import BinaryIO, ClassVar
 
 from timenet._fs import staged_directory
 from timenet.errors import TimeNetDatasetNotFoundError, TimeNetFileError, TimeNetRegistryError
@@ -22,25 +22,25 @@
 from timenet.registry.shared import complete_copy
 from timenet.registry.version import DatasetVersion
 from timenet.sources import FoundInRegistry, NotFound, NotServed, ReleaseLookup
-from timenet.types import Access, DatasetMetadata, Domain, License, Task, Version
+from timenet.types import Access, DatasetFilter, DatasetMetadata, ListedRelease, Version
 from timenet.types.access import access_requirements, effective_access
 from timenet.types.references import DatasetRef
 
 
-T = TypeVar("T")
-
 # Reports download progress: called with each file (or chunk) of bytes as it lands. A ``download_release``
 # implementation calls it so a caller can render a progress bar; ``None`` downloads silently.
 ProgressCallback = Callable[[int], None]
 
 
 class BaseRegistry(ABC):
-    """A searchable source of TimeF datasets, manifests, and files.
+    """A registry of TimeF manifests, files, and searchable dataset metadata.
 
     :meth:`open_release` and :meth:`download_release` handle one release without its payload
     sources. :func:`~timenet.registry.copy_release` copies a release with its payload sources.
     """
 
+    has_catalog: ClassVar[bool] = True
+    """Whether :meth:`list_datasets` and :meth:`search` work. Without a catalog, both raise ``NotImplementedError``."""
     _read_only: bool = False
 
     def __str__(self) -> str:
@@ -240,98 +240,17 @@ def _stage_file(self, dataset_id: str, version: str, relpath: str, staging: Path
         with self.open_file(dataset_id, version, relpath) as source, destination.open("wb") as sink:
             shutil.copyfileobj(source, sink)
 
-    def search(  # ruff: ignore[too-many-arguments]
-        self,
-        *,
-        query: str | list[str] | None = None,
-        domain: Domain | list[Domain] | None = None,
-        task: type[Task] | list[type[Task]] | None = None,
-        license: License | list[License] | None = None,
-        time_series_spec: str | list[str] | None = None,
-        dataset_id: str | list[str] | None = None,
-        tag: str | list[str] | None = None,
-        limit: int = 100,
-    ) -> list[DatasetMetadata]:
-        """Filter datasets by any combination of criteria.
-
-        Each filter accepts a single value or a list of values. The method ignores filters set to
-        ``None``. It combines all other filters with AND logic. The filters ``query``, ``domain``,
-        ``task``, ``license``, and ``dataset_id`` match a dataset when any of their values match.
-        The filters ``time_series_spec`` and ``tag`` match a dataset only when all of their values
-        match. The type filters ``task`` and ``time_series_spec`` read the manifest schema of each
-        dataset.
+    def search(self, filter: DatasetFilter) -> list[DatasetMetadata]:
+        """Find datasets that match a filter.
+
+        A ``task`` or ``time_series_spec`` filter reads the manifest of a dataset only if its card matches.
 
         Args:
-            query: Free-text terms. The method matches each term as a case-insensitive substring
-                in the dataset name, description, or tags.
-            domain: Keep datasets that share any of these domains.
-            task: Keep datasets whose schema includes any of these task classes.
-            license: Keep datasets that have any of these licenses.
-            time_series_spec: Keep datasets that declare all of these ``spec_type`` values.
-            dataset_id: Keep only these ids.
-            tag: Keep datasets that declare all of these tags.
-            limit: The maximum number of results. The default is 100.
+            filter: The filters and the limit.
 
         Returns:
-            The matching dataset metadata. The list has at most ``limit`` entries.
-
-        Raises:
-            ValueError: If ``limit`` is negative.
+            The matching dataset metadata, sorted by dataset ID, with at most ``filter.limit`` entries.
         """
-        if limit < 0:
-            raise ValueError(f"limit must be non-negative, got {limit}")
-        queries = _as_list(query)
-        domains = _as_list(domain)
-        tasks = _as_list(task)
-        licenses = _as_list(license)
-        specs = _as_list(time_series_spec)
-        ids = _as_list(dataset_id)
-        tags = _as_list(tag)
-
-        results: list[DatasetMetadata] = []
-        for metadata in self.list_datasets():
-            if len(results) >= limit:
-                break
-            if ids and metadata.dataset_id not in ids:
-                continue
-            if licenses and metadata.license not in licenses:
-                continue
-            if domains and not set(metadata.domains) & set(domains):
-                continue
-            if tags and not set(tags) <= set(metadata.tags):
-                continue
-            if queries and not _query_matches(metadata, queries):
-                continue
-            if (tasks or specs) and not self._schema_matches(metadata.dataset_id, tasks, specs):
-                continue
-            results.append(metadata)
-        return results
-
-    def _schema_matches(self, dataset_id: str, tasks: list[type[Task]], specs: list[str]) -> bool:
-        """Return whether the manifest schema of a dataset satisfies the type filters."""
-        schema = self.get_manifest(dataset_id).dataset_schema
-        if tasks and not set(tasks) & set(schema.tasks):
-            return False
-        return not (specs and not set(specs) <= {spec.spec_type for spec in schema.time_series_specs})
-
-
-def _as_list(value: T | list[T] | None) -> list[T]:
-    """Normalize a filter value to a list.
-
-    Args:
-        value: ``None``, a single value, or a list of values.
-
-    Returns:
-        An empty list, the list itself, or a list with one value.
-    """
-    if value is None:
-        return []
-    if isinstance(value, list):
-        return value
-    return [value]
-
-
-def _query_matches(metadata: DatasetMetadata, terms: list[str]) -> bool:
-    """Return whether any term is a case-insensitive substring of the dataset name, description, or tags."""
-    haystack = " ".join([metadata.name, metadata.description, *metadata.tags]).lower()
-    return any(term.lower() in haystack for term in terms)
+        # Sorted, so that the limit keeps the first matches by dataset ID, also for a catalog in another order.
+        rows = sorted(self.list_datasets(), key=lambda row: row.dataset_id)
+        return filter.select(ListedRelease(row, self) for row in rows)
diff --git a/packages/timenet/src/timenet/registry/s3.py b/packages/timenet/src/timenet/registry/s3.py
index d2d0b338b..b9b227a14 100644
--- a/packages/timenet/src/timenet/registry/s3.py
+++ b/packages/timenet/src/timenet/registry/s3.py
@@ -15,7 +15,7 @@
 from contextlib import contextmanager
 import os
 from pathlib import Path
-from typing import Any, BinaryIO
+from typing import Any, BinaryIO, ClassVar
 from urllib.parse import urlparse
 import uuid
 
@@ -26,7 +26,7 @@
 from timenet.manifest import Manifest, parse_manifest
 from timenet.registry.lazy_read import LazyReadRegistry
 from timenet.registry.version import DatasetVersion
-from timenet.types import DatasetMetadata, Version
+from timenet.types import DatasetFilter, DatasetMetadata, Version
 from timenet.types.metadata_fields import validate_dataset_id
 from timenet.types.references import DatasetRef, is_latest
 
@@ -39,6 +39,7 @@
 class S3Registry(LazyReadRegistry):
     """A registry that reads and publishes datasets under an S3 bucket or prefix."""
 
+    has_catalog: ClassVar[bool] = False
     _staging_prefix = "timenet-s3-"
 
     def __init__(self, uri: str, *, read_only: bool = False) -> None:
@@ -72,7 +73,7 @@ def list_datasets(self) -> list[DatasetMetadata]:
             "explicit org/name@version."
         )
 
-    def search(self, **_kwargs: Any) -> list[DatasetMetadata]:
+    def search(self, filter: DatasetFilter) -> list[DatasetMetadata]:
         """Unsupported: the S3 backend has no catalog.
 
         Raises:
diff --git a/packages/timenet/src/timenet/sources.py b/packages/timenet/src/timenet/sources.py
index 3689f8baa..9f847e3d0 100644
--- a/packages/timenet/src/timenet/sources.py
+++ b/packages/timenet/src/timenet/sources.py
@@ -2,10 +2,12 @@
 
 from collections.abc import Callable, Sequence
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, TypeAlias, Union
+from typing import TYPE_CHECKING, ClassVar, TypeAlias, Union
 
 from timenet import builders
-from timenet.errors import TimeNetAccessError
+from timenet._warnings import warn_at_caller
+from timenet.errors import ConnectorCardWarning, TimeNetAccessError, TimeNetInvalidCardError
+from timenet.types import DatasetMetadata
 from timenet.types.access import Access, AccessRequirement
 from timenet.types.references import LATEST_VERSION, DatasetRef
 from timenet.types.version import Version
@@ -96,12 +98,15 @@ def access_error(self, dataset_id: str) -> TimeNetAccessError:
 
 @dataclass(frozen=True)
 class InstalledConnectors:
-    """A release source backed by installed dataset connectors.
+    """A source for the datasets that installed connectors build.
 
-    Each connector provides only the version in its card.
-    All instances compare equal.
+    A connector counts as installed if a package registers it under the ``timenet.builders`` entry point group. Each
+    connector provides only the version its card declares, and all instances are equal.
     """
 
+    has_catalog: ClassVar[bool] = True
+    """Whether :meth:`list_datasets` works. It always does for installed connectors."""
+
     def __str__(self) -> str:
         if not tuple(builders.entry_points(group=builders.ENTRY_POINT_GROUP)):
             return "installed connectors (no connector package is installed)"
@@ -127,6 +132,38 @@ def resolve(self, dataset_id: str, version: Version | None) -> ReleaseLookup:  #
             return NotFound()
         return FoundInConnector(builder)
 
+    def list_datasets(self) -> list[DatasetMetadata]:
+        """Return dataset cards from installed connectors without building the datasets.
+
+        The first entry point that claims a dataset takes priority.
+        The method skips unreadable cards and emits a :class:`~timenet.errors.ConnectorCardWarning`.
+
+        Returns:
+            The cards, sorted by dataset ID.
+
+        Raises:
+            TimeNetInvalidCardError: If a backend cannot list its dataset IDs.
+        """  # ruff: ignore[docstring-extraneous-exception] - raised by the backend
+        return self._cards()
+
+    def _cards(self) -> list[DatasetMetadata]:  # ruff: ignore[no-self-use] (the listing of this source)
+        """Return the readable cards of installed connectors, sorted by dataset ID."""
+        cards: list[DatasetMetadata] = []
+        for claim in builders.claims():
+            try:
+                card = claim.backend.card(claim.dataset_id)
+            except TimeNetInvalidCardError as error:
+                warn_at_caller(
+                    ConnectorCardWarning(
+                        f"{claim.dataset_id}: the card of its installed connector cannot be read, "
+                        f"so the listing leaves it out: {error}"
+                    )
+                )
+                continue
+            if card is not None:
+                cards.append(card)
+        return sorted(cards, key=lambda card: card.dataset_id)
+
 
 Ask: TypeAlias = Callable[[ReleaseSource, str, Version | None], ReleaseLookup]
 """A function that asks one source for a release: ``ask(source, dataset_id, version)``."""
diff --git a/packages/timenet/src/timenet/types/__init__.py b/packages/timenet/src/timenet/types/__init__.py
index f1917b18f..7fcface81 100644
--- a/packages/timenet/src/timenet/types/__init__.py
+++ b/packages/timenet/src/timenet/types/__init__.py
@@ -15,7 +15,7 @@
 from timenet.types.ids import EntityId, check_id
 from timenet.types.label_sets import Label, LabelSet, LabelSetRegistry
 from timenet.types.licenses import License
-from timenet.types.metadata import DatasetMetadata, DatasetSchema
+from timenet.types.metadata import DatasetFilter, DatasetMetadata, DatasetSchema, ListedRelease
 from timenet.types.modalities import InputModality
 from timenet.types.object_kind import ObjectKind
 from timenet.types.spans import (
@@ -59,6 +59,7 @@
     "AnswerTask",
     "ArrayHandle",
     "ClassificationTask",
+    "DatasetFilter",
     "DatasetMetadata",
     "DatasetSchema",
     "Domain",
@@ -71,6 +72,7 @@
     "LazyText",
     "LazyTextList",
     "License",
+    "ListedRelease",
     "LocalizationMode",
     "MissingReason",
     "MissingValues",
diff --git a/packages/timenet/src/timenet/types/metadata.py b/packages/timenet/src/timenet/types/metadata.py
index 8dc275e09..9378bcddf 100644
--- a/packages/timenet/src/timenet/types/metadata.py
+++ b/packages/timenet/src/timenet/types/metadata.py
@@ -1,7 +1,9 @@
 """Dataset-level descriptive identity and derived type declaration."""
 
+from collections.abc import Iterable
+from dataclasses import dataclass
 from pathlib import Path
-from typing import Annotated, Any, Self
+from typing import TYPE_CHECKING, Annotated, Any, Self, TypeVar
 
 from pydantic import (
     Field,
@@ -26,6 +28,13 @@
 from timenet.types.tasks import TASKS, Task, TaskType
 
 
+if TYPE_CHECKING:
+    from timenet.registry.base import BaseRegistry
+
+
+T = TypeVar("T")
+
+
 class DatasetMetadata(TimeFModel):
     """Dataset identity, release version, source information, and access requirements.
 
@@ -194,3 +203,141 @@ class DatasetSchema(TimeFModel):
     """Type-level descriptors for the dataset's annotation keys."""
     tasks: tuple[TaskDeclaration, ...] = ()
     """Built-in ``Task`` subclasses the dataset declares, resolved from the registry."""
+
+
+@dataclass(frozen=True)
+class ListedRelease:
+    """A card of a dataset listing, with the registry that lists it."""
+
+    card: DatasetMetadata
+    """The card of the release."""
+    registry: "BaseRegistry | None"
+    """The registry that holds the manifest of the release, or ``None`` for the card of an installed connector."""
+
+
+@dataclass(frozen=True)
+class DatasetFilter:
+    """The filters of a dataset search. Filters combine with AND, and an empty filter matches every dataset."""
+
+    query: tuple[str, ...] = ()
+    """Terms of which any must appear in the name, description, or tags, ignoring case."""
+    domain: tuple[Domain, ...] = ()
+    """Domains of which the dataset must have any."""
+    task: tuple[type[Task], ...] = ()
+    """Task classes of which the schema must declare any."""
+    license: tuple[License, ...] = ()
+    """Licenses of which the dataset must have any."""
+    time_series_spec: tuple[str, ...] = ()
+    """``spec_type`` values that the schema must declare all of."""
+    dataset_id: tuple[str, ...] = ()
+    """IDs of which the dataset must have any."""
+    tag: tuple[str, ...] = ()
+    """Tags that the dataset must declare all of."""
+    limit: int = 100
+    """The maximum number of results."""
+
+    def __post_init__(self) -> None:
+        if self.limit < 0:
+            raise TimeFValidationError(f"limit must be 0 or more, got {self.limit}")
+
+    @classmethod
+    def of(  # ruff: ignore[too-many-arguments]
+        cls,
+        *,
+        query: str | list[str] | None = None,
+        domain: Domain | list[Domain] | None = None,
+        task: type[Task] | list[type[Task]] | None = None,
+        license: License | list[License] | None = None,
+        time_series_spec: str | list[str] | None = None,
+        dataset_id: str | list[str] | None = None,
+        tag: str | list[str] | None = None,
+        limit: int = 100,
+    ) -> Self:
+        """Build a filter from one value, a list, or None for each filter.
+
+        Args:
+            query: Match any term in the name, description, or tags, ignoring case.
+            domain: Match any of these domains.
+            task: Match any of these task classes in the schema.
+            license: Match any of these licenses.
+            time_series_spec: Match all of these ``spec_type`` values.
+            dataset_id: Match any of these IDs.
+            tag: Match all of these tags.
+            limit: The maximum number of results.
+
+        Returns:
+            The filter. ``None`` or an empty list sets no restriction.
+
+        Raises:
+            TimeFValidationError: If ``limit`` is negative.
+        """  # ruff: ignore[docstring-extraneous-exception] - raised by __post_init__
+        return cls(
+            query=_as_tuple(query),
+            domain=_as_tuple(domain),
+            task=_as_tuple(task),
+            license=_as_tuple(license),
+            time_series_spec=_as_tuple(time_series_spec),
+            dataset_id=_as_tuple(dataset_id),
+            tag=_as_tuple(tag),
+            limit=limit,
+        )
+
+    @property
+    def needs_schema(self) -> bool:
+        """Whether a ``task`` or ``time_series_spec`` filter is set. Both read the schema of a manifest."""
+        return bool(self.task or self.time_series_spec)
+
+    def matches_card(self, card: DatasetMetadata) -> bool:
+        """Return whether a card matches the ``query``, ``domain``, ``license``, ``dataset_id`` and ``tag`` filters."""
+        if self.dataset_id and card.dataset_id not in self.dataset_id:
+            return False
+        if self.license and card.license not in self.license:
+            return False
+        if self.domain and not set(card.domains) & set(self.domain):
+            return False
+        if not set(self.tag) <= set(card.tags):
+            return False
+        text = " ".join([card.name, card.description, *card.tags]).lower()
+        return not self.query or any(term.lower() in text for term in self.query)
+
+    def matches_schema(self, schema: DatasetSchema) -> bool:
+        """Return whether a schema matches the ``task`` and ``time_series_spec`` filters."""
+        if self.task and not set(self.task) & set(schema.tasks):
+            return False
+        return set(self.time_series_spec) <= {spec.spec_type for spec in schema.time_series_specs}
+
+    def select(self, listed: Iterable[ListedRelease]) -> list[DatasetMetadata]:
+        """Return the cards of the listed releases that match, in order, with at most ``limit`` entries.
+
+        A ``task`` or ``time_series_spec`` filter reads the schema of a release only if its card matches. A release
+        without a registry has no schema, so it never matches these filters.
+
+        Args:
+            listed: The releases to filter.
+
+        Returns:
+            The matching cards.
+        """
+        results: list[DatasetMetadata] = []
+        for release in listed:
+            if len(results) >= self.limit:
+                break
+            if not self.matches_card(release.card):
+                continue
+            if self.needs_schema:
+                if release.registry is None:
+                    continue
+                manifest = release.registry.get_manifest(release.card.dataset_id, str(release.card.dataset_version))
+                if not self.matches_schema(manifest.dataset_schema):
+                    continue
+            results.append(release.card)
+        return results
+
+
+def _as_tuple(value: T | list[T] | None) -> tuple[T, ...]:
+    """Return a filter value as a tuple: empty for ``None``, the items of a list, or the one value."""
+    if value is None:
+        return ()
+    if isinstance(value, list):
+        return tuple(value)
+    return (value,)
diff --git a/packages/timenet/tests/registry/test_registry.py b/packages/timenet/tests/registry/test_registry.py
index 625ff104b..a4ea35ec1 100644
--- a/packages/timenet/tests/registry/test_registry.py
+++ b/packages/timenet/tests/registry/test_registry.py
@@ -19,7 +19,7 @@
     open_registry,
 )
 from timenet.testing import make_dataset
-from timenet.types import Annotation, AnswerTask, Domain, License
+from timenet.types import Annotation, AnswerTask, DatasetFilter, Domain
 from timenet.types.references import DatasetRef
 from timenet.writer import TimeFWriter
 
@@ -191,66 +191,35 @@ def test_open_release_checks_the_size_of_each_file(registry_root):
 
 
 def test_search_no_filters_returns_all(registry_root):
-    assert len({m.dataset_id for m in LocalRegistry(registry_root).search()}) == 2
+    assert len({m.dataset_id for m in LocalRegistry(registry_root).search(DatasetFilter())}) == 2
 
 
 def test_search_by_domain(registry_root):
-    results = LocalRegistry(registry_root).search(domain=Domain.CARDIOLOGY)
+    results = LocalRegistry(registry_root).search(DatasetFilter(domain=(Domain.CARDIOLOGY,)))
     assert {m.dataset_id for m in results} == {"demo/ecg"}
 
 
-def test_search_by_license(registry_root):
-    assert {m.dataset_id for m in LocalRegistry(registry_root).search(license=License.MIT)} == {"demo/ecg"}
-
-
 def test_search_by_task_type_filter(registry_root):
     # only hello_world has a QA task
-    assert {m.dataset_id for m in LocalRegistry(registry_root).search(task=AnswerTask)} == {"timenet/hello-world"}
-
-
-def test_search_by_time_series_spec(registry_root):
-    assert {m.dataset_id for m in LocalRegistry(registry_root).search(time_series_spec="ecg_lead")} == {"demo/ecg"}
-
-
-def test_search_by_tag(registry_root):
-    assert {m.dataset_id for m in LocalRegistry(registry_root).search(tag="clinical")} == {"demo/ecg"}
-
-
-def test_search_by_dataset_id(registry_root):
-    assert {m.dataset_id for m in LocalRegistry(registry_root).search(dataset_id="timenet/hello-world")} == {
+    assert {m.dataset_id for m in LocalRegistry(registry_root).search(DatasetFilter(task=(AnswerTask,)))} == {
         "timenet/hello-world"
     }
 
 
-def test_search_by_query_substring(registry_root):
-    assert {m.dataset_id for m in LocalRegistry(registry_root).search(query="hello")} == {"timenet/hello-world"}
-
-
-def test_search_accepts_scalar_or_list(registry_root):
-    both = LocalRegistry(registry_root).search(domain=[Domain.CARDIOLOGY, Domain.GENERAL])
-    assert {m.dataset_id for m in both} == {"timenet/hello-world", "demo/ecg"}
+class _Reversed(LocalRegistry):
+    """A local registry whose catalog lists the datasets in reverse order of their IDs."""
 
+    def list_datasets(self):
+        return super().list_datasets()[::-1]
 
-def test_search_filters_are_anded(registry_root):
-    # cardiology AND MIT => ecg; cardiology AND CC-BY-4.0 => none
-    assert {
-        m.dataset_id for m in LocalRegistry(registry_root).search(domain=Domain.CARDIOLOGY, license=License.MIT)
-    } == {"demo/ecg"}
-    assert LocalRegistry(registry_root).search(domain=Domain.CARDIOLOGY, license=License.CC_BY_4_0) == []
 
-
-def test_search_limit(registry_root):
-    assert len(LocalRegistry(registry_root).search(limit=1)) == 1
+def test_search_returns_the_first_matches_by_dataset_id(registry_root):
+    assert [m.dataset_id for m in _Reversed(registry_root).search(DatasetFilter(limit=1))] == ["demo/ecg"]
 
 
 def test_search_limit_zero_returns_empty(registry_root):
     # the limit check must run before the append, else a zero limit still yields one row
-    assert LocalRegistry(registry_root).search(limit=0) == []
-
-
-def test_search_negative_limit_rejected(registry_root):
-    with pytest.raises(ValueError, match="non-negative"):
-        LocalRegistry(registry_root).search(limit=-1)
+    assert LocalRegistry(registry_root).search(DatasetFilter(limit=0)) == []
 
 
 # ---- put_dataset (write side) -----------------------------------------------------------------
@@ -337,7 +306,7 @@ def test_s3_registry_has_no_catalog():
     with pytest.raises(NotImplementedError):
         s3.list_datasets()
     with pytest.raises(NotImplementedError):
-        s3.search()
+        s3.search(DatasetFilter())
 
 
 def test_a_read_only_registry_refuses_put_dataset_and_a_copy_into_it(tmp_path):
diff --git a/packages/timenet/tests/registry/test_remote.py b/packages/timenet/tests/registry/test_remote.py
index 638f317cb..467dfe285 100644
--- a/packages/timenet/tests/registry/test_remote.py
+++ b/packages/timenet/tests/registry/test_remote.py
@@ -26,7 +26,7 @@
 from timenet.registry import LocalRegistry, RemoteRegistry, copy_release, open_local_version
 from timenet.sources import FoundInRegistry, NotServed
 from timenet.testing import assert_datasets_equal, make_dataset
-from timenet.types import Access, DatasetSchema, Domain, License, Version
+from timenet.types import Access, DatasetFilter, DatasetSchema, Domain, License, Version
 from timenet.types.access import AccessRequirement
 from timenet.types.parent_dataset_card import ParentDatasetCard
 from timenet.types.reference_keys import ParentAlias
@@ -172,8 +172,10 @@ def test_search_filters_by_domain(version_dir, tmp_path):
     registry, _ = _remote(version_dir, tmp_path)
     _, manifest = version_dir
     domain = manifest.metadata.domains[0]
-    assert registry.search(domain=domain)
-    assert registry.search(domain=Domain.FINANCE) == [] or manifest.metadata.domains == (Domain.FINANCE,)
+    assert registry.search(DatasetFilter(domain=(domain,)))
+    assert registry.search(DatasetFilter(domain=(Domain.FINANCE,))) == [] or manifest.metadata.domains == (
+        Domain.FINANCE,
+    )
 
 
 def test_open_file_streams_bytes(version_dir, tmp_path):
diff --git a/packages/timenet/tests/registry/test_s3.py b/packages/timenet/tests/registry/test_s3.py
index 49d89cca9..75bbca840 100644
--- a/packages/timenet/tests/registry/test_s3.py
+++ b/packages/timenet/tests/registry/test_s3.py
@@ -17,7 +17,7 @@
 from timenet.reader import TimeFReader  # ruff: ignore[module-import-not-at-top-of-file]
 from timenet.registry import LocalRegistry, S3Registry, open_local_version  # ruff: ignore[module-import-not-at-top-of-file]
 from timenet.testing import assert_datasets_equal, make_dataset  # ruff: ignore[module-import-not-at-top-of-file]
-from timenet.types import DatasetSchema, Version  # ruff: ignore[module-import-not-at-top-of-file]
+from timenet.types import DatasetFilter, DatasetSchema, Version  # ruff: ignore[module-import-not-at-top-of-file]
 from timenet.types.references import DatasetRef  # ruff: ignore[module-import-not-at-top-of-file]
 
 
@@ -133,7 +133,7 @@ def test_list_and_search_are_unsupported(tmp_path):
     with pytest.raises(NotImplementedError):
         registry.list_datasets()
     with pytest.raises(NotImplementedError):
-        registry.search()
+        registry.search(DatasetFilter())
 
 
 def test_rejects_a_non_s3_uri(tmp_path):
diff --git a/packages/timenet/tests/test_builders.py b/packages/timenet/tests/test_builders.py
index 0809f5fea..1f1727342 100644
--- a/packages/timenet/tests/test_builders.py
+++ b/packages/timenet/tests/test_builders.py
@@ -31,6 +31,18 @@ def test_installed_builder_returns_the_first_backend_whose_dataset_ids_hold_the_
     assert installed_builder("x/y") is None
 
 
+def test_installed_builder_reads_no_backend_after_the_first_that_lists_the_id(monkeypatch):
+    class _Unread(_Backend):
+        def dataset_ids(self):
+            raise AssertionError("the lookup reads no backend after the match")
+
+    card = make_dataset().metadata
+    install_backends(monkeypatch, _Backend(card.dataset_id, card=card), _Unread())
+    found = installed_builder(card.dataset_id)
+    assert found is not None
+    assert found.card == card
+
+
 def test_declared_returns_the_release_of_the_card_and_refuses_another_version(monkeypatch):
     card = make_dataset().metadata
     install_backends(monkeypatch, _Backend(card.dataset_id, card=card))
diff --git a/packages/timenet/tests/test_cli.py b/packages/timenet/tests/test_cli.py
index 4eafe0488..f943a6dcb 100644
--- a/packages/timenet/tests/test_cli.py
+++ b/packages/timenet/tests/test_cli.py
@@ -17,6 +17,7 @@
 from timenet.errors import (
     TimeFValidationError,
     TimeNetDatasetNotFoundError,
+    TimeNetInvalidCardError,
     TimeNetMissingParentError,
     TimeNetRegistryError,
 )
@@ -157,10 +158,15 @@ def test_the_local_commands_take_the_local_registry_flag(tmp_path):
     assert stored_releases(LocalRegistry(store)) == []
 
 
-def test_search_no_match(registry_root):
-    result = runner.invoke(app, ["search", "--remote-registry", str(registry_root), "--domain", "cardiology"])
+def _flags(tmp_path, remote):
+    """Return the flags of a list of sources: an empty store and ``remote``, without the connectors."""
+    return ["--local-registry", str(tmp_path / "store"), "--remote-registry", str(remote), "--no-auto-build"]
+
+
+def test_search_no_match(registry_root, tmp_path):
+    result = runner.invoke(app, ["search", "--domain", "cardiology", *_flags(tmp_path, registry_root)])
     assert result.exit_code == 0, result.output
-    assert click.unstyle(result.output) == "No datasets match.\n"
+    assert result.stdout == "No datasets match.\n"
 
 
 class _Backend:
@@ -180,6 +186,24 @@ def card(self, dataset_id):
         )
 
 
+def test_list_shows_a_warning_as_a_line_without_a_file_and_a_line_number(tmp_path, monkeypatch):
+    class _Broken(_Backend):
+        def card(self, dataset_id):
+            raise TimeNetInvalidCardError("the card cannot be read")
+
+    install_backends(monkeypatch, _Broken())
+    monkeypatch.setattr(_cli_module.ui, "rich", Console(stderr=True, width=1000))
+    flags = ["--local-registry", str(tmp_path / "store"), "--remote-registry", str(tmp_path / "empty"), "--auto-build"]
+
+    result = runner.invoke(app, ["list", *flags])
+
+    assert result.exit_code == 0, result.output
+    assert click.unstyle(result.stderr).splitlines() == [
+        "⚠️ test/gated: the card of its installed connector cannot be read, so the listing leaves it out: "
+        "the card cannot be read"
+    ]
+
+
 def test_info_shows_the_card_of_a_dataset_that_only_a_connector_has(tmp_path, monkeypatch):
     install_backends(monkeypatch, _Backend())
     store = tmp_path / "store"
@@ -454,8 +478,8 @@ def test_an_empty_flag_is_an_error(tmp_path, monkeypatch, argv):
     assert f"must not be empty (also as {argv[-2]})" in str(result.exception)
 
 
-def test_search_rejects_unknown_filter_value(registry_root):
-    assert runner.invoke(app, ["search", "--remote-registry", str(registry_root), "--domain", "bogus"]).exit_code == 2
+def test_search_rejects_unknown_filter_value(registry_root, tmp_path):
+    assert runner.invoke(app, ["search", "--domain", "bogus", *_flags(tmp_path, registry_root)]).exit_code == 2
     # The error message is a Rich-rendered panel, so assert it on the raising helper directly.
     with pytest.raises(typer.BadParameter, match="invalid --domain"):
         _enum_list(["bogus"], Domain, "--domain", [d.value for d in Domain])
diff --git a/packages/timenet/tests/test_client.py b/packages/timenet/tests/test_client.py
index 967c013e5..0f24ad853 100644
--- a/packages/timenet/tests/test_client.py
+++ b/packages/timenet/tests/test_client.py
@@ -14,11 +14,15 @@
 from timenet.dataset import Record, RegularAxis, Signal, Source, TimeFDataset
 from timenet.engine import open_parents, run_pipeline
 from timenet.errors import (
+    CatalogWarning,
+    ConnectorCardWarning,
     TimeFValidationError,
     TimeNetAccessError,
     TimeNetDatasetNotFoundError,
     TimeNetFileError,
+    TimeNetInvalidCardError,
     TimeNetMissingParentError,
+    TimeNetNoCatalogError,
     TimeNetNotBuiltError,
     TimeNetRegistryError,
     TimeNetVersionConflictError,
@@ -31,7 +35,7 @@
 from timenet.registry import TIMENET_REGISTRY_URL, LocalRegistry, RemoteRegistry, open_local_version
 from timenet.sources import InstalledConnectors
 from timenet.testing import assert_datasets_equal, install_backends, make_dataset
-from timenet.types import Access, Annotation, DatasetMetadata, License, Version
+from timenet.types import Access, Annotation, AnswerTask, DatasetMetadata, Domain, License, Version
 from timenet.types.access import AccessRequirement
 from timenet.types.parent_dataset_card import ParentDatasetCard
 from timenet.types.reference_keys import ParentAlias
@@ -309,18 +313,170 @@ def test_installed_connectors_need_a_local_registry_that_is_not_read_only(tmp_pa
     )
 
 
-def test_list_and_search_ask_the_first_registry_that_is_not_the_store(registry_root, tmp_path):
-    store = LocalRegistry(tmp_path / "store")
-    other = make_dataset()
-    other._metadata = other.metadata.model_copy(update={"dataset_id": "test/other"})
-    store.put_dataset(other)
-    client = TimeNet(sources=[store, str(registry_root), InstalledConnectors()])
+class _CardsBackend:
+    """A builder backend with a connector for each dataset ID of ``versions``, at its version.
+
+    Each card is the card of the demo dataset, with ``update`` applied.
+    """
+
+    def __init__(self, versions, **update):
+        self.versions = versions
+        self.update = update
+
+    def dataset_ids(self):
+        return list(self.versions)
+
+    def card(self, dataset_id):
+        if dataset_id not in self.versions:
+            return None
+        return make_dataset(dataset_id=dataset_id, dataset_version=self.versions[dataset_id], **self.update).metadata
+
+
+def _rows(rows):
+    return [(row.dataset_id, str(row.dataset_version)) for row in rows]
+
+
+def test_list_merges_every_source_at_the_highest_version(tmp_path, monkeypatch):
+    store, remote = LocalRegistry(tmp_path / "store"), LocalRegistry(tmp_path / "remote")
+    store.put_dataset(make_dataset(dataset_id="a/one", dataset_version="1.0.0"))
+    remote.put_dataset(make_dataset(dataset_id="a/one", dataset_version="1.1.0"))
+    remote.put_dataset(make_dataset(dataset_id="b/two", dataset_version="1.0.0"))
+    install_backends(monkeypatch, _CardsBackend({"a/one": "2.0.0", "c/three": "1.0.0"}))
+    rows = TimeNet(sources=[store, remote, InstalledConnectors()]).list()
+    assert [(row.dataset_id, str(row.dataset_version)) for row in rows] == [
+        ("a/one", "2.0.0"),
+        ("b/two", "1.0.0"),
+        ("c/three", "1.0.0"),
+    ]
+
+
+def test_list_keeps_the_row_of_the_earlier_source_for_the_same_version(tmp_path):
+    first, second = LocalRegistry(tmp_path / "first"), LocalRegistry(tmp_path / "second")
+    for registry in (first, second):
+        registry.put_dataset(make_dataset(dataset_id="a/one", dataset_version="1.0.0", name=registry.root.name))
+    assert [row.name for row in TimeNet(sources=[second, first]).list()] == ["second"]
+    assert [row.name for row in TimeNet(sources=[first, second]).search(query="demo")] == ["first"]
+
+
+class _WithoutCatalog(LocalRegistry):
+    """A local registry without a catalog, as an S3 registry is."""
+
+    has_catalog = False
+
+    def list_datasets(self):
+        raise NotImplementedError("no catalog")
+
+
+def test_list_leaves_out_a_registry_without_a_catalog_with_a_warning(tmp_path, registry_root):
+    hidden = _WithoutCatalog(tmp_path / "hidden")
+    hidden.put_dataset(make_dataset(dataset_id="a/hidden", dataset_version="1.0.0"))
+    client = TimeNet(sources=[hidden, str(registry_root)])
+    message = f"^local registry {re.escape(str(hidden.root))} has no catalog, so the list leaves it out\\.$"
+    with pytest.warns(CatalogWarning, match=message):
+        assert _rows(client.list()) == [("timenet/hello-world", "1.0.0")]
+    with pytest.warns(CatalogWarning, match=message):
+        assert _rows(client.search(query="hello")) == [("timenet/hello-world", "1.0.0")]
+
+
+def test_list_and_search_raise_if_no_source_has_a_catalog(tmp_path):
+    client = TimeNet(sources=[_WithoutCatalog(tmp_path / "hidden")])
+    message = f"^none of the sources has a catalog: local registry {re.escape(str(tmp_path / 'hidden'))}$"
+    with pytest.raises(TimeNetNoCatalogError, match=message):
+        client.list()
+    with pytest.raises(TimeNetNoCatalogError, match=message):
+        client.search()
 
-    assert [metadata.dataset_id for metadata in client.list()] == ["timenet/hello-world"]
-    assert [metadata.dataset_id for metadata in client.search(query="hello")] == ["timenet/hello-world"]
-    # Without such a registry, they ask the store.
-    alone = TimeNet(sources=[store, InstalledConnectors()])
-    assert [metadata.dataset_id for metadata in alone.list()] == ["test/other"]
+
+def test_list_and_search_pass_on_a_not_implemented_error_of_a_source_with_a_catalog(tmp_path):
+    class _Broken(_WithoutCatalog):
+        has_catalog = True
+
+    client = TimeNet(sources=[_Broken(tmp_path / "broken")])
+    with pytest.raises(NotImplementedError, match=r"^no catalog$"):
+        client.list()
+    with pytest.raises(NotImplementedError, match=r"^no catalog$"):
+        client.search()
+
+
+class _BrokenCard(_CardsBackend):
+    def card(self, dataset_id):
+        if dataset_id == "b/broken":
+            raise TimeNetInvalidCardError("the card has no name")
+        return super().card(dataset_id)
+
+
+def test_list_leaves_out_a_broken_connector_card_with_a_warning(tmp_path, monkeypatch):
+    install_backends(monkeypatch, _BrokenCard({"a/one": "1.0.0", "b/broken": "1.0.0"}))
+    client = TimeNet(sources=[LocalRegistry(tmp_path / "store"), InstalledConnectors()])
+
+    with pytest.warns(ConnectorCardWarning, match="cannot be read") as caught:
+        rows = client.list()
+
+    assert _rows(rows) == [("a/one", "1.0.0")]
+    assert [str(warning.message) for warning in caught] == [
+        "b/broken: the card of its installed connector cannot be read, so the listing leaves it out: "
+        "the card has no name"
+    ]
+    # The warning points at the caller of list().
+    assert [warning.filename for warning in caught] == [__file__]
+
+
+class _Reversed(LocalRegistry):
+    """A local registry whose catalog lists the datasets in reverse order of their IDs."""
+
+    def list_datasets(self):
+        return super().list_datasets()[::-1]
+
+
+def test_search_returns_the_first_matches_by_dataset_id_at_their_highest_version(tmp_path):
+    first, second = _Reversed(tmp_path / "first"), LocalRegistry(tmp_path / "second")
+    for dataset_id in ("a/one", "b/two", "c/three"):
+        first.put_dataset(make_dataset(dataset_id=dataset_id, dataset_version="1.0.0"))
+    second.put_dataset(make_dataset(dataset_id="b/two", dataset_version="2.0.0"))
+    second.put_dataset(make_dataset(dataset_id="d/four", dataset_version="1.0.0"))
+    client = TimeNet(sources=[first, second])
+    # The merge sorts the rows by dataset ID, so the limit keeps the first two.
+    assert _rows(client.search(query="hello", limit=2)) == [("a/one", "1.0.0"), ("b/two", "2.0.0")]
+    assert client.search(query="hello", limit=0) == []
+
+
+class _CountingReads(LocalRegistry):
+    """A local registry that records each manifest that it reads outside its listing."""
+
+    def __init__(self, root, reads):
+        super().__init__(root)
+        self.reads = reads
+
+    def list_datasets(self):
+        return LocalRegistry(self.root).list_datasets()
+
+    def get_manifest(self, dataset_id, version=None):
+        self.reads.append(f"{dataset_id}@{version}")
+        return super().get_manifest(dataset_id, version)
+
+
+def test_search_reads_the_schemas_of_the_merged_rows_only_up_to_the_limit(tmp_path):
+    reads = []
+    first, second = _CountingReads(tmp_path / "first", reads), _CountingReads(tmp_path / "second", reads)
+    first.put_dataset(make_dataset(dataset_id="a/one", domains=(Domain.CARDIOLOGY,)))
+    first.put_dataset(make_dataset(dataset_id="c/three"))
+    second.put_dataset(make_dataset(dataset_id="b/two"))
+    second.put_dataset(make_dataset(dataset_id="d/four"))
+    reads.clear()
+    rows = TimeNet(sources=[first, second]).search(domain=Domain.GENERAL, task=AnswerTask, limit=1)
+    assert _rows(rows) == [("b/two", "1.0.0")]
+    # a/one fails the domain filter, and b/two is the first match of both registries.
+    assert reads == ["b/two@1.0.0"]
+
+
+def test_search_matches_a_connector_card_by_its_card_only(tmp_path, monkeypatch):
+    install_backends(monkeypatch, _CardsBackend({"a/one": "1.0.0", "c/three": "1.0.0"}))
+    store = LocalRegistry(tmp_path / "store")
+    store.put_dataset(make_dataset(dataset_id="b/two"))
+    client = TimeNet(sources=[store, InstalledConnectors()])
+    assert _rows(client.search(dataset_id=["b/two", "c/three"])) == [("b/two", "1.0.0"), ("c/three", "1.0.0")]
+    # A connector card has no manifest, so a schema filter leaves it out.
+    assert _rows(client.search(task=AnswerTask)) == [("b/two", "1.0.0")]
 
 
 def test_get_manifest_returns_the_manifest_of_the_release_that_open_reads(tmp_path):
diff --git a/packages/timenet/tests/test_sources.py b/packages/timenet/tests/test_sources.py
index 8a0c7e518..ab4a4bf90 100644
--- a/packages/timenet/tests/test_sources.py
+++ b/packages/timenet/tests/test_sources.py
@@ -3,7 +3,9 @@
 import pytest
 
 from timenet.errors import (
+    ConnectorCardWarning,
     TimeFValidationError,
+    TimeNetInvalidCardError,
     TimeNetRegistryError,
 )
 from timenet.registry import LocalRegistry
@@ -55,6 +57,23 @@ def card(self, dataset_id):
         return self._card if dataset_id == self._card.dataset_id else None
 
 
+class _Cards:
+    """A backend with a connector for each card."""
+
+    def __init__(self, *cards):
+        self.cards = {card.dataset_id: card for card in cards}
+
+    def dataset_ids(self):
+        return list(self.cards)
+
+    def card(self, dataset_id):
+        return self.cards.get(dataset_id)
+
+
+def _card(dataset_id, name="Hello World"):
+    return make_dataset().metadata.model_copy(update={"dataset_id": dataset_id, "name": name})
+
+
 def _fail(*args, **kwargs):
     raise AssertionError("resolve() must not read every manifest")
 
@@ -136,6 +155,44 @@ def test_installed_connectors_without_a_package_answer_not_found():
     assert "no connector package is installed" in str(InstalledConnectors())
 
 
+def test_installed_connectors_list_their_cards(monkeypatch):
+    first = _Cards(_card("z/last"), _card("a/first", "First"))
+    second = _Cards(_card("a/first", "Second"), _card("m/middle"))
+    install_backends(monkeypatch, first, second)
+    # Sorted by dataset ID. The first entry point wins for an ID that two claim, as it does for a build.
+    assert [(card.dataset_id, card.name) for card in InstalledConnectors().list_datasets()] == [
+        ("a/first", "First"),
+        ("m/middle", "Hello World"),
+        ("z/last", "Hello World"),
+    ]
+
+
+def test_installed_connectors_leave_out_a_card_that_cannot_be_read(monkeypatch):
+    class Broken(_Cards):
+        def card(self, dataset_id):
+            raise TimeNetInvalidCardError("the card cannot be read")
+
+    # The second backend claims the dataset too, but the first entry point wins, as it does for a build.
+    install_backends(monkeypatch, Broken(_card("a/broken")), _Cards(_card("a/broken"), _card("b/kept")))
+    with pytest.warns(
+        ConnectorCardWarning, match=r"^a/broken: the card of its installed connector cannot be read"
+    ) as caught:
+        cards = InstalledConnectors().list_datasets()
+    assert [card.dataset_id for card in cards] == ["b/kept"]
+    # The warning points at the caller of list_datasets().
+    assert [warning.filename for warning in caught] == [__file__]
+
+
+def test_an_error_of_the_dataset_ids_of_a_backend_propagates_from_the_cards(monkeypatch):
+    class Broken(_Cards):
+        def dataset_ids(self):
+            raise TimeNetInvalidCardError("the cards cannot be listed")
+
+    install_backends(monkeypatch, Broken())
+    with pytest.raises(TimeNetInvalidCardError, match="cannot be listed"):
+        InstalledConnectors().list_datasets()
+
+
 def test_a_local_registry_answers_not_found_without_reading_every_manifest(tmp_path, monkeypatch):
     registry = LocalRegistry(tmp_path)
     registry.put_dataset(make_dataset())
diff --git a/packages/timenet/tests/types/test_metadata.py b/packages/timenet/tests/types/test_metadata.py
index dec68908c..c33a77f78 100644
--- a/packages/timenet/tests/types/test_metadata.py
+++ b/packages/timenet/tests/types/test_metadata.py
@@ -3,11 +3,14 @@
 from pydantic import ValidationError
 import pytest
 
+from timenet.errors import TimeFValidationError
 from timenet.types import (
     Access,
     AnnotationDescriptor,
     AnnotationType,
+    AnswerTask,
     ClassificationTask,
+    DatasetFilter,
     DatasetMetadata,
     DatasetSchema,
     Domain,
@@ -191,3 +194,52 @@ def test_metadata_model_validate_exposes_pydantic_validation_errors():
 def test_metadata_rejects_invalid_field_types_and_values(field, value):
     with pytest.raises(ValidationError, match=field):
         _metadata(**{field: value})
+
+
+# ---- DatasetFilter ----------------------------------------------------------------------------
+
+
+def test_dataset_filter_of_takes_one_value_a_list_or_none():
+    built = DatasetFilter.of(domain=Domain.GENERAL, tag=["demo", "sine"], query=None, limit=5)
+    assert built == DatasetFilter(domain=(Domain.GENERAL,), tag=("demo", "sine"), limit=5)
+
+
+def test_a_dataset_filter_rejects_a_negative_limit():
+    with pytest.raises(TimeFValidationError, match=r"^limit must be 0 or more, got -1$"):
+        DatasetFilter(limit=-1)
+
+
+def test_a_dataset_filter_matches_a_card_by_any_value_of_a_filter_and_every_filter():
+    card = _metadata(domains=(Domain.GENERAL,), tags=("demo", "sine"))
+    matching = [
+        DatasetFilter(),
+        DatasetFilter(query=("WORLD",)),
+        DatasetFilter(query=("nothing", "synthetic")),
+        DatasetFilter(query=("sine",)),
+        DatasetFilter(domain=(Domain.CARDIOLOGY, Domain.GENERAL)),
+        DatasetFilter(license=(License.MIT, License.CC_BY_4_0)),
+        DatasetFilter(dataset_id=("x/y", "timenet/hello-world")),
+        DatasetFilter(tag=("demo", "sine")),
+        DatasetFilter(domain=(Domain.GENERAL,), license=(License.CC_BY_4_0,)),
+    ]
+    assert [wanted.matches_card(card) for wanted in matching] == [True] * len(matching)
+    # A tag filter needs all its tags, and the filters combine with AND.
+    failing = [
+        DatasetFilter(query=("nothing",)),
+        DatasetFilter(domain=(Domain.CARDIOLOGY,)),
+        DatasetFilter(license=(License.MIT,)),
+        DatasetFilter(dataset_id=("x/y",)),
+        DatasetFilter(tag=("demo", "other")),
+        DatasetFilter(domain=(Domain.GENERAL,), license=(License.MIT,)),
+    ]
+    assert [wanted.matches_card(card) for wanted in failing] == [False] * len(failing)
+
+
+def test_a_dataset_filter_matches_a_schema_by_any_task_and_all_specs():
+    specs = tuple(TimeSeriesSpec(spec_type=kind, name=kind, unit_value=ureg.millivolt) for kind in ("ecg_lead", "ppg"))
+    schema = DatasetSchema(time_series_specs=specs, tasks=(ClassificationTask,))
+    assert DatasetFilter(task=(AnswerTask, ClassificationTask), time_series_spec=("ecg_lead", "ppg")).matches_schema(
+        schema
+    )
+    assert not DatasetFilter(task=(AnswerTask,)).matches_schema(schema)
+    assert not DatasetFilter(time_series_spec=("ecg_lead", "eeg")).matches_schema(schema)