Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/kimi_cli/soul/kimisoul.py
Original file line number Diff line number Diff line change
Expand Up @@ -607,6 +607,7 @@ def _build_slash_commands(self) -> list[SlashCommand[Any]]:
func=self._make_skill_runner(skill),
description=skill.description or "",
aliases=[],
completion_submit="insert_only",
)
)
seen_names.add(name)
Expand Down
56 changes: 49 additions & 7 deletions src/kimi_cli/ui/shell/prompt.py
Original file line number Diff line number Diff line change
Expand Up @@ -97,11 +97,13 @@ def __init__(self, available_commands: Sequence[SlashCommand[Any]]) -> None:
super().__init__()
self._available_commands = list(available_commands)
self._command_lookup: dict[str, list[SlashCommand[Any]]] = {}
self._commands_by_name: dict[str, SlashCommand[Any]] = {}
words: list[str] = []

for cmd in sorted(self._available_commands, key=lambda c: c.name):
if cmd.name not in self._command_lookup:
self._command_lookup[cmd.name] = []
self._commands_by_name[cmd.name] = cmd
words.append(cmd.name)
self._command_lookup[cmd.name].append(cmd)
for alias in cmd.aliases:
Expand Down Expand Up @@ -130,6 +132,30 @@ def should_complete(document: Document) -> bool:

return not prefix.strip() and token.startswith("/")

@staticmethod
def _completion_to_command_name(completion: Completion | None) -> str | None:
if completion is None:
return None
text = completion.text.strip()
if not text.startswith("/"):
return None
name = text[1:].strip()
if not name:
return None
return name

def command_for_completion(self, completion: Completion | None) -> SlashCommand[Any] | None:
command_name = self._completion_to_command_name(completion)
if command_name is None:
return None
return self._commands_by_name.get(command_name)

def should_auto_submit_completion(self, completion: Completion | None) -> bool:
command = self.command_for_completion(completion)
if command is None:
return True
return command.completion_submit == "auto_submit"

