Coverage for src / lexigram / ai / relay / gateway / admin / actions.py: 74%

1"""Admin action handlers for the relay gateway contributor. 

2 

3Each handler accepts ``(container, **params)`` per the admin action 

4contract, validates every parameter server-side, and returns a result 

5dict describing the audited outcome (or the validation/concurrency 

6failure). The ``container`` resolves ``RelayControlsService`` lazily. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import Any 

12 

13from lexigram.ai.relay.gateway.operations.controls import RelayControlsService 

14from lexigram.logging import get_logger 

15 

16logger = get_logger(__name__) 

17 

18_TRUE_TOKENS = {"true", "1", "yes", "on"} 

19_FALSE_TOKENS = {"false", "0", "no", "off"} 

20 

21 

22def _coerce_bool(raw: object, default: bool | None = None) -> bool | None: 

23 """Coerce a bool-like value to a strict boolean. 

24 

25 Args: 

26 raw: Value to coerce; bools, strings, and numbers accepted. 

27 default: Value returned when ``raw`` is ``None``. ``None`` 

28 makes coercion failure return ``None`` as well. 

29 

30 Returns: 

31 The coerced boolean, or ``None`` when the value is not a 

32 recognizable boolean token. 

33 """ 

34 if isinstance(raw, bool): 

35 return raw 

36 if raw is None: 

37 return default 

38 if isinstance(raw, int) and raw in (0, 1): 

39 return bool(raw) 

40 if isinstance(raw, str): 

41 token = raw.strip().lower() 

42 if token in _TRUE_TOKENS: 

43 return True 

44 if token in _FALSE_TOKENS: 

45 return False 

46 return None 

47 

48 

49async def set_channel_state(container: Any, **params: object) -> dict[str, object]: 

50 """Enable or drain a gateway channel. 

51 

52 Args: 

53 container: Container resolver exposing ``RelayControlsService``. 

54 **params: Action parameters; ``channel`` (str) and ``enabled`` 

55 (bool-like) are required, ``actor_id`` (str) defaults to 

56 ``"admin"``. 

57 

58 Returns: 

59 Mapping describing the outcome: ``ok`` boolean, ``message``, 

60 ``echo`` (validated params), and ``raised`` for failures. 

61 """ 

62 channel = params.get("channel") 

63 if not isinstance(channel, str) or not channel.strip(): 

64 return {"ok": False, "message": "channel is required", "echo": {}} 

65 enabled = _coerce_bool(params.get("enabled")) 

66 if enabled is None: 

67 return {"ok": False, "message": "enabled must be a boolean", "echo": {}} 

68 actor_id = str(params.get("actor_id") or "admin") 

69 echo = {"channel": channel, "enabled": enabled, "actor_id": actor_id} 

70 try: 

71 controls = await container.resolve(RelayControlsService) 

72 await controls.set_channel_state( 

73 channel=channel, enabled=enabled, actor_id=actor_id 

74 ) 

75 except ValueError as exc: 

76 logger.warning("relay_admin.set_channel_state.rejected", error=str(exc)) 

77 return {"ok": False, "message": str(exc), "echo": echo} 

78 except Exception as exc: # noqa: BLE001 

79 logger.error("relay_admin.set_channel_state.failed", error=str(exc)) 

80 return {"ok": False, "message": str(exc), "echo": echo} 

81 return { 

82 "ok": True, 

83 "message": f"channel {channel!r} {'enabled' if enabled else 'drained'}", 

84 "echo": echo, 

85 } 

86 

87 

88async def force_cancel_stream(container: Any, **params: object) -> dict[str, object]: 

89 """Force-cancel an in-flight upstream stream. 

90 

91 Args: 

92 container: Container resolver exposing ``RelayControlsService``. 

93 **params: Action parameters; ``stream_id`` (str) is required, 

94 ``actor_id`` (str) defaults to ``"admin"``. 

95 

96 Returns: 

97 Mapping describing the outcome: ``ok`` boolean, ``message``, 

98 and ``echo`` with the validated stream identifier. 

99 """ 

100 stream_id = params.get("stream_id") 

101 if not isinstance(stream_id, str) or not stream_id.strip(): 

102 return {"ok": False, "message": "stream_id is required", "echo": {}} 

103 actor_id = str(params.get("actor_id") or "admin") 

104 echo = {"stream_id": stream_id, "actor_id": actor_id} 

105 try: 

106 controls = await container.resolve(RelayControlsService) 

107 await controls.force_cancel_stream( 

108 stream_id=stream_id, 

109 actor_id=actor_id, 

110 ) 

111 except ValueError as exc: 

112 logger.warning("relay_admin.force_cancel_stream.rejected", error=str(exc)) 

113 return {"ok": False, "message": str(exc), "echo": echo} 

114 except Exception as exc: # noqa: BLE001 

115 logger.error("relay_admin.force_cancel_stream.failed", error=str(exc)) 

116 return {"ok": False, "message": str(exc), "echo": echo} 

117 return {"ok": True, "message": f"stream {stream_id!r} cancelled", "echo": echo} 

118 

119 

120__all__ = ["force_cancel_stream", "set_channel_state"]