Coverage for src / lexigram / ai / relay / gateway / operations / auto_test.py: 97%
66 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""Background auto-tester for relay gateway channel health.
3``RelayChannelAutoTester`` periodically runs one channel health sweep and
4translates probe outcomes into runtime selection state: channels that
5continuously fail are disabled at runtime, and channels that the tester
6itself took down are re-enabled once their probe comes back healthy.
7The tester never restores a channel a human admin drained through the
8permissioned actuator surface — only its own disables are tracked and
9reversed.
10"""
12from __future__ import annotations
14import asyncio
16from lexigram.ai.relay.gateway.channels import RelayChannelRegistry
17from lexigram.ai.relay.gateway.operations.health import RelayHealthService
18from lexigram.contracts.ai.relay import RelayChannelHealth
19from lexigram.logging import get_logger
21__all__ = ["RelayChannelAutoTester"]
23logger = get_logger(__name__)
25_HEALTHY_STATUS = "healthy"
26_UNHEALTHY_STATUSES = frozenset({"failed"})
27"""Channel statuses treated by the auto-tester as failing probes."""
30class RelayChannelAutoTester:
31 """Automatically disable failing channels and restore recovered ones.
33 The tester runs as its own ``asyncio.Task``, taking one health
34 snapshot per interval and turning probe outcomes into runtime
35 transitions via ``RelayChannelRegistry.set_runtime_enabled``. Its
36 own disable decisions are journaled in ``_disabled_by_tester`` so a
37 channel drained by a human through the actuator controls is never
38 silently restored.
40 Args:
41 health: The health service that produces per-channel snapshots.
42 registry: The channel registry whose runtime enable flags this
43 tester mutates.
44 interval_seconds: Whole seconds between two consecutive sweeps.
45 Must be positive.
46 """
48 def __init__(
49 self,
50 health: RelayHealthService,
51 registry: RelayChannelRegistry,
52 interval_seconds: float,
53 ) -> None:
54 """Bind the auto-tester to its health service and registry.
56 Args:
57 health: Health service whose snapshots drive the sweep.
58 registry: Channel registry receiving runtime transitions.
59 interval_seconds: Delay between sweeps, in whole seconds.
60 """
61 self._health = health
62 self._registry = registry
63 self._interval_seconds = interval_seconds
64 self._disabled_by_tester: set[str] = set()
65 self._task: asyncio.Task[None] | None = None
67 @property
68 def is_running(self) -> bool:
69 """Return whether a sweep loop is currently scheduled.
71 Returns:
72 ``True`` when ``start()`` has scheduled a task that has not
73 gone away, ``False`` otherwise.
74 """
75 return self._task is not None and not self._task.done()
77 async def start(self) -> None:
78 """Start the periodic sweep; a no-op when one is already running.
80 The first sweep runs immediately after scheduling, then the loop
81 sleeps ``interval_seconds`` between iterations.
82 """
83 if self.is_running:
84 return
85 self._task = asyncio.create_task(self._sweep_loop(), name="relay-auto-test")
87 async def stop(self) -> None:
88 """Cancel the sweep loop and await its completion.
90 Idempotent: calling stop when nothing is running is a no-op.
91 """
92 task = self._task
93 self._task = None
94 if task is None:
95 return
96 task.cancel()
97 try:
98 await task
99 except asyncio.CancelledError:
100 pass
102 async def sweep(self) -> None:
103 """Run one probe sweep and apply the resulting transitions.
105 The snapshots come from the health service as-is; the sweep does
106 not probe channels on its own. An exception raised while the
107 health service probes a channel is caught and logged per sweep,
108 and the loop continues to its next iteration.
110 Example:
111 ```python
112 await tester.sweep()
113 ```
114 """
115 try:
116 snapshots = await self._health.channel_health()
117 except Exception as exc: # noqa: BLE001
118 logger.warning(
119 "relay_gateway_channel_auto_sweep_failed",
120 error=str(exc),
121 )
122 return
123 await self._apply(list(snapshots))
125 async def _sweep_loop(self) -> None:
126 """Repeat ``sweep()`` then idle ``interval_seconds`` forever."""
127 while True:
128 try:
129 await self.sweep()
130 except Exception as exc: # noqa: BLE001
131 logger.error(
132 "relay_gateway_channel_auto_loop_failed",
133 error=str(exc),
134 )
135 await asyncio.sleep(self._interval_seconds)
137 async def _apply(self, snapshots: list[RelayChannelHealth]) -> None:
138 """Diff *snapshots* against the journal and update the registry.
140 Args:
141 snapshots: One snapshot per configured channel, in config
142 order, from the health service.
143 """
144 by_name = {snapshot.channel: snapshot for snapshot in snapshots}
145 for channel_name in list(self._disabled_by_tester):
146 snapshot = by_name.get(channel_name)
147 if snapshot is not None and snapshot.status == _HEALTHY_STATUS:
148 self._recover(channel_name)
149 for snapshot in snapshots:
150 if (
151 snapshot.status in _UNHEALTHY_STATUSES
152 and snapshot.channel not in self._disabled_by_tester
153 ):
154 self._disable(snapshot.channel)
156 def _disable(self, channel_name: str) -> None:
157 """Take *channel_name* out of service and journal the decision.
159 Args:
160 channel_name: The channel to disable at runtime.
161 """
162 self._registry.set_runtime_enabled(channel_name, False)
163 self._disabled_by_tester.add(channel_name)
164 logger.info(
165 "relay_gateway_channel_auto_disabled",
166 channel=channel_name,
167 reason="probe_failed",
168 )
170 def _recover(self, channel_name: str) -> None:
171 """Restore *channel_name* at runtime; the journal entry is removed.
173 Args:
174 channel_name: The channel previously disabled by this tester.
175 """
176 self._registry.set_runtime_enabled(channel_name, True)
177 self._disabled_by_tester.discard(channel_name)
178 logger.info(
179 "relay_gateway_channel_auto_reenabled",
180 channel=channel_name,
181 reason="probe_recovered",
182 )