« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Passthrough relay service: non-chat endpoint kinds, no conversion.
3:class:`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.
10"""
12from __future__ import annotations
14from collections.abc import Mapping
15from dataclasses import dataclass
16import time
17from typing import Any, Literal, cast
19from lexigram.ai.relay.gateway.channels import RelayChannelRegistry
20from lexigram.ai.relay.gateway.config import RelayGatewayConfig
21from lexigram.ai.relay.gateway.errors import (
22 auth_denied,
23 billing_error_to_gateway,
24 with_request_id,
25)
26from lexigram.ai.relay.gateway.passthrough_body import (
27 _JSON_CONTENT_TYPE,
28 _as_relay_body,
29 _is_json_content_type,
30 _multipart_boundary,
31 rewrite_multipart_form_field,
32)
33from lexigram.ai.relay.gateway.passthrough_result import RelayPassthroughResult
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 RelayGatewayRequest,
48 RelayGatewayResult,
49 RelayRequestPayload,
50 RelayUsage,
51 UpstreamRequest,
52 UpstreamResponse,
53)
54from lexigram.contracts.ai.relay.gateway import RelayGatewayErrorCode
55from lexigram.contracts.auth.guard import AuthorizerProtocol
56from lexigram.contracts.core.result import Err, Ok, Result
57from lexigram.logging import get_logger
58from lexigram.serialization import dumps
60__all__ = ["PassthroughService"]
62logger = get_logger(__name__)
64_ENDPOINT_PATHS: dict[str, str] = {
65 "embeddings": "/v1/embeddings",
66}
67"""Endpoint kinds to upstream path segments served by this relay.
69Every registered kind uses the OpenAI-shaped ``/v1/<kind>`` path; future
70kinds with provider-specific shapes (multipart audio, binary images)
71extend this table in their own plans.
72"""
75@dataclass(frozen=True, slots=True)
76class _PassthroughPayloadCarrier:
77 """Billing-admission carrier for a passthrough request body.
79 Passthrough bodies do not belong to any chat wire DTO, so the
80 shared billing pipeline receives a transparent carrier: prompt
81 estimation counts the serialized body (same estimate function as the
82 chat path) and the requested output budget is unknown, so zero is
83 reserved. The carrier quacks like a ``RelayRequestPayload`` at the
84 only call site the billing pipeline uses (``to_dict``).
85 """
87 body: dict[str, Any]
89 def to_dict(self) -> dict[str, Any]:
90 """Return the passthrough body.
92 Returns:
93 A shallow copy of the forwarded request body.
94 """
95 return dict(self.body)
98class PassthroughService:
99 """Endpoint-kind mapping: call one method, no conversion.
101 The service is stateless between requests and never includes
102 payloads or upstream details in error messages; errors are always
103 safe ``RelayGatewayError`` values. Authorization, billing, channel
104 selection, and upstream transport reuse the chat pipeline's
105 dependencies unchanged.
107 Attributes:
108 _registry: Deterministic channel selector.
109 _upstream: HTTP transport adapter.
110 _config: Gateway configuration (channel table and model suffixes).
111 _authorizer: Optional authorization check before dispatch.
112 _billing: Optional billing lifecycle; when ``None`` admission and
113 settlement are skipped.
114 """
116 def __init__(
117 self,
118 registry: RelayChannelRegistry,
119 upstream: HTTPUpstreamAdapter,
120 config: RelayGatewayConfig,
121 *,
122 authorizer: AuthorizerProtocol | None = None,
123 billing: RelayBillingProtocol | None = None,
124 ) -> None:
125 """Bind the service to its dependencies.
127 Args:
128 registry: Channel selection registry.
129 upstream: Upstream transport adapter; handles credential
130 injection per channel through its configured provider.
131 config: Static gateway configuration.
132 authorizer: Optional authorizer; when ``None`` authorization
133 is skipped.
134 billing: Optional billing lifecycle; when ``None`` the
135 passthrough runs without admission control or settlement.
136 """
137 self._registry = registry
138 self._upstream = upstream
139 self._config = config
140 self._authorizer = authorizer
141 self._billing = billing
143 async def handle(
144 self, kind: str, request: RelayGatewayRequest
145 ) -> Result[RelayPassthroughResult, RelayGatewayError]:
146 """Run the passthrough lifecycle for one request.
148 Dependencies run in fixed order: authorize, select channel by
149 endpoint kind, reserve billing capacity, call upstream with the
150 caller's body verbatim, settle billing, assemble result. Any
151 failure short-circuits the pipeline.
153 Args:
154 kind: The endpoint kind being served (e.g. ``"embeddings"``).
155 request: The passthrough gateway request; ``payload`` is
156 either a ``RelayPassthroughBody`` (JSON or raw multipart)
157 or a plain JSON mapping forwarded by legacy callers, and
158 ``source`` is a conventional marker (``OPENAI_CHAT``)
159 never used for conversion.
161 Returns:
162 ``Ok(RelayPassthroughResult)`` on success, or
163 ``Err(RelayGatewayError)`` on the first failure. Unexpected
164 exceptions from dependencies never escape: they are logged
165 and mapped to a generic ``CONVERSION_FAILED`` error.
166 """
167 started = time.monotonic()
168 logger.info(
169 "relay_passthrough_request_accepted",
170 request_id=request.request_id,
171 tenant_id=request.tenant_id,
172 endpoint=kind,
173 model=request.model,
174 )
175 try:
176 result, channel_name = await self._dispatch(kind, request)
177 except Exception as exc:
178 logger.warning(
179 "relay_passthrough_unexpected_error",
180 request_id=request.request_id,
181 endpoint=kind,
182 error=str(exc),
183 )
184 error = self._unexpected_error(request.request_id)
185 self._log_request_completed(request, kind, "", error, started)
186 return Err(error)
187 if result.is_err():
188 self._log_request_completed(
189 request, kind, channel_name, result.unwrap_err(), started
190 )
191 return result
192 outcome = result.unwrap()
193 self._log_request_completed(request, kind, channel_name, outcome, started)
194 return result
196 async def _dispatch(
197 self,
198 kind: str,
199 request: RelayGatewayRequest,
200 ) -> tuple[Result[RelayPassthroughResult, RelayGatewayError], str]:
201 """Run the ordered dependency pipeline for one request.
203 Returns:
204 ``tuple`` of the pipeline result and the selected channel
205 name. The channel name is ``""`` when selection failed
206 before a channel was chosen.
207 """
208 if kind not in _ENDPOINT_PATHS:
209 return (
210 Err(
211 RelayGatewayError(
212 code=RelayGatewayErrorCode.INVALID_REQUEST,
213 message="unsupported endpoint kind",
214 status_code=400,
215 request_id=request.request_id,
216 retryable=False,
217 )
218 ),
219 "",
220 )
221 if self._authorizer is not None:
222 allowed = await self._authorizer.authorize(
223 user=request.tenant_id,
224 action="relay.invoke",
225 resource=request.model,
226 )
227 if not allowed:
228 return Err(auth_denied(request.request_id)), ""
229 selected = self._registry.select_for_endpoint(
230 kind=kind,
231 model=request.model,
232 )
233 if selected.is_err():
234 return (
235 Err(with_request_id(selected.unwrap_err(), request.request_id)),
236 "",
237 )
238 channel = selected.unwrap()
239 logger.info(
240 "relay_passthrough_channel_selected",
241 request_id=request.request_id,
242 endpoint=kind,
243 channel=channel.name,
244 model=request.model,
245 )
246 billing = self._billing
247 reservation: RelayUsageReservation | None = None
248 if billing is not None:
249 admitted = await self._reserve(request, billing, channel)
250 if admitted.is_err():
251 return Err(admitted.unwrap_err()), channel.name
252 reservation = admitted.unwrap()
253 body = _as_relay_body(request.payload)
254 outbound_model = channel.resolve_model(
255 request.model
256 ) + self._config.model_suffix.get(channel.name, "")
257 body_data = body.data
258 if isinstance(body_data, Mapping):
259 outbound = dict(body_data)
260 outbound["model"] = outbound_model
261 upstream_response = await self._call_upstream(
262 kind, channel, outbound, body.content_type, request
263 )
264 else:
265 content_type = body.content_type
266 boundary = _multipart_boundary(content_type)
267 outbound_raw = body_data
268 if boundary is not None:
269 outbound_raw = rewrite_multipart_form_field(
270 outbound_raw, boundary, "model", outbound_model
271 )
272 upstream_response = await self._call_upstream(
273 kind, channel, outbound_raw, content_type, request
274 )
275 if upstream_response.is_err():
276 if billing is not None and reservation is not None:
277 await self._settle_failed(billing, reservation)
278 return (
279 Err(
280 with_request_id(upstream_response.unwrap_err(), request.request_id)
281 ),
282 channel.name,
283 )
284 resp = upstream_response.unwrap()
285 payload = resp.payload
286 if isinstance(payload, Mapping) and _is_json_content_type(
287 resp.headers.get("content-type", _JSON_CONTENT_TYPE)
288 ):
289 if billing is not None and reservation is not None:
290 await self._settle(
291 billing,
292 reservation,
293 self._usage_from_response(payload),
294 status="completed",
295 )
296 return (
297 Ok(
298 RelayPassthroughResult(
299 status_code=resp.status_code,
300 headers={**resp.headers, "x-request-id": request.request_id},
301 payload=payload,
302 stream=None,
303 metadata=None,
304 body=dumps(payload),
305 content_type=resp.headers.get(
306 "content-type", _JSON_CONTENT_TYPE
307 ),
308 )
309 ),
310 channel.name,
311 )
312 if isinstance(payload, bytes):
313 if billing is not None and reservation is not None:
314 await self._settle(billing, reservation, None, status="completed")
315 return (
316 Ok(
317 RelayPassthroughResult(
318 status_code=resp.status_code,
319 headers={**resp.headers, "x-request-id": request.request_id},
320 payload=None,
321 stream=None,
322 metadata=None,
323 body=payload,
324 content_type=resp.headers.get("content-type", ""),
325 )
326 ),
327 channel.name,
328 )
329 if billing is not None and reservation is not None:
330 await self._settle_failed(billing, reservation)
331 return (
332 Err(
333 RelayGatewayError(
334 code=RelayGatewayErrorCode.UPSTREAM_MALFORMED,
335 message="malformed upstream response",
336 status_code=502,
337 request_id=request.request_id,
338 retryable=False,
339 )
340 ),
341 channel.name,
342 )
344 async def _reserve(
345 self,
346 request: RelayGatewayRequest,
347 billing: RelayBillingProtocol,
348 channel: RelayChannel,
349 ) -> Result[RelayUsageReservation, RelayGatewayError]:
350 """Reserve billing capacity before the upstream call.
352 The passthrough body is wrapped in a transparent carrier so the
353 shared billing pipeline can estimate prompt tokens from the
354 serialized body; the output budget is unknown and reserved as
355 zero. Billing denials short-circuit the pipeline and are
356 classified through :func:`billing_error_to_gateway`.
358 Args:
359 request: The passthrough request being dispatched.
360 billing: The billing lifecycle to reserve through.
361 channel: The selected channel.
363 Returns:
364 ``Ok(reservation)`` when admission is proven, or
365 ``Err(RelayGatewayError)`` carrying the classified failure.
366 """
367 scope = RelayUsageScope(
368 tenant_id=request.tenant_id,
369 model=request.model,
370 channel=channel.name,
371 )
372 body = _as_relay_body(request.payload)
373 body_data = body.data
374 if isinstance(body_data, Mapping):
375 carrier_body: dict[str, JsonValue] = dict(body_data)
376 else:
377 carrier_body = {}
378 carrier = _PassthroughPayloadCarrier(carrier_body)
379 admitted = await billing.pre_consume(
380 request.request_id,
381 scope,
382 cast("RelayRequestPayload", carrier),
383 )
384 if admitted.is_err():
385 error = admitted.unwrap_err()
386 logger.warning(
387 "relay_passthrough_billing_denied",
388 request_id=request.request_id,
389 channel=channel.name,
390 code=error.code,
391 error=error.message,
392 )
393 return Err(billing_error_to_gateway(error, request.request_id))
394 return Ok(admitted.unwrap())
396 async def _settle(
397 self,
398 billing: RelayBillingProtocol,
399 reservation: RelayUsageReservation,
400 usage: RelayUsage | None,
401 *,
402 status: Literal["completed", "failed", "cancelled", "truncated"],
403 ) -> None:
404 """Settle the reservation exactly once without failing the response.
406 Settlement failures are logged and never propagate: the response
407 path has already completed by the time accounting runs.
409 Args:
410 billing: The billing lifecycle to settle through.
411 reservation: The reservation granted by ``pre_consume``.
412 usage: The usage extracted from the upstream response, or
413 ``None`` when the response omits it.
414 status: Terminal lifecycle status of the attempt.
415 """
416 result = RelayConvertResult[Any](
417 value=None,
418 source=RelayFormat.OPENAI_CHAT,
419 target=RelayFormat.OPENAI_CHAT,
420 converter_id="passthrough",
421 quality=ConversionQuality.GOOD,
422 usage=usage,
423 )
424 settled = await billing.settle(reservation, result, status=status)
425 if settled.is_err():
426 error = settled.unwrap_err()
427 logger.warning(
428 "relay_passthrough_settle_failed",
429 request_id=reservation.request_id,
430 status=status,
431 code=error.code,
432 error=error.message,
433 )
435 async def _settle_failed(
436 self,
437 billing: RelayBillingProtocol,
438 reservation: RelayUsageReservation,
439 ) -> None:
440 """Settle a failed attempt without usage through the billing pipeline.
442 Args:
443 billing: The billing lifecycle to settle through.
444 reservation: The reservation granted by ``pre_consume``.
445 """
446 await self._settle(billing, reservation, None, status="failed")
448 async def _call_upstream(
449 self,
450 kind: str,
451 channel: RelayChannel,
452 payload: Mapping[str, JsonValue] | bytes,
453 content_type: str,
454 request: RelayGatewayRequest,
455 ) -> Result[UpstreamResponse, RelayGatewayError]:
456 """Send the passthrough body to the selected channel's endpoint.
458 Uses the same ``HTTPUpstreamAdapter`` as the chat path, so
459 channel-credential injection applies unchanged. JSON bodies go
460 out as their decoded dict; raw bodies (multipart) travel through
461 the adapter's payload slot as opaque bytes with their content
462 type header intact, so the binary parts reach the provider
463 untouched.
465 Args:
466 kind: The endpoint kind being served, selecting the wire path.
467 channel: The selected channel.
468 payload: The caller's body with the model substituted; a
469 decoded JSON object or raw body bytes.
470 content_type: The outbound content type header value.
471 request: The original gateway request.
473 Returns:
474 ``Ok(UpstreamResponse)`` or ``Err`` as returned by the
475 adapter; the adapter already normalizes transport failures.
476 """
477 url = self._endpoint_url(kind, channel)
478 logger.info(
479 "relay_passthrough_upstream_started",
480 request_id=request.request_id,
481 channel=channel.name,
482 method="POST",
483 url=url,
484 )
485 upstream = await self._upstream.request(
486 UpstreamRequest(
487 request_id=request.request_id,
488 method="POST",
489 url=url,
490 headers={"content-type": content_type},
491 payload=cast("Mapping[str, JsonValue]", payload),
492 timeout_seconds=channel.timeout_seconds,
493 channel_name=channel.name,
494 )
495 )
496 if upstream.is_err():
497 err = upstream.unwrap_err()
498 logger.warning(
499 "relay_passthrough_upstream_failed",
500 request_id=request.request_id,
501 channel=channel.name,
502 code=err.code,
503 status_code=err.status_code,
504 error=str(err),
505 )
506 return upstream
508 def _endpoint_url(self, kind: str, channel: RelayChannel) -> str:
509 """Build the endpoint URL for *kind* on *channel*.
511 Args:
512 kind: The endpoint kind being served.
513 channel: The selected channel.
515 Returns:
516 ``<channel base>/v1/<kind>`` for a registered kind; the
517 kind was validated against ``_ENDPOINT_PATHS`` before the
518 channel call, so this never misses.
519 """
520 base = channel.upstream_base_url.rstrip("/")
521 return f"{base}{_ENDPOINT_PATHS[kind]}"
523 @staticmethod
524 def _usage_from_response(payload: Mapping[str, Any]) -> RelayUsage | None:
525 """Extract normalized usage from an OpenAI-shaped response body.
527 Args:
528 payload: The upstream response body.
530 Returns:
531 ``RelayUsage`` when the body carries an integer
532 ``prompt_tokens`` count, otherwise ``None`` (the billing
533 pipeline records usage as missing).
534 """
535 usage = payload.get("usage")
536 if not isinstance(usage, dict):
537 return None
538 prompt = usage.get("prompt_tokens")
539 if not isinstance(prompt, int):
540 return None
541 completion = usage.get("completion_tokens", 0)
542 if not isinstance(completion, int):
543 completion = 0
544 return RelayUsage(prompt_tokens=prompt, completion_tokens=completion)
546 @staticmethod
547 def _unexpected_error(request_id: str) -> RelayGatewayError:
548 """Build the generic error for unexpected dependency failures."""
549 return RelayGatewayError(
550 code=RelayGatewayErrorCode.CONVERSION_FAILED,
551 message="Unexpected relay gateway failure",
552 status_code=500,
553 request_id=request_id,
554 retryable=False,
555 )
557 def _log_request_completed(
558 self,
559 request: RelayGatewayRequest,
560 kind: str,
561 channel_name: str,
562 outcome: RelayGatewayResult | RelayGatewayError,
563 started: float,
564 ) -> None:
565 """Emit the terminal request-completed event for any outcome.
567 Args:
568 request: The original gateway request.
569 kind: The endpoint kind that was served.
570 channel_name: Selected channel name (or ``""`` when unknown).
571 outcome: The success result or the error that ended the flow.
572 started: Monotonic start time used to compute the duration.
573 """
574 logger.info(
575 "relay_passthrough_request_completed",
576 request_id=request.request_id,
577 tenant_id=request.tenant_id,
578 endpoint=kind,
579 channel=channel_name,
580 status_code=outcome.status_code,
581 code=outcome.code if isinstance(outcome, RelayGatewayError) else "OK",
582 duration_ms=round((time.monotonic() - started) * 1000, 2),
583 )