Module adcp.reporting.receipts.transport

Request-local lossless receipt JSON, before transport/model normalization.

Functions

def a2a_receipt_has_invalid_unicode(body: bytes | None) ‑> bool

Expand source code

def a2a_receipt_has_invalid_unicode(body: bytes | None) -> bool:
    """Scope the upstream protobuf codec error boundary to this exact raw route."""
    if body is None or _a2a_receipt_invocation(body) is None:
        return False
    try:
        pending = [_raw_json(body)]
        while pending:
            value = pending.pop()
            if isinstance(value, str) and any(0xD800 <= ord(c) <= 0xDFFF for c in value):
                return True
            if type(value) is dict:
                pending.extend(value)
                pending.extend(value.values())
            elif type(value) is list:
                pending.extend(value)
    except (ValueError, TypeError, RecursionError):
        # Failed parsing establishes no Unicode diagnosis; ordinary validation still applies.
        pass
    return False

Scope the upstream protobuf codec error boundary to this exact raw route.

def a2a_receipt_parameters(body: bytes | None, *, task: str = 'sync_reporting_receipts') ‑> dict[str, typing.Any] | None

Expand source code

def a2a_receipt_parameters(body: bytes | None, *, task: str = TASK) -> dict[str, Any] | None:
    """Distinguish an exact receipt route from invalid receipt parameters.

    None means no uniquely identified standard invocation. An empty dictionary
    means that invocation's parameters are invalid and must reach the ordinary
    whole-shape rejection. Neither case can authorize a domain write.
    """
    invocation = _a2a_receipt_invocation(body, task=task)
    if invocation is None:
        return None
    try:
        params = invocation.get("parameters")
        if type(params) is dict:
            return _parameters(params, task)
    except (ValueError, TypeError, RecursionError):
        # The receipt route is known; return invalid parameters for ordinary rejection.
        pass
    return {}

Distinguish an exact receipt route from invalid receipt parameters.

None means no uniquely identified standard invocation. An empty dictionary means that invocation's parameters are invalid and must reach the ordinary whole-shape rejection. Neither case can authorize a domain write.

def mcp_receipt_parameters(body: bytes | None, *, task: str = 'sync_reporting_receipts') ‑> dict[str, typing.Any]

Expand source code

def mcp_receipt_parameters(body: bytes | None, *, task: str = TASK) -> dict[str, Any]:
    """The already selected tools/call must match this exact receipt invocation."""
    try:
        if body is None:
            return {}
        envelope = _raw_json(body)
        if (
            type(envelope) is dict
            and envelope.get("jsonrpc") == "2.0"
            and envelope.get("method") == "tools/call"
            and envelope["params"]["name"] == task
            and type(envelope["params"].get("arguments")) is dict
        ):
            return _parameters(envelope["params"]["arguments"], task)
    except (ValueError, TypeError, KeyError, RecursionError):
        # The selected MCP route must receive invalid parameters, never a partial body.
        pass
    return {}

The already selected tools/call must match this exact receipt invocation.

def receipt_body_receive(scope: dict[str, Any],
receive: Callable[[], Awaitable[Any]],
*,
limit: int = 10485760) ‑> Callable[[], Awaitable[typing.Any]]

Expand source code

def receipt_body_receive(
    scope: dict[str, Any],
    receive: Callable[[], Awaitable[Any]],
    *,
    limit: int = MAX_RECEIPT_BODY_BYTES,
) -> Callable[[], Awaitable[Any]]:
    """Tee only the bytes delivered to this mounted HTTP request's decoder.

    No caller metadata, global cache, request ID lookup or auth shortcut is
    involved. The existing upstream body cap still applies first.
    """
    captured: bytearray | None = bytearray()

    async def capture() -> Any:
        nonlocal captured
        message = await receive()
        if message.get("type") == "http.request":
            chunk = message.get("body", b"")
            if captured is not None:
                if len(captured) + len(chunk) > limit:
                    captured = None
                else:
                    captured.extend(chunk)
            if not message.get("more_body", False):
                scope[RAW_RECEIPT_BODY_SCOPE_KEY] = (
                    bytes(captured) if captured is not None else None
                )
        return message

    return capture

Tee only the bytes delivered to this mounted HTTP request's decoder.

No caller metadata, global cache, request ID lookup or auth shortcut is involved. The existing upstream body cap still applies first.