Coverage for src / lexigram / ai / relay / gateway / operations / health.py: 99%
84 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""Channel health aggregation for the relay gateway operations surface.
3``RelayHealthService`` turns the static channel table and optional
4runtime probes into stable ``RelayChannelHealth`` snapshots. Probe
5failures and timeouts are bounded per channel, and never leak upstream
6URLs or credentials into the snapshot. Registry diagnostics expose
7converter capabilities as a failed dependency when no converter registry
8is registered.
9"""
11from __future__ import annotations
13import asyncio
14from collections.abc import Sequence
15from dataclasses import dataclass
16from datetime import datetime
17from typing import Literal, Protocol, runtime_checkable
19from lexigram.ai.relay.gateway.channels import RelayChannelRegistry
20from lexigram.contracts.ai.relay import (
21 RelayChannel,
22 RelayChannelHealth,
23 RelayGatewayError,
24 RelayPolicyStoreProtocol,
25 RelayRegistryDiagnostics,
26 RelayRegistryProtocol,
27)
28from lexigram.primitives import clock
30__all__ = [
31 "CONVERTER_ID",
32 "RelayChannelCheckerProtocol",
33 "RelayChannelProbeResult",
34 "RelayHealthService",
35]
37CONVERTER_ID = "relay-converter"
38"""Diagnostics identifier for the built-in relay converter."""
40_MIN_PROBE_TIMEOUT_SECONDS = 0.001
41"""Probe timeout floor so degenerate timeouts stay bounded."""
43Status = Literal["healthy", "degraded", "unavailable", "failed"]
46@dataclass(frozen=True, slots=True)
47class RelayChannelProbeResult:
48 """Outcome of probing one upstream channel.
50 Attributes:
51 ok: Whether the upstream responded within the bound.
52 latency_ms: Observed latency, or ``None`` when unknown.
53 failure: Human-readable failure reason, or ``None``.
54 """
56 ok: bool
57 latency_ms: float | None = None
58 failure: str | None = None
61@runtime_checkable
62class RelayChannelCheckerProtocol(Protocol):
63 """Bounded upstream probe for a single channel.
65 Implementations are free to ping any endpoint, but must not embed
66 credentials in the probe; the health service records only the status
67 values, never ``upstream_base_url`` or query strings.
68 """
70 async def check(self, channel: RelayChannel) -> RelayChannelProbeResult | None:
71 """Probe *channel* and report a result.
73 Args:
74 channel: The channel to probe.
76 Returns:
77 The probe result, or ``None`` when the checker has no signal
78 for this channel.
79 """
80 ...
83class RelayHealthService:
84 """Aggregate per-channel health and converter diagnostics.
86 Status rules, evaluated in order per channel:
88 - ``enabled=False`` config flag -> ``unavailable``
89 (``channel_disabled``); the model count still reflects aliases.
90 - Runtime policy drained the channel -> ``unavailable``
91 (``drained``).
92 - No checker registered -> ``unavailable`` (``dependency_missing``).
93 - Probe returns ``None`` -> ``unavailable`` (``no_probe_result``).
94 - Probe fails or exceeds the channel timeout -> ``failed``
95 (``probe_failed`` / ``probe_timeout``), counting one failure.
96 - Probe ok but latency at/above the degradation threshold ->
97 ``degraded`` (``high_latency``).
98 - Otherwise -> ``healthy``.
100 The failure precedence ``failed > degraded > unavailable > healthy``
101 holds because disabled/missing cases are decided before probing, and
102 failures are decided before latency thresholds.
103 """
105 def __init__(
106 self,
107 registry: RelayChannelRegistry,
108 checker: RelayChannelCheckerProtocol | None = None,
109 converter: RelayRegistryProtocol | None = None,
110 policy: RelayPolicyStoreProtocol | None = None,
111 degraded_latency_ms: float = 200.0,
112 ) -> None:
113 """Bind the health service to its dependencies.
115 Args:
116 registry: Static channel table; the only source of channels.
117 checker: Optional upstream probe. ``None`` means every
118 channel is reported ``unavailable``.
119 converter: Optional converter registry used by
120 ``registry_diagnostics``. ``None`` makes diagnostics a
121 failed dependency.
122 policy: Optional runtime policy store. A channel drained
123 through the store is reported ``unavailable`` with
124 detail code ``drained``. ``None`` disables the check.
125 degraded_latency_ms: Latency at/above which a working probe
126 is reported ``degraded``. Defaults to 200 ms.
127 """
128 self._registry = registry
129 self._checker = checker
130 self._converter = converter
131 self._policy = policy
132 self._degraded_latency_ms = degraded_latency_ms
134 async def channel_health(self) -> Sequence[RelayChannelHealth]:
135 """Return a health snapshot per configured channel.
137 Channels are reported in configuration order; every channel
138 gets exactly one snapshot.
140 Returns:
141 One snapshot per channel, in configuration order.
142 """
143 checked_at = clock.now()
144 drained: set[str] = set()
145 if self._policy is not None:
146 snapshot = await self._policy.load()
147 drained = {
148 name
149 for name, enabled in snapshot.enabled_channels.items()
150 if not enabled
151 }
152 snapshots: list[RelayChannelHealth] = []
153 for channel in self._registry.channels:
154 snapshots.append(
155 await self._snapshot(channel, checked_at, channel.name in drained)
156 )
157 return snapshots
159 async def registry_diagnostics(self) -> RelayRegistryDiagnostics:
160 """Return converter capability diagnostics.
162 Returns:
163 Converter identifier, version, mapper ids, and supported
164 route pairs.
166 Raises:
167 RelayGatewayError: With ``DEPENDENCY_UNAVAILABLE`` when no
168 converter registry is registered.
169 """
170 if self._converter is None:
171 raise RelayGatewayError(
172 code="DEPENDENCY_UNAVAILABLE",
173 message="converter registry is not registered",
174 status_code=503,
175 request_id="",
176 )
177 return RelayRegistryDiagnostics(
178 converter_id=CONVERTER_ID,
179 converter_version=self._converter.converter_version(),
180 mapper_ids=self._converter.mapper_ids(),
181 supported_routes=self._converter.converter_routes(),
182 )
184 async def _snapshot(
185 self,
186 channel: RelayChannel,
187 checked_at: datetime,
188 drained: bool = False,
189 ) -> RelayChannelHealth:
190 """Build the snapshot for one channel."""
191 if not channel.enabled:
192 return self._build(channel, checked_at, "unavailable", "channel_disabled")
193 if drained:
194 return self._build(channel, checked_at, "unavailable", "drained")
195 if self._checker is None:
196 return self._build(channel, checked_at, "unavailable", "dependency_missing")
197 status: Status = "healthy"
198 detail: str | None = None
199 latency: float | None = None
200 failures = 0
201 try:
202 probe = await self._probe(channel)
203 except TimeoutError:
204 probe = None
205 status = "failed"
206 detail = "probe_timeout"
207 failures = 1
208 if probe is None and status == "healthy":
209 status = "unavailable"
210 detail = "no_probe_result"
211 elif probe is not None and not probe.ok:
212 status = "failed"
213 detail = "probe_failed"
214 failures = 1
215 elif probe is not None:
216 latency = probe.latency_ms
217 if latency is not None and latency >= self._degraded_latency_ms:
218 status = "degraded"
219 detail = "high_latency"
220 return RelayChannelHealth(
221 channel=channel.name,
222 target=channel.target_format,
223 status=status,
224 model_count=len(channel.models),
225 latency_ms_p50=latency,
226 latency_ms_p95=latency,
227 failure_count=failures,
228 checked_at=checked_at,
229 detail_code=detail,
230 )
232 async def _probe(self, channel: RelayChannel) -> RelayChannelProbeResult | None:
233 """Run the probe for *channel*, bounded by its timeout.
235 Args:
236 channel: The channel to probe.
238 Returns:
239 The probe result, or ``None`` when the checker has no
240 signal for the channel.
242 Raises:
243 asyncio.TimeoutError: When the probe exceeds the channel
244 timeout; converted by the caller into a failed status.
245 """
246 checker = self._checker
247 if checker is None:
248 return None
249 timeout = max(channel.timeout_seconds, _MIN_PROBE_TIMEOUT_SECONDS)
250 return await asyncio.wait_for(checker.check(channel), timeout=timeout)
252 @staticmethod
253 def _build(
254 channel: RelayChannel,
255 checked_at: datetime,
256 status: Status,
257 detail: str | None,
258 ) -> RelayChannelHealth:
259 """Build a probe-free snapshot."""
260 return RelayChannelHealth(
261 channel=channel.name,
262 target=channel.target_format,
263 status=status,
264 model_count=len(channel.models),
265 latency_ms_p50=None,
266 latency_ms_p95=None,
267 failure_count=0,
268 checked_at=checked_at,
269 detail_code=detail,
270 )