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. 

2 

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""" 

9 

10from __future__ import annotations 

11 

12from collections.abc import Sequence 

13from typing import TYPE_CHECKING, Any, cast 

14 

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 

48 

49if TYPE_CHECKING: 

50 from lexigram.contracts.admin.errors import AdminError 

51 

52__all__ = ["RelayGatewayAdminContributor"] 

53 

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) 

92 

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) 

137 

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) 

147 

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) 

198 

199 

200class RelayGatewayAdminContributor(BaseAdminContributor): 

201 """Admin contributor for the relay gateway. 

202 

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 """ 

208 

209 name = "relay-gateway" 

210 display_name = "Relay Gateway" 

211 group = "ai" 

212 icon = "shuffle" 

213 priority = 57 

214 

215 required_permissions = frozenset( 

216 {PERMISSION_READ, PERMISSION_CHANNEL_CONTROL, PERMISSION_POLICY_CONTROL} 

217 ) 

218 

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 

224 

225 async def on_admin_boot(self, container: Any) -> None: 

226 """Resolve relay services from the DI container. 

227 

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 

244 

245 def get_dashboard_widgets(self) -> Sequence[DashboardWidgetDefinition]: 

246 return list(_WIDGETS) 

247 

248 def get_navigation_items(self) -> Sequence[NavigationContribution]: 

249 return list(_NAV_ITEMS) 

250 

251 def get_health_definitions(self) -> Sequence[AdminHealthDefinition]: 

252 return list(_HEALTH_DEFS) 

253 

254 def get_actions(self) -> Sequence[AdminActionDefinition]: 

255 return list(_ACTIONS) 

256 

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. 

263 

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. 

267 

268 Args: 

269 action_name: Name of the action to execute. 

270 params: Parameters forwarded to the action handler. 

271 

272 Returns: 

273 The handler's result mapping. 

274 

275 Raises: 

276 LookupError: Unknown action name. 

277 RuntimeError: Contributor booted without a container. 

278 """ 

279 from importlib import import_module as _import_module 

280 

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) 

291 

292 def get_routes(self) -> Sequence[AdminRouteSpec]: 

293 """Return the widget and health render endpoints. 

294 

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 

319 

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 ] 

367 

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. 

375 

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. 

381 

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)) 

400 

401 async def render_health_check( 

402 self, 

403 check_name: str, 

404 ) -> Result[str, AdminError]: 

405 """Render the aggregate channel health check. 

406 

407 Args: 

408 check_name: Name of the health check; only 

409 ``relay.channels`` is served. 

410 

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) 

431 

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 ) 

464 

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 

472 

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)) 

524 

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 ) 

557 

558 

559async def _render_for_widget(request: object) -> str: # noqa: ARG001 

560 """Placeholder route handler for widget endpoints. 

561 

562 Widget content is rendered by the admin dashboard through 

563 ``render_widget``; the route spec only proves registration. 

564 """ 

565 return "" 

566 

567 

568async def _render_for_health(request: object) -> str: # noqa: ARG001 

569 """Placeholder route handler for health check endpoints. 

570 

571 Health content is rendered by the admin dashboard through 

572 ``render_health_check``; the route spec only proves registration. 

573 """ 

574 return "" 

575 

576 

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 ) 

589 

590 

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 ) 

617 

618 

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 ) 

632 

633 

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 ) 

648 

649 

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 "-" 

653 

654 

655def _iso(dt: Any) -> str: 

656 """Format a datetime for display.""" 

657 return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else "-" 

658 

659 

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("_", " ")