Coverage for src / lexigram / ai / relay / gateway / passthrough.py: 97%
220 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""Passthrough relay lifecycle for non-chat endpoint kinds.
3``PassthroughService`` relays endpoint kinds that do not fit the
4chat-focused conversion engine (starting with embeddings) through the
5generalized parts of the chat pipeline — channel selection by endpoint
6kind, authorization, billing admission and settlement, and the upstream
7HTTP adapter — while skipping ``RelayConverterProtocol`` entirely: the
8wire format in is the wire format out, with the model alias
9substituted only when the channel config declares a suffix. Bodies are
10carried by :class:`RelayPassthroughBody` (decoded JSON or raw bytes
11with a content type) and upstream responses by
12:class:`RelayPassthroughResult` (verbatim bytes plus the upstream
13content type) — JSON responses are still returned decoded through the
14``payload`` accessor so the embeddings wire path is byte-for-byte
15unchanged, while non-JSON responses ride in ``body`` uninterpreted.
16The upstream response is returned verbatim, unvalidated beyond the
17adapter's existing malformed-body handling.
18"""
20from __future__ import annotations
22from collections.abc import AsyncIterator, Iterator, Mapping
23from dataclasses import dataclass
24import time
25from typing import Any, Literal, cast
27from lexigram.ai.relay.gateway.channels import RelayChannelRegistry
28from lexigram.ai.relay.gateway.config import RelayGatewayConfig
29from lexigram.ai.relay.gateway.errors import (
30 auth_denied,
31 billing_error_to_gateway,
32 with_request_id,
33)
34from lexigram.ai.relay.gateway.upstream import HTTPUpstreamAdapter
35from lexigram.contracts.ai.governance import (
36 RelayBillingProtocol,
37 RelayUsageReservation,
38 RelayUsageScope,
39)
40from lexigram.contracts.ai.relay import (
41 ConversionQuality,
42 JsonValue,
43 RelayChannel,
44 RelayConvertResult,
45 RelayFormat,
46 RelayGatewayError,
47 RelayGatewayMetadata,
48 RelayGatewayRequest,
49 RelayGatewayResult,
50 RelayRequestPayload,
51 RelayUsage,
52 RelayWireEvent,
53 UpstreamRequest,
54 UpstreamResponse,
55)
56from lexigram.contracts.ai.relay.gateway import RelayGatewayErrorCode
57from lexigram.contracts.auth.guard import AuthorizerProtocol
58from lexigram.contracts.core.result import Err, Ok, Result
59from lexigram.logging import get_logger
60from lexigram.serialization import dumps
62__all__ = [
63 "PassthroughService",
64 "RelayPassthroughBody",
65 "RelayPassthroughResult",
66 "rewrite_multipart_form_field",
67]
69logger = get_logger(__name__)
71_JSON_CONTENT_TYPE = "application/json"
72_FORM_FIELD_HEADER_MARKER = b'name="'
73_FORM_FIELD_HEADER_SUFFIX = b'"'
74"""Multipart ``Content-Disposition`` attribute delimiters used by the field rewrite."""
76_ENDPOINT_PATHS: dict[str, str] = {
77 "embeddings": "/v1/embeddings",
78}
79"""Endpoint kinds to upstream path segments served by this relay.
81Every registered kind uses the OpenAI-shaped ``/v1/<kind>`` path; future
82kinds with provider-specific shapes (multipart audio, binary images)
83extend this table in their own plans.
84"""
87@dataclass(frozen=True, slots=True)
88class RelayPassthroughBody(Mapping[str, JsonValue]):
89 """One forwarded gateway request body: decoded JSON or raw bytes.
91 The two constructors are the entire surface: :meth:`json` wraps a
92 decoded JSON object (content type ``application/json``) and
93 :meth:`raw` wraps arbitrary bytes with their content type, so
94 ``multipart/form-data`` requests travel through the same
95 ``RelayGatewayRequest.payload`` field as JSON bodies. The mapping
96 facade (``__getitem__``/``__iter__``/``__len__``) delegates to the
97 JSON dict for ``json`` bodies and raises ``TypeError`` for raw
98 bodies — the passthrough pipeline branches on whether ``data`` is a
99 mapping and never treats raw content as JSON.
101 Attributes:
102 data: The decoded JSON object for ``json`` bodies, or the raw
103 body bytes for ``raw`` bodies.
104 content_type: Outbound content type header value; ``json``
105 bodies always carry ``application/json``.
106 """
108 data: Mapping[str, JsonValue] | bytes
109 content_type: str
111 @classmethod
112 def json(cls, payload: Mapping[str, JsonValue]) -> RelayPassthroughBody:
113 """Wrap a decoded JSON object request body.
115 Args:
116 payload: The decoded JSON object to forward.
118 Returns:
119 A JSON body carrying ``application/json`` as its content
120 type; the object is shallow-copied so later mutation of the
121 source never leaks into the frozen body.
122 """
123 return cls(dict(payload), _JSON_CONTENT_TYPE)
125 @classmethod
126 def raw(cls, data: bytes, content_type: str) -> RelayPassthroughBody:
127 """Wrap a raw (e.g. ``multipart/form-data``) request body.
129 Args:
130 data: The raw body bytes to forward verbatim.
131 content_type: The body's content type header (boundary
132 parameter included for multipart bodies).
134 Returns:
135 A raw body carrying *content_type* unchanged.
136 """
137 return cls(data, content_type)
139 def __getitem__(self, key: str) -> JsonValue:
140 """Return one JSON field for ``json`` bodies.
142 Raises:
143 TypeError: If the body is raw bytes, which are not JSON.
144 """
145 data = self.data
146 if not isinstance(data, Mapping):
147 raise TypeError("raw passthrough bodies are not JSON mappings")
148 return data[key]
150 def __iter__(self) -> Iterator[str]:
151 """Iterate the JSON field names for ``json`` bodies.
153 Raises:
154 TypeError: If the body is raw bytes, which are not JSON.
155 """
156 data = self.data
157 if not isinstance(data, Mapping):
158 raise TypeError("raw passthrough bodies are not JSON mappings")
159 return iter(data)
161 def __len__(self) -> int:
162 """Return the JSON field count for ``json`` bodies.
164 Raises:
165 TypeError: If the body is raw bytes, which are not JSON.
166 """
167 data = self.data
168 if not isinstance(data, Mapping):
169 raise TypeError("raw passthrough bodies are not JSON mappings")
170 return len(data)
173@dataclass(frozen=True, slots=True, init=False)
174class RelayPassthroughResult(RelayGatewayResult):
175 """One passthrough upstream response, decoded when JSON, verbatim otherwise.
177 Extends the gateway result carrier with the two fields the passthrough
178 wire paths need: the upstream body and its content type. JSON
179 responses keep their decoded object on ``payload`` (byte-for-byte the
180 Plan J shape) and additionally populate ``body`` with the serialized
181 bytes; non-JSON responses carry the raw bytes in ``body`` with
182 ``payload`` left ``None``. Callers read ``body`` regardless of the
183 response shape; the constructor is ``(body, content_type,
184 status_code)`` with the inherited gateway fields (headers, payload)
185 as optional keywords so existing relay route code keeps compiling.
187 Attributes:
188 body: The response body bytes (serialized JSON for JSON
189 responses, the upstream bytes verbatim otherwise).
190 content_type: The upstream ``content-type`` header value.
191 """
193 body: bytes = b""
194 content_type: str = ""
196 def __init__(
197 self,
198 body: bytes = b"",
199 content_type: str = "",
200 status_code: int = 200,
201 *,
202 headers: Mapping[str, str] | None = None,
203 payload: Mapping[str, JsonValue] | None = None,
204 stream: AsyncIterator[RelayWireEvent] | None = None,
205 metadata: RelayGatewayMetadata | None = None,
206 ) -> None:
207 """Bind the passthrough result fields.
209 Args:
210 body: The response body bytes.
211 content_type: The upstream content-type header.
212 status_code: The upstream HTTP status code.
213 headers: Response headers to relay; defaults to empty.
214 payload: Decoded JSON object for JSON responses; ``None``
215 for raw bodies.
216 stream: Never used by passthrough; always ``None``.
217 metadata: Never used by passthrough; always ``None``.
218 """
219 object.__setattr__(self, "body", body)
220 object.__setattr__(self, "content_type", content_type)
221 object.__setattr__(self, "status_code", status_code)
222 object.__setattr__(self, "headers", headers if headers is not None else {})
223 object.__setattr__(self, "payload", payload)
224 object.__setattr__(self, "stream", stream)
225 object.__setattr__(self, "metadata", metadata)
228def rewrite_multipart_form_field(
229 body: bytes,
230 boundary: str,
231 field: str,
232 value: str,
233) -> bytes:
234 """Rewrite one named form field's value in a multipart body.
236 Narrow boundary-aware rewrite (not a general multipart parser): the
237 body is split on the ``--<boundary>`` framing marker and the first
238 part whose ``Content-Disposition`` header carries
239 ``name="<field>"`` has its value content swapped in place; every
240 other byte — headers, other parts, the closing marker — is left
241 untouched. A body without the field (or without the boundary
242 marker) is returned unchanged; that is not an error, some
243 passthrough endpoints resolve the model from the URL path or a
244 channel default instead of a body field.
246 Args:
247 body: The raw ``multipart/form-data`` body bytes.
248 boundary: The boundary token from the content-type header.
249 field: The form field name to rewrite (e.g. ``"model"``).
250 value: The replacement field value.
252 Returns:
253 The body with the named field's value replaced, or the body
254 unchanged when the field is absent.
255 """
256 marker = b"--" + boundary.encode("utf-8")
257 target = (
258 _FORM_FIELD_HEADER_MARKER + field.encode("utf-8") + _FORM_FIELD_HEADER_SUFFIX
259 )
260 replacement = value.encode("utf-8")
261 segments = body.split(marker)
262 if len(segments) < 2:
263 return body
264 for index in range(1, len(segments) - 1):
265 part = segments[index]
266 separator = part.find(b"\r\n\r\n")
267 if separator < 0:
268 continue
269 headers = part[2:separator].lower()
270 if target not in headers:
271 continue
272 value_end = len(part) - 2 if part.endswith(b"\r\n") else len(part)
273 segments[index] = part[: separator + 4] + replacement + part[value_end:]
274 return marker.join(segments)
275 return body
278def _as_relay_body(payload: Mapping[str, JsonValue]) -> RelayPassthroughBody:
279 """Normalize a gateway request payload into a relay passthrough body.
281 Bodies already carrying the relay carrier pass through unchanged;
282 plain JSON mappings (legacy callers) are wrapped as JSON bodies.
284 Args:
285 payload: The ``RelayGatewayRequest.payload`` value.
287 Returns:
288 The payload as a :class:`RelayPassthroughBody`.
289 """
290 if isinstance(payload, RelayPassthroughBody):
291 return payload
292 return RelayPassthroughBody.json(dict(payload))
295def _multipart_boundary(content_type: str) -> str | None:
296 """Extract the ``boundary`` parameter from a content-type header.
298 Args:
299 content_type: The raw content-type header value.
301 Returns:
302 The boundary token without surrounding quotes, or ``None`` when
303 the header carries no boundary parameter.
304 """
305 for parameter in content_type.split(";"):
306 key, separator, raw_value = parameter.strip().partition("=")
307 if separator and key.lower() == "boundary":
308 return raw_value.strip().strip('"')
309 return None
312def _is_json_content_type(content_type: str) -> bool:
313 """Tell whether a content-type value denotes JSON.
315 Args:
316 content_type: A content-type header value.
318 Returns:
319 ``True`` for the exact ``application/json`` media type and for
320 any ``*+json`` suffix variant; ``False`` otherwise.
321 """
322 media_type = content_type.partition(";")[0].strip().lower()
323 return media_type == _JSON_CONTENT_TYPE or media_type.endswith("+json")
326@dataclass(frozen=True, slots=True)
327class _PassthroughPayloadCarrier:
328 """Billing-admission carrier for a passthrough request body.
330 Passthrough bodies do not belong to any chat wire DTO, so the
331 shared billing pipeline receives a transparent carrier: prompt
332 estimation counts the serialized body (same estimate function as the
333 chat path) and the requested output budget is unknown, so zero is
334 reserved. The carrier quacks like a ``RelayRequestPayload`` at the
335 only call site the billing pipeline uses (``to_dict``).
336 """
338 body: dict[str, Any]
340 def to_dict(self) -> dict[str, Any]:
341 """Return the passthrough body.
343 Returns:
344 A shallow copy of the forwarded request body.
345 """
346 return dict(self.body)
349class PassthroughService:
350 """Endpoint-kind mapping: call one method, no conversion.
352 The service is stateless between requests and never includes
353 payloads or upstream details in error messages; errors are always
354 safe ``RelayGatewayError`` values. Authorization, billing, channel
355 selection, and upstream transport reuse the chat pipeline's
356 dependencies unchanged.
358 Attributes:
359 _registry: Deterministic channel selector.
360 _upstream: HTTP transport adapter.
361 _config: Gateway configuration (channel table and model suffixes).
362 _authorizer: Optional authorization check before dispatch.
363 _billing: Optional billing lifecycle; when ``None`` admission and
364 settlement are skipped.
365 """
367 def __init__(
368 self,
369 registry: RelayChannelRegistry,
370 upstream: HTTPUpstreamAdapter,
371 config: RelayGatewayConfig,
372 *,
373 authorizer: AuthorizerProtocol | None = None,
374 billing: RelayBillingProtocol | None = None,
375 ) -> None:
376 """Bind the service to its dependencies.
378 Args:
379 registry: Channel selection registry.
380 upstream: Upstream transport adapter; handles credential
381 injection per channel through its configured provider.
382 config: Static gateway configuration.
383 authorizer: Optional authorizer; when ``None`` authorization
384 is skipped.
385 billing: Optional billing lifecycle; when ``None`` the
386 passthrough runs without admission control or settlement.
387 """
388 self._registry = registry
389 self._upstream = upstream
390 self._config = config
391 self._authorizer = authorizer
392 self._billing = billing
394 async def handle(
395 self, kind: str, request: RelayGatewayRequest
396 ) -> Result[RelayPassthroughResult, RelayGatewayError]:
397 """Run the passthrough lifecycle for one request.
399 Dependencies run in fixed order: authorize, select channel by
400 endpoint kind, reserve billing capacity, call upstream with the
401 caller's body verbatim, settle billing, assemble result. Any
402 failure short-circuits the pipeline.
404 Args:
405 kind: The endpoint kind being served (e.g. ``"embeddings"``).
406 request: The passthrough gateway request; ``payload`` is
407 either a ``RelayPassthroughBody`` (JSON or raw multipart)
408 or a plain JSON mapping forwarded by legacy callers, and
409 ``source`` is a conventional marker (``OPENAI_CHAT``)
410 never used for conversion.
412 Returns:
413 ``Ok(RelayPassthroughResult)`` on success, or
414 ``Err(RelayGatewayError)`` on the first failure. Unexpected
415 exceptions from dependencies never escape: they are logged
416 and mapped to a generic ``CONVERSION_FAILED`` error.
417 """
418 started = time.monotonic()
419 logger.info(
420 "relay_passthrough_request_accepted",
421 request_id=request.request_id,
422 tenant_id=request.tenant_id,
423 endpoint=kind,
424 model=request.model,
425 )
426 try:
427 result, channel_name = await self._dispatch(kind, request)
428 except Exception as exc:
429 logger.warning(
430 "relay_passthrough_unexpected_error",
431 request_id=request.request_id,
432 endpoint=kind,
433 error=str(exc),
434 )
435 error = self._unexpected_error(request.request_id)
436 self._log_request_completed(request, kind, "", error, started)
437 return Err(error)
438 if result.is_err():
439 self._log_request_completed(
440 request, kind, channel_name, result.unwrap_err(), started
441 )
442 return result
443 outcome = result.unwrap()
444 self._log_request_completed(request, kind, channel_name, outcome, started)
445 return result
447 async def _dispatch(
448 self,
449 kind: str,
450 request: RelayGatewayRequest,
451 ) -> tuple[Result[RelayPassthroughResult, RelayGatewayError], str]:
452 """Run the ordered dependency pipeline for one request.
454 Returns:
455 ``tuple`` of the pipeline result and the selected channel
456 name. The channel name is ``""`` when selection failed
457 before a channel was chosen.
458 """
459 if kind not in _ENDPOINT_PATHS:
460 return (
461 Err(
462 RelayGatewayError(
463 code=RelayGatewayErrorCode.INVALID_REQUEST,
464 message="unsupported endpoint kind",
465 status_code=400,
466 request_id=request.request_id,
467 retryable=False,
468 )
469 ),
470 "",
471 )
472 if self._authorizer is not None:
473 allowed = await self._authorizer.authorize(
474 user=request.tenant_id,
475 action="relay.invoke",
476 resource=request.model,
477 )
478 if not allowed:
479 return Err(auth_denied(request.request_id)), ""
480 selected = self._registry.select_for_endpoint(
481 kind=kind,
482 model=request.model,
483 )
484 if selected.is_err():
485 return (
486 Err(with_request_id(selected.unwrap_err(), request.request_id)),
487 "",
488 )
489 channel = selected.unwrap()
490 logger.info(
491 "relay_passthrough_channel_selected",
492 request_id=request.request_id,
493 endpoint=kind,
494 channel=channel.name,
495 model=request.model,
496 )
497 billing = self._billing
498 reservation: RelayUsageReservation | None = None
499 if billing is not None:
500 admitted = await self._reserve(request, billing, channel)
501 if admitted.is_err():
502 return Err(admitted.unwrap_err()), channel.name
503 reservation = admitted.unwrap()
504 body = _as_relay_body(request.payload)
505 outbound_model = request.model + self._config.model_suffix.get(channel.name, "")
506 body_data = body.data
507 if isinstance(body_data, Mapping):
508 outbound = dict(body_data)
509 outbound["model"] = outbound_model
510 upstream_response = await self._call_upstream(
511 kind, channel, outbound, body.content_type, request
512 )
513 else:
514 content_type = body.content_type
515 boundary = _multipart_boundary(content_type)
516 outbound_raw = body_data
517 if boundary is not None:
518 outbound_raw = rewrite_multipart_form_field(
519 outbound_raw, boundary, "model", outbound_model
520 )
521 upstream_response = await self._call_upstream(
522 kind, channel, outbound_raw, content_type, request
523 )
524 if upstream_response.is_err():
525 if billing is not None and reservation is not None:
526 await self._settle_failed(billing, reservation)
527 return (
528 Err(
529 with_request_id(upstream_response.unwrap_err(), request.request_id)
530 ),
531 channel.name,
532 )
533 resp = upstream_response.unwrap()
534 payload = resp.payload
535 if isinstance(payload, Mapping) and _is_json_content_type(
536 resp.headers.get("content-type", _JSON_CONTENT_TYPE)
537 ):
538 if billing is not None and reservation is not None:
539 await self._settle(
540 billing,
541 reservation,
542 self._usage_from_response(payload),
543 status="completed",
544 )
545 return (
546 Ok(
547 RelayPassthroughResult(
548 status_code=resp.status_code,
549 headers={**resp.headers, "x-request-id": request.request_id},
550 payload=payload,
551 stream=None,
552 metadata=None,
553 body=dumps(payload),
554 content_type=resp.headers.get(
555 "content-type", _JSON_CONTENT_TYPE
556 ),
557 )
558 ),
559 channel.name,
560 )
561 if isinstance(payload, bytes):
562 if billing is not None and reservation is not None:
563 await self._settle(billing, reservation, None, status="completed")
564 return (
565 Ok(
566 RelayPassthroughResult(
567 status_code=resp.status_code,
568 headers={**resp.headers, "x-request-id": request.request_id},
569 payload=None,
570 stream=None,
571 metadata=None,
572 body=payload,
573 content_type=resp.headers.get("content-type", ""),
574 )
575 ),
576 channel.name,
577 )
578 if billing is not None and reservation is not None:
579 await self._settle_failed(billing, reservation)
580 return (
581 Err(
582 RelayGatewayError(
583 code=RelayGatewayErrorCode.UPSTREAM_MALFORMED,
584 message="malformed upstream response",
585 status_code=502,
586 request_id=request.request_id,
587 retryable=False,
588 )
589 ),
590 channel.name,
591 )
593 async def _reserve(
594 self,
595 request: RelayGatewayRequest,
596 billing: RelayBillingProtocol,
597 channel: RelayChannel,
598 ) -> Result[RelayUsageReservation, RelayGatewayError]:
599 """Reserve billing capacity before the upstream call.
601 The passthrough body is wrapped in a transparent carrier so the
602 shared billing pipeline can estimate prompt tokens from the
603 serialized body; the output budget is unknown and reserved as
604 zero. Billing denials short-circuit the pipeline and are
605 classified through :func:`billing_error_to_gateway`.
607 Args:
608 request: The passthrough request being dispatched.
609 billing: The billing lifecycle to reserve through.
610 channel: The selected channel.
612 Returns:
613 ``Ok(reservation)`` when admission is proven, or
614 ``Err(RelayGatewayError)`` carrying the classified failure.
615 """
616 scope = RelayUsageScope(
617 tenant_id=request.tenant_id,
618 model=request.model,
619 channel=channel.name,
620 )
621 body = _as_relay_body(request.payload)
622 body_data = body.data
623 if isinstance(body_data, Mapping):
624 carrier_body: dict[str, JsonValue] = dict(body_data)
625 else:
626 carrier_body = {}
627 carrier = _PassthroughPayloadCarrier(carrier_body)
628 admitted = await billing.pre_consume(
629 request.request_id,
630 scope,
631 cast("RelayRequestPayload", carrier),
632 )
633 if admitted.is_err():
634 error = admitted.unwrap_err()
635 logger.warning(
636 "relay_passthrough_billing_denied",
637 request_id=request.request_id,
638 channel=channel.name,
639 code=error.code,
640 error=error.message,
641 )
642 return Err(billing_error_to_gateway(error, request.request_id))
643 return Ok(admitted.unwrap())
645 async def _settle(
646 self,
647 billing: RelayBillingProtocol,
648 reservation: RelayUsageReservation,
649 usage: RelayUsage | None,
650 *,
651 status: Literal["completed", "failed", "cancelled", "truncated"],
652 ) -> None:
653 """Settle the reservation exactly once without failing the response.
655 Settlement failures are logged and never propagate: the response
656 path has already completed by the time accounting runs.
658 Args:
659 billing: The billing lifecycle to settle through.
660 reservation: The reservation granted by ``pre_consume``.
661 usage: The usage extracted from the upstream response, or
662 ``None`` when the response omits it.
663 status: Terminal lifecycle status of the attempt.
664 """
665 result = RelayConvertResult[Any](
666 value=None,
667 source=RelayFormat.OPENAI_CHAT,
668 target=RelayFormat.OPENAI_CHAT,
669 converter_id="passthrough",
670 quality=ConversionQuality.GOOD,
671 usage=usage,
672 )
673 settled = await billing.settle(reservation, result, status=status)
674 if settled.is_err():
675 error = settled.unwrap_err()
676 logger.warning(
677 "relay_passthrough_settle_failed",
678 request_id=reservation.request_id,
679 status=status,
680 code=error.code,
681 error=error.message,
682 )
684 async def _settle_failed(
685 self,
686 billing: RelayBillingProtocol,
687 reservation: RelayUsageReservation,
688 ) -> None:
689 """Settle a failed attempt without usage through the billing pipeline.
691 Args:
692 billing: The billing lifecycle to settle through.
693 reservation: The reservation granted by ``pre_consume``.
694 """
695 await self._settle(billing, reservation, None, status="failed")
697 async def _call_upstream(
698 self,
699 kind: str,
700 channel: RelayChannel,
701 payload: Mapping[str, JsonValue] | bytes,
702 content_type: str,
703 request: RelayGatewayRequest,
704 ) -> Result[UpstreamResponse, RelayGatewayError]:
705 """Send the passthrough body to the selected channel's endpoint.
707 Uses the same ``HTTPUpstreamAdapter`` as the chat path, so
708 channel-credential injection applies unchanged. JSON bodies go
709 out as their decoded dict; raw bodies (multipart) travel through
710 the adapter's payload slot as opaque bytes with their content
711 type header intact, so the binary parts reach the provider
712 untouched.
714 Args:
715 kind: The endpoint kind being served, selecting the wire path.
716 channel: The selected channel.
717 payload: The caller's body with the model substituted; a
718 decoded JSON object or raw body bytes.
719 content_type: The outbound content type header value.
720 request: The original gateway request.
722 Returns:
723 ``Ok(UpstreamResponse)`` or ``Err`` as returned by the
724 adapter; the adapter already normalizes transport failures.
725 """
726 url = self._endpoint_url(kind, channel)
727 logger.info(
728 "relay_passthrough_upstream_started",
729 request_id=request.request_id,
730 channel=channel.name,
731 method="POST",
732 url=url,
733 )
734 upstream = await self._upstream.request(
735 UpstreamRequest(
736 request_id=request.request_id,
737 method="POST",
738 url=url,
739 headers={"content-type": content_type},
740 payload=cast("Mapping[str, JsonValue]", payload),
741 timeout_seconds=channel.timeout_seconds,
742 channel_name=channel.name,
743 )
744 )
745 if upstream.is_err():
746 err = upstream.unwrap_err()
747 logger.warning(
748 "relay_passthrough_upstream_failed",
749 request_id=request.request_id,
750 channel=channel.name,
751 code=err.code,
752 status_code=err.status_code,
753 error=str(err),
754 )
755 return upstream
757 def _endpoint_url(self, kind: str, channel: RelayChannel) -> str:
758 """Build the endpoint URL for *kind* on *channel*.
760 Args:
761 kind: The endpoint kind being served.
762 channel: The selected channel.
764 Returns:
765 ``<channel base>/v1/<kind>`` for a registered kind; the
766 kind was validated against ``_ENDPOINT_PATHS`` before the
767 channel call, so this never misses.
768 """
769 base = channel.upstream_base_url.rstrip("/")
770 return f"{base}{_ENDPOINT_PATHS[kind]}"
772 @staticmethod
773 def _usage_from_response(payload: Mapping[str, Any]) -> RelayUsage | None:
774 """Extract normalized usage from an OpenAI-shaped response body.
776 Args:
777 payload: The upstream response body.
779 Returns:
780 ``RelayUsage`` when the body carries an integer
781 ``prompt_tokens`` count, otherwise ``None`` (the billing
782 pipeline records usage as missing).
783 """
784 usage = payload.get("usage")
785 if not isinstance(usage, dict):
786 return None
787 prompt = usage.get("prompt_tokens")
788 if not isinstance(prompt, int):
789 return None
790 completion = usage.get("completion_tokens", 0)
791 if not isinstance(completion, int):
792 completion = 0
793 return RelayUsage(prompt_tokens=prompt, completion_tokens=completion)
795 @staticmethod
796 def _unexpected_error(request_id: str) -> RelayGatewayError:
797 """Build the generic error for unexpected dependency failures."""
798 return RelayGatewayError(
799 code=RelayGatewayErrorCode.CONVERSION_FAILED,
800 message="Unexpected relay gateway failure",
801 status_code=500,
802 request_id=request_id,
803 retryable=False,
804 )
806 def _log_request_completed(
807 self,
808 request: RelayGatewayRequest,
809 kind: str,
810 channel_name: str,
811 outcome: RelayGatewayResult | RelayGatewayError,
812 started: float,
813 ) -> None:
814 """Emit the terminal request-completed event for any outcome.
816 Args:
817 request: The original gateway request.
818 kind: The endpoint kind that was served.
819 channel_name: Selected channel name (or ``""`` when unknown).
820 outcome: The success result or the error that ended the flow.
821 started: Monotonic start time used to compute the duration.
822 """
823 logger.info(
824 "relay_passthrough_request_completed",
825 request_id=request.request_id,
826 tenant_id=request.tenant_id,
827 endpoint=kind,
828 channel=channel_name,
829 status_code=outcome.status_code,
830 code=outcome.code if isinstance(outcome, RelayGatewayError) else "OK",
831 duration_ms=round((time.monotonic() - started) * 1000, 2),
832 )