Coverage for src / lexigram / ai / relay / gateway / admin / contributor.py: 85%
148 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""Admin contributor for the relay gateway operations surface.
3The contributor surfaces channel health, route metrics, active streams,
4and runtime policy into the admin dashboard, plus permissioned control
5actions (channel drain/enable and stream cancellation). Dependencies
6are resolved lazily from the DI container at boot; every surface
7renders an explicit unavailable state when a dependency is missing.
8"""
10from __future__ import annotations
12from collections.abc import Sequence
13from typing import TYPE_CHECKING, Any, cast
15from lexigram.ai.relay.gateway.operations.controls import (
16 PERMISSION_CHANNEL_CONTROL,
17 PERMISSION_POLICY_CONTROL,
18 PERMISSION_READ,
19 PERMISSION_STREAM_CONTROL,
20 RelayControlsService,
21)
22from lexigram.ai.relay.gateway.operations.health import RelayHealthService
23from lexigram.ai.relay.gateway.operations.metrics import RelayMetricsService
24from lexigram.contracts.admin.contributor import BaseAdminContributor
25from lexigram.contracts.admin.errors import (
26 HealthCheckNotFoundError,
27 WidgetNotFoundError,
28)
29from lexigram.contracts.admin.route_spec import AdminRouteSpec
30from lexigram.contracts.admin.types import (
31 ActionParameterField,
32 ActionParameterSchema,
33 AdminActionDefinition,
34 AdminHealthDefinition,
35 DashboardWidgetDefinition,
36 ManagementPageDefinition,
37 NavigationContribution,
38 PageCategory,
39 WidgetCategory,
40 WidgetParams,
41 WidgetSize,
42 WidgetViewModel,
43)
44from lexigram.contracts.ai.relay import RelayGatewayError, TimeWindow
45from lexigram.primitives import clock
46from lexigram.result import Err, Ok, Result
47from lexigram.ui import Card, el, render_to_string
49if TYPE_CHECKING:
50 from lexigram.contracts.admin.errors import AdminError
52__all__ = ["RelayGatewayAdminContributor"]
54_WIDGETS: tuple[DashboardWidgetDefinition, ...] = (
55 DashboardWidgetDefinition(
56 name="channel_health",
57 title="Channel Health",
58 contributor="relay-gateway",
59 render_endpoint="/admin/relay-gateway/widgets/channel_health",
60 size=WidgetSize.LARGE,
61 category=WidgetCategory.HEALTH,
62 refresh_interval_seconds=15,
63 permission=PERMISSION_READ,
64 icon="heart-pulse",
65 description="Per-channel health snapshots from probe status.",
66 ),
67 DashboardWidgetDefinition(
68 name="route_activity",
69 title="Route Activity",
70 contributor="relay-gateway",
71 render_endpoint="/admin/relay-gateway/widgets/route_activity",
72 size=WidgetSize.LARGE,
73 category=WidgetCategory.METRICS,
74 refresh_interval_seconds=30,
75 permission=PERMISSION_READ,
76 icon="activity",
77 description="Per-route conversion and failure metrics.",
78 ),
79 DashboardWidgetDefinition(
80 name="active_streams",
81 title="Active Streams",
82 contributor="relay-gateway",
83 render_endpoint="/admin/relay-gateway/widgets/active_streams",
84 size=WidgetSize.MEDIUM,
85 category=WidgetCategory.METRICS,
86 refresh_interval_seconds=10,
87 permission=PERMISSION_READ,
88 icon="radio",
89 description="In-flight upstream streams, oldest first.",
90 ),
91)
93_NAV_ITEMS: tuple[NavigationContribution, ...] = (
94 NavigationContribution(
95 label="Relay Gateway",
96 url="/admin/relay-gateway/overview",
97 icon="shuffle",
98 group="ai",
99 order=30,
100 permission=PERMISSION_READ,
101 children=(
102 NavigationContribution(
103 label="Overview",
104 url="/admin/relay-gateway/overview",
105 icon="gauge",
106 group="ai",
107 order=10,
108 permission=PERMISSION_READ,
109 ),
110 NavigationContribution(
111 label="Routes",
112 url="/admin/relay-gateway/routes",
113 icon="activity",
114 group="ai",
115 order=20,
116 permission=PERMISSION_READ,
117 ),
118 NavigationContribution(
119 label="Streams",
120 url="/admin/relay-gateway/streams",
121 icon="radio",
122 group="ai",
123 order=30,
124 permission=PERMISSION_READ,
125 ),
126 NavigationContribution(
127 label="Settings",
128 url="/admin/relay-gateway/settings",
129 icon="settings",
130 group="ai",
131 order=40,
132 permission=PERMISSION_READ,
133 ),
134 ),
135 ),
136)
138_HEALTH_DEFS: tuple[AdminHealthDefinition, ...] = (
139 AdminHealthDefinition(
140 name="relay.channels",
141 contributor="relay-gateway",
142 component="Relay Channels",
143 check_endpoint="/admin/relay-gateway/health/channels",
144 description="Aggregates per-channel probe status.",
145 ),
146)
148_ACTIONS: tuple[AdminActionDefinition, ...] = (
149 AdminActionDefinition(
150 name="set_channel_state",
151 title="Set Channel State",
152 contributor="relay-gateway",
153 handler="lexigram.ai.relay.gateway.admin.actions:set_channel_state",
154 icon="toggle-right",
155 confirmation_message="Enable or drain this channel for new requests?",
156 category="operations",
157 permission=PERMISSION_CHANNEL_CONTROL,
158 parameter_schema=ActionParameterSchema(
159 fields=(
160 ActionParameterField(
161 name="channel",
162 type_hint="str",
163 required=True,
164 description="Channel name from the gateway channel table.",
165 ),
166 ActionParameterField(
167 name="enabled",
168 type_hint="bool",
169 required=True,
170 description="Whether the channel accepts new requests.",
171 ),
172 ),
173 description="Enable or drain a gateway channel.",
174 ),
175 ),
176 AdminActionDefinition(
177 name="force_cancel_stream",
178 title="Force Cancel Stream",
179 contributor="relay-gateway",
180 handler="lexigram.ai.relay.gateway.admin.actions:force_cancel_stream",
181 icon="x-circle",
182 confirmation_message="This terminates the upstream request immediately.",
183 destructive=True,
184 category="operations",
185 permission=PERMISSION_STREAM_CONTROL,
186 parameter_schema=ActionParameterSchema(
187 fields=(
188 ActionParameterField(
189 name="stream_id",
190 type_hint="str",
191 required=True,
192 description="Identifier of the stream session to cancel.",
193 ),
194 ),
195 ),
196 ),
197)
200class RelayGatewayAdminContributor(BaseAdminContributor):
201 """Admin contributor for the relay gateway.
203 Provides channel-health and route-activity widgets, management
204 pages for overview, routes, streams, and settings, and two
205 permissioned operations actions. Registered via the
206 ``lexigram.admin.contributors`` entry point.
207 """
209 name = "relay-gateway"
210 display_name = "Relay Gateway"
211 group = "ai"
212 icon = "shuffle"
213 priority = 57
215 required_permissions = frozenset(
216 {PERMISSION_READ, PERMISSION_CHANNEL_CONTROL, PERMISSION_POLICY_CONTROL}
217 )
219 def __init__(self) -> None:
220 self._container: Any = None
221 self._health: RelayHealthService | None = None
222 self._metrics: RelayMetricsService | None = None
223 self._controls: RelayControlsService | None = None
225 async def on_admin_boot(self, container: Any) -> None:
226 """Resolve relay services from the DI container.
228 Args:
229 container: The DI container resolver.
230 """
231 self._container = container
232 try:
233 self._health = await container.resolve(RelayHealthService)
234 except Exception:
235 self._health = None
236 try:
237 self._metrics = await container.resolve(RelayMetricsService)
238 except Exception:
239 self._metrics = None
240 try:
241 self._controls = await container.resolve(RelayControlsService)
242 except Exception:
243 self._controls = None
245 def get_dashboard_widgets(self) -> Sequence[DashboardWidgetDefinition]:
246 return list(_WIDGETS)
248 def get_navigation_items(self) -> Sequence[NavigationContribution]:
249 return list(_NAV_ITEMS)
251 def get_health_definitions(self) -> Sequence[AdminHealthDefinition]:
252 return list(_HEALTH_DEFS)
254 def get_actions(self) -> Sequence[AdminActionDefinition]:
255 return list(_ACTIONS)
257 async def execute_action(
258 self,
259 action_name: str,
260 params: dict[str, object],
261 ) -> object:
262 """Dispatch an action to its lazy-loaded handler.
264 Handlers run with the container captured at boot; a container is
265 required. Every handler performs server-side parameter
266 validation before invoking the control service.
268 Args:
269 action_name: Name of the action to execute.
270 params: Parameters forwarded to the action handler.
272 Returns:
273 The handler's result mapping.
275 Raises:
276 LookupError: Unknown action name.
277 RuntimeError: Contributor booted without a container.
278 """
279 from importlib import import_module as _import_module
281 registry = {action.name: action for action in _ACTIONS}
282 definition = registry.get(action_name)
283 if definition is None:
284 raise LookupError(f"unknown relay-gateway action {action_name!r}")
285 module_path, _, handler_name = definition.handler.partition(":")
286 module = _import_module(module_path)
287 handler = getattr(module, handler_name)
288 if self._container is None:
289 raise RuntimeError("contributor has no container; on_admin_boot required")
290 return await handler(self._container, **params)
292 def get_routes(self) -> Sequence[AdminRouteSpec]:
293 """Return the widget and health render endpoints.
295 Returns:
296 One route spec per dashboard widget and health check.
297 """
298 routes: list[AdminRouteSpec] = [
299 AdminRouteSpec(
300 path=cast("str", w.render_endpoint),
301 method="GET",
302 handler=_render_for_widget,
303 name=f"widgets.{w.name}",
304 permissions=frozenset({PERMISSION_READ}),
305 )
306 for w in _WIDGETS
307 ]
308 routes += [
309 AdminRouteSpec(
310 path=cast("str", h.check_endpoint),
311 method="GET",
312 handler=_render_for_health,
313 name=f"health.{h.name}",
314 permissions=frozenset({PERMISSION_READ}),
315 )
316 for h in _HEALTH_DEFS
317 ]
318 return routes
320 def get_management_pages(self) -> Sequence[ManagementPageDefinition]:
321 return [
322 ManagementPageDefinition(
323 name="relay_gateway_overview",
324 title="Relay Gateway Overview",
325 contributor="relay-gateway",
326 route_path="/relay-gateway/overview",
327 handler="lexigram.ai.relay.gateway.admin.pages:RelayGatewayOverviewPage",
328 category=PageCategory.AI,
329 icon="shuffle",
330 description="Gateway channels, converters, and active streams",
331 order=10,
332 ),
333 ManagementPageDefinition(
334 name="relay_gateway_routes",
335 title="Relay Routes",
336 contributor="relay-gateway",
337 route_path="/relay-gateway/routes",
338 handler="lexigram.ai.relay.gateway.admin.pages:RelayGatewayRoutesPage",
339 category=PageCategory.AI,
340 icon="activity",
341 description="Per-route conversion and failure metrics",
342 order=20,
343 ),
344 ManagementPageDefinition(
345 name="relay_gateway_streams",
346 title="Relay Streams",
347 contributor="relay-gateway",
348 route_path="/relay-gateway/streams",
349 handler="lexigram.ai.relay.gateway.admin.pages:RelayGatewayStreamsPage",
350 category=PageCategory.AI,
351 icon="radio",
352 description="In-flight upstream streams",
353 order=30,
354 ),
355 ManagementPageDefinition(
356 name="relay_gateway_settings",
357 title="Relay Settings",
358 contributor="relay-gateway",
359 route_path="/relay-gateway/settings",
360 handler="lexigram.ai.relay.gateway.admin.pages:RelayGatewaySettingsPage",
361 category=PageCategory.AI,
362 icon="settings",
363 description="Runtime routing policy",
364 order=40,
365 ),
366 ]
368 async def render_widget(
369 self,
370 widget_name: str,
371 params: WidgetParams,
372 resolver: Any = None,
373 ) -> Result[WidgetViewModel, AdminError]:
374 """Render a named widget using handler registry dispatch.
376 Args:
377 widget_name: Name of the widget to render.
378 params: Typed widget parameters.
379 resolver: Optional container override; unused, services were
380 resolved at boot.
382 Returns:
383 Ok(WidgetViewModel) with rendered HTML on success;
384 Err(WidgetNotFoundError) for unknown widget names.
385 """
386 renderers = {
387 "channel_health": self._render_channel_health,
388 "route_activity": self._render_route_activity,
389 "active_streams": self._render_active_streams,
390 }
391 renderer = renderers.get(widget_name)
392 if renderer is None:
393 not_found: Result[WidgetViewModel, AdminError] = cast(
394 "Result[WidgetViewModel, AdminError]",
395 Err(WidgetNotFoundError("relay-gateway", widget_name)),
396 )
397 return not_found
398 html = await renderer(params)
399 return Ok(WidgetViewModel(body=html))
401 async def render_health_check(
402 self,
403 check_name: str,
404 ) -> Result[str, AdminError]:
405 """Render the aggregate channel health check.
407 Args:
408 check_name: Name of the health check; only
409 ``relay.channels`` is served.
411 Returns:
412 Ok(html) with the aggregate snapshot; Err when the check is
413 unknown or the health service is unavailable.
414 """
415 if check_name != "relay.channels":
416 not_found: Result[str, AdminError] = cast(
417 "Result[str, AdminError]",
418 Err(HealthCheckNotFoundError("relay-gateway", check_name)),
419 )
420 return not_found
421 if self._health is None:
422 unavailable: Result[str, AdminError] = cast(
423 "Result[str, AdminError]",
424 Err(HealthCheckNotFoundError("relay-gateway", check_name)),
425 )
426 return unavailable
427 snapshots = await self._health.channel_health()
428 parts = [f"{snap.channel}: {snap.status}" for snap in snapshots]
429 body = ", ".join(parts) if parts else "no channels configured"
430 return Ok(body)
432 async def _render_channel_health(self, params: WidgetParams) -> str:
433 """Render per-channel health as a status table."""
434 if self._health is None:
435 return render_to_string(
436 _unavailable("Channel health requires RelayHealthService.")
437 )
438 snapshots = await self._health.channel_health()
439 rows = []
440 for snap in snapshots:
441 badge = _status_badge(snap.status)
442 rows.append(
443 el(
444 "tr",
445 el("td", snap.channel, class_="py-1.5 pr-3 font-medium"),
446 el("td", snap.target.value, class_="py-1.5 pr-3"),
447 el("td", badge, class_="py-1.5 pr-3"),
448 el("td", str(snap.model_count), class_="py-1.5 pr-3"),
449 el(
450 "td",
451 _ms(snap.latency_ms_p50),
452 class_="py-1.5 pr-3",
453 ),
454 el("td", _details(snap.detail_code), class_="py-1.5"),
455 )
456 )
457 table = _table(
458 ["Channel", "Target", "Status", "Models", "P50", "Detail"],
459 rows,
460 )
461 return render_to_string(
462 Card(title="Channel Health", content=render_to_string(table))
463 )
465 async def _render_route_activity(self, params: WidgetParams) -> str:
466 """Render routed metrics for the widget window."""
467 if self._metrics is None:
468 return render_to_string(
469 _unavailable("Route metrics service is unavailable.")
470 )
471 from datetime import timedelta
473 window = TimeWindow(
474 start=clock.now() - timedelta(minutes=params.time_window_minutes),
475 end=clock.now(),
476 )
477 rows = []
478 try:
479 routes = await self._metrics.route_metrics(window)
480 except RelayGatewayError as exc:
481 return render_to_string(_unavailable(exc.message))
482 for route in routes:
483 badges = (
484 el("span", "conv-loss", class_="mr-1"),
485 el("span", "stream-fail", class_="mr-1"),
486 )
487 rows.append(
488 el(
489 "tr",
490 el("td", route.source.value, class_="py-1.5 pr-3 font-medium"),
491 el("td", route.target.value, class_="py-1.5 pr-3"),
492 el("td", str(route.request_count), class_="py-1.5 pr-3"),
493 el(
494 "td",
495 str(route.unsupported_count),
496 class_="py-1.5 pr-3",
497 ),
498 el(
499 "td",
500 str(route.stream_failure_count),
501 class_="py-1.5 pr-3",
502 ),
503 el("td", _quality_badge(route.quality.value), class_="py-1.5"),
504 )
505 )
506 empty = el(
507 "p",
508 "No route activity in this window.",
509 class_="text-sm text-[var(--muted-foreground)] py-4",
510 )
511 table = _table(
512 [
513 "Route",
514 "Target",
515 "Requests",
516 "Unsupported",
517 "Stream Failures",
518 "Quality",
519 ],
520 rows,
521 )
522 body = render_to_string(table) if rows else render_to_string(empty)
523 return render_to_string(Card(title="Route Activity", content=body))
525 async def _render_active_streams(self, params: WidgetParams) -> str:
526 """Render the in-flight stream registry."""
527 if self._controls is None:
528 return render_to_string(_unavailable("Controls service is unavailable."))
529 streams = self._controls.active_streams()
530 if not streams:
531 return render_to_string(
532 Card(
533 title="Active Streams",
534 content=render_to_string(
535 el(
536 "p",
537 "No streams in flight.",
538 class_="text-sm text-[var(--muted-foreground)] py-4",
539 )
540 ),
541 )
542 )
543 rows = [
544 el(
545 "tr",
546 el("td", s.stream_id, class_="py-1.5 pr-3 font-mono text-xs"),
547 el("td", s.channel, class_="py-1.5 pr-3"),
548 el("td", s.model, class_="py-1.5 pr-3"),
549 el("td", _iso(s.started_at), class_="py-1.5"),
550 )
551 for s in streams
552 ]
553 table = _table(["Stream", "Channel", "Model", "Started"], rows)
554 return render_to_string(
555 Card(title="Active Streams", content=render_to_string(table))
556 )
559async def _render_for_widget(request: object) -> str: # noqa: ARG001
560 """Placeholder route handler for widget endpoints.
562 Widget content is rendered by the admin dashboard through
563 ``render_widget``; the route spec only proves registration.
564 """
565 return ""
568async def _render_for_health(request: object) -> str: # noqa: ARG001
569 """Placeholder route handler for health check endpoints.
571 Health content is rendered by the admin dashboard through
572 ``render_health_check``; the route spec only proves registration.
573 """
574 return ""
577def _unavailable(message: str) -> Any:
578 """Render an explicit unavailable dependency message."""
579 return Card(
580 title="Unavailable",
581 content=render_to_string(
582 el(
583 "p",
584 message,
585 class_="text-sm text-[var(--muted-foreground)] py-4",
586 )
587 ),
588 )
591def _table(headers: list[str], rows: list[Any]) -> Any:
592 """Build a styled table element."""
593 return el(
594 "table",
595 el(
596 "thead",
597 el(
598 "tr",
599 *[
600 el(
601 "th",
602 h,
603 class_=(
604 "text-left text-xs font-semibold "
605 "text-[var(--muted-foreground)] uppercase tracking-wider "
606 "pb-1 pr-3"
607 ),
608 scope_="col",
609 )
610 for h in headers
611 ],
612 ),
613 ),
614 el("tbody", *rows, class_="divide-y divide-[var(--border)]"),
615 class_="w-full",
616 )
619def _status_badge(status: str) -> Any:
620 """Render a status badge for a channel health snapshot."""
621 colors = {
622 "healthy": "bg-green-100 text-green-700",
623 "degraded": "bg-yellow-100 text-yellow-700",
624 "unavailable": "bg-gray-100 text-gray-500",
625 "failed": "bg-red-100 text-red-700",
626 }
627 return el(
628 "span",
629 status,
630 class_=f"inline-block px-2 py-0.5 rounded text-xs font-medium {colors.get(status, 'bg-gray-100 text-gray-500')}",
631 )
634def _quality_badge(quality: str) -> Any:
635 """Render a conversion quality badge."""
636 colors = {
637 "native": "bg-green-100 text-green-700",
638 "preserved": "bg-blue-100 text-blue-700",
639 "lossless": "bg-emerald-100 text-emerald-700",
640 "lossy": "bg-yellow-100 text-yellow-700",
641 "discouraged": "bg-red-100 text-red-700",
642 }
643 return el(
644 "span",
645 quality,
646 class_=f"inline-flex px-2 py-0.5 rounded text-xs font-medium {colors.get(quality, 'bg-gray-100 text-gray-500')}",
647 )
650def _ms(value: float | None) -> str:
651 """Format a latency value as milliseconds."""
652 return f"{value:.1f} ms" if value is not None else "-"
655def _iso(dt: Any) -> str:
656 """Format a datetime for display."""
657 return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else "-"
660def _details(detail_code: str | None) -> str:
661 """Humanize the machine-readable detail code."""
662 if detail_code is None:
663 return "-"
664 return detail_code.replace("_", " ")