Return ``"exit"`` to end the REPL, ``"new_session"`` to swap sessions, or ``None`` to continue with the current session.
(
cmd: str,
kb_dir: Path,
session: ChatSession,
style: Style,
)
| 740 | |
| 741 | |
| 742 | async def _handle_slash( |
| 743 | cmd: str, |
| 744 | kb_dir: Path, |
| 745 | session: ChatSession, |
| 746 | style: Style, |
| 747 | ) -> str | None: |
| 748 | """Return ``"exit"`` to end the REPL, ``"new_session"`` to swap sessions, |
| 749 | or ``None`` to continue with the current session.""" |
| 750 | parts = cmd.split(maxsplit=1) |
| 751 | head = parts[0].lower() |
| 752 | arg = parts[1].strip() if len(parts) > 1 else "" |
| 753 | # Strip surrounding quotes (user may type /add '/path/to file') |
| 754 | if len(arg) >= 2 and arg[0] == arg[-1] and arg[0] in ("'", '"'): |
| 755 | arg = arg[1:-1] |
| 756 | elif arg and arg[0] in ("'", '"'): |
| 757 | arg = arg[1:] |
| 758 | |
| 759 | if head in ("/exit", "/quit"): |
| 760 | _fmt(style, ("class:header", "Bye. Thanks for using OpenKB.\n\n")) |
| 761 | return "exit" |
| 762 | |
| 763 | if head == "/help": |
| 764 | _fmt(style, ("class:slash.help", _HELP_TEXT + "\n")) |
| 765 | return None |
| 766 | |
| 767 | if head == "/clear": |
| 768 | old_id = session.id |
| 769 | _fmt( |
| 770 | style, |
| 771 | ("class:slash.ok", f"Started new session (previous: {old_id})\n"), |
| 772 | ) |
| 773 | return "new_session" |
| 774 | |
| 775 | if head == "/save": |
| 776 | if not session.user_turns: |
| 777 | _fmt(style, ("class:error", "Nothing to save yet.\n")) |
| 778 | return None |
| 779 | from openkb.locks import kb_ingest_lock |
| 780 | |
| 781 | with kb_ingest_lock(kb_dir / ".openkb"): |
| 782 | path = _save_transcript(kb_dir, session, arg or None) |
| 783 | _fmt(style, ("class:slash.ok", f"Saved to {path}\n")) |
| 784 | return None |
| 785 | |
| 786 | if head == "/status": |
| 787 | from openkb.cli import print_status |
| 788 | from openkb.locks import kb_read_lock |
| 789 | |
| 790 | with kb_read_lock(kb_dir / ".openkb"): |
| 791 | print_status(kb_dir) |
| 792 | return None |
| 793 | |
| 794 | if head == "/list": |
| 795 | from openkb.cli import print_list |
| 796 | from openkb.locks import kb_read_lock |
| 797 | |
| 798 | with kb_read_lock(kb_dir / ".openkb"): |
| 799 | print_list(kb_dir) |