« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Typed gateway configuration for the relay gateway.
3Holds the static channel table plus model-suffix and provider-options
4metadata. The suffix and option maps are consumed at conversion time by
5the service layer; channel selection never reads them. The auto-test
6flags control the background channel health sweep (see
7:class:`~lexigram.ai.relay.gateway.operations.auto_test.RelayChannelAutoTester`).
8"""
10from __future__ import annotations
12from collections.abc import Mapping
13from dataclasses import dataclass, field
14from typing import Any, Literal
16from lexigram.contracts.ai.relay import JsonValue, RelayChannel, RelayFormat
17from lexigram.serialization import loads
19__all__ = ["RelayGatewayConfig"]
21_CHANNEL_FIELDS = {
22 "name",
23 "upstream_base_url",
24 "target_format",
25 "models",
26 "capabilities",
27 "endpoint_kinds",
28 "priority",
29 "weight",
30 "enabled",
31 "timeout_seconds",
32 "model_map",
33}
34"""Channel keys accepted by :meth:`RelayGatewayConfig.from_mapping`."""
37@dataclass(frozen=True, slots=True)
38class RelayGatewayConfig:
39 """Static configuration backing ``RelayChannelRegistry`` selection.
41 Attributes:
42 channels: The ordered channel configurations. Selection filters
43 before sorting, so order is never observable in the result
44 except as the stable ``name`` tiebreak. Duplicate names are
45 rejected.
46 model_suffix: Channel name to a suffix (e.g. ``":thinking"``)
47 appended to the outbound model alias at the service layer.
48 Selection does not use this field.
49 provider_options: Channel name to provider-specific options merged
50 into ``RelayConversionContext`` at conversion time. Selection
51 does not use this field.
52 auto_test_channels: When ``True`` the provider starts a background
53 channel auto-tester that periodically probes every channel
54 and disables failed ones, re-enabling them on recovery.
55 Defaults to ``False`` (disabled).
56 auto_test_interval_seconds: Delay between auto-test sweeps in
57 seconds. Must be positive when defined. Defaults to ``600``.
58 max_upstream_retries: Number of retry attempts across *other*
59 channels after a retryable upstream failure on the buffered
60 path. Defaults to ``0`` (single attempt, today's behavior).
61 load_balancing: Channel-selection mode. ``"deterministic"``
62 (default) keeps today's name-sort tiebreak; ``"weighted"``
63 breaks ties among equal-priority eligible channels by
64 weighted-random pick driven by each channel's ``weight``.
65 job_ttl_seconds: Age in seconds after which a relay job record
66 (``POST /v1/videos`` style job relay) is evicted from the
67 in-memory job registry on its next poll. Must be positive.
68 Defaults to ``3600`` (one hour).
69 require_auth: When ``True`` the inbound relay routes require a
70 bound ``RelayAuthVerifierProtocol``; ``False`` is an explicit
71 opt-out for local/dev use only.
72 rate_limits: Model name (or ``"*"`` for the token-wide rule) to
73 a ``{"max": int, "window_seconds": int}`` budget. Empty
74 (default) disables the rate-limit guard entirely.
75 auto_disable_on_failures: When ``True`` the gateway tracks
76 consecutive upstream failures per channel and takes a channel
77 out of service at runtime once ``failover_failure_threshold``
78 is reached, restoring it after the next successful dispatch.
79 Defaults to ``False`` (disabled).
80 failover_failure_threshold: Number of consecutive failures that
81 disable a channel when ``auto_disable_on_failures`` is on.
82 Defaults to ``3``.
83 """
85 channels: tuple[RelayChannel, ...] = ()
86 model_suffix: Mapping[str, str] = field(default_factory=dict)
87 provider_options: Mapping[str, Mapping[str, JsonValue]] = field(
88 default_factory=dict
89 )
90 auto_test_channels: bool = False
91 auto_test_interval_seconds: int = 600
92 max_upstream_retries: int = 0
93 load_balancing: Literal["deterministic", "weighted"] = "deterministic"
94 job_ttl_seconds: int = 3600
95 require_auth: bool = True
96 rate_limits: Mapping[str, Mapping[str, int]] = field(default_factory=dict)
97 auto_disable_on_failures: bool = False
98 failover_failure_threshold: int = 3
100 def __post_init__(self) -> None:
101 """Reject duplicate names, bad auto-test intervals, and negative retries."""
102 names = [channel.name for channel in self.channels]
103 if len(names) != len(set(names)):
104 raise ValueError("duplicate channel names in RelayGatewayConfig")
105 if self.auto_test_interval_seconds <= 0:
106 raise ValueError("auto_test_interval_seconds must be a positive integer")
107 if self.max_upstream_retries < 0:
108 raise ValueError("max_upstream_retries must be non-negative")
109 if self.load_balancing not in ("deterministic", "weighted"):
110 raise ValueError("load_balancing must be 'deterministic' or 'weighted'")
111 if self.job_ttl_seconds <= 0:
112 raise ValueError("job_ttl_seconds must be a positive integer")
113 if self.failover_failure_threshold < 1:
114 raise ValueError("failover_failure_threshold must be a positive integer")
116 @classmethod
117 def from_string(cls, config_str: str) -> RelayGatewayConfig:
118 """Build the configuration from a JSON document passed directly.
120 Parsing is fully local — no network, file, or store access — so
121 unit tests and embedded hosts can hand the gateway its whole
122 configuration as text (the same document shape as
123 :meth:`from_mapping`).
125 Args:
126 config_str: JSON document with a ``"channels"`` list and
127 optional gateway fields.
129 Returns:
130 A validated ``RelayGatewayConfig``.
132 Raises:
133 TypeError: When the document is valid JSON but not an
134 object.
135 ValueError: When the document is malformed JSON or contains
136 invalid keys/values (same as :meth:`from_mapping`).
137 """
138 try:
139 data = loads(config_str)
140 except ValueError as exc:
141 raise ValueError("gateway config is not valid JSON") from exc
142 if not isinstance(data, Mapping):
143 raise TypeError("gateway config must be a JSON object")
144 return cls.from_mapping(data)
146 @classmethod
147 def from_mapping(cls, data: Mapping[str, Any]) -> RelayGatewayConfig:
148 """Build the configuration from a JSON/TOML-style mapping.
150 Channel entries accept the fields of :class:`RelayChannel`
151 (``target_format`` as the format member name, e.g.
152 ``"OPENAI_CHAT"``), and top-level keys mirror the remaining
153 attributes of this class. Unknown channel keys and top-level
154 ``"channels"`` types raise ``ValueError`` with the offending
155 key named.
157 Args:
158 data: Mapping with a ``"channels"`` list and optional
159 gateway fields.
161 Returns:
162 A validated ``RelayGatewayConfig``.
164 Raises:
165 TypeError: On malformed channel entries or a non-list
166 ``"channels"`` value.
167 ValueError: On unknown channel keys or invalid top-level
168 values.
169 """
170 raw_channels = data.get("channels", ())
171 if isinstance(raw_channels, Mapping):
172 raise TypeError("channels must be a list of channel mappings")
173 channels: list[RelayChannel] = []
174 for index, entry in enumerate(raw_channels):
175 if not isinstance(entry, Mapping):
176 raise TypeError(f"channels[{index}] must be an object")
177 unknown = set(entry) - _CHANNEL_FIELDS
178 if unknown:
179 raise ValueError(
180 f"channels[{index}] has unknown keys: {sorted(unknown)}"
181 )
182 channels.append(cls._channel_from_mapping(entry))
183 allowed = {
184 "channels",
185 "model_suffix",
186 "provider_options",
187 "auto_test_channels",
188 "auto_test_interval_seconds",
189 "max_upstream_retries",
190 "load_balancing",
191 "job_ttl_seconds",
192 "require_auth",
193 "rate_limits",
194 "auto_disable_on_failures",
195 "failover_failure_threshold",
196 }
197 unknown = set(data) - allowed
198 if unknown:
199 raise ValueError(f"unknown gateway config keys: {sorted(unknown)}")
200 return cls(
201 channels=tuple(channels),
202 model_suffix=dict(data.get("model_suffix", {}) or {}),
203 provider_options=dict(data.get("provider_options", {}) or {}),
204 auto_test_channels=bool(data.get("auto_test_channels", False)),
205 auto_test_interval_seconds=int(data.get("auto_test_interval_seconds", 600)),
206 max_upstream_retries=int(data.get("max_upstream_retries", 0)),
207 load_balancing=data.get("load_balancing", "deterministic"),
208 job_ttl_seconds=int(data.get("job_ttl_seconds", 3600)),
209 require_auth=bool(data.get("require_auth", True)),
210 rate_limits=dict(data.get("rate_limits", {}) or {}),
211 auto_disable_on_failures=bool(data.get("auto_disable_on_failures", False)),
212 failover_failure_threshold=int(data.get("failover_failure_threshold", 3)),
213 )
215 @staticmethod
216 def _channel_from_mapping(entry: Mapping[str, Any]) -> RelayChannel:
217 """Build one ``RelayChannel`` from a validated mapping entry."""
218 missing = {"name", "upstream_base_url", "target_format", "models"} - set(entry)
219 if missing:
220 raise ValueError(f"channel object missing keys: {sorted(missing)}")
221 raw_format = entry["target_format"]
222 try:
223 target_format = RelayFormat[raw_format] # member name
224 except (KeyError, TypeError):
225 try:
226 target_format = RelayFormat(raw_format) # member value
227 except ValueError as exc:
228 raise ValueError(
229 f"unknown target_format {raw_format!r}; "
230 f"expected one of {[m.value for m in RelayFormat]}"
231 ) from exc
232 return RelayChannel(
233 name=str(entry["name"]),
234 upstream_base_url=str(entry["upstream_base_url"]),
235 target_format=target_format,
236 models=tuple(entry["models"]),
237 capabilities=frozenset(entry.get("capabilities", ()) or ()),
238 endpoint_kinds=frozenset(entry.get("endpoint_kinds", ()) or ()),
239 priority=int(entry.get("priority", 100)),
240 weight=int(entry.get("weight", 100)),
241 enabled=bool(entry.get("enabled", True)),
242 timeout_seconds=float(entry.get("timeout_seconds", 60.0)),
243 model_map=dict(entry.get("model_map", {}) or {}),
244 )