diff --git a/services/product-composition-api/src/orgmetra_product_composition/asgi_request_body.py b/services/product-composition-api/src/orgmetra_product_composition/asgi_request_body.py
new file mode 100644
index 000000000..c2f7412ea
--- /dev/null
+++ b/services/product-composition-api/src/orgmetra_product_composition/asgi_request_body.py
@@ -0,0 +1,111 @@
+"""Consume ASGI HTTP request bodies with explicit bounds and lifecycle failure semantics."""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+import inspect
+
+from .asgi_transport import CompositionTransportError
+
+_MAX_REQUEST_BODY_BYTES = 16 * 1024 * 1024
+_MAX_REQUEST_BODY_RECEIVE_EVENTS = 4096
+
+
+class CompositionRequestBodyError(CompositionTransportError):
+    """Raised when ASGI request-body evidence is malformed or its local contract is invalid."""
+
+
+class CompositionRequestBodyTooLargeError(CompositionRequestBodyError):
+    """Raised when accumulated request content exceeds the configured hard byte limit."""
+
+
+class CompositionRequestBodyTooManyEventsError(CompositionRequestBodyError):
+    """Raised when a request body would require more receive events than the configured budget."""
+
+
+class CompositionClientDisconnectedError(CompositionTransportError):
+    """Raised when the ASGI server reports that the client disconnected before body completion."""
+
+
+async def read_bounded_http_request_body(
+    receive: Callable[[], Awaitable[object]],
+    *,
+    max_body_bytes: int,
+    max_receive_events: int,
+) -> bytes:
+    """Return one detached HTTP request body without inventing timeout or failure ownership.
+
+    ASGI defines omitted ``body`` as ``b""`` and omitted ``more_body`` as ``False``. Those defaults
+    are accepted, while supplied values must use exact built-in ``bytes`` and ``bool`` values.
+    Both accumulated bytes and receive-event count are bounded: the event budget prevents an
+    otherwise byte-bounded request from consuming unbounded CPU/list overhead through empty or
+    tiny ``more_body=True`` chunks. Once the budget is spent, no additional ``receive()`` call is
+    made. The injected receive capability must be callable and a normal invocation result must be
+    awaitable as ASGI requires. A non-callable capability or normal non-awaitable return is
+    demonstrably malformed and fails closed. Exceptions raised while invoking a callable, and
+    exceptions or cancellation raised while awaiting a valid result, remain caller/server lifecycle
+    authority and propagate unchanged. ``http.disconnect`` remains a distinct lifecycle signal so a
+    future host can stop work instead of trying to serialize an HTTP error to a peer that is already
+    gone.
+    """
+
+    if type(max_body_bytes) is not int or not 0 <= max_body_bytes <= _MAX_REQUEST_BODY_BYTES:
+        raise CompositionRequestBodyError(
+            "max_body_bytes must be an exact integer between 0 and 16777216"
+        )
+    if (
+        type(max_receive_events) is not int
+        or not 1 <= max_receive_events <= _MAX_REQUEST_BODY_RECEIVE_EVENTS
+    ):
+        raise CompositionRequestBodyError(
+            "max_receive_events must be an exact integer between 1 and 4096"
+        )
+    if not callable(receive):
+        raise CompositionRequestBodyError("ASGI receive must be callable")
+
+    chunks: list[bytes] = []
+    accumulated_bytes = 0
+    received_events = 0
+    while True:
+        if received_events >= max_receive_events:
+            raise CompositionRequestBodyTooManyEventsError(
+                "ASGI HTTP request body exceeds the configured receive event limit"
+            )
+        pending_event = receive()
+        if not inspect.isawaitable(pending_event):
+            raise CompositionRequestBodyError("ASGI receive result must be awaitable")
+        event = await pending_event
+        received_events += 1
+        if type(event) is not dict:
+            raise CompositionRequestBodyError("ASGI receive event must be an exact built-in dict")
+
+        event_type = event.get("type")
+        if type(event_type) is not str:
+            raise CompositionRequestBodyError("ASGI receive event type must be an exact built-in str")
+        if event_type == "http.disconnect":
+            raise CompositionClientDisconnectedError(
+                "client disconnected before the HTTP request body completed"
+            )
+        if event_type != "http.request":
+            raise CompositionRequestBodyError(
+                "ASGI receive event must be http.request or http.disconnect"
+            )
+
+        body = event.get("body", b"")
+        more_body = event.get("more_body", False)
+        if type(body) is not bytes:
+            raise CompositionRequestBodyError("ASGI http.request body must be exact built-in bytes")
+        if type(more_body) is not bool:
+            raise CompositionRequestBodyError(
+                "ASGI http.request more_body must be an exact built-in bool"
+            )
+
+        accumulated_bytes += len(body)
+        if accumulated_bytes > max_body_bytes:
+            raise CompositionRequestBodyTooLargeError(
+                "ASGI HTTP request body exceeds the configured byte limit"
+            )
+        if body:
+            chunks.append(body)
+        if not more_body:
+            return b"".join(chunks)
diff --git a/services/product-composition-api/tests/test_asgi_request_body_lifecycle.py b/services/product-composition-api/tests/test_asgi_request_body_lifecycle.py
new file mode 100644
index 000000000..c40962641
--- /dev/null
+++ b/services/product-composition-api/tests/test_asgi_request_body_lifecycle.py
@@ -0,0 +1,309 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Iterable
+
+import pytest
+
+from orgmetra_product_composition.asgi_request_body import (
+    CompositionClientDisconnectedError,
+    CompositionRequestBodyError,
+    CompositionRequestBodyTooLargeError,
+    read_bounded_http_request_body,
+)
+
+
+class _ReceiveSequence:
+    """Return exact ASGI receive events in order and expose how many were consumed."""
+
+    def __init__(self, events: Iterable[object]) -> None:
+        """Detach the supplied event sequence for deterministic one-shot consumption."""
+
+        self._events = iter(tuple(events))
+        self.calls = 0
+
+    async def __call__(self) -> object:
+        """Return the next queued event and count the receive call."""
+
+        self.calls += 1
+        return next(self._events)
+
+
+def test_empty_request_body_accepts_asgi_defaults() -> None:
+    """Accept an omitted body and omitted more_body as the ASGI empty-body defaults."""
+
+    receive = _ReceiveSequence(({"type": "http.request"},))
+
+    assert asyncio.run(
+        read_bounded_http_request_body(receive, max_body_bytes=0, max_receive_events=1)
+    ) == b""
+    assert receive.calls == 1
+
+
+def test_chunked_request_body_is_detached_and_stops_after_terminal_chunk() -> None:
+    """Join exact bytes and do not consume a later event after more_body becomes false."""
+
+    receive = _ReceiveSequence(
+        (
+            {"type": "http.request", "body": b"abc", "more_body": True},
+            {"type": "http.request", "body": b"def", "more_body": False},
+            {"type": "http.disconnect"},
+        )
+    )
+
+    assert asyncio.run(
+        read_bounded_http_request_body(receive, max_body_bytes=6, max_receive_events=2)
+    ) == b"abcdef"
+    assert receive.calls == 2
+
+
+def test_request_body_size_limit_accepts_exact_boundary() -> None:
+    """Permit a body whose accumulated byte length equals the configured hard limit."""
+
+    receive = _ReceiveSequence(
+        (
+            {"type": "http.request", "body": b"ab", "more_body": True},
+            {"type": "http.request", "body": b"cd"},
+        )
+    )
+
+    assert asyncio.run(
+        read_bounded_http_request_body(receive, max_body_bytes=4, max_receive_events=2)
+    ) == b"abcd"
+
+
+def test_request_body_size_limit_fails_before_waiting_for_more_chunks() -> None:
+    """Reject an oversized partial body immediately even when the sender advertises more data."""
+
+    receive = _ReceiveSequence(
+        (
+            {"type": "http.request", "body": b"abc", "more_body": True},
+            {"type": "http.request", "body": b"never-consumed"},
+        )
+    )
+
+    with pytest.raises(CompositionRequestBodyTooLargeError, match="exceeds"):
+        asyncio.run(
+            read_bounded_http_request_body(receive, max_body_bytes=2, max_receive_events=2)
+        )
+    assert receive.calls == 1
+
+
+def test_receive_event_limit_accepts_terminal_event_at_exact_boundary() -> None:
+    """Allow the terminal body event to consume the final configured receive-event slot."""
+
+    receive = _ReceiveSequence(
+        (
+            {"type": "http.request", "body": b"a", "more_body": True},
+            {"type": "http.request", "body": b"b", "more_body": False},
+        )
+    )
+
+    assert asyncio.run(
+        read_bounded_http_request_body(receive, max_body_bytes=2, max_receive_events=2)
+    ) == b"ab"
+    assert receive.calls == 2
+
+
+def test_receive_event_limit_fails_before_waiting_beyond_budget() -> None:
+    """Bound empty/tiny chunk CPU work and refuse another receive once the event budget is spent."""
+
+    receive = _ReceiveSequence(
+        (
+            {"type": "http.request", "body": b"", "more_body": True},
+            {"type": "http.request", "body": b"", "more_body": True},
+            {"type": "http.request", "body": b"never-consumed", "more_body": False},
+        )
+    )
+
+    with pytest.raises(CompositionRequestBodyError, match="event limit"):
+        asyncio.run(
+            read_bounded_http_request_body(receive, max_body_bytes=16, max_receive_events=2)
+        )
+    assert receive.calls == 2
+
+
+def test_client_disconnect_is_distinct_from_malformed_transport() -> None:
+    """Surface a peer disconnect as lifecycle state rather than an HTTP error candidate."""
+
+    receive = _ReceiveSequence(({"type": "http.disconnect"},))
+
+    with pytest.raises(CompositionClientDisconnectedError, match="disconnected"):
+        asyncio.run(
+            read_bounded_http_request_body(receive, max_body_bytes=16, max_receive_events=1)
+        )
+
+
+@pytest.mark.parametrize(
+    "event",
+    (
+        None,
+        [],
+        {"type": 1},
+        {"type": "websocket.receive"},
+        {"type": "http.request", "body": bytearray(b"abc")},
+        {"type": "http.request", "more_body": 1},
+    ),
+)
+def test_malformed_receive_events_fail_closed(event: object) -> None:
+    """Reject non-ASGI or non-exact built-in event fields before body bytes are trusted."""
+
+    receive = _ReceiveSequence((event,))
+
+    with pytest.raises(CompositionRequestBodyError):
+        asyncio.run(
+            read_bounded_http_request_body(receive, max_body_bytes=16, max_receive_events=1)
+        )
+
+
+@pytest.mark.parametrize("max_body_bytes", (True, -1, 16 * 1024 * 1024 + 1))
+def test_body_limit_configuration_is_bounded_and_exact(max_body_bytes: object) -> None:
+    """Reject ambiguous or operationally unbounded body-limit configuration before receive I/O."""
+
+    receive = _ReceiveSequence(({"type": "http.request"},))
+
+    with pytest.raises(CompositionRequestBodyError, match="max_body_bytes"):
+        asyncio.run(
+            read_bounded_http_request_body(
+                receive,
+                max_body_bytes=max_body_bytes,  # type: ignore[arg-type]
+                max_receive_events=1,
+            )
+        )
+    assert receive.calls == 0
+
+
+@pytest.mark.parametrize("max_receive_events", (True, 0, 4097))
+def test_receive_event_limit_configuration_is_bounded_and_exact(
+    max_receive_events: object,
+) -> None:
+    """Reject ambiguous or excessive receive-event budgets before request-body I/O."""
+
+    receive = _ReceiveSequence(({"type": "http.request"},))
+
+    with pytest.raises(CompositionRequestBodyError, match="max_receive_events"):
+        asyncio.run(
+            read_bounded_http_request_body(
+                receive,
+                max_body_bytes=16,
+                max_receive_events=max_receive_events,  # type: ignore[arg-type]
+            )
+        )
+    assert receive.calls == 0
+
+
+def test_receive_must_be_callable_before_any_io() -> None:
+    """Reject a non-callable receive boundary before attempting request-body lifecycle I/O."""
+
+    with pytest.raises(CompositionRequestBodyError, match="receive"):
+        asyncio.run(
+            read_bounded_http_request_body(
+                object(),  # type: ignore[arg-type]
+                max_body_bytes=16,
+                max_receive_events=1,
+            )
+        )
+
+
+def test_receive_result_must_be_awaitable_before_event_validation() -> None:
+    """Reject a callable that violates the ASGI awaitable receive contract without leaking TypeError."""
+
+    class _SynchronousReceive:
+        """Return an event synchronously to model a malformed injected ASGI receive capability."""
+
+        def __init__(self) -> None:
+            """Track how many times the malformed capability is invoked."""
+
+            self.calls = 0
+
+        def __call__(self) -> object:
+            """Return a non-awaitable event object instead of the required awaitable."""
+
+            self.calls += 1
+            return {"type": "http.request"}
+
+    receive = _SynchronousReceive()
+    with pytest.raises(CompositionRequestBodyError, match="awaitable"):
+        asyncio.run(
+            read_bounded_http_request_body(
+                receive,  # type: ignore[arg-type]
+                max_body_bytes=16,
+                max_receive_events=1,
+            )
+        )
+    assert receive.calls == 1
+
+
+def test_receive_preserves_synchronous_invocation_failure() -> None:
+    """Preserve a server failure raised before an awaitable exists instead of reclassifying it."""
+
+    failure = RuntimeError("synchronous receive server failure")
+
+    class _RaisingSynchronousReceive:
+        """Raise before returning an awaitable to model a synchronous server/runtime failure."""
+
+        def __init__(self) -> None:
+            """Track the single receive capability invocation."""
+
+            self.calls = 0
+
+        def __call__(self) -> object:
+            """Raise the server/runtime failure without manufacturing capability evidence."""
+
+            self.calls += 1
+            raise failure
+
+    receive = _RaisingSynchronousReceive()
+    with pytest.raises(RuntimeError) as caught:
+        asyncio.run(
+            read_bounded_http_request_body(
+                receive,  # type: ignore[arg-type]
+                max_body_bytes=16,
+                max_receive_events=1,
+            )
+        )
+
+    assert receive.calls == 1
+    assert caught.value is failure
+
+
+def test_receive_await_exception_remains_server_lifecycle_authority() -> None:
+    """Do not reclassify an exception raised after a valid receive awaitable has been returned."""
+
+    class _FailingAsyncReceive:
+        """Return a valid coroutine whose execution fails at the await boundary."""
+
+        async def __call__(self) -> object:
+            """Raise only while the valid receive awaitable is executing."""
+
+            raise RuntimeError("await-side server failure")
+
+    with pytest.raises(RuntimeError, match="await-side server failure"):
+        asyncio.run(
+            read_bounded_http_request_body(
+                _FailingAsyncReceive(),
+                max_body_bytes=16,
+                max_receive_events=1,
+            )
+        )
+
+
+def test_receive_cancellation_propagates_without_timeout_reclassification() -> None:
+    """Preserve task cancellation so callers can distinguish cancellation from protocol failure."""
+
+    class _CancelledReceive:
+        """Raise the task-cancellation signal when the ASGI server cancels receive."""
+
+        async def __call__(self) -> object:
+            """Raise CancelledError without wrapping it in a composition transport exception."""
+
+            raise asyncio.CancelledError
+
+    with pytest.raises(asyncio.CancelledError):
+        asyncio.run(
+            read_bounded_http_request_body(
+                _CancelledReceive(),
+                max_body_bytes=16,
+                max_receive_events=1,
+            )
+        )