fix(sdk): isolate metadata requests and bound streamed responses

Author: nkavian

README.md

@@ -203,9 +203,10 @@ The Agent module also provides:
 - Conditional request and representation caching with injectable `Cache` and `Transport` protocols.
 
 Default fallback cache lifetimes are four hours for Service documents, one hour for Collections,
-and five minutes for Offerings. HTTP cache directives take precedence. Provide distinct `transport`
-and `supporting_transport` instances when protocol resources and linked schemas require different
-credentials or network policy.
+and five minutes for Offerings. HTTP cache directives take precedence. `ServiceClient` uses a
+separate anonymous transport for linked schemas and OpenAPI documents, even when its primary
+`transport` has authentication configured. An explicit `supporting_transport` override must also
+send these requests anonymously; it must not share the primary transport's credentials or cookies.
 
 ### Search across Services
 
@@ -269,6 +270,12 @@ must survive process restarts or share storage across workers. A custom `Transpo
 asynchronous `send()` and `aclose()` methods. Caller-provided caches and transports remain owned by
 the caller.
 
+`HttpRequest.maximum_response_bytes` gives a custom transport the response budget. Enforce it while
+reading, rather than buffering the complete response first. The built-in transport closes responses
+on overflow, read failure, and cancellation. A successful response exceeding its budget raises
+`TransportError` with `code="RESPONSE_LIMIT_EXCEEDED"`; `ServiceClient` preserves that code on
+`AgentError`. Oversized error bodies are discarded while retaining the HTTP status and headers.
+
 The built-in HTTP transport resolves and validates every destination before connecting, pins the
 connection to a validated public address, does not inherit proxy settings from the environment, and
 sends supporting-document requests without credentials. A custom transport must preserve those ODP
@@ -348,7 +355,9 @@ Directory results filter unrecognized enrollment, payment, and trust descriptors
 strict validation for recognized descriptors.
 
 Individual Offering and Collection GETs default to full representations; list and search operations
-default to terse items. A Catalog receives the requested language in `CatalogRequest.language` and
+default to terse items. The Service handler writes `odp_version` on standalone resources and page
+envelopes, omitting it from embedded page items without changing the Catalog's models.
+A Catalog receives the requested language in `CatalogRequest.language` and
 declares the language it actually returns on each resource. Static catalogs do not translate content.
 
 Refinement parsing detects duplicate JSON values without guessing the type of a string. Comparing
@@ -367,6 +376,10 @@ Protocol models preserve additive members in `model.additional` and round-trip t
 `model.to_dict()`. Parsing remains strict for normative constraints and fields that prohibit unknown
 members.
 
+Directory records retain additional metadata such as branding and MCP endpoints. These are discovery
+hints, not authorization or authoritative routing data. The default Agent factory uses the record's
+`service_origin` and retrieves that Service's own document before making catalog requests.
+
 Handle the narrowest error that the application can act upon and use the role's base error for the
 remaining failures:
 

src/offering_protocol/agent/client.py

@@ -91,6 +91,10 @@ class TraversalOptions:
 class AgentError(RuntimeError):
     """Base error for Service discovery operations."""
 
+    def __init__(self, message: str, *, code: str | None = None) -> None:
+        super().__init__(message)
+        self.code = code
+
 
 class UnsupportedOperationError(AgentError):
     def __init__(self, operation: Operation) -> None:
@@ -132,7 +136,12 @@ def __init__(
         self._cache_partition = cache_partition
         self._owns_transport = transport is None
         self._transport = transport or HttpxTransport(allow_local_network=allow_local_network)
-        self._supporting_transport = supporting_transport or self._transport
+        self._owns_supporting_transport = supporting_transport is None
+        self._supporting_transport = (
+            supporting_transport
+            if supporting_transport is not None
+            else HttpxTransport(allow_local_network=allow_local_network)
+        )
 
     async def __aenter__(self) -> ServiceClient:
         return self
@@ -141,8 +150,12 @@ async def __aexit__(self, *args: object) -> None:
         await self.aclose()
 
     async def aclose(self) -> None:
-        if self._owns_transport:
-            await self._transport.aclose()
+        try:
+            if self._owns_transport:
+                await self._transport.aclose()
+        finally:
+            if self._owns_supporting_transport:
+                await self._supporting_transport.aclose()
 
     async def inspect(self) -> Inspection:
         requested_url = f"{self.service_origin}/.well-known/odp"
@@ -399,7 +412,9 @@ async def _request_cached(
                 headers["if-none-match"] = cached.etag
             if cached.last_modified:
                 headers["if-modified-since"] = cached.last_modified
-        response, final_url = await self._request_raw(method, request_target, body, headers)
+        response, final_url = await self._request_raw(
+            method, request_target, body, headers, maximum_bytes
+        )
         if response.status == 304:
             if cached is None:
                 raise AgentError("ODP response returned 304 without a cached representation")
@@ -435,7 +450,7 @@ async def _request_cached(
         return _FetchedResponse(response.body, final_url, Freshness.FETCHED)
 
     async def _request_raw(
-        self, method: str, target: str, body: bytes, conditional: dict[str, str]
+        self, method: str, target: str, body: bytes, conditional: dict[str, str], maximum_bytes: int
     ) -> tuple[HttpResponse, str]:
         redirect_origin = derive_service_origin(target)
         for redirects in range(_MAXIMUM_REDIRECTS + 1):
@@ -445,9 +460,11 @@ async def _request_raw(
             if body:
                 headers["content-type"] = MEDIA_TYPE
             try:
-                response = await self._transport.send(HttpRequest(method, target, headers, body))
+                response = await self._transport.send(
+                    HttpRequest(method, target, headers, body, maximum_response_bytes=maximum_bytes)
+                )
             except TransportError as error:
-                raise AgentError(f"ODP Service request failed: {error}") from error
+                raise AgentError(f"ODP Service request failed: {error}", code=error.code) from error
             if response.status not in {301, 302, 303, 307, 308}:
                 return response, target
             if redirects == _MAXIMUM_REDIRECTS:
@@ -499,10 +516,17 @@ async def _supporting_json(
             visited.add(current)
             try:
                 response = await self._supporting_transport.send(
-                    HttpRequest("GET", current, {"accept": accept, **conditional})
+                    HttpRequest(
+                        "GET",
+                        current,
+                        {"accept": accept, **conditional},
+                        maximum_response_bytes=maximum_bytes,
+                    )
                 )
             except TransportError as error:
-                raise AgentError(f"ODP supporting document request failed: {error}") from error
+                raise AgentError(
+                    f"ODP supporting document request failed: {error}", code=error.code
+                ) from error
             if response.status in {301, 302, 303, 307, 308}:
                 if redirects == _MAXIMUM_REDIRECTS:
                     raise AgentError("ODP supporting document exceeded five redirects")
@@ -544,7 +568,9 @@ async def _supporting_json(
                     response.headers,
                 )
             if len(response.body) > maximum_bytes:
-                raise AgentError("ODP supporting document exceeds its byte limit")
+                raise AgentError(
+                    "ODP supporting document exceeds its byte limit", code="RESPONSE_LIMIT_EXCEEDED"
+                )
             content_type = response.headers.get("content-type", "").split(";", 1)[0].strip().lower()
             if content_type not in media_types:
                 raise AgentError("ODP supporting document has an unsupported media type")
@@ -643,7 +669,7 @@ def _consume(response: HttpResponse, maximum_bytes: int, maximum_depth: int) ->
     if not 200 <= response.status < 300:
         raise ServiceRequestError(response.status, _problem_message(response), response.headers)
     if len(response.body) > maximum_bytes:
-        raise AgentError("ODP response exceeds its byte limit")
+        raise AgentError("ODP response exceeds its byte limit", code="RESPONSE_LIMIT_EXCEEDED")
     content_type = response.headers.get("content-type", "").split(";", 1)[0].strip().lower()
     if content_type != MEDIA_TYPE:
         raise AgentError(f"ODP response must use {MEDIA_TYPE}")

src/offering_protocol/directory/client.py

@@ -42,19 +42,6 @@
 _RFC_3339 = re.compile(
     r"^\d{4}-\d{2}-\d{2}[Tt]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[Zz]|[+-]\d{2}:\d{2})$"
 )
-#: Service Document members a Directory result may echo that this client does not validate.
-#:
-#: They are dropped rather than passed through, because a caller reading them off a record has no
-#: way to tell they were never checked. `http` is the one that matters most: a caller could build
-#: request URLs from an `endpoint_base` a Directory made up.
-_UNVERIFIED_MEMBERS = (
-    "branding",
-    "http",
-    "mcp",
-    "odp_version",
-    "payment_origins",
-    "search_capabilities",
-)
 
 
 class DirectoryError(RuntimeError):
@@ -288,8 +275,7 @@ def _read_service(entry: object) -> dict[str, object]:
     document = _validate_as_service_document(item)
     # `protocols` is reinstated from the validated document only when something survived agent
     # filtering, so a block naming nothing this ODP version knows does not pass straight through.
-    for member in (*_UNVERIFIED_MEMBERS, "protocols"):
-        item.pop(member, None)
+    item.pop("protocols", None)
     if document.protocols is not None:
         item["protocols"] = document.protocols.model_dump(mode="json", exclude_defaults=True)
     item["operations"] = [operation.model_dump(mode="json") for operation in document.operations]

src/offering_protocol/directory/transport.py

@@ -20,6 +20,7 @@ class HttpRequest:
     url: str
     headers: dict[str, str] = field(default_factory=dict)
     body: bytes = b""
+    maximum_response_bytes: int = 524_288
 
 
 @dataclass(frozen=True, slots=True)
@@ -32,6 +33,10 @@ class HttpResponse:
 class TransportError(RuntimeError):
     """Raised when the HTTP transport cannot complete a request."""
 
+    def __init__(self, message: str, *, code: str | None = None) -> None:
+        super().__init__(message)
+        self.code = code
+
 
 class Transport(Protocol):
     async def send(self, request: HttpRequest) -> HttpResponse: ...
@@ -49,6 +54,8 @@ def __init__(
 
     async def send(self, request: HttpRequest) -> HttpResponse:
         try:
+            if request.maximum_response_bytes <= 0:
+                raise ValueError("maximum_response_bytes must be positive")
             target, hostname, host_header = await _pinned_target(
                 request.url, self._allow_local_network
             )
@@ -78,14 +85,36 @@ async def send(self, request: HttpRequest) -> HttpResponse:
                 content=request.body,
                 extensions={"sni_hostname": hostname},
             )
-            response = await client.send(outgoing, follow_redirects=False)
+            response = await client.send(outgoing, follow_redirects=False, stream=True)
+            try:
+                body = bytearray()
+                if request.method.upper() != "HEAD" and not 300 <= response.status_code < 400:
+                    success = 200 <= response.status_code < 300
+                    maximum = (
+                        request.maximum_response_bytes
+                        if success
+                        else min(request.maximum_response_bytes, 16_384)
+                    )
+                    async for chunk in response.aiter_bytes():
+                        if len(chunk) > maximum - len(body):
+                            if success:
+                                raise TransportError(
+                                    "HTTP response exceeds its byte limit",
+                                    code="RESPONSE_LIMIT_EXCEEDED",
+                                )
+                            # Preserve the HTTP failure without exposing truncated error JSON.
+                            body.clear()
+                            break
+                        body.extend(chunk)
+                return HttpResponse(
+                    status=response.status_code,
+                    headers={name.lower(): value for name, value in response.headers.items()},
+                    body=bytes(body),
+                )
+            finally:
+                await response.aclose()
         except (httpx.HTTPError, OSError, ValueError) as error:
             raise TransportError(f"HTTP transport failed: {error}") from error
-        return HttpResponse(
-            status=response.status_code,
-            headers={name.lower(): value for name, value in response.headers.items()},
-            body=response.content,
-        )
 
     async def aclose(self) -> None:
         if self._client is not None:

src/offering_protocol/service/service.py

@@ -555,6 +555,11 @@ def _problem(
 
 
 def _encode(value: object) -> bytes:
+    if isinstance(value, Page):
+        document = value.model_dump(mode="json", by_alias=True, exclude_unset=True)
+        for item in document["items"]:
+            item.pop("odp_version", None)
+        return json.dumps(document, separators=(",", ":"), ensure_ascii=False).encode()
     if hasattr(value, "model_dump_json"):
         encoded = value.model_dump_json(by_alias=True, exclude_unset=True)
         return cast(str, encoded).encode()

tests/test_agent.py

@@ -276,15 +276,18 @@ async def test_builds_agent_friendly_offering_details_without_invoking_action()
     transport = QueueTransport(
         response(SERVICE_DOCUMENT),
         response(ACTION_OFFERING),
-        response(schema, content_type="application/schema+json"),
     )
+    supporting = QueueTransport(response(schema, content_type="application/schema+json"))
     details = await ServiceClient(
-        "https://demo.inflowpay.ai", transport=transport
+        "https://demo.inflowpay.ai", transport=transport, supporting_transport=supporting
     ).get_offering_details("rubber-plant")
     assert details.actions[0].http is not None
     assert details.actions[0].http.url == "https://demo.inflowpay.ai/actions/purchase"
     assert details.attribute_schema is not None
-    assert len(transport.requests) == 3
+    assert len(transport.requests) == 2
+    assert len(supporting.requests) == 1
+    assert [request.maximum_response_bytes for request in transport.requests] == [65_536, 524_288]
+    assert supporting.requests[0].maximum_response_bytes == 262_144
 
 
 DIRECTORY_PAGE = """{
@@ -447,6 +450,8 @@ async def test_resolves_http_action_request_schema() -> None:
         transport=QueueTransport(
             response(SERVICE_DOCUMENT),
             response(offering),
+        ),
+        supporting_transport=QueueTransport(
             response(schema, content_type="application/schema+json"),
             response(schema, content_type="application/schema+json"),
             response(schema, content_type="application/schema+json"),
@@ -476,22 +481,26 @@ async def test_resolves_openapi_action_and_rejects_invalid_openapi() -> None:
       "openapi":"3.1.0","paths":{"/purchase":{"post":{
       "operationId":"purchasePlant","responses":{}}}}
     }"""
+    supporting = QueueTransport(response(openapi, content_type="application/json"))
     client = ServiceClient(
         "https://demo.inflowpay.ai",
         transport=QueueTransport(
             response(document),
             response(offering),
-            response(openapi, content_type="application/json"),
         ),
+        supporting_transport=supporting,
     )
     resolved = await client.resolve_action("plant", "purchase")
     assert resolved.operation == {"operationId": "purchasePlant", "responses": {}}
+    assert supporting.requests[0].maximum_response_bytes == 1_048_576
 
     invalid_client = ServiceClient(
         "https://demo.inflowpay.ai",
         transport=QueueTransport(
             response(document),
             response(offering),
+        ),
+        supporting_transport=QueueTransport(
             response('{"openapi":"3.0.0"}', content_type="application/json"),
         ),
     )
@@ -522,6 +531,8 @@ async def test_offering_details_report_unusable_actions_and_attributes() -> None
         transport=QueueTransport(
             response(SERVICE_DOCUMENT),
             response(offering),
+        ),
+        supporting_transport=QueueTransport(
             response(schema, content_type="application/schema+json"),
         ),
     )
@@ -564,7 +575,7 @@ async def test_supporting_document_security_and_cache_edges() -> None:
     revalidating = ServiceClient(
         "https://demo.inflowpay.ai",
         cache=cache,
-        transport=QueueTransport(response(b"", status=304)),
+        supporting_transport=QueueTransport(response(b"", status=304)),
     )
     assert await revalidating._supporting_json(
         "https://schemas.example/a", "schema", "application/json", {"application/json"}, 100
@@ -721,9 +732,8 @@ async def test_agent_remaining_traversal_and_supporting_document_edges() -> None
 async def test_action_resolution_boundaries(monkeypatch: pytest.MonkeyPatch) -> None:
     details = await ServiceClient(
         "https://demo.inflowpay.ai",
-        transport=QueueTransport(
-            response(SERVICE_DOCUMENT), response(ACTION_OFFERING), response(b"", status=500)
-        ),
+        transport=QueueTransport(response(SERVICE_DOCUMENT), response(ACTION_OFFERING)),
+        supporting_transport=QueueTransport(response(b"", status=500)),
     ).get_offering_details("rubber-plant")
     assert details.attribute_schema is None
     assert details.issues[0].scope.value == "attribute_schema"
@@ -798,6 +808,8 @@ async def fake_details(client: ServiceClient, identifier: str) -> OfferingDetail
                 )
             ),
             response(offering),
+        ),
+        supporting_transport=QueueTransport(
             response(duplicate, content_type="application/json"),
         ),
     )