Coverage for  / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / infra / telemetry.py: 100%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Telemetry adapter interfaces and default implementations.""" 

5 

6from __future__ import annotations 

7 

8from enum import Enum 

9from typing import Any 

10 

11 

12class TelemetryEvent(str, Enum): 

13 """Standardized telemetry event names.""" 

14 

15 CLI_STARTED = "cli_started" 

16 CLI_ERROR = "cli_error" 

17 CLI_INTERRUPTED = "cli_interrupted" 

18 CLI_SYSTEM_EXIT = "cli_system_exit" 

19 CLI_UNEXPECTED_ERROR = "cli_unexpected_error" 

20 CLI_SHUTDOWN_FAILED = "cli_shutdown_failed" 

21 ENGINE_INITIALIZED = "engine_initialized" 

22 ENGINE_SHUTDOWN = "engine_shutdown" 

23 PLUGINS_LIST_COMMAND = "cmd/plugins/list" 

24 PLUGINS_LIST_COMMAND_FAILED = "cmd/err/plugins/list" 

25 PLUGINS_INFO_COMMAND = "cmd/plugins/info" 

26 PLUGINS_INFO_COMMAND_FAILED = "cmd/err/plugins/info" 

27 PLUGINS_INFO_NOT_FOUND = "cmd/err/plugins/info/not_found" 

28 PLUGINS_INSTALL_COMMAND = "cmd/plugins/install" 

29 PLUGINS_INSTALL_COMMAND_FAILED = "cmd/err/plugins/install" 

30 PLUGINS_UNINSTALL_COMMAND = "cmd/plugins/uninstall" 

31 PLUGINS_UNINSTALL_COMMAND_FAILED = "cmd/err/plugins/uninstall" 

32 PLUGINS_UNINSTALL_NOT_FOUND = "cmd/err/plugins/uninstall/not_found" 

33 PLUGINS_CHECK_COMMAND = "cmd/plugins/check" 

34 PLUGINS_CHECK_COMMAND_FAILED = "cmd/err/plugins/check" 

35 PLUGINS_CHECK_NOT_FOUND = "cmd/err/plugins/check/not_found" 

36 PLUGINS_SCAFFOLD_COMMAND = "cmd/plugins/scaffold" 

37 PLUGINS_SCAFFOLD_COMMAND_FAILED = "cmd/err/plugins/scaffold" 

38 PLUGINS_SCAFFOLD_DIR_EXISTS = "cmd/err/plugins/scaffold/dir_exists" 

39 CONFIG_COMMAND = "cmd/config" 

40 CONFIG_COMMAND_FAILED = "cmd/err/config" 

41 AUDIT_COMMAND = "cmd/audit" 

42 AUDIT_COMMAND_FAILED = "cmd/err/audit" 

43 DOCTOR_COMMAND = "cmd/doctor" 

44 DOCTOR_COMMAND_FAILED = "cmd/err/doctor" 

45 VERSION_COMMAND = "cmd/version" 

46 VERSION_COMMAND_FAILED = "cmd/err/version" 

47 STATUS_COMMAND = "cmd/status" 

48 STATUS_COMMAND_FAILED = "cmd/err/status" 

49 SLEEP_COMMAND = "cmd/test/sleep" 

50 SLEEP_COMMAND_FAILED = "cmd/err/test/sleep" 

51 HISTORY_COMMAND = "cmd/history" 

52 HISTORY_COMMAND_FAILED = "cmd/err/history" 

53 REPL_COMMAND = "cmd/repl" 

54 REPL_EXIT = "cmd/repl/exit" 

55 REPL_COMMAND_NOT_FOUND = "cmd/err/repl/not_found" 

56 DEV_COMMAND_EXECUTED = "cmd/dev" 

57 DEV_COMMAND_FAILED = "cmd/err/dev" 

58 MEMORY_COMMAND_EXECUTED = "cmd/memory" 

59 MEMORY_COMMAND_FAILED = "cmd/err/memory" 

60 HELP_COMMAND = "cmd/help" 

61 HELP_COMMAND_FAILED = "cmd/err/help" 

62 PLUGIN_STARTED = "plugin_started" 

63 PLUGIN_SHUTDOWN = "plugin_shutdown" 

64 PLUGIN_INSTALLED = "plugin_installed" 

65 PLUGIN_LOADED = "plugin_loaded" 

66 PLUGIN_CLI_REGISTERED = "plugin_cli_registered" 

67 PLUGIN_LOAD_FAILED = "plugin_load_failed" 

68 

69 

70class NoopTelemetry: 

71 """No-op telemetry adapter.""" 

72 

73 def event(self, name: str | TelemetryEvent, payload: dict[str, Any]) -> None: 

74 """Record a telemetry event and do nothing.""" 

75 return None 

76 

77 def flush(self) -> None: 

78 """Flush buffered events (no-op).""" 

79 return None 

80 

81 def enable(self) -> None: 

82 """Enable telemetry (no-op).""" 

83 return None 

84 

85 

86class LoggingTelemetry: 

87 """Telemetry adapter that logs events via an observability sink.""" 

88 

89 def __init__(self, observability: Any) -> None: 

90 """Initialize with an observability sink.""" 

91 self._observability = observability 

92 self._buffer: list[tuple[str, dict[str, Any]]] = [] 

93 

94 def event(self, name: str | TelemetryEvent, payload: dict[str, Any]) -> None: 

95 """Record and log a telemetry event.""" 

96 event = name.value if isinstance(name, TelemetryEvent) else str(name) 

97 self._buffer.append((event, payload)) 

98 self._observability.log( 

99 "debug", f"telemetry:{event}", extra={"event": event, **payload} 

100 ) 

101 

102 def flush(self) -> None: 

103 """Clear buffered telemetry events.""" 

104 self._buffer.clear() 

105 

106 def enable(self) -> None: 

107 """Enable telemetry (no-op for logging adapter).""" 

108 return None 

109 

110 

111__all__ = [ 

112 "TelemetryEvent", 

113 "NoopTelemetry", 

114 "LoggingTelemetry", 

115]