Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/commands/repl.py: 93%
314 statements
« prev ^ index » next coverage.py v7.10.4, created at 2025-08-19 23:36 +0000
1# SPDX-License-Identifier: MIT
2# Copyright © 2025 Bijan Mousavi
4"""Implements the interactive Read-Eval-Print Loop (REPL) for the Bijux CLI.
6This module provides a rich, interactive shell for executing Bijux CLI commands.
7It enhances the user experience with features like persistent command history,
8context-aware tab-completion, and a colorized prompt. Users can chain multiple
9commands on a single line using semicolons. The REPL can also operate in a
10non-interactive mode to process commands piped from stdin.
12The REPL itself operates in a human-readable format. When executing commands,
13it respects global flags like `--format` or `--quiet` for those specific
14invocations.
16Exit Codes:
17 * `0`: The REPL session was exited cleanly (e.g., via `exit`, `quit`,
18 Ctrl+D, or a caught signal).
19 * `2`: An invalid flag was provided to the `repl` command itself
20 (e.g., `--format=json`).
21"""
23from __future__ import annotations
25import asyncio
26from collections.abc import Iterator
27from contextlib import suppress
28import json
29import os
30from pathlib import Path
31import re
32import shlex
33import signal
34import sys
35from types import FrameType
36from typing import Any
38from prompt_toolkit.buffer import Buffer
39from prompt_toolkit.completion import CompleteEvent, Completer, Completion
40from prompt_toolkit.document import Document
41from prompt_toolkit.formatted_text import ANSI
42from prompt_toolkit.key_binding.key_processor import KeyPressEvent
43from rapidfuzz import process as rf_process
44import typer
46from bijux_cli.commands.utilities import emit_error_and_exit, validate_common_flags
47from bijux_cli.core.constants import (
48 HELP_DEBUG,
49 HELP_FORMAT_HELP,
50 HELP_NO_PRETTY,
51 HELP_QUIET,
52 HELP_VERBOSE,
53)
55GLOBAL_OPTS = [
56 "-q",
57 "--quiet",
58 "-v",
59 "--verbose",
60 "-f",
61 "--format",
62 "--pretty",
63 "--no-pretty",
64 "-d",
65 "--debug",
66 "-h",
67 "--help",
68]
70repl_app = typer.Typer(
71 name="repl",
72 help="Starts an interactive shell with history and tab-completion.",
73 add_completion=False,
74)
76_ansi_re = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
79def _filter_control(text: str) -> str:
80 """Removes ANSI control sequences from a string.
82 Args:
83 text (str): The input string that may contain ANSI escape codes.
85 Returns:
86 str: A cleaned version of the string with ANSI codes removed.
87 """
88 return _ansi_re.sub("", text)
91_semicolon_re = re.compile(
92 r"""
93 ;
94 (?=(?:[^'"]|'[^']*'|"[^"]*")*$)
95""",
96 re.VERBOSE,
97)
100def _exit_on_signal(_signum: int, _frame: FrameType | None = None) -> None:
101 """Exits the process cleanly when a watched signal is received.
103 Args:
104 _signum (int): The signal number that triggered the handler (unused).
105 _frame (FrameType | None): The current stack frame (unused).
106 """
107 sys.exit(0)
110def _split_segments(input_text: str) -> Iterator[str]:
111 """Splits input text into individual, non-empty command segments.
113 Commands are separated by newlines or by semicolons that are not inside
114 quotes.
116 Args:
117 input_text (str): The raw input text.
119 Yields:
120 str: A cleaned, non-empty command segment.
121 """
122 clean = _filter_control(input_text)
123 for ln in clean.splitlines():
124 for part in _semicolon_re.split(ln):
125 seg = part.strip()
126 if seg:
127 yield seg
130def _known_commands() -> list[str]:
131 """Loads the list of known CLI commands.
133 It attempts to load commands from a `spec.json` file located in the
134 project structure. If the file is not found or is invalid, it returns a
135 hard-coded default list of commands.
137 Returns:
138 list[str]: A list of known command names.
139 """
140 here = Path(__file__).resolve()
141 for p in [here, *here.parents]:
142 spec = p.parent / "spec.json"
143 if spec.is_file(): 143 ↛ 141line 143 didn't jump to line 141 because the condition on line 143 was always true
144 with suppress(Exception):
145 data = json.loads(spec.read_text())
146 cmds = data.get("commands")
147 if isinstance(cmds, list): 147 ↛ 148line 147 didn't jump to line 148 because the condition on line 147 was never true
148 return cmds
149 return [
150 "audit",
151 "config",
152 "dev",
153 "docs",
154 "doctor",
155 "help",
156 "history",
157 "memory",
158 "plugins",
159 "repl",
160 "sleep",
161 "status",
162 "version",
163 ]
166def _suggest(cmd: str) -> str | None:
167 """Suggests a command based on fuzzy matching.
169 Args:
170 cmd (str): The user-provided command input to evaluate.
172 Returns:
173 str | None: A hint message (e.g., " Did you mean 'status'?") if a
174 sufficiently similar command is found, otherwise None.
175 """
176 choices = _known_commands()
177 best, score, _ = rf_process.extractOne(cmd, choices)
178 if score >= 60 and best != cmd:
179 return f" Did you mean '{best}'?"
180 return None
183def _invoke(tokens: list[str], *, repl_quiet: bool) -> int:
184 """Runs a single CLI command invocation within the REPL sandbox.
186 It prepares the arguments, invokes the command via `CliRunner`, and handles
187 the printing of output based on the quiet flags.
189 Args:
190 tokens (list[str]): The shell-split tokens for the command.
191 repl_quiet (bool): If True, all stdout/stderr from the invocation
192 is suppressed.
194 Returns:
195 int: The exit code returned by the command invocation.
196 """
197 from typer.testing import CliRunner
199 from bijux_cli.cli import app as _root_app
201 env = {**os.environ, "PS1": ""}
203 head = tokens[0] if tokens else ""
205 if head in _JSON_CMDS and not {"--pretty", "--no-pretty", "-f", "--format"} & set(
206 tokens
207 ):
208 tokens.append("--no-pretty")
210 if (
211 head == "config"
212 and len(tokens) > 1
213 and tokens[1] == "list"
214 and "--no-pretty" not in tokens
215 and "--pretty" not in tokens
216 ):
217 tokens.append("--no-pretty")
219 result = CliRunner().invoke(_root_app, tokens, env=env)
221 sub_quiet = any(t in ("-q", "--quiet") for t in tokens)
222 should_print = not repl_quiet and not sub_quiet
224 if head == "history":
225 with suppress(Exception):
226 data = json.loads(result.stdout or "{}")
227 if data.get("entries", []) == []:
228 if should_print:
229 pretty = json.dumps(data, indent=2) + "\n"
230 sys.stdout.write(pretty)
231 sys.stderr.write(result.stderr or "")
232 return result.exit_code
234 if should_print:
235 sys.stdout.write(result.stdout or "")
236 sys.stderr.write(result.stderr or "")
238 return result.exit_code
241def _run_piped(repl_quiet: bool) -> None:
242 """Processes piped input commands in non-interactive mode.
244 Reads from stdin, splits commands, and executes them sequentially. This
245 mode is activated when stdin is not a TTY or the `--quiet` flag is used.
247 Args:
248 repl_quiet (bool): If True, suppresses prompts and error messages.
250 Returns:
251 None:
252 """
253 for raw_line in sys.stdin.read().splitlines():
254 line = raw_line.rstrip()
256 if not line or line.lstrip().startswith("#"):
257 if not repl_quiet:
258 sys.stderr.write(_filter_control(str(get_prompt())) + "\n")
259 sys.stderr.flush()
260 continue
262 if line.lstrip().startswith(";"):
263 bad = line.lstrip(";").strip()
264 hint = _suggest(bad)
265 msg = f"No such command '{bad}'." + (hint or "")
266 if not repl_quiet:
267 print(msg, file=sys.stderr)
268 continue
270 for seg in _split_segments(line):
271 seg = seg.strip()
272 if not seg or seg.startswith("#"): 272 ↛ 273line 272 didn't jump to line 273 because the condition on line 272 was never true
273 continue
275 lo = seg.lower()
276 if lo in {"exit", "quit"}:
277 sys.exit(0)
279 if seg == "docs":
280 if not repl_quiet:
281 print("Available topics: …")
282 continue
283 if seg.startswith("docs "):
284 if not repl_quiet:
285 print(seg.split(None, 1)[1])
286 continue
288 if seg.startswith("-"):
289 bad = seg.lstrip("-")
290 hint = _suggest(bad)
291 msg = f"No such command '{bad}'." + (hint or "")
292 if not repl_quiet:
293 print(msg, file=sys.stderr)
294 continue
296 try:
297 tokens = shlex.split(seg)
298 except ValueError:
299 continue
300 if not tokens:
301 continue
303 head = tokens[0]
305 if head == "config":
306 sub = tokens[1:]
308 def _emit(
309 msg: str,
310 failure: str,
311 subcommand: list[str] = sub,
312 ) -> None:
313 """Emits a JSON error for a `config` subcommand.
315 Args:
316 msg (str): The error message.
317 failure (str): A short failure code (e.g., "parse").
318 subcommand (list[str]): The subcommand tokens.
319 """
320 if repl_quiet:
321 return
323 error_obj = {
324 "error": msg,
325 "code": 2,
326 "failure": failure,
327 "command": f"config {subcommand[0] if subcommand else ''}".strip(),
328 "format": "json",
329 }
330 print(json.dumps(error_obj))
332 if not sub:
333 pass
334 elif sub[0] == "set" and len(sub) == 1:
335 _emit("Missing argument: KEY=VALUE required", "missing_argument")
336 continue
337 elif sub[0] in {"get", "unset"} and len(sub) == 1: 337 ↛ 341line 337 didn't jump to line 341 because the condition on line 337 was always true
338 _emit("Missing argument: key required", "missing_argument")
339 continue
341 if head not in _known_commands():
342 hint = _suggest(head)
343 msg = f"No such command '{head}'."
344 if hint:
345 msg += hint
346 if not repl_quiet:
347 print(msg, file=sys.stderr)
348 continue
349 else:
350 _invoke(tokens, repl_quiet=repl_quiet)
352 sys.exit(0)
355def get_prompt() -> str | ANSI:
356 """Returns the REPL prompt string.
358 The prompt is styled with ANSI colors unless `NO_COLOR` or a test mode
359 environment variable is set.
361 Returns:
362 str | ANSI: The prompt string, which may include ANSI color codes.
363 """
364 if os.environ.get("BIJUXCLI_TEST_MODE") == "1" or os.environ.get("NO_COLOR") == "1":
365 return "bijux> "
366 return ANSI("\x1b[36mbijux> \x1b[0m")
369_JSON_CMDS = {
370 "audit",
371 "doctor",
372 "history",
373 "memory",
374 "plugins",
375 "status",
376 "version",
377}
378_BUILTINS = ("exit", "quit")
381class CommandCompleter(Completer):
382 """Provides context-aware tab-completion for the REPL."""
384 def __init__(self, main_app: typer.Typer) -> None:
385 """Initializes the completer.
387 Args:
388 main_app (typer.Typer): The root Typer application whose commands
389 and options will be used for completion suggestions.
390 """
391 self.main_app = main_app
392 self._cmd_map = self._collect(main_app)
393 self._BUILTINS = _BUILTINS
395 def _collect(
396 self,
397 app: typer.Typer,
398 path: list[str] | None = None,
399 ) -> dict[tuple[str, ...], Any]:
400 """Recursively collects all commands from a Typer application.
402 Args:
403 app (typer.Typer): The Typer application to scan.
404 path (list[str] | None): The accumulated command path for recursion.
406 Returns:
407 dict[tuple[str, ...], Any]: A mapping from command-path tuples to
408 the corresponding Typer app or command object.
409 """
410 path = path or []
411 out: dict[tuple[str, ...], Any] = {}
412 for cmd in getattr(app, "registered_commands", []):
413 out[tuple(path + [cmd.name])] = cmd
414 for grp in getattr(app, "registered_groups", []):
415 out[tuple(path + [grp.name])] = grp.typer_instance
416 out.update(self._collect(grp.typer_instance, path + [grp.name]))
417 return out
419 def _find(
420 self,
421 words: list[str],
422 ) -> tuple[Any | None, list[str]]:
423 """Finds the best-matching command or group for the given tokens.
425 Args:
426 words (list[str]): The list of tokens from the input buffer.
428 Returns:
429 tuple[Any | None, list[str]]: A tuple containing the matched
430 command/group object and the list of remaining tokens.
431 """
432 for i in range(len(words), 0, -1):
433 key = tuple(words[:i])
434 if key in self._cmd_map:
435 return self._cmd_map[key], words[i:]
436 return None, words
438 def get_completions( # pyright: ignore[reportIncompatibleMethodOverride]
439 self,
440 document: Document,
441 _complete_event: CompleteEvent,
442 ) -> Iterator[Completion]:
443 """Yields completion suggestions for the current input.
445 Args:
446 document (Document): The current `prompt_toolkit` document.
447 _complete_event (CompleteEvent): The completion event (unused).
449 Yields:
450 Completion: A `prompt_toolkit` `Completion` object.
451 """
452 text = document.text_before_cursor
453 try:
454 words: list[str] = shlex.split(text)
455 except ValueError:
456 return
457 if text.endswith(" ") or not text:
458 words.append("")
459 current = words[-1]
461 found = False
463 if current.startswith("-"):
464 for opt in GLOBAL_OPTS:
465 if opt.startswith(current):
466 found = True
467 yield Completion(opt, start_position=-len(current))
469 cmd_obj, _rem = self._find(words[:-1])
470 if cmd_obj is None:
471 for b in self._BUILTINS:
472 if b.startswith(current):
473 found = True
474 yield Completion(b, start_position=-len(current))
476 if cmd_obj is None:
477 for key in self._cmd_map:
478 if len(key) == 1 and key[0].startswith(current):
479 found = True
480 yield Completion(key[0], start_position=-len(current))
481 return
483 is_group = hasattr(cmd_obj, "registered_commands") or hasattr(
484 cmd_obj, "registered_groups"
485 )
486 if is_group:
487 names = [c.name for c in getattr(cmd_obj, "registered_commands", [])]
488 names += [g.name for g in getattr(cmd_obj, "registered_groups", [])]
489 for n in names:
490 if n.startswith(current):
491 found = True
492 yield Completion(n, start_position=-len(current))
494 if (not is_group) and hasattr(cmd_obj, "params"): 494 ↛ 495line 494 didn't jump to line 495 because the condition on line 494 was never true
495 for param in cmd_obj.params:
496 for opt in (*param.opts, *(getattr(param, "secondary_opts", []) or [])):
497 if opt.startswith(current):
498 found = True
499 yield Completion(opt, start_position=-len(current))
501 if "--help".startswith(current):
502 found = True
503 yield Completion("--help", start_position=-len(current))
505 if not found:
506 if ( 506 ↛ 512line 506 didn't jump to line 512 because the condition on line 506 was never true
507 len(words) >= 3
508 and words[0] == "config"
509 and words[1] == "set"
510 and words[2] == ""
511 ):
512 yield Completion("KEY=VALUE", display="KEY=VALUE", start_position=0)
513 elif current == "": 513 ↛ 514line 513 didn't jump to line 514 because the condition on line 513 was never true
514 yield Completion("DUMMY", display="DUMMY", start_position=0)
517async def _run_interactive() -> None:
518 """Starts the interactive REPL session.
520 This function configures and runs a `prompt_toolkit` session, providing
521 an interactive shell for the user. It handles user input asynchronously.
523 Returns:
524 None:
525 """
526 from importlib import import_module
527 import os
528 from pathlib import Path
529 import shlex
530 import subprocess # nosec B404
531 import sys
533 from prompt_toolkit import PromptSession
534 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
535 from prompt_toolkit.history import FileHistory
536 from prompt_toolkit.key_binding import KeyBindings
537 from prompt_toolkit.output import ColorDepth
539 cli_mod = import_module("bijux_cli.cli")
540 app = cli_mod.build_app()
542 kb = KeyBindings()
544 @kb.add("tab")
545 def _(event: KeyPressEvent) -> None:
546 """Handles Tab key presses for completion.
548 Args:
549 event (KeyPressEvent): The `prompt_toolkit` key press event.
550 """
551 buf = event.app.current_buffer
552 if buf.complete_state:
553 buf.complete_next()
554 else:
555 buf.start_completion(select_first=True)
557 @kb.add("enter")
558 def _(event: KeyPressEvent) -> None:
559 """Handles Enter key presses to submit or accept completions.
561 Args:
562 event (KeyPressEvent): The `prompt_toolkit` key press event.
563 """
564 buf: Buffer = event.app.current_buffer
565 state = buf.complete_state
566 if state: 566 ↛ 572line 566 didn't jump to line 572 because the condition on line 566 was always true
567 comp: Completion | None = state.current_completion
568 if comp: 568 ↛ 570line 568 didn't jump to line 570 because the condition on line 568 was always true
569 buf.apply_completion(comp)
570 buf.complete_state = None
571 else:
572 buf.validate_and_handle()
574 history_file = os.environ.get(
575 "BIJUXCLI_HISTORY_FILE",
576 str(Path.home() / ".bijux" / ".repl_history"),
577 )
579 session: PromptSession[str] = PromptSession(
580 get_prompt(),
581 history=FileHistory(history_file),
582 completer=CommandCompleter(app),
583 auto_suggest=AutoSuggestFromHistory(),
584 color_depth=ColorDepth.DEPTH_1_BIT,
585 enable_history_search=True,
586 complete_while_typing=False,
587 key_bindings=kb,
588 )
590 cli_bin = os.environ.get("BIJUXCLI_BIN") or sys.argv[0]
592 while True:
593 try:
594 line = await session.prompt_async()
595 except (EOFError, KeyboardInterrupt):
596 print("\nExiting REPL.")
597 return
599 for seg in _split_segments(line):
600 lower = seg.lower()
601 if lower in ("exit", "quit"):
602 print("Exiting REPL.")
603 return
604 if not seg.strip() or seg.startswith("#"):
605 continue
607 try:
608 tokens = shlex.split(seg)
609 except ValueError:
610 continue
612 head = tokens[0]
613 if seg == "docs":
614 print("Available topics: ...")
615 continue
616 if seg.startswith("docs "):
617 print(seg.split(None, 1)[1])
618 continue
619 if seg == "memory list":
620 subprocess.run( # noqa: S603 # nosec B603
621 [cli_bin, *tokens], env=os.environ
622 )
623 continue
625 if head not in _known_commands():
626 hint = _suggest(head)
627 msg = f"No such command '{head}'."
628 if hint:
629 msg += hint
630 print(msg, file=sys.stderr)
631 continue
633 _invoke(tokens, repl_quiet=False)
636@repl_app.callback(invoke_without_command=True)
637def main(
638 ctx: typer.Context,
639 quiet: bool = typer.Option(False, "-q", "--quiet", help=HELP_QUIET),
640 verbose: bool = typer.Option(False, "-v", "--verbose", help=HELP_VERBOSE),
641 fmt: str = typer.Option("human", "-f", "--format", help=HELP_FORMAT_HELP),
642 pretty: bool = typer.Option(True, "--pretty/--no-pretty", help=HELP_NO_PRETTY),
643 debug: bool = typer.Option(False, "-d", "--debug", help=HELP_DEBUG),
644) -> None:
645 """Defines the entrypoint for the `bijux repl` command.
647 This function initializes the REPL environment. It validates flags, sets
648 up signal handlers for clean shutdown, and dispatches to either the
649 non-interactive (piped) mode or the interactive async prompt loop.
651 Args:
652 ctx (typer.Context): The Typer context for the CLI.
653 quiet (bool): If True, forces non-interactive mode and suppresses
654 prompts and command output.
655 verbose (bool): If True, enables verbose output for subcommands.
656 fmt (str): The desired output format. Only "human" is supported for
657 the REPL itself.
658 pretty (bool): If True, enables pretty-printing for subcommands.
659 debug (bool): If True, enables debug diagnostics for subcommands.
661 Returns:
662 None:
663 """
664 if ctx.invoked_subcommand:
665 return
667 command = "repl"
668 effective_include_runtime = (verbose or debug) and not quiet
670 fmt_lower = fmt.strip().lower()
672 if fmt_lower != "human":
673 validate_common_flags(
674 fmt_lower,
675 command,
676 quiet,
677 include_runtime=effective_include_runtime,
678 )
679 emit_error_and_exit(
680 "REPL only supports human format.",
681 code=2,
682 failure="format",
683 command=command,
684 fmt=fmt_lower,
685 quiet=quiet,
686 include_runtime=effective_include_runtime,
687 debug=debug,
688 )
690 for sig in (
691 signal.SIGINT,
692 signal.SIGTERM,
693 signal.SIGHUP,
694 signal.SIGQUIT,
695 signal.SIGUSR1,
696 ):
697 with suppress(Exception):
698 signal.signal(sig, _exit_on_signal)
700 if quiet or not sys.stdin.isatty():
701 _run_piped(quiet)
702 else:
703 try:
704 asyncio.get_event_loop()
705 except RuntimeError:
706 asyncio.set_event_loop(asyncio.new_event_loop())
708 asyncio.run(_run_interactive())
711if __name__ == "__main__":
712 repl_app()