diff --git a/.github/ci/unit-paths.txt b/.github/ci/unit-paths.txt
index a63108a68..e2c4f96cc 100644
--- a/.github/ci/unit-paths.txt
+++ b/.github/ci/unit-paths.txt
@@ -2,15 +2,16 @@ src/tests/adapters/test_bounded_postgres_read.py
 src/tests/adapters/test_codex_writer.py
 src/tests/adapters/test_control_plane_spend_lease.py
 src/tests/adapters/test_hashed_bearer_auth.py
-src/tests/adapters/test_postgres_p1.py
 src/tests/adapters/test_minio_store.py
 src/tests/adapters/test_mistral_ocr.py
 src/tests/adapters/test_openrouter.py
+src/tests/adapters/test_postgres_p1.py
 src/tests/adapters/test_selfhost_forget.py
 src/tests/adapters/test_selfhost_git.py
 src/tests/adapters/test_selfhost_purge.py
 src/tests/adapters/test_selfhost_stores.py
 src/tests/adapters/test_sentry.py
+src/tests/adapters/test_source_handle.py
 src/tests/adapters/test_telemetry.py
 src/tests/benchmarks/test_beam_answer_agent.py
 src/tests/benchmarks/test_beam_official_score.py
diff --git a/decisions.md b/decisions.md
index 80d5bdf14..bd1cec927 100644
--- a/decisions.md
+++ b/decisions.md
@@ -4793,3 +4793,70 @@ unread; fixing only the sort order while keeping the flag.
 rest of D21 (connected-components-to-gather, HAC distance-cut, nDR
 incremental re-decision, `merge_events`, `merged_into`, `resolution_exclusions`)
 is unchanged. Does not change D95, D99, or D100.
+## D104. A converter receives a source handle, never the file's bytes
+
+**Decision.** The converter contract takes a **source handle** — a read-only
+capability on one immutable, already-hashed source — in place of the whole file
+as a `bytes` value (amends D65's `convert(bytes, mime, hints)` and, through it,
+D38/D57). The handle offers four reads, every one of them bounded: **stream** the source
+in bounded chunks, **read a half-open byte range** `[start, end)`,
+**materialise** it to a temporary local file, or **read it bounded** into
+memory after naming the limit. There is no operation that reads a source of
+unknown size — the hazard was never bytes in memory, it was bytes in memory
+without anyone having decided how many. Because naming a large limit is still
+naming one, a handle may carry a **read ceiling** that refuses any single read
+above it; without a ceiling the property is friction rather than a guarantee,
+and the decision says so rather than overclaiming. A stream is opened for the
+duration of a block and released at its end, since the resource underneath is a
+descriptor or a connection and garbage-collection timing is not a lifetime
+contract. Reads never come back short: a range returns exactly what was asked
+for or raises, because a parser handed a truncated header cannot tell it from a
+valid one. Recorded length is verified alongside recorded hash — bytes can hash
+correctly and still be described by a wrong size, which would let two access
+paths disagree about one source. `ObjectStorePort` grows the matching
+`open_stream` and `read_range` so the handle has something to sit on, and the
+existing `read_bytes` stays for the small text objects every shipped route
+converts.
+
+**Context.** A text file fits in memory and a video does not. Under the
+bytes-first contract the client, the HTTP process, the object store, and the
+converter each materialise a complete copy of the same file, so two concurrent
+large ingests can exhaust a worker that would have processed either one alone.
+That is an availability property, not an optimisation. Requiring every read to
+carry a bound is the load-bearing part of the decision: a contract that offers
+an unbounded read as the convenient option gets unbounded reads, and the
+failure surfaces on the first large file in production rather than in review.
+
+**Bounds live in materialisation**, because that is the operation that can
+exhaust a host. The caller declares what it can afford and an oversized source
+is refused before any byte moves, costing one comparison rather than a filled
+disk. The written bytes are verified against the recorded content hash — an
+immutable object cannot legitimately change, so a mismatch is corruption or a
+wrong key and never retryable, and without the check a truncated read becomes a
+short document that looks complete, which is the silent-loss failure D65's
+coverage rules exist to prevent. The recorded size is a claim rather than a
+guarantee, so the write also counts what it actually writes and refuses a
+source that exceeds both the declaration and the accepted bound. The temporary
+file's lifetime belongs to the handle, so a route that fails mid-decode cannot
+leak the file it asked for.
+
+**Not solved here.** Nothing reserves worker-wide temporary disk, so two
+concurrent materialisations of one source each pass their own bound and can
+together exceed the volume. Aggregate admission belongs to whatever schedules
+the work; this is a stated boundary, not an oversight.
+
+**Consequences.** One contract serves both deployment shapes: a self-host
+directory tree and a cloud object store present the same three operations, so
+no route branches on where bytes live. Decoders that require a seekable path
+are served by bounded materialisation rather than by reimplementing them.
+Design home: `plan/designs/media_design.md` §2.1. Interacts with D38 (router),
+D57 (Markdown coordinate system), D61 (port boundary), D65 (media routes).
+
+**Alternatives.** *Keep bytes and raise the worker's memory* — rejected: it
+scales the blast radius with source size and solves neither resumability nor
+cancellation. *Hand routes a raw provider object* — rejected: it puts a storage
+SDK type in the converter contract, which D61 exists to prevent. *Give the
+handle an unbounded `read_all()` for convenience* — rejected: it becomes the
+path of least resistance and reintroduces exactly the failure this decision
+removes. `read_bounded(max_bytes=...)` serves the same routes while forcing the
+caller to state a limit, which is the property that actually matters.
diff --git a/plan/designs/media_design.md b/plan/designs/media_design.md
index 743571a87..84d533ba6 100644
--- a/plan/designs/media_design.md
+++ b/plan/designs/media_design.md
@@ -103,7 +103,8 @@ Notes an implementer needs:
   (the route prompt/adapter enforces it), and it is what makes §5's disclosure a property of
   *ranges the converter wrote*, not a per-claim judgment anyone has to make downstream.
 - **The generalized converter contract** (refines D38 a second time):
-  `convert(bytes, mime, hints) → { document.md, source_map, derived_assets[], manifest }` —
+  `convert(source, mime, hints) → { document.md, source_map, derived_assets[], manifest }` —
+  where `source` is a **source handle** (§2.1), not the file's bytes —
   the *page map* generalizes to a **source map** (§4), `derived_assets` are the `media/`
   children with their locators, and the `manifest` is the route's **complete self-account**,
   with required fields (nullable only where a capability is genuinely absent):
@@ -118,6 +119,104 @@ Notes an implementer needs:
   regions the tool could not read) — a conversion that silently drops ten minutes of a
   recording is the same lie as a silent top-k.
 
