feat(media): a converter receives a source handle, never the file's bytes (D104)

.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

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.

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

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:

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: