fix: decoder cache keys, msgspec field claims, unread middleware responses

docs/decoders.md

@@ -33,10 +33,11 @@ Both clients take `decoders: Sequence[ResponseDecoder] | None = None`. The list
 - Order is preference. With `decoders=[CsvDecoder(), PydanticDecoder()]`, pydantic only sees the types the CSV decoder declined. When two decoders could both handle a type, the earlier one wins.
 - `decoders=None` uses the installed extras: pydantic then msgspec when both are installed, whichever one is installed, or no decoders at all. Passing a list replaces the defaults, so include the built-ins you still want: `decoders=[CsvDecoder(), PydanticDecoder()]`.
 - If `response_model=` is set and no decoder accepts it, the client raises `MissingDecoderError` before sending the request. That means nothing handles the type, and the fix is to install an extra or pass `decoders=[...]`. A `DecodeError` instead means a decoder ran and the body didn't fit the type, which points at the server or the model. See [Errors](errors.md).
+- `MsgspecDecoder` accepts a `Struct`, dataclass, `TypedDict` or `NamedTuple` only if msgspec can decode every field type itself, nested fields included. A field typed as a pydantic model or some other class would need a `dec_hook`, which `MsgspecDecoder` doesn't take, so it declines the model and you get `MissingDecoderError` before the request goes out.
 
 ## One sync protocol for both clients
 
-Middleware comes in sync and async versions, but there is only one `ResponseDecoder` protocol, used by both clients. `decode` is synchronous because the body has already been read when it runs, so there is nothing to await. The same decoder works with either client.
+Middleware comes in sync and async versions, but there is only one `ResponseDecoder` protocol, used by both clients. `decode` is synchronous because the client has the whole body by the time it runs, so there is nothing to await. If a middleware returns a streamed response it never read, the client reads it before decoding. In `AsyncClient` that stream also has to support sync iteration, as `httpx2.ByteStream` does; otherwise the middleware should `await response.aread()` before returning it. The same decoder works with either client.
 
 ## Writing your own
 
@@ -131,5 +132,5 @@ This decoder makes two passes: `json.loads`, then `structure`. The built-ins dec
 
 ## See also
 
