diff --git a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py
index d39c865..b0fa5b4 100644
--- a/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py
+++ b/src/us_equity_snapshot_pipelines/tqqq_r1_snapshot.py
@@ -3,16 +3,18 @@
 from __future__ import annotations
 
 import ctypes
+import csv
 import errno
 import hashlib
 import json
 import math
 import os
 import shutil
+import stat
 import sys
 import tempfile
 from dataclasses import dataclass
-from io import BytesIO
+from io import StringIO
 from pathlib import Path
 from typing import Any
 
@@ -26,6 +28,15 @@
 MODE = "core_only"
 OUTPUT_FILENAMES = ("prices.csv", "manifest.json", "validation.json", "sha256sums.json")
 _MAX_FLOAT_INTEGER_DECIMAL = format(sys.float_info.max, ".0f")
+_MEMBER_BYTE_LIMITS = {
+    "prices.csv": 16 * 1024 * 1024,
+    "manifest.json": 1024 * 1024,
+    "validation.json": 1024 * 1024,
+    "sha256sums.json": 1024 * 1024,
+}
+_MAX_TOTAL_MEMBER_BYTES = 18 * 1024 * 1024
+_MAX_CSV_ROWS = 500_000
+_MAX_JSON_NUMBER_DIGITS = 128
 
 
 class SnapshotValidationError(ValueError):
@@ -193,6 +204,27 @@ def _write_prices(path: Path, prices: pd.DataFrame) -> None:
     output.to_csv(path, index=False, lineterminator="\n", float_format="%.17g")
 
 
+def _parse_prices_csv(raw: bytes) -> pd.DataFrame:
+    """Parse the single accepted CSV lexical form before semantic normalization."""
+    try:
+        reader = csv.reader(StringIO(raw.decode("utf-8"), newline=""), strict=True)
+        if next(reader, None) != ["session", "symbol", PRICE_FIELD]:
+            _invalid("prices.csv must contain exact columns")
+        rows: list[list[str]] = []
+        for row in reader:
+            if len(rows) >= _MAX_CSV_ROWS or len(row) != 3:
+                _invalid("invalid prices.csv")
+            rows.append(row)
+    except (UnicodeDecodeError, csv.Error, ValueError) as exc:
+        raise SnapshotValidationError("invalid prices.csv") from exc
+    return _normalized_prices(
+        pd.DataFrame(rows, columns=["session", "symbol", PRICE_FIELD]),
+        require_exact_columns=True,
+        require_canonical_symbols=True,
+        require_canonical_sessions=True,
+    )
+
+
 def _parse_json_object(raw: bytes, name: str) -> dict[str, Any]:
     def no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
         result: dict[str, Any] = {}
@@ -202,19 +234,129 @@ def no_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
             result[key] = value
         return result
 
+    def parse_integer(value: str) -> int:
+        if len(value) > _MAX_JSON_NUMBER_DIGITS:
+            raise ValueError("integer exceeds resource limit")
+        return int(value)
+
+    def parse_float(value: str) -> float:
+        if len(value) > _MAX_JSON_NUMBER_DIGITS:
+            raise ValueError("float exceeds resource limit")
+        parsed = float(value)
+        if not math.isfinite(parsed):
+            raise ValueError("non-finite")
+        return parsed
+
     try:
         value = json.loads(
             raw.decode("utf-8"),
             object_pairs_hook=no_duplicates,
+            parse_int=parse_integer,
+            parse_float=parse_float,
             parse_constant=lambda _: (_ for _ in ()).throw(ValueError("non-finite")),
         )
-    except (UnicodeDecodeError, ValueError, json.JSONDecodeError) as exc:
+    except (UnicodeDecodeError, ValueError, json.JSONDecodeError, OverflowError, RecursionError) as exc:
         raise SnapshotValidationError(f"invalid {name}") from exc
     if type(value) is not dict:
         _invalid(f"invalid {name}")
     return value
 
 
+def _require_descriptor_capabilities() -> None:
+    required_flags = ("O_DIRECTORY", "O_NOFOLLOW", "O_NONBLOCK")
+    if (
+        sys.platform not in {"darwin"} and not sys.platform.startswith("linux")
+    ) or any(not hasattr(os, flag) for flag in required_flags):
+        _invalid("required descriptor capability unavailable")
+    if os.scandir not in os.supports_fd or os.open not in os.supports_dir_fd:
+        _invalid("required descriptor capability unavailable")
+
+
+def _stable_member_identity(member: os.stat_result) -> tuple[int, int, int, int, int, int]:
+    return (
+        member.st_dev,
+        member.st_ino,
+        member.st_mode,
+        member.st_size,
+        member.st_mtime_ns,
+        member.st_ctime_ns,
+    )
+
+
+def _read_member_from_root(root_fd: int, name: str, remaining_bytes: int) -> bytes:
+    flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | getattr(os, "O_CLOEXEC", 0)
+    try:
+        member_fd = os.open(name, flags, dir_fd=root_fd)
+    except OSError as exc:
+        raise SnapshotValidationError("unable to read snapshot members") from exc
+    try:
+        try:
+            before = os.fstat(member_fd)
+        except OSError as exc:
+            raise SnapshotValidationError("unable to read snapshot members") from exc
+        if not stat.S_ISREG(before.st_mode):
+            _invalid("snapshot members must be regular non-symlink files")
+        if before.st_size > _MEMBER_BYTE_LIMITS[name]:
+            _invalid("snapshot member exceeds size limit")
+        if before.st_size > remaining_bytes:
+            _invalid("snapshot members exceed total size limit")
+        raw = bytearray()
+        read_limit = before.st_size + 1
+        while len(raw) < read_limit:
+            try:
+                chunk = os.read(member_fd, read_limit - len(raw))
+            except OSError as exc:
+                raise SnapshotValidationError("unable to read snapshot members") from exc
+            if not chunk:
+                break
+            raw.extend(chunk)
+        try:
+            after = os.fstat(member_fd)
+        except OSError as exc:
+            raise SnapshotValidationError("unable to read snapshot members") from exc
+        if len(raw) != before.st_size or _stable_member_identity(before) != _stable_member_identity(after):
+            _invalid("snapshot member changed during read")
+        return bytes(raw)
+    finally:
+        os.close(member_fd)
+
+
+def _read_snapshot_members(output: Path) -> dict[str, bytes]:
+    _require_descriptor_capabilities()
+    root_flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_NONBLOCK | getattr(os, "O_CLOEXEC", 0)
+    try:
+        root_fd = os.open(output, root_flags)
+    except OSError as exc:
+        raise SnapshotValidationError("unable to read snapshot members") from exc
+    try:
+        try:
+            root_stat = os.fstat(root_fd)
+        except OSError as exc:
+            raise SnapshotValidationError("unable to read snapshot members") from exc
+        if not stat.S_ISDIR(root_stat.st_mode):
+            _invalid("snapshot root must be a directory")
+        names: set[str] = set()
+        try:
+            with os.scandir(root_fd) as entries:
+                for entry_count, entry in enumerate(entries, start=1):
+                    if entry_count > len(OUTPUT_FILENAMES):
+                        _invalid("unexpected output files")
+                    names.add(entry.name)
+        except OSError as exc:
+            raise SnapshotValidationError("unable to read snapshot members") from exc
+        if names != set(OUTPUT_FILENAMES):
+            _invalid("unexpected output files")
+        members: dict[str, bytes] = {}
+        total_size = 0
+        for name in OUTPUT_FILENAMES:
+            raw = _read_member_from_root(root_fd, name, _MAX_TOTAL_MEMBER_BYTES - total_size)
+            total_size += len(raw)
+            members[name] = raw
+        return members
+    finally:
+        os.close(root_fd)
+
+
 def _publish_noreplace(source: Path, destination: Path) -> None:
     """Publish a directory without clobbering an existing destination."""
     if sys.platform.startswith("linux"):
