Coverage for  / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / cli / repl / ui.py: 92%

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""UI helpers for the REPL prompt and interactive loop.""" 

5 

6from __future__ import annotations 

7 

8from contextlib import suppress 

9import os 

10import shlex 

11import signal 

12import sys 

13from types import FrameType 

14 

15from prompt_toolkit.formatted_text import ANSI 

16from prompt_toolkit.key_binding.key_processor import KeyPressEvent 

17 

18from bijux_cli.cli.core.constants import ( 

19 ENV_BIN, 

20 ENV_HISTORY_FILE, 

21 ENV_NO_COLOR, 

22 ENV_TEST_MODE, 

23) 

24from bijux_cli.cli.repl.completion import CommandCompleter 

25from bijux_cli.cli.repl.execution import _invoke 

26from bijux_cli.cli.repl.parsing import _known_commands, _split_segments, _suggest 

27from bijux_cli.core.enums import ExitCode, OutputFormat 

28from bijux_cli.core.exit_policy import ExitIntent, ExitIntentError 

29 

30 

31def _exit_on_signal(_signum: int, _frame: FrameType | None = None) -> None: 

32 """Exits the process cleanly when a watched signal is received.""" 

33 raise ExitIntentError( 

34 ExitIntent( 

35 code=ExitCode.SUCCESS, 

36 stream=None, 

37 payload=None, 

38 fmt=OutputFormat.JSON, 

39 pretty=False, 

40 show_traceback=False, 

41 ) 

42 ) 

43 

44 

45def get_prompt() -> str | ANSI: 

46 """Returns the REPL prompt string.""" 

47 if os.environ.get(ENV_TEST_MODE) == "1" or os.environ.get(ENV_NO_COLOR) == "1": 

48 return "bijux> " 

49 return ANSI("\x1b[36mbijux> \x1b[0m") 

50 

51 

52async def _run_interactive() -> None: 

53 """Starts the interactive REPL session.""" 

54 from importlib import import_module 

55 from pathlib import Path 

56 import subprocess # nosec B404 

57 

58 from prompt_toolkit import PromptSession 

59 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory 

60 from prompt_toolkit.buffer import Buffer 

61 from prompt_toolkit.completion import Completion 

62 from prompt_toolkit.history import FileHistory 

63 from prompt_toolkit.key_binding import KeyBindings 

64 from prompt_toolkit.output import ColorDepth 

65 

66 cli_mod = import_module("bijux_cli.cli.root") 

67 build_app = getattr(cli_mod, "build_app", None) 

68 app = build_app() if callable(build_app) else cli_mod.app 

69 

70 kb = KeyBindings() 

71 

72 @kb.add("tab") 

73 def _(event: KeyPressEvent) -> None: 

74 """Handles Tab key presses for completion.""" 

75 buf = event.app.current_buffer 

76 if buf.complete_state: 

77 buf.complete_next() 

78 else: 

79 buf.start_completion(select_first=True) 

80 

81 @kb.add("enter") 

82 def _(event: KeyPressEvent) -> None: 

83 """Handles Enter key presses to submit or accept completions.""" 

84 buf: Buffer = event.app.current_buffer 

85 state = buf.complete_state 

86 if state: 86 ↛ 92line 86 didn't jump to line 92 because the condition on line 86 was always true

87 comp: Completion | None = state.current_completion 

88 if comp: 88 ↛ 90line 88 didn't jump to line 90 because the condition on line 88 was always true

89 buf.apply_completion(comp) 

90 buf.complete_state = None 

91 else: 

92 buf.validate_and_handle() 

93 

94 history_file = os.environ.get( 

95 ENV_HISTORY_FILE, 

96 str(Path.home() / ".bijux" / ".repl_history"), 

97 ) 

98 

99 session: PromptSession[str] = PromptSession( 

100 get_prompt(), 

101 history=FileHistory(history_file), 

102 completer=CommandCompleter(app), 

103 auto_suggest=AutoSuggestFromHistory(), 

104 color_depth=ColorDepth.DEPTH_1_BIT, 

105 enable_history_search=True, 

106 complete_while_typing=False, 

107 key_bindings=kb, 

108 ) 

109 

110 cli_bin = os.environ.get(ENV_BIN) or sys.argv[0] 

111 

112 while True: 

113 try: 

114 line = await session.prompt_async() 

115 except (EOFError, KeyboardInterrupt): 

116 print("\nExiting REPL.") 

117 return 

118 

119 for seg in _split_segments(line): 

120 lower = seg.lower() 

121 if lower in ("exit", "quit"): 

122 print("Exiting REPL.") 

123 return 

124 if not seg.strip() or seg.startswith("#"): 

125 continue 

126 

127 try: 

128 tokens = shlex.split(seg) 

129 except ValueError: 

130 continue 

131 

132 head = tokens[0] 

133 if seg == "docs": 

134 print("Available topics: ...") 

135 continue 

136 if seg.startswith("docs "): 

137 print(seg.split(None, 1)[1]) 

138 continue 

139 if seg == "memory list": 

140 subprocess.run( # noqa: S603 # nosec B603 

141 [cli_bin, *tokens], env=os.environ 

142 ) 

143 continue 

144 

145 if head not in _known_commands(): 

146 hint = _suggest(head) 

147 msg = f"No such command '{head}'." 

148 if hint: 

149 msg += hint 

150 print(msg, file=sys.stderr) 

151 continue 

152 

153 _invoke(tokens, repl_quiet=False) 

154 

155 

156def register_signal_handlers() -> None: 

157 """Register REPL signal handlers for a clean exit.""" 

158 for sig in ( 

159 signal.SIGINT, 

160 signal.SIGTERM, 

161 signal.SIGHUP, 

162 signal.SIGQUIT, 

163 signal.SIGUSR1, 

164 ): 

165 with suppress(Exception): 

166 signal.signal(sig, _exit_on_signal)