-- [`src/httpware/decoders/`](https://github.com/modern-python/httpware/tree/main/src/httpware/decoders): the built-in decoders, including how they cache `can_decode` results and parsers per type.
+- [`src/httpware/decoders/`](https://github.com/modern-python/httpware/tree/main/src/httpware/decoders): the built-in decoders, including how each caches one parser, or one rejection, per type.
 - [Quickstart: typed responses](index.md#typed-responses): `response_model=` with the default decoders.

src/httpware/decoders/_caching.py

@@ -1,30 +1,52 @@
-"""Shared memoizing get-or-build cache for the decoder adapters (Seam C).
+"""One cache per decoder, keyed so that union member order counts (Seam C).
 
-Imports neither pydantic nor msgspec — safe for both decoder modules to
-import without crossing Seam C's per-extra import isolation.
+Imports neither pydantic nor msgspec, so both decoder modules can use it
+without crossing Seam C's per-extra import isolation.
 """
 
 import typing
+from collections.abc import Callable, Hashable
 
 
 V = typing.TypeVar("V")
 
+_Cache: typing.TypeAlias = dict[Hashable, tuple[object, V | None]]
 
-def _get_or_build(cache: dict[type, V], model: type, build: typing.Callable[[], V]) -> V:
-    """Return cache[model], building and memoizing it via build() on miss.
 
-    Unhashable models bypass the cache entirely: dict.get(model) raises
-    TypeError, in which case this returns a fresh, uncached build() every
-    call rather than raising. A real cached value (bool verdict or a
-    built adapter/decoder) is never None, so a None result unambiguously
-    means "not cached yet" — callers must not store None as a value.
+def _key(model: object) -> Hashable:
+    """Return a key that tells `int | str` from `str | int`, which compare equal.
+
+    A class keys by identity; anything else by itself plus the keys of its
+    `__args__`, in order.
+    """
+    if isinstance(model, type):
+        return model
+    return (model, *map(_key, getattr(model, "__args__", ())))
+
+
+def _lookup(cache: "_Cache[V]", model: object, build: Callable[[typing.Any], V | None]) -> V | None:
+    """Return the cached `build(model)`, building it on a miss; None means `model` cannot be decoded.
+
+    `build` must not raise. Concurrent first calls may build twice; the last write wins.
     """
     try:
-        cached = cache.get(model)
-    except TypeError:
-        return build()
-    if cached is not None:
-        return cached
-    result = build()
-    cache[model] = result
-    return result
+        key = _key(model)
+        entry = cache.get(key)
+    except Exception:  # noqa: BLE001 — unhashable models fall back to a repr key
+        return _lookup_by_repr(cache, model, build)
+    if entry is None:
+        entry = cache[key] = (model, build(model))
+    return entry[1]
+
+
+def _lookup_by_repr(cache: "_Cache[V]", model: object, build: Callable[[typing.Any], V | None]) -> V | None:
+    """Key an unhashable `model` by its repr; the equality check keeps two models with one repr apart."""
+    try:
+        key = repr(model)
+        entry = cache.get(key)
+        if entry is not None and entry[0] == model:
+            return entry[1]
+    except Exception:  # noqa: BLE001 — a model that cannot be keyed is built uncached
+        return build(model)
+    entry = cache[key] = (model, build(model))
+    return entry[1]

src/httpware/decoders/_resolver.py

@@ -36,9 +36,9 @@ def __init__(self, decoder: ResponseDecoder, model: type[T]) -> None:
         self._model = model
 
     def decode(self, response: httpx2.Response) -> T:
-        """Decode `response.content` into `model`, wrapping any failure as `DecodeError`."""
+        """Read the body if needed and decode it into `model`, wrapping any failure as `DecodeError`."""
         try:
-            return self._decoder.decode(response.content, self._model)
+            return self._decoder.decode(response.read(), self._model)
         except Exception as exc:
             raise DecodeError(response=response, model=self._model, original=exc) from exc
 

src/httpware/decoders/msgspec.py

@@ -1,10 +1,11 @@
 """MsgspecDecoder — opt-in ResponseDecoder backed by a per-instance msgspec.json.Decoder cache."""
 
 import typing
+from collections.abc import Iterator
 from typing import TypeVar
 
 from httpware._internal import import_checker
-from httpware.decoders._caching import _get_or_build
+from httpware.decoders._caching import _Cache, _lookup
 
 
 if import_checker.is_msgspec_installed:
@@ -16,36 +17,50 @@
 T = TypeVar("T")
 
 
-def _contains_custom_type(info: "msgspec.inspect.Type") -> bool:
-    """Return True if `info` is a CustomType or nests one in its parameters.
+def _child_types(info: "msgspec.inspect.Type") -> Iterator["msgspec.inspect.Type"]:
+    """Yield the types nested in `info`: container parameters and the field types of a Struct-like model."""
+    for name in info.__struct_fields__:
+        value = getattr(info, name)
+        if isinstance(value, msgspec.inspect.Type):
+            yield value
+        elif isinstance(value, tuple):
+            for item in value:
+                if isinstance(item, msgspec.inspect.Type):
+                    yield item
+                elif isinstance(item, msgspec.inspect.Field):
+                    yield item.type
+
 
-    Walks generic-container parameterization (list/dict/set/tuple/union element
-    types) by visiting any attribute that is itself a `msgspec.inspect.Type` or a
-    tuple of them. It deliberately does NOT descend into `StructType`/dataclass
-    fields: those expose `fields` as `Field` objects (not `Type`), so the walk
-    stops at the boundary of a type msgspec natively owns. That boundary is what
-    makes the walk both correct (a Struct is a valid target) and safe against
-    infinite recursion on self-referential struct definitions.
+def _is_native(info: "msgspec.inspect.Type") -> bool:
+    """Return True iff no `CustomType` appears anywhere in the type graph rooted at `info`.
+
+    The graph is cyclic for self-referential models, so each node is visited once.
     """
-    if not import_checker.is_msgspec_installed:
-        raise ImportError(MISSING_DEPENDENCY_MESSAGE)
-    if isinstance(info, msgspec.inspect.CustomType):
-        return True
-    for name in dir(info):
-        if name.startswith("_"):
+    seen: set[int] = set()
+    stack = [info]
+    while stack:
+        node = stack.pop()
+        if isinstance(node, msgspec.inspect.CustomType):
+            return False
+        if id(node) in seen:
             continue
-        value = getattr(info, name, None)
-        if isinstance(value, msgspec.inspect.Type):
-            if _contains_custom_type(value):
-                return True
-        elif (
-            isinstance(value, tuple)
-            and value
-            and all(isinstance(item, msgspec.inspect.Type) for item in value)
-            and any(_contains_custom_type(item) for item in value)
-        ):
-            return True
-    return False
+        seen.add(id(node))
+        stack.extend(_child_types(node))
+    return True
+
+
+def _build_adapter(model: object) -> "msgspec.json.Decoder[typing.Any] | None":
+    """Return a `msgspec.json.Decoder` for `model`, or None when msgspec cannot decode it natively.
+
+    msgspec builds a Decoder for almost any class through its `CustomType`
+    fallback, so the type graph is checked for one first, field types included.
+    """
+    try:
+        if not _is_native(msgspec.inspect.type_info(model)):
+            return None
+        return msgspec.json.Decoder(model)
+    except Exception:  # noqa: BLE001 — can_decode is a probe; any failure means no
+        return None
 
 
 class MsgspecDecoder:
@@ -56,49 +71,25 @@ class MsgspecDecoder:
     `find_spec` check), but instantiating the decoder raises `ImportError`.
     """
 
-    _msgspec_decoders: dict[type, "msgspec.json.Decoder[typing.Any]"]
-    _can_decode_results: dict[type, bool]
+    _adapters: "_Cache[msgspec.json.Decoder[typing.Any]]"
 
     def __init__(self) -> None:
         if not import_checker.is_msgspec_installed:
             raise ImportError(MISSING_DEPENDENCY_MESSAGE)
-        self._msgspec_decoders = {}
-        self._can_decode_results = {}
-
-    def _get_msgspec_decoder(self, model: type[T]) -> "msgspec.json.Decoder[T]":
-        return _get_or_build(self._msgspec_decoders, model, lambda: msgspec.json.Decoder(model))
+        self._adapters = {}
 
     def can_decode(self, model: type) -> bool:
         """Return True iff msgspec natively understands `model` end-to-end.
 
-        The verdict is memoized per `model`: the probe below (an uncached
-        `type_info` call plus a recursive tree walk) runs once per type, not on
-        every dispatch. Unhashable models skip the cache and probe fresh.
-        """
-        return _get_or_build(self._can_decode_results, model, lambda: self._probe_can_decode(model))
-
-    def _probe_can_decode(self, model: type) -> bool:
-        """Decide whether msgspec natively decodes `model` (the uncached path).
-
-        msgspec builds a Decoder for almost any class via a generic CustomType
-        fallback; the Decoder constructor does NOT raise on unsupported types
-        (e.g. pydantic.BaseModel, or a container parameterized by one). We walk
-        msgspec.inspect.type_info and reject if a CustomType appears anywhere in
-        the type tree, so MissingDecoderError fires before a request is sent.
+        Rejects a model if any type in it, including the field types of a
+        Struct, dataclass, TypedDict or NamedTuple, would need a `dec_hook`.
+        The adapter, or the rejection, is cached per `model`.
         """
-        try:
-            info = msgspec.inspect.type_info(model)
-        except Exception:  # noqa: BLE001 — can_decode is a probe; any failure means no
-            return False
-        if _contains_custom_type(info):
-            return False
-        try:
-            self._get_msgspec_decoder(model)
-        except Exception:  # noqa: BLE001 — can_decode is a probe; any failure means no
-            return False
-        return True
+        return _lookup(self._adapters, model, _build_adapter) is not None
 
     def decode(self, content: bytes, model: type[T]) -> T:
         """Validate `content` as JSON against `model` in a single parse pass."""
-        decoder = self._get_msgspec_decoder(model)
-        return decoder.decode(content)
+        adapter = _lookup(self._adapters, model, _build_adapter)
+        if adapter is None:
+            adapter = msgspec.json.Decoder(model)  # decoding then raises msgspec's own error
+        return adapter.decode(content)

src/httpware/decoders/pydantic.py

@@ -11,7 +11,7 @@ class entirely when `is_pydantic_installed` is False, so `AsyncClient()` does
 from typing import TypeVar
 
 from httpware._internal import import_checker
-from httpware.decoders._caching import _get_or_build
+from httpware.decoders._caching import _Cache, _lookup
 
 
 if import_checker.is_pydantic_installed:
@@ -25,6 +25,14 @@ class entirely when `is_pydantic_installed` is False, so `AsyncClient()` does
 T = TypeVar("T")
 
 
+def _build_adapter(model: typing.Any) -> "TypeAdapter[typing.Any] | None":  # noqa: ANN401
+    """Return a `TypeAdapter` for `model`, or None when pydantic cannot build a schema for it."""
+    try:
+        return TypeAdapter(model)
+    except Exception:  # noqa: BLE001 — can_decode is a probe; any failure means no
+        return None
+
+
 class PydanticDecoder:
     """Decode raw response bytes into `model` via a per-instance cached `pydantic.TypeAdapter`.
 
@@ -34,42 +42,25 @@ class PydanticDecoder:
     `ImportError`.
     """
 
-    _adapters: dict[type, "TypeAdapter[typing.Any]"]
-    _can_decode_results: dict[type, bool]
+    _adapters: "_Cache[TypeAdapter[typing.Any]]"
 
     def __init__(self) -> None:
         if not import_checker.is_pydantic_installed:
             raise ImportError(MISSING_DEPENDENCY_MESSAGE)
         self._adapters = {}
-        self._can_decode_results = {}
-
-    def _get_adapter(self, model: type[T]) -> "TypeAdapter[T]":
-        return _get_or_build(self._adapters, model, lambda: TypeAdapter(model))
 
     def can_decode(self, model: type) -> bool:
         """Return True iff pydantic can build a schema for `model`.
 
-        The verdict is memoized per `model` so a rejection (which costs a
-        `PydanticSchemaGenerationError` round-trip) is not re-probed on every
-        dispatch. Unhashable models skip the cache and probe fresh.
-        """
-        return _get_or_build(self._can_decode_results, model, lambda: self._probe_can_decode(model))
-
-    def _probe_can_decode(self, model: type) -> bool:
-        """Decide whether pydantic can build a schema for `model` (uncached).
-
-        Probes via `_get_adapter`; subsequent calls (including `decode`) reuse
-        the cached `TypeAdapter`. Rejects `msgspec.Struct` subclasses —
-        pydantic raises `PydanticSchemaGenerationError` (a `TypeError`) when
-        building a schema for them.
+        The adapter, or the rejection, is cached per `model`, so `decode` reuses
+        what this built. `msgspec.Struct` subclasses are rejected: pydantic
+        cannot build a schema for them.
         """
-        try:
-            self._get_adapter(model)
-        except Exception:  # noqa: BLE001 — can_decode is a probe; any failure means no
-            return False
-        return True
+        return _lookup(self._adapters, model, _build_adapter) is not None
 
     def decode(self, content: bytes, model: type[T]) -> T:
         """Validate `content` as JSON against `model` in a single parse pass."""
-        adapter = self._get_adapter(model)
+        adapter = _lookup(self._adapters, model, _build_adapter)
+        if adapter is None:
+            adapter = TypeAdapter(model)  # raises pydantic's own schema error
         return adapter.validate_json(content)

tests/test_decoder_resolver.py

@@ -1,13 +1,15 @@
 """Seam-level tests for _DecoderResolver / _BoundDecoder.
 
-Drives resolve() and _BoundDecoder.decode() directly with a fake decoder list —
-no client, no MockTransport. Covers resolution, the pre-flight MissingDecoderError
-(with registered_names + ordering), first-match-wins, and DecodeError wrapping.
+Drives resolve() and _BoundDecoder.decode() directly with a fake decoder list.
+Covers resolution, the pre-flight MissingDecoderError (with registered_names +
+ordering), first-match-wins, and DecodeError wrapping. The last tests run both
+clients over MockTransport to decode a response a middleware built but never read.
 """
 
 import httpx2
 import pytest
 
+from httpware import AsyncClient, Client
 from httpware.decoders._resolver import _DecoderResolver
 from httpware.errors import DecodeError, MissingDecoderError
 
@@ -83,3 +85,48 @@ def test_decode_failure_wraps_as_decode_error() -> None:
     assert ei.value.model is _Wanted
     assert ei.value.original is boom
     assert ei.value.__cause__ is boom
+
+
+class _EchoDecoder:
+    def can_decode(self, model: type) -> bool:
+        return model is bytes
+
+    def decode(self, content: bytes, model: type) -> object:  # noqa: ARG002 — echoes the body
+        return content
+
+
+def _unread_response() -> httpx2.Response:
+    return httpx2.Response(200, stream=httpx2.ByteStream(b"streamed"))
+
+
+def test_bound_decoder_reads_an_unread_response() -> None:
+    bound = _DecoderResolver((_EchoDecoder(),)).resolve(bytes)
+    assert bound.decode(_unread_response()) == b"streamed"
+
+
+def _never_called(request: httpx2.Request) -> httpx2.Response:  # pragma: no cover — the request must not go out
+    raise AssertionError(request.url)
+
+
+def test_client_decodes_unread_response_from_middleware() -> None:
+    def synthesize(request: httpx2.Request, next: object) -> httpx2.Response:  # noqa: A002, ARG001
+        return _unread_response()
+
+    client = Client(
+        httpx2_client=httpx2.Client(transport=httpx2.MockTransport(_never_called)),
+        decoders=[_EchoDecoder()],
+        middleware=[synthesize],
+    )
+    assert client.get("https://example.test/x", response_model=bytes) == b"streamed"
+
+
+async def test_async_client_decodes_unread_response_from_middleware() -> None:
+    async def synthesize(request: httpx2.Request, next: object) -> httpx2.Response:  # noqa: A002, ARG001
+        return _unread_response()
+
+    client = AsyncClient(
+        httpx2_client=httpx2.AsyncClient(transport=httpx2.MockTransport(_never_called)),
+        decoders=[_EchoDecoder()],
+        middleware=[synthesize],
+    )
+    assert await client.get("https://example.test/x", response_model=bytes) == b"streamed"