@@ -274,28 +416,20 @@ def verify_tqqq_r1_snapshot(
     expected_manifest_sha256: str,
 ) -> SnapshotResult:
     output = Path(output_dir)
-    if output.is_symlink():
-        _invalid("snapshot root symlink is not allowed")
-    try:
-        names = tuple(sorted(path.name for path in output.iterdir())) if output.is_dir() else ()
-    except OSError as exc:
-        raise SnapshotValidationError("unable to read snapshot members") from exc
-    if names != tuple(sorted(OUTPUT_FILENAMES)):
-        _invalid(f"unexpected output files: {names}")
-    if any(not (output / name).is_file() or (output / name).is_symlink() for name in OUTPUT_FILENAMES):
-        _invalid("snapshot members must be regular non-symlink files")
-    try:
-        members = {name: (output / name).read_bytes() for name in OUTPUT_FILENAMES}
-    except OSError as exc:
-        raise SnapshotValidationError("unable to read snapshot members") from exc
-
-    member_hashes = {name: hashlib.sha256(content).hexdigest() for name, content in members.items()}
-    if type(expected_manifest_sha256) is not str or member_hashes["manifest.json"] != expected_manifest_sha256:
+    if (
+        type(expected_manifest_sha256) is not str
+        or len(expected_manifest_sha256) != 64
+        or any(character not in "0123456789abcdef" for character in expected_manifest_sha256)
+    ):
+        _invalid("invalid trusted manifest hash")
+    members = _read_snapshot_members(output)
+    manifest_sha256 = hashlib.sha256(members["manifest.json"]).hexdigest()
+    if manifest_sha256 != expected_manifest_sha256:
         _invalid("trusted manifest hash mismatch")
 
-    sums = _parse_json_object(members["sha256sums.json"], "sha256sums")
     manifest = _parse_json_object(members["manifest.json"], "manifest")
-    validation = _parse_json_object(members["validation.json"], "validation")
+    sums = _parse_json_object(members["sha256sums.json"], "sha256sums")
+    member_hashes = {name: hashlib.sha256(content).hexdigest() for name, content in members.items()}
     if set(sums) != {"prices.csv", "manifest.json", "validation.json"}:
         _invalid("invalid sha256sums")
     for name, expected in sums.items():
@@ -303,12 +437,8 @@ def verify_tqqq_r1_snapshot(
             _invalid(f"hash mismatch: {name}")
 
     try:
-        prices = _normalized_prices(
-            pd.read_csv(BytesIO(members["prices.csv"]), dtype={PRICE_FIELD: "string"}),
-            require_exact_columns=True,
-            require_canonical_symbols=True,
-            require_canonical_sessions=True,
-        )
+        validation = _parse_json_object(members["validation.json"], "validation")
+        prices = _parse_prices_csv(members["prices.csv"])
     except (UnicodeDecodeError, ValueError, pd.errors.EmptyDataError, pd.errors.ParserError) as exc:
         if isinstance(exc, SnapshotValidationError):
             raise
@@ -330,7 +460,7 @@ def verify_tqqq_r1_snapshot(
     expected_validation = {"valid": True, "row_count": len(prices), "symbols": list(SYMBOLS)}
     if not _has_exact_type(validation, expected_validation):
         _invalid("invalid validation")
-    return SnapshotResult(output_dir=output, manifest_sha256=member_hashes["manifest.json"])
+    return SnapshotResult(output_dir=output, manifest_sha256=manifest_sha256)
 
 
 def materialize_tqqq_r1_snapshot(
diff --git a/tests/test_tqqq_r1_snapshot.py b/tests/test_tqqq_r1_snapshot.py
index 8ae631d..12ba57a 100644
--- a/tests/test_tqqq_r1_snapshot.py
+++ b/tests/test_tqqq_r1_snapshot.py
@@ -2,8 +2,12 @@
 
 import hashlib
 import json
+import os
+import socket
+import tempfile
 from datetime import datetime, timezone
 from pathlib import Path
+from types import SimpleNamespace
 
 import numpy as np
 import pandas as pd
@@ -42,6 +46,16 @@ def _refresh_trusted_metadata(output_dir: Path) -> str:
     return hashlib.sha256(manifest_path.read_bytes()).hexdigest()
 
 
+def _rewrite_sha256sums(output_dir: Path) -> None:
+    _write_json(
+        output_dir / "sha256sums.json",
+        {
+            name: hashlib.sha256((output_dir / name).read_bytes()).hexdigest()
+            for name in ("prices.csv", "manifest.json", "validation.json")
+        },
+    )
+
+
 def test_materialize_writes_deterministic_immutable_artifacts(tmp_path: Path) -> None:
     result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
 
@@ -264,3 +278,221 @@ def test_materialize_rejects_python_int_beyond_digit_limit(tmp_path: Path) -> No
 
     with pytest.raises(snapshot.SnapshotValidationError):
         snapshot.materialize_tqqq_r1_snapshot(prices, tmp_path / "snapshot")
+
+
+def test_verify_rejects_four_field_csv_row_without_pandas_parser(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    (result.output_dir / "prices.csv").write_text(
+        "session,symbol,adjusted_close\n"
+        "2010-01-04,QQQ,45.25,unexpected\n"
+        "2010-01-04,TQQQ,10.5\n"
+        "2010-01-05,QQQ,46\n"
+        "2010-01-05,TQQQ,11\n",
+        encoding="utf-8",
+    )
+    expected_manifest_sha256 = _refresh_trusted_metadata(result.output_dir)
+
+    def pandas_parser_must_not_run(*args: object, **kwargs: object) -> pd.DataFrame:
+        pytest.fail("trusted CSV verification must not use pandas.read_csv")
+
+    monkeypatch.setattr(snapshot.pd, "read_csv", pandas_parser_must_not_run)
+
+    with pytest.raises(snapshot.SnapshotValidationError, match="invalid prices.csv"):
+        snapshot.verify_tqqq_r1_snapshot(result.output_dir, expected_manifest_sha256=expected_manifest_sha256)
+
+
+def test_verify_fails_closed_without_posix_descriptor_capabilities(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    monkeypatch.setattr(snapshot.sys, "platform", "win32")
+
+    with pytest.raises(snapshot.SnapshotValidationError, match="descriptor"):
+        snapshot.verify_tqqq_r1_snapshot(result.output_dir, expected_manifest_sha256=result.manifest_sha256)
+
+
+def test_verify_rejects_oversized_member_before_csv_parse(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    (result.output_dir / "prices.csv").write_bytes(b"session,symbol,adjusted_close\n" + b"x" * (17 * 1024 * 1024))
+    expected_manifest_sha256 = _refresh_trusted_metadata(result.output_dir)
+
+    def pandas_parser_must_not_run(*args: object, **kwargs: object) -> pd.DataFrame:
+        pytest.fail("oversized member reached CSV parser")
+
+    monkeypatch.setattr(snapshot.pd, "read_csv", pandas_parser_must_not_run)
+
+    with pytest.raises(snapshot.SnapshotValidationError, match="size limit"):
+        snapshot.verify_tqqq_r1_snapshot(result.output_dir, expected_manifest_sha256=expected_manifest_sha256)
+
+
+def test_verify_normalizes_deep_json_recursion(tmp_path: Path) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    (result.output_dir / "manifest.json").write_bytes(b'{"x":' * 2_000 + b"0" + b"}" * 2_000)
+    manifest_sha256 = hashlib.sha256((result.output_dir / "manifest.json").read_bytes()).hexdigest()
+    _write_json(
+        result.output_dir / "sha256sums.json",
+        {
+            "prices.csv": hashlib.sha256((result.output_dir / "prices.csv").read_bytes()).hexdigest(),
+            "manifest.json": manifest_sha256,
+            "validation.json": hashlib.sha256((result.output_dir / "validation.json").read_bytes()).hexdigest(),
+        },
+    )
+
+    with pytest.raises(snapshot.SnapshotValidationError, match="invalid manifest"):
+        snapshot.verify_tqqq_r1_snapshot(
+            result.output_dir,
+            expected_manifest_sha256=manifest_sha256,
+        )
+
+
+def test_verify_authenticates_manifest_before_json_parse(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+
+    def json_parser_must_not_run(*args: object, **kwargs: object) -> dict[str, object]:
+        pytest.fail("untrusted manifest reached a JSON parser")
+
+    monkeypatch.setattr(snapshot, "_parse_json_object", json_parser_must_not_run)
+
+    with pytest.raises(snapshot.SnapshotValidationError, match="trusted manifest hash mismatch"):
+        snapshot.verify_tqqq_r1_snapshot(result.output_dir, expected_manifest_sha256="0" * 64)
+
+
+def test_verify_rejects_fifth_entry_before_opening_members(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    (result.output_dir / "unexpected.txt").write_text("extra", encoding="utf-8")
+
+    def member_reader_must_not_run(*args: object, **kwargs: object) -> bytes:
+        pytest.fail("directory admission opened a member")
+
+    monkeypatch.setattr(snapshot, "_read_member_from_root", member_reader_must_not_run)
+
+    with pytest.raises(snapshot.SnapshotValidationError, match="unexpected output files"):
+        snapshot.verify_tqqq_r1_snapshot(result.output_dir, expected_manifest_sha256=result.manifest_sha256)
+
+
+def test_verify_rejects_member_symlink_by_descriptor(tmp_path: Path) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    replacement = tmp_path / "replacement.csv"
+    replacement.write_text((result.output_dir / "prices.csv").read_text(encoding="utf-8"), encoding="utf-8")
+    (result.output_dir / "prices.csv").unlink()
+    (result.output_dir / "prices.csv").symlink_to(replacement)
+
+    with pytest.raises(snapshot.SnapshotValidationError):
+        snapshot.verify_tqqq_r1_snapshot(result.output_dir, expected_manifest_sha256=result.manifest_sha256)
+
+
+def test_verify_rejects_root_symlink_by_descriptor(tmp_path: Path) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    alias = tmp_path / "snapshot-alias"
+    alias.symlink_to(result.output_dir, target_is_directory=True)
+
+    with pytest.raises(snapshot.SnapshotValidationError):
+        snapshot.verify_tqqq_r1_snapshot(alias, expected_manifest_sha256=result.manifest_sha256)
+
+
+def test_verify_rejects_member_identity_change_during_descriptor_read(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    original_fstat = snapshot.os.fstat
+    regular_fstat_calls = 0
+
+    def unstable_fstat(fd: int) -> os.stat_result | SimpleNamespace:
+        nonlocal regular_fstat_calls
+        observed = original_fstat(fd)
+        if snapshot.stat.S_ISREG(observed.st_mode):
+            regular_fstat_calls += 1
+            if regular_fstat_calls == 2:
+                return SimpleNamespace(
+                    st_dev=observed.st_dev,
+                    st_ino=observed.st_ino,
+                    st_mode=observed.st_mode,
+                    st_size=observed.st_size,
+                    st_mtime_ns=observed.st_mtime_ns + 1,
+                    st_ctime_ns=observed.st_ctime_ns,
+                )
+        return observed
+
+    monkeypatch.setattr(snapshot.os, "fstat", unstable_fstat)
+
+    with pytest.raises(snapshot.SnapshotValidationError, match="changed during read"):
+        snapshot.verify_tqqq_r1_snapshot(result.output_dir, expected_manifest_sha256=result.manifest_sha256)
+
+
+@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="POSIX FIFO capability unavailable")
+def test_verify_rejects_fifo_without_reading_it(tmp_path: Path) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    (result.output_dir / "prices.csv").unlink()
+    os.mkfifo(result.output_dir / "prices.csv")
+
+    with pytest.raises(snapshot.SnapshotValidationError, match="regular non-symlink"):
+        snapshot.verify_tqqq_r1_snapshot(result.output_dir, expected_manifest_sha256=result.manifest_sha256)
+
+
+@pytest.mark.skipif(not hasattr(socket, "AF_UNIX"), reason="Unix-domain socket capability unavailable")
+def test_verify_rejects_socket_without_reading_it() -> None:
+    with tempfile.TemporaryDirectory(dir="/tmp", prefix="tq-") as directory:
+        result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), Path(directory) / "snapshot")
+        prices_path = result.output_dir / "prices.csv"
+        prices_path.unlink()
+        listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        try:
+            listener.bind(str(prices_path))
+
+            with pytest.raises(snapshot.SnapshotValidationError):
+                snapshot.verify_tqqq_r1_snapshot(result.output_dir, expected_manifest_sha256=result.manifest_sha256)
+        finally:
+            listener.close()
+
+
+def test_verify_rejects_total_member_limit_before_parse(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    (result.output_dir / "prices.csv").write_bytes(b"p" * (16 * 1024 * 1024))
+    (result.output_dir / "manifest.json").write_bytes(b"m" * (1024 * 1024))
+    (result.output_dir / "validation.json").write_bytes(b"v" * (1024 * 1024))
+
+    def json_parser_must_not_run(*args: object, **kwargs: object) -> dict[str, object]:
+        pytest.fail("oversized aggregate reached a JSON parser")
+
+    monkeypatch.setattr(snapshot, "_parse_json_object", json_parser_must_not_run)
+
+    with pytest.raises(snapshot.SnapshotValidationError, match="total size limit"):
+        snapshot.verify_tqqq_r1_snapshot(
+            result.output_dir,
+            expected_manifest_sha256=hashlib.sha256((result.output_dir / "manifest.json").read_bytes()).hexdigest(),
+        )
+
+
+@pytest.mark.parametrize(
+    "manifest_bytes",
+    [
+        b'{"x":1,"x":2}',
+        b'{"x":NaN}',
+        b'{"x":Infinity}',
+        b'{"x":1e10000}',
+        b'{"x":' + b"9" * 129 + b"}",
+    ],
+    ids=["duplicate-key", "nan", "infinity", "overflow-float", "oversized-integer"],
+)
+def test_verify_normalizes_strict_json_rejections(tmp_path: Path, manifest_bytes: bytes) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    (result.output_dir / "manifest.json").write_bytes(manifest_bytes)
+    _rewrite_sha256sums(result.output_dir)
+
+    with pytest.raises(snapshot.SnapshotValidationError, match="invalid manifest"):
+        snapshot.verify_tqqq_r1_snapshot(
+            result.output_dir,
+            expected_manifest_sha256=hashlib.sha256(manifest_bytes).hexdigest(),
+        )
+
+
+def test_verify_accepts_legal_short_descriptor_reads(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    result = snapshot.materialize_tqqq_r1_snapshot(_fixture_prices(), tmp_path / "snapshot")
+    original_read = snapshot.os.read
+
+    def short_read(fd: int, size: int) -> bytes:
+        return original_read(fd, min(size, 7))
+
+    monkeypatch.setattr(snapshot.os, "read", short_read)
+
+    assert snapshot.verify_tqqq_r1_snapshot(
+        result.output_dir, expected_manifest_sha256=result.manifest_sha256
+    ) == result