Coverage for src / lexigram / ai / relay / gateway / job_passthrough.py: 98%
192 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""Job-relay passthrough lifecycle for job-based media generation.
3``JobPassthroughService`` relays endpoint kinds whose upstream protocol is
4submit-then-poll (starting with video generation) through the same
5channel-selection, authorization, billing-admission/settlement, and
6credential-injection pipeline as ``PassthroughService``, plus channel
7affinity across the two or more HTTP calls of one job.
9``submit`` selects a channel by endpoint kind, reserves and settles
10billing once, forwards the caller's body to the channel's submit path,
11and stores a :class:`RelayJobRecord` mapping a gateway-issued job id to
12the upstream's own job id and the owning channel. The submit response
13has its ``id`` field rewritten to the gateway-issued id, so the caller
14never observes the upstream job id or which channel served it.
15``status`` looks the record up (evicting TTL-expired records), routes the
16poll to the *same* channel through the same credential injection as the
17submit, authorizes the caller but never re-bills, and rewrites the
18response ``id`` back to the gateway-issued id. Records live in an
19in-memory registry: a gateway process restart loses in-flight job
20mappings, and a caller polling a lost or expired job receives the same
21``MODEL_NOT_FOUND``-family error as an unknown id.
22"""
24from __future__ import annotations
26from collections.abc import Callable, Mapping
27from dataclasses import dataclass
28import time
29from typing import Any, Literal, cast
31from lexigram.ai.relay.gateway.channels import RelayChannelRegistry
32from lexigram.ai.relay.gateway.config import RelayGatewayConfig
33from lexigram.ai.relay.gateway.errors import (
34 auth_denied,
35 billing_error_to_gateway,
36 with_request_id,
37)
38from lexigram.ai.relay.gateway.job_registry import RelayJobRecord, RelayJobRegistry
39from lexigram.ai.relay.gateway.upstream import HTTPUpstreamAdapter
40from lexigram.contracts.ai.governance import (
41 RelayBillingProtocol,
42 RelayUsageReservation,
43 RelayUsageScope,
44)
45from lexigram.contracts.ai.relay import (
46 ConversionQuality,
47 RelayChannel,
48 RelayConvertResult,
49 RelayFormat,
50 RelayGatewayError,
51 RelayGatewayRequest,
52 RelayGatewayResult,
53 RelayRequestPayload,
54 RelayUsage,
55 UpstreamRequest,
56 UpstreamResponse,
57)
58from lexigram.contracts.ai.relay.gateway import RelayGatewayErrorCode
59from lexigram.contracts.auth.guard import AuthorizerProtocol
60from lexigram.contracts.core.result import Err, Ok, Result
61from lexigram.logging import get_logger
63__all__ = ["JobPassthroughService"]
65logger = get_logger(__name__)
67_JOB_ENDPOINT_PATHS: dict[str, str] = {
68 "video_generation": "/v1/videos",
69}
70"""Endpoint kinds to upstream submit path segments served by job relay.
72Every registered kind uses the OpenAI-shaped ``/v1/<kind>`` submit path;
73the status poll for a record reuses its stored kind's path. Music
74generation is deliberately absent: no ``lexigram-multimedia-music``
75provider exists to relay to (see the async job-relay plan).
76"""
79@dataclass(frozen=True, slots=True)
80class _JobPayloadCarrier:
81 """Billing-admission carrier for a job-relay request body.
83 Identical in purpose to the passthrough carrier: the body is not a
84 chat wire DTO, so prompt estimation counts the serialized body and
85 the output budget is reserved as zero. The carrier quacks like a
86 ``RelayRequestPayload`` at the only call site the billing pipeline
87 uses (``to_dict``).
88 """
90 body: dict[str, Any]
92 def to_dict(self) -> dict[str, Any]:
93 """Return the job-relay request body.
95 Returns:
96 A shallow copy of the forwarded request body.
97 """
98 return dict(self.body)
101class JobPassthroughService:
102 """Submit/poll relay lifecycle with per-job channel affinity.
104 The service is stateful in exactly one dimension: it stores the
105 channel name and upstream job id of every submitted job in the
106 injected ``RelayJobRegistry`` so subsequent polls route to the same
107 channel, and it never includes payloads or upstream details in error
108 messages; errors are always safe ``RelayGatewayError`` values.
110 Attributes:
111 job_registry: Stores the channel-affinity mapping for in-flight
112 jobs; also evicts TTL-expired records on lookup.
113 _registry: Deterministic channel selector.
114 _upstream: HTTP transport adapter; handles credential injection
115 per channel through its configured provider.
116 _config: Gateway configuration (channel table, model suffixes,
117 and the job TTL).
118 _authorizer: Optional authorization check before dispatch.
119 _billing: Optional billing lifecycle; when ``None`` jobs run
120 without admission control or settlement.
121 """
123 def __init__(
124 self,
125 registry: RelayChannelRegistry,
126 job_registry: RelayJobRegistry,
127 upstream: HTTPUpstreamAdapter,
128 config: RelayGatewayConfig,
129 *,
130 authorizer: AuthorizerProtocol | None = None,
131 billing: RelayBillingProtocol | None = None,
132 clock: Callable[[], float] = time.monotonic,
133 ) -> None:
134 """Bind the service to its dependencies.
136 Args:
137 registry: Channel selection registry.
138 job_registry: Job record store providing channel affinity
139 and TTL eviction.
140 upstream: Upstream transport adapter; handles credential
141 injection per channel through its configured provider.
142 config: Static gateway configuration.
143 authorizer: Optional authorizer; when ``None`` authorization
144 is skipped.
145 billing: Optional billing lifecycle; when ``None`` jobs run
146 without admission control or settlement.
147 clock: Callable returning the current monotonic time,
148 stamped on created job records. Defaults to
149 ``time.monotonic``; tests inject a fake consistent with
150 the job registry's.
151 """
152 self.job_registry = job_registry
153 self._registry = registry
154 self._upstream = upstream
155 self._config = config
156 self._authorizer = authorizer
157 self._billing = billing
158 self._clock = clock
160 async def submit(
161 self, kind: str, request: RelayGatewayRequest
162 ) -> Result[RelayGatewayResult, RelayGatewayError]:
163 """Submit a job: authorize, select, reserve, forward, record.
165 Runs the ordered dependency pipeline, then stores a
166 ``RelayJobRecord`` mapping the gateway-issued job id to the
167 upstream's job id and the owning channel, and returns the
168 upstream response with its ``id`` field rewritten to the
169 gateway-issued id. Billing is admitted and settled exactly once,
170 here; a subsequent ``status`` call never re-charges.
172 Args:
173 kind: The endpoint kind being served (e.g.
174 ``"video_generation"``).
175 request: The gateway request; the body is forwarded to the
176 selected channel's submit path verbatim.
178 Returns:
179 ``Ok(RelayGatewayResult)`` carrying the id-rewritten upstream
180 response on success, or ``Err(RelayGatewayError)`` on the
181 first failure. Unexpected exceptions never escape: they are
182 logged and mapped to a generic ``CONVERSION_FAILED`` error.
183 """
184 started = time.monotonic()
185 logger.info(
186 "relay_job_submit_accepted",
187 request_id=request.request_id,
188 tenant_id=request.tenant_id,
189 endpoint=kind,
190 model=request.model,
191 )
192 try:
193 result, channel_name = await self._dispatch_submit(kind, request)
194 except Exception as exc:
195 logger.warning(
196 "relay_job_submit_unexpected_error",
197 request_id=request.request_id,
198 endpoint=kind,
199 error=str(exc),
200 )
201 result = Err(self._unexpected_error(request.request_id))
202 channel_name = ""
203 outcome: RelayGatewayResult | RelayGatewayError = (
204 result.unwrap_err() if result.is_err() else result.unwrap()
205 )
206 self._log_request_completed(
207 request, kind, "submit", channel_name, outcome, started
208 )
209 return result
211 async def status(
212 self, kind: str, gateway_job_id: str, request: RelayGatewayRequest
213 ) -> Result[RelayGatewayResult, RelayGatewayError]:
214 """Poll a submitted job on the channel that owns it.
216 Looks up the job record by the gateway-issued id (silently
217 evicting TTL-expired records), authorizes the caller, and polls
218 the owning channel's status path with the stored upstream job id
219 — the record's channel is reused, never re-selected. The
220 response ``id`` is rewritten back to the gateway-issued id.
221 Billing never runs here: only ``submit`` charges.
223 Args:
224 kind: The endpoint kind the job was submitted through.
225 gateway_job_id: The gateway-issued job id returned by
226 ``submit``.
227 request: The gateway request; only identity and request id
228 are used.
230 Returns:
231 ``Ok(RelayGatewayResult)`` carrying the id-rewritten status
232 response on success, or ``Err(RelayGatewayError)``. Unknown
233 and TTL-expired job ids both map to the same
234 ``MODEL_NOT_FOUND``-family error ``select_for_endpoint``
235 produces for an unmatched model. Unexpected exceptions
236 never escape: they are logged and mapped to a generic
237 ``CONVERSION_FAILED`` error.
238 """
239 started = time.monotonic()
240 logger.info(
241 "relay_job_status_accepted",
242 request_id=request.request_id,
243 endpoint=kind,
244 )
245 try:
246 result, channel_name = await self._dispatch_status(
247 kind, gateway_job_id, request
248 )
249 except Exception as exc:
250 logger.warning(
251 "relay_job_status_unexpected_error",
252 request_id=request.request_id,
253 endpoint=kind,
254 error=str(exc),
255 )
256 result = Err(self._unexpected_error(request.request_id))
257 channel_name = ""
258 outcome: RelayGatewayResult | RelayGatewayError = (
259 result.unwrap_err() if result.is_err() else result.unwrap()
260 )
261 self._log_request_completed(
262 request, kind, "status", channel_name, outcome, started
263 )
264 return result
266 async def _dispatch_submit(
267 self,
268 kind: str,
269 request: RelayGatewayRequest,
270 ) -> tuple[Result[RelayGatewayResult, RelayGatewayError], str]:
271 """Run the ordered submit pipeline for one job.
273 Returns:
274 ``tuple`` of the pipeline result and the selected channel
275 name. The channel name is ``""`` when selection failed
276 before a channel was chosen.
277 """
278 if kind not in _JOB_ENDPOINT_PATHS:
279 return (
280 Err(
281 RelayGatewayError(
282 code=RelayGatewayErrorCode.INVALID_REQUEST,
283 message="unsupported endpoint kind",
284 status_code=400,
285 request_id=request.request_id,
286 retryable=False,
287 )
288 ),
289 "",
290 )
291 if self._authorizer is not None:
292 allowed = await self._authorizer.authorize(
293 user=request.tenant_id,
294 action="relay.invoke",
295 resource=request.model,
296 )
297 if not allowed:
298 return Err(auth_denied(request.request_id)), ""
299 selected = self._registry.select_for_endpoint(
300 kind=kind,
301 model=request.model,
302 )
303 if selected.is_err():
304 return (
305 Err(with_request_id(selected.unwrap_err(), request.request_id)),
306 "",
307 )
308 channel = selected.unwrap()
309 logger.info(
310 "relay_job_channel_selected",
311 request_id=request.request_id,
312 endpoint=kind,
313 channel=channel.name,
314 model=request.model,
315 )
316 billing = self._billing
317 reservation: RelayUsageReservation | None = None
318 if billing is not None:
319 admitted = await self._reserve(request, billing, channel)
320 if admitted.is_err():
321 return Err(admitted.unwrap_err()), channel.name
322 reservation = admitted.unwrap()
323 outbound_model = request.model + self._config.model_suffix.get(channel.name, "")
324 outbound = dict(request.payload)
325 outbound["model"] = outbound_model
326 upstream_response = await self._call_upstream(
327 "POST",
328 self._job_url(kind, channel),
329 channel,
330 outbound,
331 request,
332 )
333 if upstream_response.is_err():
334 if billing is not None and reservation is not None:
335 await self._settle_failed(billing, reservation)
336 return (
337 Err(
338 with_request_id(upstream_response.unwrap_err(), request.request_id)
339 ),
340 channel.name,
341 )
342 resp = upstream_response.unwrap()
343 if resp.payload is None:
344 if billing is not None and reservation is not None:
345 await self._settle_failed(billing, reservation)
346 return Err(self._malformed_error(request.request_id)), channel.name
347 upstream_job_id = self._extract_job_id(resp.payload)
348 if upstream_job_id is None:
349 if billing is not None and reservation is not None:
350 await self._settle_failed(billing, reservation)
351 return Err(self._missing_id_error(request.request_id)), channel.name
352 gateway_job_id = self.job_registry.put(
353 RelayJobRecord(
354 channel_name=channel.name,
355 upstream_job_id=upstream_job_id,
356 endpoint_kind=kind,
357 created_at=self._clock(),
358 )
359 )
360 rewritten = dict(resp.payload or {})
361 rewritten["id"] = gateway_job_id
362 if billing is not None and reservation is not None:
363 await self._settle(
364 billing,
365 reservation,
366 self._usage_from_response(rewritten),
367 status="completed",
368 )
369 return (
370 Ok(
371 RelayGatewayResult(
372 status_code=resp.status_code,
373 headers={**resp.headers, "x-request-id": request.request_id},
374 payload=rewritten,
375 stream=None,
376 metadata=None,
377 )
378 ),
379 channel.name,
380 )
382 async def _dispatch_status(
383 self,
384 kind: str,
385 gateway_job_id: str,
386 request: RelayGatewayRequest,
387 ) -> tuple[Result[RelayGatewayResult, RelayGatewayError], str]:
388 """Run the ordered status pipeline for one job.
390 Channel affinity holds here: the channel is read from the stored
391 record, never re-selected, and the upstream job id from the
392 record is inserted into the poll path. Unknown, expired, and
393 channel-gone records all map to the same not-found family as an
394 unserved model.
396 Returns:
397 ``tuple`` of the pipeline result and the polled channel
398 name, ``""`` when no record or channel was found.
399 """
400 if kind not in _JOB_ENDPOINT_PATHS:
401 return (
402 Err(
403 RelayGatewayError(
404 code=RelayGatewayErrorCode.INVALID_REQUEST,
405 message="unsupported endpoint kind",
406 status_code=400,
407 request_id=request.request_id,
408 retryable=False,
409 )
410 ),
411 "",
412 )
413 if self._authorizer is not None:
414 allowed = await self._authorizer.authorize(
415 user=request.tenant_id,
416 action="relay.invoke",
417 resource=request.model,
418 )
419 if not allowed:
420 return Err(auth_denied(request.request_id)), ""
421 record = self.job_registry.get(gateway_job_id)
422 if record is None:
423 return (
424 Err(self._job_not_found_error(gateway_job_id, request.request_id)),
425 "",
426 )
427 channel = self._channel_by_name(record.channel_name)
428 if channel is None:
429 logger.warning(
430 "relay_job_channel_missing",
431 request_id=request.request_id,
432 job_id=gateway_job_id,
433 channel=record.channel_name,
434 )
435 return (
436 Err(
437 RelayGatewayError(
438 code=RelayGatewayErrorCode.MODEL_NOT_FOUND,
439 message="no relay channel available for this job",
440 status_code=404,
441 request_id=request.request_id,
442 retryable=False,
443 )
444 ),
445 "",
446 )
447 upstream_response = await self._call_upstream(
448 "GET",
449 self._job_url(
450 record.endpoint_kind,
451 channel,
452 job_id=record.upstream_job_id,
453 ),
454 channel,
455 {},
456 request,
457 )
458 if upstream_response.is_err():
459 return (
460 Err(
461 with_request_id(upstream_response.unwrap_err(), request.request_id)
462 ),
463 channel.name,
464 )
465 resp = upstream_response.unwrap()
466 if resp.payload is None:
467 return (
468 Err(self._malformed_error(request.request_id)),
469 channel.name,
470 )
471 rewritten = dict(resp.payload)
472 if "id" in rewritten:
473 rewritten["id"] = gateway_job_id
474 return (
475 Ok(
476 RelayGatewayResult(
477 status_code=resp.status_code,
478 headers={**resp.headers, "x-request-id": request.request_id},
479 payload=rewritten,
480 stream=None,
481 metadata=None,
482 )
483 ),
484 channel.name,
485 )
487 async def _reserve(
488 self,
489 request: RelayGatewayRequest,
490 billing: RelayBillingProtocol,
491 channel: RelayChannel,
492 ) -> Result[RelayUsageReservation, RelayGatewayError]:
493 """Reserve billing capacity before the upstream submit call.
495 The job-relay body is wrapped in a transparent carrier so the
496 shared billing pipeline can estimate prompt tokens from the
497 serialized body; the output budget is unknown and reserved as
498 zero. Billing denials fail the submit and are classified
499 through :func:`billing_error_to_gateway`.
501 Args:
502 request: The job submit request being dispatched.
503 billing: The billing lifecycle to reserve through.
504 channel: The selected channel.
506 Returns:
507 ``Ok(reservation)`` when admission is proven, or
508 ``Err(RelayGatewayError)`` carrying the classified failure.
509 """
510 scope = RelayUsageScope(
511 tenant_id=request.tenant_id,
512 model=request.model,
513 channel=channel.name,
514 )
515 carrier = _JobPayloadCarrier(dict(request.payload))
516 admitted = await billing.pre_consume(
517 request.request_id,
518 scope,
519 cast("RelayRequestPayload", carrier),
520 )
521 if admitted.is_err():
522 error = admitted.unwrap_err()
523 logger.warning(
524 "relay_job_billing_denied",
525 request_id=request.request_id,
526 channel=channel.name,
527 code=error.code,
528 error=error.message,
529 )
530 return Err(billing_error_to_gateway(error, request.request_id))
531 return Ok(admitted.unwrap())
533 async def _settle(
534 self,
535 billing: RelayBillingProtocol,
536 reservation: RelayUsageReservation,
537 usage: RelayUsage | None,
538 *,
539 status: Literal["completed", "failed", "cancelled", "truncated"],
540 ) -> None:
541 """Settle the reservation exactly once without failing the response.
543 Settlement failures are logged and never propagate: the
544 response path has already completed by the time accounting runs.
546 Args:
547 billing: The billing lifecycle to settle through.
548 reservation: The reservation granted by ``pre_consume``.
549 usage: The usage extracted from the upstream submit response,
550 or ``None`` when the response omits it.
551 status: Terminal lifecycle status of the job attempt.
552 """
553 result = RelayConvertResult[Any](
554 value=None,
555 source=RelayFormat.OPENAI_CHAT,
556 target=RelayFormat.OPENAI_CHAT,
557 converter_id="job_passthrough",
558 quality=ConversionQuality.GOOD,
559 usage=usage,
560 )
561 settled = await billing.settle(reservation, result, status=status)
562 if settled.is_err():
563 error = settled.unwrap_err()
564 logger.warning(
565 "relay_job_settle_failed",
566 request_id=reservation.request_id,
567 status=status,
568 code=error.code,
569 error=error.message,
570 )
572 async def _settle_failed(
573 self,
574 billing: RelayBillingProtocol,
575 reservation: RelayUsageReservation,
576 ) -> None:
577 """Settle a failed submit attempt without usage.
579 Args:
580 billing: The billing lifecycle to settle through.
581 reservation: The reservation granted by ``pre_consume``.
582 """
583 await self._settle(billing, reservation, None, status="failed")
585 async def _call_upstream(
586 self,
587 method: str,
588 url: str,
589 channel: RelayChannel,
590 payload: dict[str, Any] | None,
591 request: RelayGatewayRequest,
592 ) -> Result[UpstreamResponse, RelayGatewayError]:
593 """Send one upstream call through the channel's credential injection.
595 Uses the same ``HTTPUpstreamAdapter`` as the chat path, so
596 channel-credential injection applies unchanged. The status poll
597 carries no body.
599 Args:
600 method: HTTP method of the call (``"POST"`` for submit,
601 ``"GET"`` for status).
602 url: Fully-resolved upstream URL.
603 channel: The channel the call is pinned to.
604 payload: The JSON body to send, or ``None`` for body-less
605 calls.
606 request: The originating gateway request.
608 Returns:
609 ``Ok(UpstreamResponse)`` or ``Err`` as returned by the
610 adapter; the adapter already normalizes transport failures.
611 """
612 logger.info(
613 "relay_job_upstream_started",
614 request_id=request.request_id,
615 channel=channel.name,
616 method=method,
617 url=url,
618 )
619 upstream = await self._upstream.request(
620 UpstreamRequest(
621 request_id=request.request_id,
622 method=method,
623 url=url,
624 headers={"content-type": "application/json"}
625 if method == "POST"
626 else {},
627 payload=payload if payload is not None else {},
628 timeout_seconds=channel.timeout_seconds,
629 channel_name=channel.name,
630 )
631 )
632 if upstream.is_err():
633 err = upstream.unwrap_err()
634 logger.warning(
635 "relay_job_upstream_failed",
636 request_id=request.request_id,
637 channel=channel.name,
638 code=err.code,
639 status_code=err.status_code,
640 error=str(err),
641 )
642 return upstream
644 def _channel_by_name(self, name: str) -> RelayChannel | None:
645 """Return the configured channel with *name*, or ``None`` when gone.
647 Args:
648 name: Channel name stored on a job record.
650 Returns:
651 The matching channel from the registry's static table, or
652 ``None`` when the channel was removed from configuration
653 since the job was submitted.
654 """
655 for channel in self._registry.channels:
656 if channel.name == name:
657 return channel
658 return None
660 def _job_url(
661 self,
662 kind: str,
663 channel: RelayChannel,
664 *,
665 job_id: str | None = None,
666 ) -> str:
667 """Build the upstream URL for a job call on *channel*.
669 Args:
670 kind: The endpoint kind selecting the path segment; only
671 registered kinds reach this call.
672 channel: The channel the call is pinned to.
673 job_id: The upstream job id to append for status polls;
674 ``None`` for the submit path.
676 Returns:
677 ``<channel base>/v1/<kind>`` for submits, and the same path
678 with ``/<job_id>`` appended for status polls.
679 """
680 base = channel.upstream_base_url.rstrip("/")
681 url = f"{base}{_JOB_ENDPOINT_PATHS[kind]}"
682 if job_id is not None:
683 url = f"{url}/{job_id}"
684 return url
686 @staticmethod
687 def _extract_job_id(payload: Mapping[str, Any]) -> str | None:
688 """Extract the upstream job id from a submit response body.
690 Args:
691 payload: The upstream submit response body.
693 Returns:
694 The non-empty string ``id`` field, or ``None`` when absent
695 or not a non-empty string — the caller treats that as a
696 mapping error rather than recording a null job id.
697 """
698 job_id = payload.get("id")
699 if isinstance(job_id, str) and job_id:
700 return job_id
701 return None
703 @staticmethod
704 def _missing_id_error(request_id: str) -> RelayGatewayError:
705 """Build the mapping error for a submit response without a job id."""
706 return RelayGatewayError(
707 code=RelayGatewayErrorCode.UPSTREAM_MALFORMED,
708 message="upstream submit response missing job id",
709 status_code=502,
710 request_id=request_id,
711 retryable=False,
712 )
714 @staticmethod
715 def _job_not_found_error(gateway_job_id: str, request_id: str) -> RelayGatewayError:
716 """Build the not-found error for an unknown or expired job id."""
717 return RelayGatewayError(
718 code=RelayGatewayErrorCode.MODEL_NOT_FOUND,
719 message=f"no relay job found for id {gateway_job_id}",
720 status_code=404,
721 request_id=request_id,
722 retryable=False,
723 )
725 @staticmethod
726 def _malformed_error(request_id: str) -> RelayGatewayError:
727 """Build the gateway error for a malformed upstream response body."""
728 return RelayGatewayError(
729 code=RelayGatewayErrorCode.UPSTREAM_MALFORMED,
730 message="malformed upstream response",
731 status_code=502,
732 request_id=request_id,
733 retryable=False,
734 )
736 @staticmethod
737 def _usage_from_response(payload: Mapping[str, Any]) -> RelayUsage | None:
738 """Extract normalized usage from an OpenAI-shaped response body.
740 Args:
741 payload: The upstream submit response body.
743 Returns:
744 ``RelayUsage`` when the body carries an integer
745 ``prompt_tokens`` count, otherwise ``None`` (the billing
746 pipeline records usage as missing).
747 """
748 usage = payload.get("usage")
749 if not isinstance(usage, dict):
750 return None
751 prompt = usage.get("prompt_tokens")
752 if not isinstance(prompt, int):
753 return None
754 completion = usage.get("completion_tokens", 0)
755 if not isinstance(completion, int):
756 completion = 0
757 return RelayUsage(prompt_tokens=prompt, completion_tokens=completion)
759 @staticmethod
760 def _unexpected_error(request_id: str) -> RelayGatewayError:
761 """Build the generic error for unexpected dependency failures."""
762 return RelayGatewayError(
763 code=RelayGatewayErrorCode.CONVERSION_FAILED,
764 message="Unexpected relay gateway failure",
765 status_code=500,
766 request_id=request_id,
767 retryable=False,
768 )
770 def _log_request_completed(
771 self,
772 request: RelayGatewayRequest,
773 kind: str,
774 phase: str,
775 channel_name: str,
776 outcome: RelayGatewayResult | RelayGatewayError,
777 started: float,
778 ) -> None:
779 """Emit the terminal completed event for any outcome.
781 Args:
782 request: The original gateway request.
783 kind: The endpoint kind that was served.
784 phase: ``"submit"`` or ``"status"``.
785 channel_name: Selected channel name (or ``""`` when unknown).
786 outcome: The success result or the error that ended the flow.
787 started: Monotonic start time used to compute the duration.
788 """
789 logger.info(
790 "relay_job_request_completed",
791 request_id=request.request_id,
792 tenant_id=request.tenant_id,
793 endpoint=kind,
794 phase=phase,
795 channel=channel_name,
796 status_code=outcome.status_code,
797 code=outcome.code if isinstance(outcome, RelayGatewayError) else "OK",
798 duration_ms=round((time.monotonic() - started) * 1000, 2),
799 )