@override
def get_completions(
self, document: Document, complete_event: CompleteEvent
Expand Down Expand Up @@ -1222,30 +1248,44 @@ def __init__(
self._last_history_content = history_entries[-1].content

# Build completers
self._agent_slash_completer = SlashCommandCompleter(agent_mode_slash_commands)
self._shell_slash_completer = SlashCommandCompleter(shell_mode_slash_commands)
self._agent_mode_completer = merge_completers(
[
SlashCommandCompleter(agent_mode_slash_commands),
self._agent_slash_completer,
# TODO(kaos): we need an async KaosFileMentionCompleter
LocalFileMentionCompleter(KaosPath.cwd().unsafe_to_local_path()),
],
deduplicate=True,
)
self._shell_mode_completer = SlashCommandCompleter(shell_mode_slash_commands)
self._shell_mode_completer = self._shell_slash_completer

# Build key bindings
_kb = KeyBindings()

def _current_or_first_completion(buff: Buffer) -> Completion | None:
complete_state = buff.complete_state
if complete_state is None or not complete_state.completions:
return None
return complete_state.current_completion or complete_state.completions[0]

def _accept_completion(buff: Buffer) -> None:
"""Accept the current or first completion, suppressing re-completion."""
completion = buff.complete_state.current_completion # type: ignore[union-attr]
if not completion:
completion = buff.complete_state.completions[0] # type: ignore[union-attr]
completion = _current_or_first_completion(buff)
if completion is None:
return
self._suppress_auto_completion = True
try:
buff.apply_completion(completion)
finally:
self._suppress_auto_completion = False

def _slash_completion_should_submit(buff: Buffer) -> bool:
completion = _current_or_first_completion(buff)
if self._mode == PromptMode.SHELL:
return self._shell_slash_completer.should_auto_submit_completion(completion)
return self._agent_slash_completer.should_auto_submit_completion(completion)

def _is_slash_completion() -> bool:
"""True when the active completion menu is for a slash command."""
buff = self._session.default_buffer
Expand All @@ -1260,9 +1300,11 @@ def _is_slash_completion() -> bool:

@_kb.add("enter", filter=_slash_completion_filter)
def _(event: KeyPressEvent) -> None:
"""Slash command completion: accept and submit in one step."""
"""Slash command completion: accept, then submit based on command policy."""
should_submit = _slash_completion_should_submit(event.current_buffer)
_accept_completion(event.current_buffer)
event.current_buffer.validate_and_handle()
if should_submit:
event.current_buffer.validate_and_handle()

@_kb.add("enter", filter=_non_slash_completion_filter)
def _(event: KeyPressEvent) -> None:
Expand Down
5 changes: 4 additions & 1 deletion src/kimi_cli/utils/slashcmd.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
import re
from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass
from typing import overload
from typing import Literal, overload

CompletionSubmitPolicy = Literal["auto_submit", "insert_only"]


@dataclass(frozen=True, slots=True, kw_only=True)
Expand All @@ -10,6 +12,7 @@ class SlashCommand[F: Callable[..., None | Awaitable[None]]]:
description: str
func: F
aliases: list[str]
completion_submit: CompletionSubmitPolicy = "auto_submit"

def slash_name(self):
"""/name (aliases)"""
Expand Down
8 changes: 5 additions & 3 deletions tests/core/test_kimisoul_slash_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,8 @@ def test_flow_skill_registers_skill_and_flow_commands(runtime: Runtime, tmp_path
)
soul = KimiSoul(agent, context=Context(file_backend=tmp_path / "history.jsonl"))

command_names = {cmd.name for cmd in soul.available_slash_commands}
assert "skill:flow-skill" in command_names
assert "flow:flow-skill" in command_names
commands = {cmd.name: cmd for cmd in soul.available_slash_commands}
assert "skill:flow-skill" in commands
assert "flow:flow-skill" in commands
assert commands["skill:flow-skill"].completion_submit == "insert_only"
assert commands["flow:flow-skill"].completion_submit == "auto_submit"
98 changes: 98 additions & 0 deletions tests/e2e/test_slash_completion_enter_tmux.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,12 +44,45 @@ def _wait_for_pane_text(session: str, text: str, *, timeout: float = 15.0) -> st
time.sleep(0.1)


def _assert_pane_text_absent(session: str, text: str, *, timeout: float = 2.0) -> None:
deadline = time.monotonic() + timeout
while True:
pane = _capture_pane(session)
if text in pane:
raise AssertionError(f"Unexpected text {text!r} appeared.\nPane contents:\n{pane}")
if time.monotonic() >= deadline:
return
time.sleep(0.1)


def _write_skill(tmp_path: Path, *, name: str) -> Path:
skill_root = tmp_path / "skills"
skill_dir = skill_root / name
skill_dir.mkdir(parents=True, exist_ok=True)
skill_dir.joinpath("SKILL.md").write_text(
"\n".join(
[
"---",
f"name: {name}",
"description: tmux slash completion test skill",
"---",
"",
"Use this skill for slash completion tmux tests.",
"",
]
),
encoding="utf-8",
)
return skill_root


def _start_tmux_shell(
*,
session: str,
config_path: Path,
work_dir: Path,
home_dir: Path,
extra_args: list[str] | None = None,
columns: int = 120,
lines: int = 40,
) -> None:
Expand All @@ -74,6 +107,8 @@ def _start_tmux_shell(
"--work-dir",
str(work_dir),
]
if extra_args:
command_parts.extend(extra_args)
command = shlex.join(command_parts)
env_prefix = " ".join(f"{key}={shlex.quote(value)}" for key, value in env.items())
shell_command = f"cd {shlex.quote(str(repo_root()))} && {env_prefix} {command}"
Expand Down Expand Up @@ -134,3 +169,66 @@ def test_slash_completion_single_enter_executes(tmp_path: Path) -> None:
time.sleep(0.1)
finally:
_tmux("kill-session", "-t", session_name, check=False)


def test_skill_completion_enter_inserts_only_then_executes(tmp_path: Path) -> None:
config_path = write_scripted_config(tmp_path, ["text: skill command executed"])
work_dir = make_work_dir(tmp_path)
home_dir = make_home_dir(tmp_path)
skill_name = "tmux-skill"
skills_dir = _write_skill(tmp_path, name=skill_name)
session_name = f"kimi-tmux-skill-enter-{uuid.uuid4().hex[:8]}"

try:
_start_tmux_shell(
session=session_name,
config_path=config_path,
work_dir=work_dir,
home_dir=home_dir,
extra_args=["--skills-dir", str(skills_dir)],
)
_wait_for_pane_text(session_name, "Welcome to Kimi Code CLI!")
_wait_for_pane_text(session_name, "── input")

_tmux("send-keys", "-t", f"{session_name}:0.0", f"/skill:{skill_name[:4]}", "")
_wait_for_pane_text(session_name, f"/skill:{skill_name}", timeout=5.0)

# Enter should only insert skill slash command from completion, not submit.
_tmux("send-keys", "-t", f"{session_name}:0.0", "Enter")
_assert_pane_text_absent(session_name, "skill command executed", timeout=2.0)
_wait_for_pane_text(session_name, f"/skill:{skill_name}", timeout=2.0)

# After appending user request and pressing Enter, command executes normally.
_tmux("send-keys", "-t", f"{session_name}:0.0", " fix login", "Enter")
_wait_for_pane_text(session_name, "skill command executed", timeout=10.0)
finally:
_tmux("kill-session", "-t", session_name, check=False)


def test_skill_completion_tab_does_not_submit(tmp_path: Path) -> None:
config_path = write_scripted_config(tmp_path, ["text: tab path executed"])
work_dir = make_work_dir(tmp_path)
home_dir = make_home_dir(tmp_path)
skill_name = "tmux-tab-skill"
skills_dir = _write_skill(tmp_path, name=skill_name)
session_name = f"kimi-tmux-skill-tab-{uuid.uuid4().hex[:8]}"

try:
_start_tmux_shell(
session=session_name,
config_path=config_path,
work_dir=work_dir,
home_dir=home_dir,
extra_args=["--skills-dir", str(skills_dir)],
)
_wait_for_pane_text(session_name, "Welcome to Kimi Code CLI!")
_wait_for_pane_text(session_name, "── input")

_tmux("send-keys", "-t", f"{session_name}:0.0", "/skill:tmux-tab", "")
_wait_for_pane_text(session_name, f"/skill:{skill_name}", timeout=5.0)
_tmux("send-keys", "-t", f"{session_name}:0.0", "Tab")

# Tab completion should not trigger slash execution by itself.
_assert_pane_text_absent(session_name, "tab path executed", timeout=2.0)
finally:
_tmux("kill-session", "-t", session_name, check=False)
25 changes: 24 additions & 1 deletion tests/ui_and_conv/test_slash_completer.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,13 +25,17 @@ def _noop(app: object, args: str) -> None:


def _make_command(
name: str, *, aliases: Iterable[str] = ()
name: str,
*,
aliases: Iterable[str] = (),
completion_submit: str = "auto_submit",
) -> SlashCommand[Callable[[object, str], None]]:
return SlashCommand(
name=name,
description=f"{name} command",
func=_noop,
aliases=list(aliases),
completion_submit=completion_submit,
)


Expand Down Expand Up @@ -100,6 +104,25 @@ def test_completion_display_uses_canonical_command_name():
assert completions[0].display_meta_text == "help command"


def test_completion_submit_policy_uses_command_metadata():
completer = SlashCommandCompleter(
[
_make_command("sessions"),
_make_command("skill:test", completion_submit="insert_only"),
]
)

completions = _completions(completer, "/ski")
skill_completion = next(c for c in completions if c.text == "/skill:test")
assert completer.should_auto_submit_completion(skill_completion) is False

completions = _completions(completer, "/ses")
sessions_completion = next(c for c in completions if c.text == "/sessions")
assert completer.should_auto_submit_completion(sessions_completion) is True
assert completer.should_auto_submit_completion(Completion(text="/unknown", start_position=0))
assert completer.should_auto_submit_completion(None)


def test_wrap_to_width_respects_width():
lines = _wrap_to_width(
"Help address review issue comments on the open GitHub PR",
Expand Down
Loading