From 0994a3586dfbe8ca0f4a7f16234747114ee26118 Mon Sep 17 00:00:00 2001 From: lyteen Date: Sat, 30 May 2026 23:12:24 +0800 Subject: [PATCH] [feat] Fuzzy /command Resolution & Custom Aliases --- plugins/cow_cli/cow_cli.py | 178 +++++++++++++++++++++++++++++++------ 1 file changed, 149 insertions(+), 29 deletions(-) diff --git a/plugins/cow_cli/cow_cli.py b/plugins/cow_cli/cow_cli.py index b7dc6371..f3fee96a 100644 --- a/plugins/cow_cli/cow_cli.py +++ b/plugins/cow_cli/cow_cli.py @@ -23,6 +23,7 @@ from plugins import Plugin, Event, EventContext, EventAction from bridge.context import ContextType from bridge.reply import Reply, ReplyType from common.log import logger +from config import conf from cli import __version__ @@ -44,6 +45,23 @@ CLI_ONLY_COMMANDS = {"start", "stop", "restart"} # Commands that can only run from chat (need access to in-process memory) CHAT_ONLY_COMMANDS = set() # context is allowed in both, but behaves differently +# Convenience shorthands for the slash form only. Values are *full command +# strings* so an alias can carry arguments (e.g. "cc" -> "context clear"). +# These shorthands are deliberate and NOT derivable from prefix/typo rules: +# - "c" is a prefix of cancel/config/context (ambiguous) -> needs explicit map +# - "cc" is a prefix of nothing and expands to a command + argument +# - "s" would otherwise be an ambiguous prefix (skill/start/status/stop) +# Users may override / extend these via config.json "command_aliases". +DEFAULT_ALIASES = { + "c": "cancel", + "cc": "context clear", + "ctx": "context", + "h": "help", + "s": "status", + "cfg": "config", + "k": "knowledge", +} + @plugins.register( name="cow_cli", @@ -57,8 +75,51 @@ class CowCliPlugin(Plugin): def __init__(self): super().__init__() self.handlers[Event.ON_HANDLE_CONTEXT] = self.on_handle_context + self.aliases = self._build_aliases() logger.debug("[CowCli] initialized") + def reload(self): + """Rebuild the alias table (e.g. after config changes).""" + self.aliases = self._build_aliases() + + @staticmethod + def _build_aliases() -> dict: + """Merge DEFAULT_ALIASES with optional config.json ``command_aliases``. + + User-supplied entries (keys lowercased / stripped) override defaults. + An alias whose target's first token is not a known command is dropped + with a warning, so a bad alias can never create a dead command. An + alias key that shadows a real command is kept but warned about — the + exact-command stage in ``_resolve`` runs first, so the command wins. + """ + merged = dict(DEFAULT_ALIASES) + try: + overrides = conf().get("command_aliases", {}) or {} + except Exception as e: + logger.warning(f"[CowCli] could not read command_aliases from config: {e}") + overrides = {} + if not isinstance(overrides, dict): + logger.warning(f"[CowCli] command_aliases must be an object, got {type(overrides).__name__}; ignoring") + overrides = {} + for key, value in overrides.items(): + if not isinstance(key, str) or not isinstance(value, str): + logger.warning(f"[CowCli] ignoring non-string alias entry: {key!r} -> {value!r}") + continue + k = key.strip().lower() + if k: + merged[k] = value.strip() + + valid = {} + for key, value in merged.items(): + head = value.split(None, 1)[0].lower() if value.strip() else "" + if head not in KNOWN_COMMANDS: + logger.warning(f"[CowCli] dropping alias '/{key}' -> '{value}': unknown command '{head}'") + continue + if key in KNOWN_COMMANDS: + logger.warning(f"[CowCli] alias '/{key}' shadows a real command; the command takes precedence") + valid[key] = value + return valid + def on_handle_context(self, e_context: EventContext): if e_context["context"].type != ContextType.TEXT: return @@ -68,28 +129,24 @@ class CowCliPlugin(Plugin): if parsed is None: return - cmd, args = parsed + token, user_args = parsed + result = self._resolve(token, user_args) + kind = result[0] - if cmd not in KNOWN_COMMANDS: - # Slash-prefixed near-miss: looks like a typo of a real command. - # Intercept with a hint so we don't burn an LLM round on "/momory". - suggestion = self._suggest_command(cmd) - if suggestion is None: - return - hint = f"未知命令: /{cmd}" - if suggestion: - hint += f"\n你是不是想输入 /{suggestion} ?" - hint += "\n发送 /help 查看全部命令。" - e_context["reply"] = Reply(ReplyType.TEXT, hint) - e_context.action = EventAction.BREAK_PASS + if kind == "passthrough": + # Not a command and not a near-typo: let the agent handle it. return - logger.info(f"[CowCli] intercepted command: {cmd} {args}") + if kind == "run": + _, cmd, args = result + logger.info(f"[CowCli] intercepted command: {cmd} {args}") + reply_text = self._dispatch(cmd, args, e_context) + elif kind == "ambiguous": + reply_text = self._ambiguous_hint(token, result[1]) + else: # "typo" + reply_text = self._typo_hint(token, result[1]) - result = self._dispatch(cmd, args, e_context) - - reply = Reply(ReplyType.TEXT, result) - e_context["reply"] = reply + e_context["reply"] = Reply(ReplyType.TEXT, reply_text) e_context.action = EventAction.BREAK_PASS def _parse_command(self, content: str): @@ -175,6 +232,69 @@ class CowCliPlugin(Plugin): return known return None + def _resolve(self, token: str, user_args: str): + """Resolve a parsed slash token to an action. + + Precedence (first hit wins): + 1. exact command -> ("run", cmd, args) + 2. alias (exact key) -> expand to a full command string, merge args + 3. unique prefix -> ("run", cmd, args) + 4. ambiguous prefix (>1) -> ("ambiguous", sorted_candidates) + 5. typo (edit-distance <=1) -> ("typo", suggestion_or_None) + 6. no match -> ("passthrough",) + + Pure function of (token, user_args, KNOWN_COMMANDS, self.aliases) so it + is trivially unit-testable. Note the `cow ` form never reaches stages + 2-5: `_parse_command` only returns known tokens for it, so it stays + strict and alias/prefix matching applies to the `/` form only. + """ + token = token.lower() + + # 1. exact command (a real command can never be shadowed by an alias) + if token in KNOWN_COMMANDS: + return ("run", token, user_args) + + # 2. alias -> full command string; merge alias args with user args. + # Alias expansion is applied at most once (no alias -> alias chains). + if token in self.aliases: + parts = self.aliases[token].split(None, 1) + cmd = parts[0].lower() + alias_args = parts[1] if len(parts) > 1 else "" + merged = f"{alias_args} {user_args}".strip() if user_args else alias_args + return ("run", cmd, merged) + + # 3 / 4. prefix match + candidates = sorted(c for c in KNOWN_COMMANDS if c.startswith(token)) + if len(candidates) == 1: + return ("run", candidates[0], user_args) + if len(candidates) > 1: + return ("ambiguous", candidates) + + # 5. typo (keeps its own len>=3 + edit-distance<=1 guards) + suggestion = self._suggest_command(token) + if suggestion is not None: + return ("typo", suggestion) + + # 6. nothing matched + return ("passthrough",) + + @staticmethod + def _typo_hint(token: str, suggestion) -> str: + hint = f"未知命令: /{token}" + if suggestion: + hint += f"\n你是不是想输入 /{suggestion} ?" + hint += "\n发送 /help 查看全部命令。" + return hint + + @staticmethod + def _ambiguous_hint(token: str, candidates) -> str: + options = " ".join(f"/{c}" for c in candidates) + return ( + f"命令不明确: /{token}\n" + f"可能想输入: {options}\n" + "发送 /help 查看全部命令。" + ) + # ------------------------------------------------------------------ # Command dispatch # ------------------------------------------------------------------ @@ -190,17 +310,17 @@ class CowCliPlugin(Plugin): parsed = self._parse_command(query.strip()) if parsed is None: return None - cmd, args = parsed - if cmd not in KNOWN_COMMANDS: - suggestion = self._suggest_command(cmd) - if suggestion is None: - return None - hint = f"未知命令: /{cmd}" - if suggestion: - hint += f"\n你是不是想输入 /{suggestion} ?" - hint += "\n发送 /help 查看全部命令。" - return hint - return self._dispatch(cmd, args, e_context=None, session_id=session_id) + token, user_args = parsed + result = self._resolve(token, user_args) + kind = result[0] + if kind == "passthrough": + return None + if kind == "run": + _, cmd, args = result + return self._dispatch(cmd, args, e_context=None, session_id=session_id) + if kind == "ambiguous": + return self._ambiguous_hint(token, result[1]) + return self._typo_hint(token, result[1]) # "typo" def _dispatch(self, cmd: str, args: str, e_context: EventContext, session_id: str = "") -> str: if cmd in CLI_ONLY_COMMANDS: