84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
"""
|
||
pipeline-core: SlashCommand — 可插拔 slash 命令注册表
|
||
|
||
slash 命令 = 产线/角色/通用 的快捷指令，用户以 "/" 开头输入时直接触发，
|
||
不进入 LLM 循环（类似 Hermes CLI 的 /new /model /tools /skills）。
|
||
|
||
命令可插拔：通用命令在 core 注册，产线命令在能力包（sdlc_ability）注册，
|
||
不同产线可有不同的 slash 命名。
|
||
|
||
依赖方向：pipeline-core（本文件，纯定义 + 注册表）← pipeline-service（注册通用命令）
|
||
← sdlc_ability（注册 SDLC 命令）
|
||
"""
|
||
|
||
import logging
|
||
from dataclasses import dataclass, field
|
||
from typing import Callable, Dict, List, Optional
|
||
|
||
logger = logging.getLogger("pipeline.slash")
|
||
|
||
|
||
@dataclass
|
||
class SlashCommand:
|
||
"""单个 slash 命令。handler 签名：async def handler(args, ctx) -> str
|
||
- args: 命令后的参数字符串（不含命令名）
|
||
- ctx: AgentExecutor 上下文 dict（project_id/user_id/pipeline_id/workspace_dir/config/executor）
|
||
"""
|
||
name: str # 不含 "/" 前缀，如 "new"
|
||
description: str = ""
|
||
handler: Optional[Callable] = None
|
||
scope: str = "global" # global / pipeline / role
|
||
scope_id: str = "" # pipeline_id 或 pipeline_id:role
|
||
args_hint: str = "" # 参数提示，如 "[task_id]"
|
||
|
||
|
||
# ── 全局注册表 ──
|
||
|
||
_commands: Dict[str, SlashCommand] = {}
|
||
|
||
|
||
def register_slash_command(cmd: SlashCommand):
|
||
"""注册一个 slash 命令（幂等：重复注册覆盖）。"""
|
||
_commands[cmd.name] = cmd
|
||
logger.debug(f"SlashCommand registered: /{cmd.name} ({cmd.scope})")
|
||
|
||
|
||
def get_slash_commands(pipeline_id: str = "", role: str = "") -> Dict[str, SlashCommand]:
|
||
"""取当前上下文可见的 slash 命令（通用 + 产线 + 角色）。"""
|
||
result: Dict[str, SlashCommand] = {}
|
||
for name, cmd in _commands.items():
|
||
if cmd.scope == "global":
|
||
result[name] = cmd
|
||
elif cmd.scope == "pipeline" and pipeline_id and cmd.scope_id == pipeline_id:
|
||
result[name] = cmd
|
||
elif cmd.scope == "role" and pipeline_id and role \
|
||
and cmd.scope_id == f"{pipeline_id}:{role}":
|
||
result[name] = cmd
|
||
return result
|
||
|
||
|
||
def resolve_slash_command(text: str, pipeline_id: str = "",
|
||
role: str = "") -> Optional[SlashCommand]:
|
||
"""从用户输入解析 slash 命令（text 以 "/" 开头）。"""
|
||
if not text or not text.startswith("/"):
|
||
return None
|
||
body = text[1:].strip()
|
||
if not body:
|
||
return None
|
||
# 命令名 = 第一个空格前的 token
|
||
name = body.split()[0] if " " in body else body
|
||
visible = get_slash_commands(pipeline_id, role)
|
||
return visible.get(name)
|
||
|
||
|
||
def parse_slash_args(text: str) -> str:
|
||
"""提取命令名之后的参数字符串。"""
|
||
body = text[1:].strip() if text.startswith("/") else text.strip()
|
||
parts = body.split(None, 1)
|
||
return parts[1] if len(parts) > 1 else ""
|
||
|
||
|
||
def list_commands() -> List[SlashCommand]:
|
||
"""列出所有已注册命令。"""
|
||
return list(_commands.values())
|