+
+### 2.1 The source handle — why a converter is not handed bytes
+
+*Decision: D104.*
+
+A text file fits in memory, so the original converter contract took the whole
+file as a `bytes` value. A video does not. If the contract keeps demanding
+bytes, then the client, the HTTP process, the object store, and the converter
+each hold a complete copy of the same multi-gigabyte file, and the host runs
+out of memory before any model work starts. That is a correctness and
+availability property, not an optimisation: two concurrent large ingests can
+take down a worker that would happily have processed either one alone.
+
+So a route receives a **source handle** — a read-only capability on one
+immutable, already-hashed source — and decides for itself how much of it to
+bring into memory. The handle offers four ways to read, every one of them
+bounded:
+
+- **stream it** — receive the source in order, one bounded chunk at a time,
+  which is what hashing, copying, and demuxing need;
+- **read a range** — take the half-open byte interval `[start, end)`, which is
+  what container parsing needs to read a header or an index without the body;
+  half-open because every other interval in the system is (§4), so nobody has
+  to remember which end is inclusive; and
+- **materialise it** — write the source to a temporary local file and hand
+  over the path. Decoders like FFmpeg want a seekable file rather than a
+  stream, and refusing to provide one would mean reimplementing them; and
+- **read it bounded** — take the whole source into memory, having first said
+  how much memory that is allowed to be. A Markdown passthrough has no use for
+  a stream, and pretending otherwise would push every small route into writing
+  its own accumulation loop.
+
+There is deliberately **no method that reads a source of unknown size**. Every
+read either bounds itself structurally (a chunk, a range) or requires the
+caller to name a limit up front. The distinction matters: the hazard was never
+"bytes in memory", it was *bytes in memory without anyone having decided how
+many*. A contract that offers an unbounded whole-file read as the convenient
+option gets unbounded whole-file reads, and the failure only shows up on the
+first large file in production.
+
+Naming a limit is not by itself a guarantee, because a caller can name a large
+one — asking for the range `[0, size)` is a whole-file read with extra steps.
+A handle can therefore carry a **read ceiling**: the largest single range,
+chunk, or bounded read it will serve, refused above that regardless of what the
+caller asks for. Without a ceiling the property is ergonomic friction; with one
+it is enforced. Deployments handling media set it; the small-text routes that
+have no size problem do not need it.
+
+Reading is also a **resource**, not just bytes. Underneath a stream is a file
+descriptor or an HTTP connection, so a stream is opened for the duration of a
+block and released when the block ends. Returning a bare iterator would leave a
+caller who stops early holding that resource until garbage collection, and
+collection timing is not a lifetime contract — enough abandoned reads exhaust
+the descriptor limit or the connection pool.
+
+Reads never come back short. A range returns exactly the bytes it was asked
+for or it raises — a container parser handed a truncated header cannot tell it
+from a valid one, and will produce confident nonsense rather than an error.
+That check belongs here rather than in each route, because every route would
+otherwise have to remember to write it.
+
+The recorded **length** is checked as well as the recorded hash, and they catch
+different lies. Bytes can hash correctly while being described by a wrong size;
+if only the hash were verified, materialisation would happily produce 5,000
+bytes for a source recorded as 10 while a range read past byte 10 was refused —
+two access paths disagreeing about the length of one source.
+
+**Not solved here:** nothing reserves worker-wide temporary disk. Two
+concurrent materialisations of one 6 GiB source each pass a 6 GiB bound and
+together need 12 GiB. Aggregate admission belongs to whatever schedules the
+work; the handle bounds one read at a time, and says so rather than implying
+otherwise.
+
+Materialisation is where the bounds live, because it is the operation that can
+actually exhaust a host. Three properties are enforced by the handle rather
+than trusted to each route:
+
+- **the caller declares what it can afford**, and a source larger than that is
+  refused before a single byte moves — an oversized source costs one integer
+  comparison, not a filled disk;
+- **the written bytes are verified against the source's content hash.** An
+  immutable object cannot legitimately change, so a mismatch is corruption or
+  a wrong key, never a retryable condition. Without this check a truncated
+  read becomes a short document that looks complete, which is precisely the
+  silent-loss failure the coverage rules in §2 exist to prevent; and
+- **the file is removed when the route is done with it**, whether it returned
+  or raised. A converter that fails mid-decode cannot leak the file it asked
+  for, because it never owned the file's lifetime.
+
+The recorded size is a claim, not a guarantee, so the streaming write also
+counts what it actually writes and refuses if the source turns out to be
+larger than both the declaration and the accepted bound.
+
+One handle contract serves both deployment shapes. A self-host deployment
+reads from a directory tree and a cloud deployment reads from object storage,
+but a route sees the same three operations either way, so no route contains a
+branch on where the bytes live.
+
 ## 3. What "already works" and stays untouched
 
 The representation flows the standard pipeline with **no media-specific machinery**: blocks
diff --git a/src/rememberstack/adapters/selfhost/minio.py b/src/rememberstack/adapters/selfhost/minio.py
index c529bf68e..7f7bca450 100644
--- a/src/rememberstack/adapters/selfhost/minio.py
+++ b/src/rememberstack/adapters/selfhost/minio.py
@@ -1,5 +1,7 @@
 """S3-compatible MinIO object storage for the self-host profile."""
 
+from collections.abc import Iterator
+from contextlib import contextmanager
 from typing import cast
 from typing import NotRequired
 from typing import Protocol
@@ -15,6 +17,7 @@
 from rememberstack.model import ObjectAlreadyExistsError
 from rememberstack.model import ObjectKey
 from rememberstack.model import ObjectKeyEscapesRootError
+from rememberstack.model import SourceRangeError
 
 
 class MinIOSettings(BaseSettings):
@@ -31,8 +34,8 @@ class MinIOSettings(BaseSettings):
 class _StreamingBody(Protocol):
     """The two response-body operations used by this adapter."""
 
-    def read(self) -> bytes:
-        """Read the complete response body."""
+    def read(self, amt: int | None = None) -> bytes:
+        """Read the whole body, or at most ``amt`` bytes when given."""
         ...
 
     def close(self) -> None:
@@ -77,8 +80,10 @@ def create_bucket(self, *, Bucket: str) -> object:
         """Create one bucket."""
         ...
 
