@@ -7,7 +7,7 @@
 import os
 import re
 import sys
-from typing import IO, Union, cast
+from typing import IO, Any, Union, cast
 import warnings
 
 import click
@@ -35,6 +35,25 @@ class TruncationNotice:
     """Number of validation results omitted from this group."""
 
 
+class ExistingPath(click.Path):
+    """
+    Like ``click.Path(exists=True)``, but also accepting broken symbolic links,
+    such as annexed files whose content is not present, which ``dandi validate``
+    knows how to handle (see ``--missing-file-content``)
+    """
+
+    def convert(
+        self, value: Any, param: click.Parameter | None, ctx: click.Context | None
+    ) -> Any:
+        if not os.path.lexists(value):
+            self.fail(
+                f"{self.name.title()} {os.fsdecode(value)!r} does not exist.",
+                param,
+                ctx,
+            )
+        return super().convert(value, param, ctx)
+
+
 STRUCTURED_FORMATS = ("json", "json_pp", "json_lines", "yaml")
 
 _EXT_TO_FORMAT = {
@@ -217,8 +236,13 @@ def validate_bids(
     "in a datalad dataset without fetched data). 'error' (default) emits a "
     "concise error per file, 'skip' skips each such file with a warning, "
     "'only-non-data' skips content-dependent validators but still validates "
-    "path layout.",
-    type=click.Choice(["error", "only-non-data", "skip"], case_sensitive=True),
+    "path layout, 'stream' streams the content of annexed files from the URLs "
+    "registered for them in git-annex so that content-dependent validators run "
+    "without the files having to be downloaded (requires fsspec; install with "
+    "`pip install 'dandi[extras]'`).",
+    type=click.Choice(
+        ["error", "only-non-data", "skip", "stream"], case_sensitive=True
+    ),
     default="error",
 )
 @click.option(
@@ -228,7 +252,7 @@ def validate_bids(
     multiple=True,
     default=(),
 )
-@click.argument("paths", nargs=-1, type=click.Path(exists=True, dir_okay=True))
+@click.argument("paths", nargs=-1, type=ExistingPath(dir_okay=True))
 @click.pass_context
 @devel_debug_option()
 @map_to_click_exceptions
 
@@ -996,3 +996,60 @@ def test_validate_missing_file_content_no_broken_symlinks(tmp_path: Path) -> Non
     assert r_default.exit_code == r_skip.exit_code == 0
     assert "FILE_CONTENT_MISSING" not in r_default.output
     assert "FILE_CONTENT_MISSING" not in r_skip.output
+
+
+@pytest.mark.ai_generated
+def test_validate_missing_file_content_stream(
+    tmp_path: Path, simple3_nwb: Path
+) -> None:
+    """--missing-file-content=stream validates annexed content via registered URLs."""
+    from ...tests.fixtures import annex_key_for_file, make_annexed_dandiset
+    from ...tests.skip import skipif
+
+    skipif.no_git()
+    pytest.importorskip("fsspec")
+    ds = tmp_path / "ds"
+    make_annexed_dandiset(
+        ds,
+        {
+            "sub-001/sub-001.nwb": (
+                annex_key_for_file(simple3_nwb),
+                [simple3_nwb.as_uri()],
+            )
+        },
+    )
+    out = tmp_path / "out.jsonl"
+    r = CliRunner().invoke(
+        validate,
+        [
+            "--missing-file-content",
+            "stream",
+            "--min-severity",
+            "INFO",
+            "-f",
+            "json_lines",
+            "-o",
+            str(out),
+            str(ds),
+        ],
+    )
+    assert "Traceback" not in r.output
+    ids = [rec.id for rec in load_validation_jsonl([str(out)])]
+    assert "DANDI.FILE_CONTENT_STREAMED" in ids
+    assert "DANDI.FILE_CONTENT_MISSING" not in ids
+    # simple3_nwb lacks a subject_id, which only content-based checks notice
+    assert "NWBI.check_subject_id_exists" in ids
+    assert r.exit_code == 1
+
+
+@pytest.mark.ai_generated
+def test_validate_broken_symlink_path_argument(tmp_path: Path) -> None:
+    """A broken symlink can be given directly as a path to validate."""
+    ds = _make_dandiset_with_broken_symlinks(tmp_path)
+    r = CliRunner().invoke(validate, [str(ds / "sub-001" / "sub-001.nwb")])
+    assert r.exit_code == 1
+    assert "does not exist" not in r.output
+    assert "FILE_CONTENT_MISSING" in r.output
+    r = CliRunner().invoke(validate, [str(ds / "sub-001" / "nonexistent.nwb")])
+    assert r.exit_code == 2
+    assert "does not exist" in r.output
@@ -10,6 +10,7 @@
 from pathlib import Path
 import re
 from threading import Lock
+import traceback
 from typing import IO, Any, Generic
 from xml.etree.ElementTree import fromstring
 
@@ -32,7 +33,7 @@
     set_asset_schema_key,
 )
 from dandi.metadata.core import get_default_metadata
-from dandi.misctypes import DUMMY_DANDI_ETAG, Digest, LocalReadableFile, P
+from dandi.misctypes import DUMMY_DANDI_ETAG, Digest, LocalReadableFile, P, Readable
 from dandi.utils import post_upload_size_check, pre_upload_size_check, yaml_load
 from dandi.validate._types import (
     ORIGIN_INTERNAL_DANDI,
@@ -320,12 +321,22 @@ class LocalFileAsset(LocalAsset):
     an asset of a Dandiset
     """
 
+    #: An alternative source for the asset's content, used instead of the file
+    #: at `filepath` when set.  This is used to stream the content of an
+    #: annexed file (e.g., in a DataLad dataset) that is not present locally
+    #: from a URL registered in git-annex; see `dandi.support.annex`.
+    content_source: Readable | None = None
+
+    def _content(self) -> Path | Readable:
+        """The source to read the asset's content from"""
+        return self.content_source if self.content_source is not None else self.filepath
+
     def get_metadata(
         self,
         digest: Digest | None = None,
         ignore_errors: bool = True,
     ) -> BareAsset:
-        metadata = get_default_metadata(self.filepath, digest=digest)
+        metadata = get_default_metadata(self._content(), digest=digest)
         metadata.path = self.path
         return metadata
 
@@ -515,7 +526,7 @@ def get_metadata(
         from dandi.metadata.nwb import nwb2asset
 
         try:
-            metadata = nwb2asset(self.filepath, digest=digest)
+            metadata = nwb2asset(self._content(), digest=digest)
         except Exception as e:
             lgr.warning(
                 "Failed to extract NWB metadata from %s: %s: %s",
@@ -524,7 +535,7 @@ def get_metadata(
                 str(e),
             )
             if ignore_errors:
-                metadata = get_default_metadata(self.filepath, digest=digest)
+                metadata = get_default_metadata(self._content(), digest=digest)
             else:
                 raise
         metadata.path = self.path
@@ -555,12 +566,16 @@ def get_validation_errors(
             pass
         else:
             # Avoid heavy import by importing within function:
-            from nwbinspector import Importance, inspect_nwbfile, load_config
+            from nwbinspector import Importance, load_config
 
             # Avoid heavy import by importing within function:
             from dandi.pynwb_utils import validate as pynwb_validate
 
-            errors.extend(pynwb_validate(self.filepath, devel_debug=devel_debug))
+            errors.extend(
+                pynwb_validate(
+                    self.filepath, devel_debug=devel_debug, readable=self.content_source
+                )
+            )
             if schema_version is not None:
                 errors.extend(
                     super().get_validation_errors(
@@ -576,9 +591,7 @@ def get_validation_errors(
                         validator_version=str(_get_nwb_inspector_version()),
                     )
 
-                    for error in inspect_nwbfile(
-                        nwbfile_path=self.filepath,
-                        skip_validate=True,
+                    for error in self._inspect_nwbfile(
                         config=load_config(filepath_or_keyword="dandi"),
                         importance_threshold=Importance.BEST_PRACTICE_VIOLATION,
                     ):
@@ -623,6 +636,57 @@ def get_validation_errors(
             )
         return errors
 
+    def _inspect_nwbfile(self, **kwargs: Any) -> Iterator[Any]:
+        """
+        Yield the messages from inspecting the file with nwbinspector, reading
+        the content from `content_source` if it is set (in which case what
+        `nwbinspector.inspect_nwbfile()` does for a local file is replicated
+        here for the streamed content)
+        """
+        # Avoid heavy import by importing within function:
+        from nwbinspector import (
+            Importance,
+            InspectorMessage,
+            inspect_nwbfile,
+            inspect_nwbfile_object,
+        )
+
+        if self.content_source is None:
+            yield from inspect_nwbfile(
+                nwbfile_path=self.filepath, skip_validate=True, **kwargs
+            )
+            return
+
+        # Avoid heavy import by importing within function:
+        import h5py
+        from pynwb import NWBHDF5IO
+
+        from dandi.pynwb_utils import open_readable
+
+        try:
+            with open_readable(self.content_source) as fp, h5py.File(
+                fp, "r"
+            ) as h5, NWBHDF5IO(file=h5, mode="r", load_namespaces=True) as io:
+                nwbfile = io.read()
+                for message in inspect_nwbfile_object(nwbfile_object=nwbfile, **kwargs):
+                    if message is not None:
+                        message.file_path = str(self.filepath)
+                        yield message
+        except Exception as e:
+            # Report the failure the same way inspect_nwbfile() does for a
+            # local file that PyNWB cannot read
+            yield InspectorMessage(
+                message=traceback.format_exc(),
+                importance=Importance.ERROR,
+                check_function_name=(
+                    f"During io.read(), an error occurred: "
+                    f"{type(e).__module__}.{type(e).__name__}. "
+                    "This indicates that PyNWB was unable to read the file. "
+                    "See the traceback message for more details."
+                ),
+                file_path=str(self.filepath),
+            )
+
 
 class VideoAsset(LocalFileAsset):
     pass
 
@@ -250,7 +250,7 @@ def get_metadata(
     ) -> BareAsset:
         metadata = self.bids_dataset_description.get_asset_metadata(self)
         start_time = end_time = datetime.now().astimezone()
-        add_common_metadata(metadata, self.filepath, start_time, end_time, digest)
+        add_common_metadata(metadata, self._content(), start_time, end_time, digest)
         metadata.path = self.path
         return metadata
 
 
@@ -616,8 +616,9 @@ def rename_if_matched(_name: str, obj: Any) -> None:
         f.visititems(rename_if_matched)
 
 
-@validate_cache.memoize_path
-def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResult]:
+def validate(
+    path: str | Path, devel_debug: bool = False, *, readable: Readable | None = None
+) -> list[ValidationResult]:
     """Run validation on a file and return errors
 
     In case of an exception being thrown, an error message added to the
@@ -626,8 +627,35 @@ def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResu
     Parameters
     ----------
     path: str or Path
+      Path of the file, as reported in the returned results
+    devel_debug: bool, optional
+      Whether to re-raise exceptions instead of reporting them as errors
+    readable: Readable, optional
+      If given, the file's content is read from this `Readable` instead of from
+      ``path`` (e.g., to stream the content of an annexed file which is not
+      present locally); the results are then not cached.
     """
+    if readable is not None:
+        return _validate(path, readable, devel_debug=devel_debug)
+    # The memoizing decorator hides the return type from mypy
+    return cast(
+        "list[ValidationResult]", _validate_cached(path, devel_debug=devel_debug)
+    )
+
+
+@validate_cache.memoize_path
+def _validate_cached(
+    path: str | Path, devel_debug: bool = False
+) -> list[ValidationResult]:
+    return _validate(path, None, devel_debug=devel_debug)
+
+
+def _validate(
+    path: str | Path, readable: Readable | None, devel_debug: bool = False
+) -> list[ValidationResult]:
     path = str(path)  # Might come in as pathlib's PATH
+    # What to read the content from:
+    source: str | Readable = readable if readable is not None else path
     errors: list[ValidationResult] = []
 
     # To overcome
@@ -639,7 +667,7 @@ def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResu
     )
     version = None
     try:
-        version = get_nwb_version(path, sanitize=False)
+        version = get_nwb_version(source, sanitize=False)
     except Exception:
         # we just will not remove any errors, it is required so should be some
         pass
@@ -685,7 +713,12 @@ def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResu
                     )
 
     try:
-        if Version(pynwb.__version__) >= Version("3.0.0"):
+        if readable is not None:
+            with open_readable(readable) as fp, h5py.File(fp, "r") as h5, NWBHDF5IO(
+                file=h5, mode="r", load_namespaces=True
+            ) as reader:
+                error_outputs = pynwb.validate(io=reader)
+        elif Version(pynwb.__version__) >= Version("3.0.0"):
             error_outputs = pynwb.validate(path=path)
         elif Version(pynwb.__version__) >= Version(
             "2.2.0"
@@ -695,6 +728,9 @@ def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResu
         else:  # Fallback if an older version
             with pynwb.NWBHDF5IO(path=path, mode="r", load_namespaces=True) as reader:
                 error_outputs = pynwb.validate(io=reader)
+        if isinstance(error_outputs, tuple):
+            # Older pynwb versions return (errors, status) for some call forms
+            error_outputs = error_outputs[0]
     except Exception as exc:
         if devel_debug:
             raise