-    def get_object(self, *, Bucket: str, Key: str) -> _GetObjectOutput:
-        """Read one object."""
+    def get_object(
+        self, *, Bucket: str, Key: str, Range: str | None = None
+    ) -> _GetObjectOutput:
+        """Read one object, optionally restricted to an HTTP byte range."""
         ...
 
     def put_object(
@@ -154,6 +159,68 @@ def read_bytes(self, *, key: ObjectKey) -> bytes:
         finally:
             body.close()
 
+    @contextmanager
+    def open_stream(
+        self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024
+    ) -> Iterator[Iterator[bytes]]:
+        """Open an ordered chunked read, releasing the HTTP body on exit."""
+        if chunk_bytes <= 0:
+            raise SourceRangeError(f"chunk_bytes must be positive, got {chunk_bytes}")
+        response = self._client.get_object(
+            Bucket=self._bucket, Key=_validated_key(key=key)
+        )
+        body = response["Body"]
+        try:
+
+            def chunks() -> Iterator[bytes]:
+                """Yield successive reads until the body is exhausted."""
+                while chunk := body.read(chunk_bytes):
+                    yield chunk
+
+            yield chunks()
+        finally:
+            body.close()
+
+    def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes:
+        """Read the half-open byte interval ``[start, end)`` of one object.
+
+        S3 ranges are inclusive on both ends, so the exclusive `end` becomes
+        ``end - 1`` on the wire. Converting here keeps every interval in the
+        system half-open (D65) regardless of the provider behind the port.
+        """
+        if start < 0 or end <= start:
+            raise SourceRangeError(
+                f"range [{start}, {end}) is empty, reversed, or negative"
+            )
+        try:
+            response = self._client.get_object(
+                Bucket=self._bucket,
+                Key=_validated_key(key=key),
+                Range=f"bytes={start}-{end - 1}",
+            )
+        except ClientError as error:
+            # A range past the end is a 416 from S3. Letting botocore's
+            # exception escape would put a provider type in the port contract
+            # that D61 exists to keep provider-free.
+            code = error.response.get("Error", {}).get("Code", "")
+            status = error.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
+            if code in {"InvalidRange", "416"} or status == 416:
+                raise SourceRangeError(
+                    f"range [{start}, {end}) is outside {key.root!r}"
+                ) from error
+            raise
+        body = response["Body"]
+        try:
+            content = body.read()
+        finally:
+            body.close()
+        if len(content) != end - start:
+            raise SourceRangeError(
+                f"range [{start}, {end}) of {key.root!r} returned {len(content)} "
+                f"bytes, not the {end - start} requested"
+            )
+        return content
+
     def write_bytes(
         self, *, key: ObjectKey, content: bytes, storage_class: str | None = None
     ) -> None:
diff --git a/src/rememberstack/adapters/selfhost/object_store.py b/src/rememberstack/adapters/selfhost/object_store.py
index 18833beb1..99bc85434 100644
--- a/src/rememberstack/adapters/selfhost/object_store.py
+++ b/src/rememberstack/adapters/selfhost/object_store.py
@@ -1,10 +1,13 @@
 """Local-filesystem object store adapter: immutable bytes under one root (D61/D62)."""
 
+from collections.abc import Iterator
+from contextlib import contextmanager
 from pathlib import Path
 
 from rememberstack.model import ObjectAlreadyExistsError
 from rememberstack.model import ObjectKey
 from rememberstack.model import ObjectKeyEscapesRootError
+from rememberstack.model import SourceRangeError
 
 
 class LocalFSObjectStore:
@@ -19,6 +22,38 @@ def read_bytes(self, *, key: ObjectKey) -> bytes:
         """Read all bytes stored under an existing object key."""
         return self._path_for(key=key).read_bytes()
 
+    @contextmanager
+    def open_stream(
+        self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024
+    ) -> Iterator[Iterator[bytes]]:
+        """Open an ordered chunked read, closing the file when the block exits."""
+        if chunk_bytes <= 0:
+            raise SourceRangeError(f"chunk_bytes must be positive, got {chunk_bytes}")
+        with self._path_for(key=key).open(mode="rb") as handle:
+
+            def chunks() -> Iterator[bytes]:
+                """Yield successive fixed-size reads until the file is spent."""
+                while chunk := handle.read(chunk_bytes):
+                    yield chunk
+
+            yield chunks()
+
+    def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes:
+        """Read the half-open byte interval ``[start, end)`` of one object."""
+        if start < 0 or end <= start:
+            raise SourceRangeError(
+                f"range [{start}, {end}) is empty, reversed, or negative"
+            )
+        with self._path_for(key=key).open(mode="rb") as handle:
+            handle.seek(start)
+            content = handle.read(end - start)
+        if len(content) != end - start:
+            raise SourceRangeError(
+                f"range [{start}, {end}) of {key.root!r} returned {len(content)} "
+                f"bytes, not the {end - start} requested"
+            )
+        return content
+
     def write_bytes(
         self, *, key: ObjectKey, content: bytes, storage_class: str | None = None
     ) -> None:
diff --git a/src/rememberstack/adapters/selfhost/source_handle.py b/src/rememberstack/adapters/selfhost/source_handle.py
new file mode 100644
index 000000000..431382d6b
--- /dev/null
+++ b/src/rememberstack/adapters/selfhost/source_handle.py
@@ -0,0 +1,216 @@
+"""A bounded source handle over any object store (D61 seam, media-safe reads).
+
+`ObjectSourceHandle` is the adapter that lets a converter read a large source
+without the whole file existing in the worker's heap. It works over any
+`ObjectStorePort`, so the same converter code runs against a local directory
+in a self-host deployment and against S3 in a managed one.
+"""
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+import hashlib
+from pathlib import Path
+import shutil
+import tempfile
+
+from rememberstack.model import SourceHashMismatchError
+from rememberstack.model import SourceIdentity
+from rememberstack.model import SourceRangeError
+from rememberstack.model import SourceSizeMismatchError
+from rememberstack.model import SourceTooLargeError
+from rememberstack.ports import ObjectStorePort
+
+_HASH_CHUNK_BYTES = 1024 * 1024
+
+
+class ObjectSourceHandle:
+    """Bounded reads of one immutable source held in an object store."""
+
+    def __init__(
+        self,
+        *,
+        store: ObjectStorePort,
+        identity: SourceIdentity,
+        temp_root: Path | None = None,
+        max_read_bytes: int | None = None,
+    ) -> None:
+        """Bind the handle to one source and the store that holds its bytes.
+
+        `temp_root` is where `materialize_seekable` writes. Leaving it None
+        uses the platform temporary directory; a deployment that bounds worker
+        disk points it at the volume it actually sized.
+
+        `max_read_bytes` is the ceiling on any single in-memory read — the
+        largest range or chunk this handle will serve. Without it, asking for
+        the range ``[0, byte_size)`` is a whole-file read with extra steps, so
+        the "no unbounded read" property is only ergonomic friction. With it,
+        the property is enforced. Left None the handle keeps the friction and
+        not the guarantee, which is the right default for the small-text routes
+        that have no size problem.
+
+        **Concurrency.** Each call reads independently; nothing here reserves
+        worker-wide temporary disk, so two concurrent materializations of a
+        6 GiB source can both pass a 6 GiB bound and together need 12 GiB.
+        Aggregate admission belongs to whatever schedules the work, and is a
+        documented non-goal of this port rather than an oversight.
+        """
+        if max_read_bytes is not None and max_read_bytes <= 0:
+            raise ValueError(
+                f"max_read_bytes must be positive when set, got {max_read_bytes}"
+            )
+        self._store = store
+        self._identity = identity
+        self._temp_root = temp_root
+        self._max_read_bytes = max_read_bytes
+
+    @property
+    def identity(self) -> SourceIdentity:
+        """The source's recorded key, content hash, size, and declared type."""
+        return self._identity
+
+    @contextmanager
+    def open_stream(
+        self, *, chunk_bytes: int = _HASH_CHUNK_BYTES
+    ) -> Iterator[Iterator[bytes]]:
+        """Open an ordered chunked read, released when the block exits."""
+        self._check_read_ceiling(requested=chunk_bytes, what="chunk")
+        with self._store.open_stream(
+            key=self._identity.object_key, chunk_bytes=chunk_bytes
+        ) as chunks:
+            yield chunks
+
+    def _check_read_ceiling(self, *, requested: int, what: str) -> None:
+        """Refuse a single read larger than this handle's configured ceiling."""
+        if self._max_read_bytes is not None and requested > self._max_read_bytes:
+            raise SourceTooLargeError(
+                f"{what} of {requested} bytes exceeds this handle's "
+                f"{self._max_read_bytes}-byte read ceiling"
+            )
+
+    def read_range(self, *, start: int, end: int) -> bytes:
+        """Read the half-open byte interval ``[start, end)`` of the source.
+
+        The range is checked against the size recorded at ingest, so a caller
+        asking past the end is refused here rather than receiving a short read
+        it might mistake for the end of meaningful content.
+        """
+        if start < 0 or end <= start:
+            raise SourceRangeError(
+                f"range [{start}, {end}) is empty, reversed, or negative"
+            )
+        if end > self._identity.byte_size:
+            raise SourceRangeError(
+                f"range [{start}, {end}) extends past the recorded source size "
+                f"{self._identity.byte_size}"
+            )
+        self._check_read_ceiling(requested=end - start, what="range")
+        content = self._store.read_range(
+            key=self._identity.object_key, start=start, end=end
+        )
+        if len(content) != end - start:
+            raise SourceRangeError(
+                f"range [{start}, {end}) of {self._identity.object_key.root!r} "
+                f"returned {len(content)} bytes, not the {end - start} requested; "
+                "the stored object is shorter than its recorded size"
+            )
+        return content
+
+    def read_bounded(self, *, max_bytes: int) -> bytes:
+        """Read the whole source into memory, refusing anything over the bound.
+
+        Small routes legitimately want every byte — a Markdown passthrough has
+        no use for a stream. This exists so they are not driven to write
+        ``b"".join(handle.open_stream())``, which is an unbounded read wearing
+        a bounded read's clothes. Naming the bound is the whole point: there is
+        still no way to ask for a source of unknown size.
+        """
+        if max_bytes < 0:
+            raise ValueError(f"max_bytes must be non-negative, got {max_bytes}")
+        if self._identity.byte_size > max_bytes:
+            raise SourceTooLargeError(
+                f"source {self._identity.object_key.root!r} is "
+                f"{self._identity.byte_size} bytes, over the {max_bytes} accepted"
+            )
+        self._check_read_ceiling(requested=self._identity.byte_size, what="read")
+        content = self._store.read_bytes(key=self._identity.object_key)
+        if len(content) != self._identity.byte_size:
+            raise SourceSizeMismatchError(
+                f"source {self._identity.object_key.root!r} holds {len(content)} "
+                f"bytes, not the {self._identity.byte_size} recorded"
+            )
+        return content
+
+    @contextmanager
+    def materialize_seekable(self, *, max_bytes: int) -> Iterator[Path]:
+        """Stream the source to a temporary file, removed when the block exits.
+
+        Three properties matter and all three are enforced here rather than
+        trusted to the caller:
+
+        * the bound is checked against the recorded size *before* any bytes
+          move, so an oversized source costs one comparison, not a filled disk;
+        * the written bytes are hashed and compared to the source's content
+          hash, so a truncated or corrupted read fails loudly instead of being
+          converted into a plausible-looking short document; and
+        * the file is removed on the way out whether the body raised or not.
+        """
+        if max_bytes < 0:
+            raise ValueError(f"max_bytes must be non-negative, got {max_bytes}")
+        if self._identity.byte_size > max_bytes:
+            raise SourceTooLargeError(
+                f"source {self._identity.object_key.root!r} is "
+                f"{self._identity.byte_size} bytes, over the {max_bytes} accepted"
+            )
+        if self._temp_root is not None:
+            self._temp_root.mkdir(parents=True, exist_ok=True)
+        directory = tempfile.mkdtemp(
+            prefix="rememberstack-source-",
+            dir=None if self._temp_root is None else str(self._temp_root),
+        )
+        # Demuxers routinely sniff the container from the extension before
+        # falling back to probing, so carrying the source's own suffix across
+        # costs nothing and avoids a class of "decoder cannot open it" bug.
+        path = Path(directory) / f"source{Path(self._identity.object_key.root).suffix}"
+        try:
+            digest = hashlib.sha256()
+            written = 0
+            # Read at most the remaining allowance plus one byte, so an object
+            # whose real length exceeds its recorded size cannot move a whole
+            # default chunk before the guard notices. The extra byte is what
+            # makes "too large" detectable at all: a read that exactly fills
+            # the allowance is indistinguishable from one that stops there.
+            chunk_bytes = min(_HASH_CHUNK_BYTES, max_bytes + 1)
+            with path.open(mode="wb") as handle:
+                with self.open_stream(chunk_bytes=chunk_bytes) as chunks:
+                    for chunk in chunks:
+                        written += len(chunk)
+                        if written > max_bytes:
+                            raise SourceTooLargeError(
+                                f"source {self._identity.object_key.root!r} exceeded "
+                                f"the {max_bytes} accepted while streaming; the "
+                                f"recorded size {self._identity.byte_size} was wrong"
+                            )
+                        digest.update(chunk)
+                        handle.write(chunk)
+            if written != self._identity.byte_size:
+                # Caught separately from the hash: bytes can hash correctly and
+                # still be described by a wrong length, which would let
+                # materialization and range reads disagree about how long the
+                # same source is.
+                raise SourceSizeMismatchError(
+                    f"source {self._identity.object_key.root!r} holds {written} "
+                    f"bytes, not the {self._identity.byte_size} recorded"
+                )
+            if digest.hexdigest() != self._identity.content_hash:
+                raise SourceHashMismatchError(
+                    f"source {self._identity.object_key.root!r} hashed to "
+                    f"{digest.hexdigest()}, not the recorded "
+                    f"{self._identity.content_hash}"
+                )
+            yield path
+        finally:
+            # rmtree, not unlink plus rmdir: a decoder handed a path routinely
+            # writes beside it (indexes, sidecars, split tracks), and a
+            # directory-not-empty error here would both leak the directory and
+            # mask whatever the route was already raising.
+            shutil.rmtree(directory, ignore_errors=True)
diff --git a/src/rememberstack/model/__init__.py b/src/rememberstack/model/__init__.py
index 073072ba7..0ba43b2bb 100644
--- a/src/rememberstack/model/__init__.py
+++ b/src/rememberstack/model/__init__.py
@@ -346,6 +346,11 @@
 from rememberstack.model.sections import SkeletonVerdict
 from rememberstack.model.sections import SnappedSection
 from rememberstack.model.sections import StructureRouteTag
+from rememberstack.model.source_handle import SourceHashMismatchError
+from rememberstack.model.source_handle import SourceIdentity
+from rememberstack.model.source_handle import SourceRangeError
+from rememberstack.model.source_handle import SourceSizeMismatchError
+from rememberstack.model.source_handle import SourceTooLargeError
 from rememberstack.model.spend_lease import SpendLeaseRefused
 from rememberstack.model.spend_lease import SpendLeaseUnavailable
 from rememberstack.model.telemetry import TelemetryAttribute
@@ -499,6 +504,11 @@
     "ObjectAlreadyExistsError",
     "ObjectKey",
     "ObjectKeyEscapesRootError",
+    "SourceHashMismatchError",
+    "SourceIdentity",
+    "SourceRangeError",
+    "SourceSizeMismatchError",
+    "SourceTooLargeError",
     "ObservationAssertion",
     "ObservationCandidate",
     "ObservationForEmbedding",
diff --git a/src/rememberstack/model/source_handle.py b/src/rememberstack/model/source_handle.py
new file mode 100644
index 000000000..acfec9ade
--- /dev/null
+++ b/src/rememberstack/model/source_handle.py
@@ -0,0 +1,66 @@
+"""Identity and failure types for bounded reads of one immutable source.
+
+A media source is too large to carry as `bytes`. These types describe what a
+converter is allowed to know about a source it has not read, and how a read
+refuses when it would exceed the bounds the caller accepted.
+"""
+
+from typing import Annotated
+
+from pydantic import BaseModel
+from pydantic import ConfigDict
+from pydantic import Field
+
+from rememberstack.model.object_store import ObjectKey
+
+Sha256Hex = Annotated[str, Field(pattern=r"^[0-9a-f]{64}$")]
+
+
+class SourceIdentity(BaseModel):
+    """What a converter may know about a source without reading it.
+
+    The hash and size are the E0 facts recorded when the raw object was
+    written, not values a converter or a caller supplies.
+
+    Deliberately no MIME field. The converter contract already carries the
+    declared type as its own argument (`convert(source, mime, hints)`), and
+    holding it in two places invites the two copies to disagree — a route
+    would then have to know which one routing used. One authority, and it is
+    not this object.
+    """
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    object_key: ObjectKey
+    content_hash: Sha256Hex
+    byte_size: int = Field(ge=0)
+
+
+class SourceTooLargeError(Exception):
+    """A read or materialization would exceed the bound the caller accepted.
+
+    Raised *before* the bytes are moved. A converter declares what it can
+    afford and is refused up front, rather than discovering the cost by
+    exhausting the worker.
+    """
+
+
+class SourceHashMismatchError(Exception):
+    """Materialized bytes did not hash to the source's recorded content hash.
+
+    Immutable objects cannot legitimately change, so this is corruption or a
+    wrong key — never a retryable condition on the same inputs.
+    """
+
+
+class SourceRangeError(Exception):
+    """A byte range that is empty, reversed, outside the source, or short."""
+
+
+class SourceSizeMismatchError(Exception):
+    """The stored object's real length is not the length recorded for it.
+
+    Distinct from a hash mismatch: the bytes may hash correctly and still be
+    described by a wrong size, which would let materialization and range reads
+    disagree about how long the same source is.
+    """
diff --git a/src/rememberstack/ports/__init__.py b/src/rememberstack/ports/__init__.py
index f0863b551..195075416 100644
--- a/src/rememberstack/ports/__init__.py
+++ b/src/rememberstack/ports/__init__.py
@@ -11,6 +11,7 @@
 from rememberstack.ports.purge import ObjectPurgePort
 from rememberstack.ports.purge import ProjectionPurgePort
 from rememberstack.ports.queue import TaskQueuePort
+from rememberstack.ports.source_handle import SourceHandlePort
 from rememberstack.ports.telemetry import TelemetryPort
 
 __all__ = (
@@ -24,6 +25,7 @@
     "PostgresReadPoolPort",
     "ObjectPurgePort",
     "ProjectionPurgePort",
+    "SourceHandlePort",
     "TaskQueuePort",
     "TelemetryPort",
 )
diff --git a/src/rememberstack/ports/object_store.py b/src/rememberstack/ports/object_store.py
index 918037b41..8e0b6bd69 100644
--- a/src/rememberstack/ports/object_store.py
+++ b/src/rememberstack/ports/object_store.py
@@ -1,5 +1,7 @@
 """D61 byte/object-key seam for immutable raw inputs, artifacts, and snapshots."""
 
+from collections.abc import Iterator
+from contextlib import AbstractContextManager
 from typing import Protocol
 from typing import runtime_checkable
 
@@ -11,7 +13,32 @@ class ObjectStorePort(Protocol):
     """Read and create immutable objects without exposing storage-provider types."""
 
     def read_bytes(self, *, key: ObjectKey) -> bytes:
-        """Read all bytes stored under an existing object key."""
+        """Read all bytes stored under an existing object key.
+
+        Correct for the small text objects every shipped route converts.
+        A caller that cannot bound the object's size uses `open_stream` or
+        `read_range` instead — see `SourceHandlePort`.
+        """
+        ...
+
+    def open_stream(
+        self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024
+    ) -> AbstractContextManager[Iterator[bytes]]:
+        """Open an ordered chunked read, released when the block exits.
+
+        A context manager rather than a bare iterator because the underlying
+        resource is a file descriptor or an HTTP body. A caller that stops
+        iterating early would otherwise hold it until garbage collection, and
+        collection timing is not a resource-lifetime contract — enough
+        abandoned reads exhaust the descriptor limit or the connection pool.
+        """
+        ...
+
+    def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes:
+        """Read the half-open byte interval ``[start, end)`` of one object.
+
+        Returns exactly ``end - start`` bytes or raises `SourceRangeError`.
+        """
         ...
 
     def write_bytes(
diff --git a/src/rememberstack/ports/source_handle.py b/src/rememberstack/ports/source_handle.py
new file mode 100644
index 000000000..4b2680852
--- /dev/null
+++ b/src/rememberstack/ports/source_handle.py
@@ -0,0 +1,75 @@
+"""Bounded read access to one immutable source, without buffering it whole.
+
+The converter contract begins with complete `bytes` (D38/D57), which is
+workable for text and unworkable for media: the client, the HTTP process, the
+object store, and the converter each materialize the whole file. This port is
+the seam that lets a route read a source by stream, by range, or as a bounded
+local file, so a large source costs bounded memory rather than its own size.
+"""
+
+from collections.abc import Iterator
+from contextlib import AbstractContextManager
+from pathlib import Path
+from typing import Protocol
+from typing import runtime_checkable
+
+from rememberstack.model import SourceIdentity
+
+
+@runtime_checkable
+class SourceHandlePort(Protocol):
+    """Read one immutable source without requiring its size in memory.
+
+    Deliberately absent: any method returning the whole source as `bytes`.
+    A caller that genuinely needs every byte streams them and decides what to
+    keep; the port never makes whole-file buffering the easy path.
+    """
+
+    @property
+    def identity(self) -> SourceIdentity:
+        """The source's recorded key, content hash, size, and declared type."""
+        ...
+
+    def open_stream(
+        self, *, chunk_bytes: int = 1024 * 1024
+    ) -> AbstractContextManager[Iterator[bytes]]:
+        """Open an ordered chunked read, released when the block exits.
+
+        A context manager rather than a bare iterator: the resource underneath
+        is a file descriptor or an HTTP body, and a caller that stops early
+        would otherwise hold it until garbage collection. Collection timing is
+        not a resource-lifetime contract.
+        """
+        ...
+
+    def read_range(self, *, start: int, end: int) -> bytes:
+        """Read the half-open byte interval ``[start, end)``.
+
+        Half-open to match every other interval in the system (D65 locators),
+        so a caller never has to remember which end is inclusive.
+
+        Returns exactly ``end - start`` bytes or raises. A short read is never
+        returned: a container parser that receives a truncated header cannot
+        tell it from a valid one and will produce confident nonsense.
+        """
+        ...
+
+    def read_bounded(self, *, max_bytes: int) -> bytes:
+        """Read the whole source into memory, refusing anything over the bound.
+
+        For routes that legitimately want every byte — a Markdown passthrough
+        has no use for a stream. The bound must be named, so this is still not
+        a way to read a source of unknown size; it exists so small routes are
+        not driven to write their own unbounded ``join`` over `open_stream`.
+        """
+        ...
+
+    def materialize_seekable(self, *, max_bytes: int) -> AbstractContextManager[Path]:
+        """Write the source to a temporary local file, removed on exit.
+
+        Decoders such as FFmpeg need a seekable path, not a stream. The bound
+        is checked against the recorded size before any bytes move, and the
+        context manager owns cleanup so a failing converter cannot leak the
+        file it asked for.
+        """
+        ...
diff --git a/src/tests/adapters/test_minio_store.py b/src/tests/adapters/test_minio_store.py
index 8156814e8..e8e1e30e5 100644
--- a/src/tests/adapters/test_minio_store.py
+++ b/src/tests/adapters/test_minio_store.py
@@ -12,6 +12,7 @@
 from rememberstack.model import ObjectAlreadyExistsError
 from rememberstack.model import ObjectKey
 from rememberstack.model import ObjectKeyEscapesRootError
+from rememberstack.model import SourceRangeError
 
 
 class _Body:
@@ -22,9 +23,9 @@ def __init__(self, *, content: bytes) -> None:
         self._stream = BytesIO(content)
         self.closed = False
 
-    def read(self) -> bytes:
-        """Read all remaining bytes."""
-        return self._stream.read()
+    def read(self, amt: int | None = None) -> bytes:
+        """Read all remaining bytes, or at most ``amt`` when given."""
+        return self._stream.read() if amt is None else self._stream.read(amt)
 
     def close(self) -> None:
         """Record connection release."""
@@ -52,9 +53,15 @@ def create_bucket(self, *, Bucket: str) -> object:
         self.buckets.add(Bucket)
         return {}
 
-    def get_object(self, *, Bucket: str, Key: str) -> _GetObjectOutput:
-        """Return one streaming body."""
-        body = _Body(content=self.objects[(Bucket, Key)][0])
+    def get_object(
+        self, *, Bucket: str, Key: str, Range: str | None = None
+    ) -> _GetObjectOutput:
+        """Return one streaming body, honoring an inclusive HTTP byte range."""
+        content = self.objects[(Bucket, Key)][0]
+        if Range is not None:
+            first, _, last = Range.removeprefix("bytes=").partition("-")
+            content = content[int(first) : int(last) + 1]
+        body = _Body(content=content)
         self.last_body = body
         return {"Body": body}
 
@@ -157,3 +164,66 @@ def _client_error(*, code: str, operation: str) -> ClientError:
         error_response={"Error": {"Code": code, "Message": code}},
         operation_name=operation,
     )
+
+
+def test_stream_reassembles_and_releases_the_connection() -> None:
+    """Chunked reads cover the object and still close the HTTP body."""
+    client = _MemoryS3()
+    store = MinIOObjectStore(bucket="raw", client=client)
+    store.ensure_bucket()
+    content = bytes(range(256)) * 8
+    store.write_bytes(key=ObjectKey("clip.mp4"), content=content)
+
+    with store.open_stream(key=ObjectKey("clip.mp4"), chunk_bytes=100) as parts:
+        chunks = list(parts)
+
+    assert b"".join(chunks) == content
+    assert max(len(chunk) for chunk in chunks) <= 100
+    assert client.last_body is not None and client.last_body.closed
+
+
+def test_range_translates_half_open_to_the_inclusive_wire_form() -> None:
+    """``[2, 5)`` must reach S3 as ``bytes=2-4``, not ``bytes=2-5``."""
+    client = _MemoryS3()
+    store = MinIOObjectStore(bucket="raw", client=client)
+    store.ensure_bucket()
+    store.write_bytes(key=ObjectKey("clip.mp4"), content=b"0123456789")
+
+    assert store.read_range(key=ObjectKey("clip.mp4"), start=2, end=5) == b"234"
+    assert client.last_body is not None and client.last_body.closed
+
+
+def test_range_past_the_object_end_raises_a_domain_error() -> None:
+    """A short provider response must not reach the caller as valid bytes."""
+    client = _MemoryS3()
+    store = MinIOObjectStore(bucket="raw", client=client)
+    store.ensure_bucket()
+    store.write_bytes(key=ObjectKey("clip.mp4"), content=b"01234")
+
+    with pytest.raises(SourceRangeError):
+        store.read_range(key=ObjectKey("clip.mp4"), start=2, end=99)
+
+
+@pytest.mark.parametrize(
+    ("start", "end"), [(5, 5), (5, 2), (-1, 4)], ids=["empty", "reversed", "negative"]
+)
+def test_degenerate_ranges_are_refused_before_any_request(start: int, end: int) -> None:
+    """A malformed range is a caller bug and never becomes a wire request."""
+    client = _MemoryS3()
+    store = MinIOObjectStore(bucket="raw", client=client)
+    store.ensure_bucket()
+    store.write_bytes(key=ObjectKey("clip.mp4"), content=b"01234")
+    client.last_body = None
+
+    with pytest.raises(SourceRangeError):
+        store.read_range(key=ObjectKey("clip.mp4"), start=start, end=end)
+    assert client.last_body is None, "a refused range must not reach the provider"
+
+
+def test_nonpositive_chunk_size_is_refused() -> None:
+    """A zero chunk would never advance; a negative one is meaningless."""
+    store = MinIOObjectStore(bucket="raw", client=_MemoryS3())
+
+    with pytest.raises(SourceRangeError):
+        with store.open_stream(key=ObjectKey("clip.mp4"), chunk_bytes=0):
+            pytest.fail("a non-positive chunk must be refused before any read")
diff --git a/src/tests/adapters/test_source_handle.py b/src/tests/adapters/test_source_handle.py
new file mode 100644
index 000000000..bb99fd988
--- /dev/null
+++ b/src/tests/adapters/test_source_handle.py
@@ -0,0 +1,397 @@
+"""Bounded source reads: streaming, ranges, size refusal, and hash verification."""
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+import hashlib
+from pathlib import Path
+
+import pytest
+
+from rememberstack.adapters.selfhost.object_store import LocalFSObjectStore
+from rememberstack.adapters.selfhost.source_handle import ObjectSourceHandle
+from rememberstack.model import ObjectKey
+from rememberstack.model import SourceHashMismatchError
+from rememberstack.model import SourceIdentity
+from rememberstack.model import SourceRangeError
+from rememberstack.model import SourceSizeMismatchError
+from rememberstack.model import SourceTooLargeError
+
+_KEY = ObjectKey("raw/recording.mp4")
+
+
+class _CountingStore(LocalFSObjectStore):
+    """A local store that records how many bytes it actually handed out."""
+
+    def __init__(self, *, root: Path) -> None:
+        """Start with an empty byte counter over the given root."""
+        super().__init__(root=root)
+        self.bytes_served = 0
+
+    @contextmanager
+    def open_stream(
+        self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024
+    ) -> Iterator[Iterator[bytes]]:
+        """Count every byte yielded so a test can assert what actually moved."""
+        with super().open_stream(key=key, chunk_bytes=chunk_bytes) as parts:
+
+            def counted() -> Iterator[bytes]:
+                """Pass chunks through, adding each to the running total."""
+                for chunk in parts:
+                    self.bytes_served += len(chunk)
+                    yield chunk
+
+            yield counted()
+
+
+def _handle(
+    *, tmp_path: Path, content: bytes, declared_size: int | None = None
+) -> ObjectSourceHandle:
+    """Build a handle over a local store holding `content` at the fixed key."""
+    store = LocalFSObjectStore(root=tmp_path / "objects")
+    store.write_bytes(key=_KEY, content=content)
+    return ObjectSourceHandle(
+        store=store,
+        identity=SourceIdentity(
+            object_key=_KEY,
+            content_hash=hashlib.sha256(content).hexdigest(),
+            byte_size=len(content) if declared_size is None else declared_size,
+        ),
+        temp_root=tmp_path / "work",
+    )
+
+
+def test_stream_reassembles_the_source_without_whole_file_chunks(
+    tmp_path: Path,
+) -> None:
+    """Streaming yields the whole source in bounded pieces, in order."""
+    content = bytes(range(256)) * 40
+    handle = _handle(tmp_path=tmp_path, content=content)
+
+    with handle.open_stream(chunk_bytes=1024) as parts:
+        chunks = list(parts)
+
+    assert b"".join(chunks) == content
+    assert max(len(chunk) for chunk in chunks) <= 1024
+    assert len(chunks) > 1, "a bounded read of a large source must be split"
+
+
+def test_range_reads_are_half_open(tmp_path: Path) -> None:
+    """``[start, end)`` matches every other interval in the system (D65)."""
+    handle = _handle(tmp_path=tmp_path, content=b"0123456789")
+
+    assert handle.read_range(start=2, end=5) == b"234"
+
+
+@pytest.mark.parametrize(
+    ("start", "end"), [(5, 5), (5, 2), (-1, 4)], ids=["empty", "reversed", "negative"]
+)
+def test_degenerate_ranges_are_refused(tmp_path: Path, start: int, end: int) -> None:
+    """An empty, reversed, or negative range is a caller bug, not a short read."""
+    handle = _handle(tmp_path=tmp_path, content=b"0123456789")
+
+    with pytest.raises(SourceRangeError):
+        handle.read_range(start=start, end=end)
+
+
+def test_range_past_the_recorded_size_is_refused(tmp_path: Path) -> None:
+    """Reading past the end must fail loudly, not return a silent short read.
+
+    A converter that mistakes a short read for the end of content produces a
+    truncated document that looks complete — the failure D65 coverage rules
+    exist to prevent.
+    """
+    handle = _handle(tmp_path=tmp_path, content=b"0123456789")
+
+    with pytest.raises(SourceRangeError):
+        handle.read_range(start=8, end=99)
+
+
+def test_materialize_gives_a_seekable_path_and_removes_it(tmp_path: Path) -> None:
+    """Decoders need a real file; the handle owns its lifetime, not the caller."""
+    content = b"video-bytes" * 100
+    handle = _handle(tmp_path=tmp_path, content=content)
+
+    with handle.materialize_seekable(max_bytes=len(content)) as path:
+        assert path.read_bytes() == content
+        materialized = path
+
+    assert not materialized.exists()
+    assert not materialized.parent.exists()
+
+
+def test_materialize_removes_the_file_even_when_the_caller_raises(
+    tmp_path: Path,
+) -> None:
+    """A converter that fails mid-decode must not leak the file it asked for."""
+    handle = _handle(tmp_path=tmp_path, content=b"payload")
+    materialized: Path | None = None
+
+    with pytest.raises(RuntimeError):
+        with handle.materialize_seekable(max_bytes=1024) as path:
+            materialized = path
+            raise RuntimeError("decoder exploded")
+
+    assert materialized is not None, "the body must have run before it raised"
+    assert not materialized.exists()
+
+
+def test_oversized_source_is_refused_before_any_bytes_move(tmp_path: Path) -> None:
+    """The bound is checked against the recorded size, so refusal costs nothing.
+
+    Asserted by counting what the store was asked for: the point is not merely
+    that no file survives, but that the store was never touched at all.
+    """
+    store = _CountingStore(root=tmp_path / "objects")
+    store.write_bytes(key=_KEY, content=b"x" * 5000)
+    handle = ObjectSourceHandle(
+        store=store,
+        identity=SourceIdentity(
+            object_key=_KEY,
+            content_hash=hashlib.sha256(b"x" * 5000).hexdigest(),
+            byte_size=5000,
+        ),
+        temp_root=tmp_path / "work",
+    )
+
+    with pytest.raises(SourceTooLargeError):
+        with handle.materialize_seekable(max_bytes=100):
+            pytest.fail("an oversized source must never be materialized")
+
+    assert store.bytes_served == 0, "refusal must precede any read"
+    assert not (tmp_path / "work").exists() or not any((tmp_path / "work").iterdir())
+
+
+def test_understated_size_is_still_caught_while_streaming(tmp_path: Path) -> None:
+    """A wrong recorded size cannot be used to smuggle past the accepted bound.
+
+    The guard must also fire *promptly*: reading a full default chunk before
+    noticing would move a megabyte for a hundred-byte allowance, so the read is
+    sized to the remaining allowance plus the one byte that makes "over" visible.
+    """
+    store = _CountingStore(root=tmp_path / "objects")
+    store.write_bytes(key=_KEY, content=b"y" * 5000)
+    handle = ObjectSourceHandle(
+        store=store,
+        identity=SourceIdentity(
+            object_key=_KEY,
+            content_hash=hashlib.sha256(b"y" * 5000).hexdigest(),
+            byte_size=10,
+        ),
+        temp_root=tmp_path / "work",
+    )
+
+    with pytest.raises(SourceTooLargeError):
+        with handle.materialize_seekable(max_bytes=100):
+            pytest.fail("the streaming guard must fire when the size was wrong")
+
+    assert store.bytes_served <= 101, (
+        f"moved {store.bytes_served} bytes for a 100-byte allowance"
+    )
+
+
+def test_corrupted_bytes_fail_the_hash_check(tmp_path: Path) -> None:
+    """Immutable objects cannot legitimately change, so a mismatch is terminal."""
+    store = LocalFSObjectStore(root=tmp_path / "objects")
+    store.write_bytes(key=_KEY, content=b"actual-content")
+    handle = ObjectSourceHandle(
+        store=store,
+        identity=SourceIdentity(
+            object_key=_KEY,
+            content_hash=hashlib.sha256(b"different-content").hexdigest(),
+            byte_size=len(b"actual-content"),
+        ),
+        temp_root=tmp_path / "work",
+    )
+
+    with pytest.raises(SourceHashMismatchError):
+        with handle.materialize_seekable(max_bytes=1024):
+            pytest.fail("a hash mismatch must not reach the converter")
+
+
+def test_nonpositive_chunk_size_is_refused(tmp_path: Path) -> None:
+    """A zero chunk would spin forever; a negative one is meaningless."""
+    handle = _handle(tmp_path=tmp_path, content=b"data")
+
+    with pytest.raises(SourceRangeError):
+        with handle.open_stream(chunk_bytes=0):
+            pytest.fail("a non-positive chunk must be refused before any read")
+
+
+def test_zero_byte_source_materializes_under_a_zero_bound(tmp_path: Path) -> None:
+    """An empty file is a valid source; `max_bytes=0` is enough room for it."""
+    handle = _handle(tmp_path=tmp_path, content=b"")
+
+    with handle.materialize_seekable(max_bytes=0) as path:
+        assert path.read_bytes() == b""
+
+
+def test_negative_bound_is_a_caller_bug_not_an_oversized_source(tmp_path: Path) -> None:
+    """A negative bound is nonsense, and must not be reported as size refusal."""
+    handle = _handle(tmp_path=tmp_path, content=b"data")
+
+    with pytest.raises(ValueError):
+        with handle.materialize_seekable(max_bytes=-1):
+            pytest.fail("a negative bound must never materialize anything")
+
+
+def test_materialized_name_keeps_the_source_suffix(tmp_path: Path) -> None:
+    """Demuxers sniff the container from the extension before probing."""
+    handle = _handle(tmp_path=tmp_path, content=b"video")
+
+    with handle.materialize_seekable(max_bytes=1024) as path:
+        assert path.suffix == ".mp4"
+
+
+def test_cleanup_survives_files_written_beside_the_source(tmp_path: Path) -> None:
+    """A decoder writing an index or sidecar must not break teardown.
+
+    Removing only the source file and then the directory would raise
+    "directory not empty" here, leaking the directory and masking whatever the
+    route itself was raising.
+    """
+    handle = _handle(tmp_path=tmp_path, content=b"video")
+
+    with handle.materialize_seekable(max_bytes=1024) as path:
+        (path.parent / "sidecar.idx").write_bytes(b"index")
+        directory = path.parent
+
+    assert not directory.exists()
+
+
+def test_materialize_without_a_temp_root_uses_the_platform_default(
+    tmp_path: Path,
+) -> None:
+    """A deployment that has not sized a work volume still gets cleanup."""
+    store = LocalFSObjectStore(root=tmp_path / "objects")
+    store.write_bytes(key=_KEY, content=b"payload")
+    handle = ObjectSourceHandle(
+        store=store,
+        identity=SourceIdentity(
+            object_key=_KEY,
+            content_hash=hashlib.sha256(b"payload").hexdigest(),
+            byte_size=len(b"payload"),
+        ),
+    )
+
+    with handle.materialize_seekable(max_bytes=1024) as path:
+        assert path.read_bytes() == b"payload"
+        directory = path.parent
+
+    assert not directory.exists()
+
+
+def test_truncated_object_fails_the_range_read(tmp_path: Path) -> None:
+    """A stored object shorter than its record must not yield a short read.
+
+    A container parser cannot distinguish a truncated header from a valid one,
+    so it would produce confident nonsense rather than an error.
+    """
+    store = LocalFSObjectStore(root=tmp_path / "objects")
+    store.write_bytes(key=_KEY, content=b"short")
+    handle = ObjectSourceHandle(
+        store=store,
+        identity=SourceIdentity(
+            object_key=_KEY,
+            content_hash=hashlib.sha256(b"short").hexdigest(),
+            byte_size=100,
+        ),
+        temp_root=tmp_path / "work",
+    )
+
+    with pytest.raises(SourceRangeError):
+        handle.read_range(start=0, end=50)
+
+
+def test_read_bounded_returns_everything_under_the_bound(tmp_path: Path) -> None:
+    """Small routes get every byte without writing their own unbounded join."""
+    handle = _handle(tmp_path=tmp_path, content=b"# heading\n")
+
+    assert handle.read_bounded(max_bytes=1024) == b"# heading\n"
+
+
+def test_read_bounded_refuses_an_oversized_source(tmp_path: Path) -> None:
+    """The bound is the point: this is not a whole-file read in disguise."""
+    handle = _handle(tmp_path=tmp_path, content=b"x" * 5000)
+
+    with pytest.raises(SourceTooLargeError):
+        handle.read_bounded(max_bytes=100)
+
+
+def test_concurrent_materializations_do_not_share_a_directory(tmp_path: Path) -> None:
+    """Two live materializations of one handle must not collide or co-delete."""
+    handle = _handle(tmp_path=tmp_path, content=b"payload")
+
+    with handle.materialize_seekable(max_bytes=1024) as first:
+        with handle.materialize_seekable(max_bytes=1024) as second:
+            assert first != second
+            assert first.parent != second.parent
+            assert first.read_bytes() == second.read_bytes() == b"payload"
+            inner = second.parent
+        assert not inner.exists()
+        assert first.exists(), "the outer materialization must survive the inner"
+
+
+def test_read_ceiling_refuses_a_whole_file_range(tmp_path: Path) -> None:
+    """With a ceiling set, `[0, byte_size)` stops being a whole-file read.
+
+    Without one the "no unbounded read" property is only friction: a caller
+    can always ask for the entire range. The ceiling turns it into a rule.
+    """
+    store = LocalFSObjectStore(root=tmp_path / "objects")
+    store.write_bytes(key=_KEY, content=b"z" * 5000)
+    handle = ObjectSourceHandle(
+        store=store,
+        identity=SourceIdentity(
+            object_key=_KEY,
+            content_hash=hashlib.sha256(b"z" * 5000).hexdigest(),
+            byte_size=5000,
+        ),
+        temp_root=tmp_path / "work",
+        max_read_bytes=1024,
+    )
+
+    with pytest.raises(SourceTooLargeError):
+        handle.read_range(start=0, end=5000)
+    with pytest.raises(SourceTooLargeError):
+        handle.read_bounded(max_bytes=10_000)
+    with pytest.raises(SourceTooLargeError):
+        with handle.open_stream(chunk_bytes=4096):
+            pytest.fail("a chunk over the ceiling must be refused")
+
+    assert handle.read_range(start=0, end=1024) == b"z" * 1024
+
+
+def test_recorded_size_must_match_what_was_actually_stored(tmp_path: Path) -> None:
+    """Bytes can hash correctly and still be described by a wrong length.
+
+    Left unchecked, materialization would succeed with 5000 bytes while a range
+    read past 10 was refused — two access paths disagreeing about one source.
+    """
+    content = b"w" * 5000
+    store = LocalFSObjectStore(root=tmp_path / "objects")
+    store.write_bytes(key=_KEY, content=content)
+    handle = ObjectSourceHandle(
+        store=store,
+        identity=SourceIdentity(
+            object_key=_KEY,
+            content_hash=hashlib.sha256(content).hexdigest(),
+            byte_size=10,
+        ),
+        temp_root=tmp_path / "work",
+    )
+
+    with pytest.raises(SourceSizeMismatchError):
+        with handle.materialize_seekable(max_bytes=10_000):
+            pytest.fail("a size mismatch must not reach the converter")
+
+
+def test_abandoning_a_stream_still_releases_the_file(tmp_path: Path) -> None:
+    """Stopping early must release the descriptor at block exit, not at GC."""
+    handle = _handle(tmp_path=tmp_path, content=b"a" * 4096)
+
+    with handle.open_stream(chunk_bytes=16) as chunks:
+        first = next(iter(chunks))
+        assert first == b"a" * 16
+
+    # Exiting the block closed the underlying file; the source stays readable.
+    assert handle.read_range(start=0, end=4) == b"aaaa"
diff --git a/src/tests/spine/test_knowledge_control_plane.py b/src/tests/spine/test_knowledge_control_plane.py
index ab5129b86..215a003b4 100644
--- a/src/tests/spine/test_knowledge_control_plane.py
+++ b/src/tests/spine/test_knowledge_control_plane.py
@@ -1,6 +1,7 @@
 """WP-6.1 acceptance: live K control plane, routing, and exact staleness."""
 
 from collections.abc import Iterator
+from contextlib import contextmanager
 from datetime import datetime
 from datetime import UTC
 from decimal import Decimal
@@ -552,6 +553,24 @@ def read_bytes(self, *, key: ObjectKey) -> bytes:
         """Return one previously archived session transcript."""
         return self.objects[key.root]
 
+    @contextmanager
+    def open_stream(
+        self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024
+    ) -> Iterator[Iterator[bytes]]:
+        """Open an ordered chunked read over the stored bytes."""
+        content = self.read_bytes(key=key)
+
+        def chunks() -> Iterator[bytes]:
+            """Yield successive fixed-size slices of the stored bytes."""
+            for offset in range(0, len(content), chunk_bytes):
+                yield content[offset : offset + chunk_bytes]
+
+        yield chunks()
+
+    def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes:
+        """Return the half-open byte interval of the stored bytes."""
+        return self.read_bytes(key=key)[start:end]
+
     def write_bytes(
         self, *, key: ObjectKey, content: bytes, storage_class: str | None = None
     ) -> None:
diff --git a/src/tests/test_port_inventory_and_conformance.py b/src/tests/test_port_inventory_and_conformance.py
index a545ba139..3d208f469 100644
--- a/src/tests/test_port_inventory_and_conformance.py
+++ b/src/tests/test_port_inventory_and_conformance.py
@@ -1,5 +1,7 @@
 """Inventory tests for D61 substrate seams plus D74 store capabilities."""
 
+from collections.abc import Iterator
+from contextlib import contextmanager
 from datetime import datetime
 from datetime import timezone
 from decimal import Decimal
@@ -12,6 +14,7 @@
 from pydantic import BaseModel
 from pydantic import SecretBytes
 
+from rememberstack.adapters.selfhost.source_handle import ObjectSourceHandle
 from rememberstack.model import AuthenticatedContext
 from rememberstack.model import EmbeddingRequest
 from rememberstack.model import EmbeddingResponse
@@ -23,6 +26,7 @@
 from rememberstack.model import ProviderCallUsage
 from rememberstack.model import PublishedMounts
 from rememberstack.model import QueueRoute
+from rememberstack.model import SourceIdentity
 from rememberstack.model import StructuredResponseModel
 from rememberstack.model import TelemetryEvent
 from rememberstack.model import UTCDateTime
@@ -32,6 +36,7 @@
 from rememberstack.ports import ModelProviderPort
 from rememberstack.ports import MountPublisherPort
 from rememberstack.ports import ObjectStorePort
+from rememberstack.ports import SourceHandlePort
 from rememberstack.ports import TaskQueuePort
 from rememberstack.ports import TelemetryPort
 import rememberstack.ports.auth as auth_module
@@ -43,6 +48,7 @@
 import rememberstack.ports.postgres_read as postgres_read_module
 import rememberstack.ports.purge as purge_module
 import rememberstack.ports.queue as queue_module
+import rememberstack.ports.source_handle as source_handle_module
 import rememberstack.ports.telemetry as telemetry_module
 
 ResponseT = TypeVar("ResponseT", bound=StructuredResponseModel)
@@ -57,6 +63,7 @@
     postgres_read_module,
     queue_module,
     purge_module,
+    source_handle_module,
     telemetry_module,
 )
 _PORT_EXPORTS = {
@@ -70,6 +77,7 @@
     "ObjectPurgePort",
     "PostgresReadPoolPort",
     "ProjectionPurgePort",
+    "SourceHandlePort",
     "TaskQueuePort",
     "TelemetryPort",
 }
@@ -92,6 +100,24 @@ def read_bytes(self, *, key: ObjectKey) -> bytes:
         """Return the bytes stored under the requested key."""
         return self.objects[key.root]
 
+    @contextmanager
+    def open_stream(
+        self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024
+    ) -> Iterator[Iterator[bytes]]:
+        """Open an ordered chunked read over the stored bytes."""
+        content = self.read_bytes(key=key)
+
+        def chunks() -> Iterator[bytes]:
+            """Yield successive fixed-size slices of the stored bytes."""
+            for offset in range(0, len(content), chunk_bytes):
+                yield content[offset : offset + chunk_bytes]
+
+        yield chunks()
+
+    def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes:
+        """Return the half-open byte interval of the stored bytes."""
+        return self.objects[key.root][start:end]
+
     def write_bytes(
         self, *, key: ObjectKey, content: bytes, storage_class: str | None = None
     ) -> None:
@@ -206,6 +232,12 @@ def announce(
 
 
 _object_store_assignment: ObjectStorePort = FakeObjectStore()
+_source_handle_assignment: SourceHandlePort = ObjectSourceHandle(
+    store=FakeObjectStore(),
+    identity=SourceIdentity(
+        object_key=ObjectKey("raw/example"), content_hash="0" * 64, byte_size=0
+    ),
+)
 _mount_assignment: MountPublisherPort = FakeMountPublisher()
 _git_assignment: KGitRemotePort = FakeKGitRemote()
 _model_assignment: ModelProviderPort = FakeModelProvider()
@@ -228,11 +260,11 @@ def _defined_protocols() -> set[type[object]]:
     return result
 
 
-def test_inventory_exports_exactly_twelve_defined_protocols() -> None:
+def test_inventory_exports_exactly_thirteen_defined_protocols() -> None:
     """Keep D61/D74 seams plus bounded PostgreSQL read admission explicit."""
     assert set(ports.__all__) == _PORT_EXPORTS
     assert {protocol.__name__ for protocol in _defined_protocols()} == _PORT_EXPORTS
-    assert len(_defined_protocols()) == 12
+    assert len(_defined_protocols()) == 13
 
 
 def test_representative_fakes_conform_structurally() -> None:
diff --git a/src/tests/workers/test_ingest_admission.py b/src/tests/workers/test_ingest_admission.py
index ac00ae989..03c504743 100644
--- a/src/tests/workers/test_ingest_admission.py
+++ b/src/tests/workers/test_ingest_admission.py
@@ -1,5 +1,7 @@
 """Fast D74 proof that ingest checks admission before persisting bytes."""
 
+from collections.abc import Iterator
+from contextlib import contextmanager
 from datetime import datetime
 from datetime import timedelta
 from datetime import timezone
@@ -56,6 +58,24 @@ def __init__(self) -> None:
     def read_bytes(self, *, key: ObjectKey) -> bytes:
         raise AssertionError(f"unexpected read of {key.root}")
 
+    @contextmanager
+    def open_stream(
+        self, *, key: ObjectKey, chunk_bytes: int = 1024 * 1024
+    ) -> Iterator[Iterator[bytes]]:
+        """Open an ordered chunked read over the stored bytes."""
+        content = self.read_bytes(key=key)
+
+        def chunks() -> Iterator[bytes]:
+            """Yield successive fixed-size slices of the stored bytes."""
+            for offset in range(0, len(content), chunk_bytes):
+                yield content[offset : offset + chunk_bytes]
+
+        yield chunks()
+
+    def read_range(self, *, key: ObjectKey, start: int, end: int) -> bytes:
+        """Return the half-open byte interval of the stored bytes."""
+        return self.read_bytes(key=key)[start:end]
+
     def write_bytes(
         self, *, key: ObjectKey, content: bytes, storage_class: str | None = None
     ) -> None: