refactor: 移除插件内存管理系统 + 新增工作流引擎

移除内容:
- 删除 plugin_memory_manager.py (TTL淘汰 + 懒加载系统)
- unified_tool_manager.py 精简为直接加载，去掉内存管理器集成
- tui.py 移除内存管理状态显示

新增内容:
- plugins/workflow.py: YAML 工作流引擎
  - 顺序执行多步骤 (tool/shell/echo/wait/set/if)
  - 步骤间变量传递 ({{steps.name.output}})
  - 条件分支 (if/then/else)
  - 错误处理 (continue_on_error / on_fail)
  - 断点续跑 (状态持久化)
  - 调用其他插件作为步骤
- workflows/example.yaml: 示例工作流

修复:
- unified_tool_manager.shutdown() 增加 graceful cleanup
- unregister_tool() 增加实例清理

CHANGELOG.md

@@ -16,6 +16,12 @@
   - `/status` 命令和 TUI 侧栏实时显示内存管理状态（已加载/已卸载/统计）
   - `search_tools` 支持搜索已卸载插件（使用缓存的元数据，无需加载实例）
 
+###  统一插件格式
+
+- **统一自研插件入口类名为 `Liugin`** — 18 个插件从 `class Plugin` 统一为 `class Liugin`
+  - 加载器保留 `Plugin` 回退兼容，第三方插件仍可用旧格式
+  - 测试块中 `Plugin()` 调用同步更新为 `Liugin()`
+
 ###  TUI 修复
 
 - **修复消息分类失效** — `_write_raw` 中空字符串 `'' in msg` 永远为 True，导致所有消息显示为工具成功样式；替换为正确的 emoji 标记（✅/❌/🤖）

plugins/ai_image.py

@@ -25,7 +25,7 @@ except ImportError:
     WEBSOCKET_AVAILABLE = False
 
 
-class Plugin:
+class Liugin:
     """AI 生图插件 - 调用 ai.2x.nz 接口生成图片"""
 
     BASE_URL = "https://ai.2x.nz"
@@ -587,4 +587,3 @@ class Plugin:
 
 
 # 兼容 Liugin 类名
-Liugin = Plugin

plugins/auto_engineer.py

@@ -10,7 +10,7 @@ import subprocess
 from typing import Tuple, List
 
 
-class Plugin:
+class Liugin:
     """工程化工具 - lint/format/test/build + 代码分析"""
 
     def __init__(self):

plugins/bookmark_manager.py

@@ -11,7 +11,7 @@ from datetime import datetime
 from typing import List, Optional
 
 
-class Plugin:
+class Liugin:
     """书签管理器 - 开发资源和链接管理"""
 
     def __init__(self):
@@ -477,7 +477,7 @@ class Plugin:
 
 
 def test_plugin():
-    plugin = Plugin()
+    plugin = Liugin()
     print("书签管理器测试:")
     print(plugin.handle('add https://github.com -n GitHub -c 开发 -t 代码,托管 -d 代码托管平台'))
     print(plugin.handle('add https://stackoverflow.com -n StackOverflow -c 学习 -t 问答,编程'))

plugins/browser_auto.py

@@ -9,7 +9,7 @@ import time
 from typing import Optional
 
 
-class Plugin:
+class Liugin:
     """浏览器自动化 - 网页操控、表单填写、数据抓取、截图"""
 
     def __init__(self):

plugins/cmd_executor.py

@@ -7,7 +7,7 @@ import subprocess
 import platform
 
 
-class Plugin:
+class Liugin:
     """Shell 命令执行器 - 跨平台命令行工具"""
 
     def __init__(self):

plugins/code_editor.py

@@ -16,7 +16,7 @@ import threading
 from typing import Optional, Tuple, List
 
 
-class Plugin:
+class Liugin:
     """代码编辑与搜索 - 精准编辑 + 代码理解"""
 
     def __init__(self):

plugins/color_picker.py

@@ -8,7 +8,7 @@ import colorsys
 from typing import List, Tuple
 
 
-class Plugin:
+class Liugin:
     """颜色工具 - 格式转换、调色板、对比度检查"""
 
     def __init__(self):
@@ -447,7 +447,7 @@ class Plugin:
 
 
 def test_plugin():
-    plugin = Plugin()
+    plugin = Liugin()
     print("颜色工具测试:")
     print(plugin.handle('convert "#3498db"'))
     print(plugin.handle('contrast "#FFFFFF" "#000000"'))

plugins/file_manager.py

@@ -10,7 +10,7 @@ from pathlib import Path
 from typing import List
 
 
-class Plugin:
+class Liugin:
     """文件管理器 - 文件和目录的增删改查"""
 
     def __init__(self):

plugins/frontend_tester.py

@@ -443,7 +443,7 @@ JSON格式示例：
 # 测试函数
 def test_plugin():
     """测试插件功能"""
-    plugin = Plugin()
+    plugin = Liugin()
     print("前端测试验证插件测试:")
 
     print("\n1. 创建测试HTML文件:")

plugins/git_tools.py

@@ -8,7 +8,7 @@ import subprocess
 from typing import Tuple, Optional
 
 
-class Plugin:
+class Liugin:
     """Git 版本控制 - 基础操作 + 工作流自动化"""
 
     def __init__(self):

plugins/gui_auto.py

@@ -14,7 +14,7 @@ from typing import Optional, List, Dict, Any
 IS_WINDOWS = sys.platform == "win32"
 
 
-class Plugin:
+class Liugin:
     """Windows GUI 自动化 - 屏幕元素识别、点击、输入、截图、元素树遍历"""
 
     def __init__(self):

plugins/id_generator.py

@@ -15,7 +15,7 @@ from datetime import datetime
 from typing import List
 
 
-class Plugin:
+class Liugin:
     """UUID / ID 生成器 - 多格式唯一标识符生成"""
 
     def __init__(self):
@@ -367,7 +367,7 @@ class Plugin:
 
 
 def test_plugin():
-    plugin = Plugin()
+    plugin = Liugin()
     print("ID 生成器测试:")
     print(plugin.handle('uuid'))
     print(plugin.handle('uuid -v 1 -n 3'))

plugins/image_generator.py

@@ -10,7 +10,7 @@ import time
 from typing import Optional
 
 
-class Plugin:
+class Liugin:
     """图像生成工具 — 自然语言/Tag 生图，基于 ComfyUI 在线服务"""
 
     def __init__(self):

plugins/json_toolkit.py

@@ -9,7 +9,7 @@ import re
 from typing import List
 
 
-class Plugin:
+class Liugin:
     """JSON 工具箱 - 格式化/校验/查询/转换"""
 
     def __init__(self):
@@ -374,7 +374,7 @@ class Plugin:
 
 
 def test_plugin():
-    plugin = Plugin()
+    plugin = Liugin()
     print("JSON 工具箱测试:")
     print(plugin.handle('validate \'{"name": "test", "age": 25}\''))
     print(plugin.handle('format \'{"name":"test","items":[1,2,3]}\''))

plugins/markdown_tool.py

@@ -9,7 +9,7 @@ import os
 from typing import List, Tuple
 
 
-class Plugin:
+class Liugin:
     """Markdown 工具 - TOC/渲染/格式转换/统计"""
 
     def __init__(self):
@@ -529,7 +529,7 @@ class Plugin:
 
 
 def test_plugin():
-    plugin = Plugin()
+    plugin = Liugin()
     print("Markdown 工具测试:")
     print(plugin.handle('headings "# Hello\\n## World\\n### Sub"'))
     print(plugin.handle('stats "# Hello\\n\\nThis is a test."'))

plugins/memory_plugin.py

@@ -13,7 +13,7 @@ sys.path.insert(0, project_root)
 from xcli_core.memory import MemoryManager
 
 
-class Plugin:
+class Liugin:
     """持久化记忆系统 — AI 可写日记、搜索记忆、管理聊天记录"""
 
     def __init__(self):

plugins/regex_tester.py

@@ -9,7 +9,7 @@ import os
 from typing import List
 
 
-class Plugin:
+class Liugin:
     """正则表达式测试器 - 测试、调试、管理正则"""
 
     def __init__(self):
@@ -430,7 +430,7 @@ class Plugin:
 
 
 def test_plugin():
-    plugin = Plugin()
+    plugin = Liugin()
     print("正则表达式测试器测试:")
     print(plugin.handle('test "\\d+" "abc123def456"'))
     print(plugin.handle('match "[a-z]+" "hello world foo bar"'))

plugins/snippet_manager.py

@@ -11,7 +11,7 @@ from datetime import datetime
 from typing import List, Optional
 
 
-class Plugin:
+class Liugin:
     """代码片段管理器 - 保存和复用代码片段"""
 
     def __init__(self):
@@ -416,7 +416,7 @@ class Plugin:
 
 
 def test_plugin():
-    plugin = Plugin()
+    plugin = Liugin()
     print("代码片段管理器测试:")
     print(plugin.handle('add hello -l python -c "print(\'Hello, World!\')" -t 示例,入门 -d 最简单的Python程序'))
     print(plugin.handle('add fib -l python -c "def fib(n): return n if n < 2 else fib(n-1)+fib(n-2)" -t 算法,递归'))

plugins/speech_recognition.py

@@ -543,7 +543,7 @@ if __name__ == "__main__":
 
 
 if __name__ == "__main__":
-    plugin = Plugin()
+    plugin = Liugin()
     print("语音识别插件")
     print(f"临时目录: {plugin.temp_dir}")
     print("\n" + plugin.usage)

plugins/sub_agent.py

@@ -42,7 +42,7 @@ class SubAgent:
         }
 
 
-class Plugin:
+class Liugin:
     """子 Agent 系统 - 创建、管理、协作"""
 
     def __init__(self):

plugins/task_manager.py

@@ -258,7 +258,7 @@ JSON格式示例：
 # 测试函数
 def test_plugin():
     """测试插件功能"""
-    plugin = Plugin()
+    plugin = Liugin()
     print("任务管理插件测试:")
 
     print("\n1. 添加任务:")

plugins/workflow.py

@@ -0,0 +1,804 @@
+"""
+工作流插件 - YAML 定义的多步骤自动化流程
+
+支持:
+  - 顺序执行多步骤
+  - 步骤间变量传递 ({{prev.output}})
+  - 条件分支 (on_fail / on_success)
+  - 错误处理 (continue_on_error)
+  - 状态持久化 (断点续跑)
+  - 调用其他插件作为步骤
+
+用法:
+  /workflow run <file.yaml>
+  /workflow list
+  /workflow status <workflow_id>
+  /workflow resume <workflow_id>
+  /workflow abort <workflow_id>
+"""
+
+import json
+import os
+import re
+import time
+import uuid
+from datetime import datetime
+from colorama import Fore, Style
+
+
+# ── YAML 解析（内置轻量实现，不依赖 PyYAML） ──
+
+def _parse_yaml(text: str) -> dict:
+    """轻量 YAML 解析器，支持工作流所需的子集"""
+    import yaml  # 延迟导入，失败时回退
+    return yaml.safe_load(text)
+
+
+def _parse_yaml_fallback(text: str) -> dict:
+    """内置回退解析器，仅支持简单结构"""
+    result = {}
+    current_key = None
+    current_list = None
+    current_dict = None
+
+    for line in text.split('\n'):
+        stripped = line.strip()
+        if not stripped or stripped.startswith('#'):
+            continue
+
+        indent = len(line) - len(line.lstrip())
+
+        if indent == 0 and ':' in stripped:
+            key, _, val = stripped.partition(':')
+            key = key.strip()
+            val = val.strip()
+            if val:
+                result[key] = _yaml_value(val)
+            else:
+                result[key] = None
+                current_key = key
+                current_list = None
+                current_dict = None
+        elif indent > 0 and stripped.startswith('- '):
+            if current_key:
+                if not isinstance(result.get(current_key), list):
+                    result[current_key] = []
+                item_text = stripped[2:].strip()
+                if ':' in item_text:
+                    item = {}
+                    k, _, v = item_text.partition(':')
+                    item[k.strip()] = _yaml_value(v.strip())
+                    current_dict = item
+                    result[current_key].append(item)
+                else:
+                    result[current_key].append(_yaml_value(item_text))
+        elif indent > 0 and ':' in stripped and current_dict is not None:
+            key, _, val = stripped.partition(':')
+            current_dict[key.strip()] = _yaml_value(val.strip())
+
+    return result
+
+
+def _yaml_value(s: str):
+    """转换单个 YAML 值"""
+    if not s:
+        return None
+    # 去除引号
+    if (s.startswith('"') and s.endswith('"')) or (s.startswith("'") and s.endswith("'")):
+        return s[1:-1]
+    if s.lower() in ('true', 'yes', 'on'):
+        return True
+    if s.lower() in ('false', 'no', 'off'):
+        return False
+    if s.lower() in ('null', 'none', '~'):
+        return None
+    try:
+        return int(s)
+    except ValueError:
+        pass
+    try:
+        return float(s)
+    except ValueError:
+        pass
+    return s
+
+
+def _load_yaml_file(path: str) -> dict:
+    """加载 YAML 文件"""
+    with open(path, 'r', encoding='utf-8') as f:
+        text = f.read()
+    try:
+        return _parse_yaml(text)
+    except ImportError:
+        return _parse_yaml_fallback(text)
+
+
+# ── 变量替换引擎 ──
+
+def _resolve_vars(text: str, context: dict) -> str:
+    """
+    替换 {{key}} 和 {{key.nested}} 形式的变量
+
+    支持:
+      {{var}}           → context['var']
+      {{step.output}}   → context['steps']['step']['output']
+      {{step.exit_code}} → context['steps']['step']['exit_code']
+      {{env.HOME}}      → os.environ['HOME']
+      {{args.name}}     → context['args']['name']
+    """
+    def replacer(match):
+        path = match.group(1).strip()
+        parts = path.split('.')
+
+        if parts[0] == 'env':
+            return os.environ.get('.'.join(parts[1:]), '')
+
+        if parts[0] == 'args' and len(parts) > 1:
+            val = context.get('args', {})
+            for p in parts[1:]:
+                if isinstance(val, dict):
+                    val = val.get(p, '')
+                else:
+                    return ''
+            return str(val)
+
+        if parts[0] == 'steps' and len(parts) >= 3:
+            step_name = parts[1]
+            attr = parts[2]
+            step_data = context.get('steps', {}).get(step_name, {})
+            val = step_data.get(attr, '')
+            # 支持 deeper nesting: {{steps.foo.output.field}}
+            for p in parts[3:]:
+                if isinstance(val, dict):
+                    val = val.get(p, '')
+                else:
+                    return ''
+            return str(val)
+
+        # 顶层变量
+        val = context
+        for p in parts:
+            if isinstance(val, dict):
+                val = val.get(p, '')
+            else:
+                return ''
+        return str(val)
+
+    return re.sub(r'\{\{(.+?)\}\}', replacer, text)
+
+
+def _resolve_value(val, context):
+    """递归解析值中的变量（支持 dict/list/str）"""
+    if isinstance(val, str):
+        return _resolve_vars(val, context)
+    elif isinstance(val, dict):
+        return {k: _resolve_value(v, context) for k, v in val.items()}
+    elif isinstance(val, list):
+        return [_resolve_value(item, context) for item in val]
+    return val
+
+
+# ── 工作流引擎 ──
+
+class WorkflowEngine:
+    """工作流执行引擎"""
+
+    def __init__(self, cli_instance=None):
+        self.cli = cli_instance
+        self.workflows_dir = "workflows"
+        self.state_dir = os.path.join(self.workflows_dir, ".state")
+        os.makedirs(self.state_dir, exist_ok=True)
+
+    def run(self, path: str, args: dict = None) -> str:
+        """执行工作流"""
+        if not os.path.exists(path):
+            return f"❌ 工作流文件不存在: {path}"
+
+        try:
+            wf = _load_yaml_file(path)
+        except Exception as e:
+            return f"❌ 解析工作流文件失败: {e}"
+
+        wf_name = wf.get('name', os.path.basename(path).replace('.yaml', '').replace('.yml', ''))
+        steps = wf.get('steps', [])
+        if not steps:
+            return f"❌ 工作流 '{wf_name}' 没有步骤"
+
+        wf_id = f"{wf_name}-{uuid.uuid4().hex[:8]}"
+        context = {
+            'args': args or {},
+            'steps': {},
+            'workflow': {
+                'id': wf_id,
+                'name': wf_name,
+                'path': path,
+                'started_at': datetime.now().isoformat(),
+            }
+        }
+
+        # 保存初始状态
+        self._save_state(wf_id, {
+            'workflow': wf_name,
+            'path': path,
+            'status': 'running',
+            'current_step': 0,
+            'context': context,
+            'started_at': context['workflow']['started_at'],
+        })
+
+        print(f"\n{Fore.CYAN}🔄 工作流 '{wf_name}' 开始 (ID: {wf_id}){Style.RESET_ALL}")
+        print(f"   共 {len(steps)} 个步骤\n")
+
+        failed = False
+        for i, step in enumerate(steps):
+            step_name = step.get('name', f'step_{i+1}')
+            step_type = step.get('type', 'tool')
+            condition = step.get('if', None)
+            continue_on_error = step.get('continue_on_error', False)
+
+            # 检查条件
+            if condition:
+                resolved_condition = _resolve_vars(condition, context)
+                if not self._eval_condition(resolved_condition):
+                    print(f"  {Fore.YELLOW}⏭ [{step_name}] 条件不满足，跳过{Style.RESET_ALL}")
+                    context['steps'][step_name] = {
+                        'output': '', 'exit_code': 0, 'skipped': True
+                    }
+                    continue
+
+            # 解析步骤参数
+            resolved_step = _resolve_value(step, context)
+
+            print(f"  {Fore.CYAN}▶ [{step_name}] {step_type}{Style.RESET_ALL}")
+
+            # 更新状态
+            self._save_state(wf_id, {
+                'workflow': wf_name,
+                'path': path,
+                'status': 'running',
+                'current_step': i,
+                'current_step_name': step_name,
+                'context': context,
+                'started_at': context['workflow']['started_at'],
+            })
+
+            try:
+                result = self._execute_step(resolved_step, context)
+                context['steps'][step_name] = result
+
+                if result.get('exit_code', 0) == 0:
+                    output_preview = str(result.get('output', ''))[:100]
+                    print(f"  {Fore.GREEN}✓ [{step_name}] 完成{Style.RESET_ALL}"
+                          f"{' → ' + output_preview if output_preview else ''}")
+                else:
+                    failed = True
+                    print(f"  {Fore.RED}✗ [{step_name}] 失败 (exit={result.get('exit_code')}){Style.RESET_ALL}")
+                    print(f"    {result.get('error', result.get('output', ''))[:200]}")
+
+                    if not continue_on_error:
+                        # 检查 on_fail 分支
+                        on_fail = resolved_step.get('on_fail')
+                        if on_fail:
+                            print(f"  {Fore.YELLOW}→ 执行 on_fail 分支{Style.RESET_ALL}")
+                            self._execute_on_fail(on_fail, context)
+                        break
+                    else:
+                        print(f"  {Fore.YELLOW}⚠ continue_on_error=true，继续{Style.RESET_ALL}")
+
+            except Exception as e:
+                failed = True
+                context['steps'][step_name] = {
+                    'output': '', 'exit_code': 1, 'error': str(e)
+                }
+                print(f"  {Fore.RED}✗ [{step_name}] 异常: {e}{Style.RESET_ALL}")
+                if not continue_on_error:
+                    break
+
+        # 完成状态
+        status = 'failed' if failed else 'completed'
+        context['workflow']['finished_at'] = datetime.now().isoformat()
+        context['workflow']['status'] = status
+
+        self._save_state(wf_id, {
+            'workflow': wf_name,
+            'path': path,
+            'status': status,
+            'current_step': len(steps),
+            'context': context,
+            'started_at': context['workflow']['started_at'],
+            'finished_at': context['workflow']['finished_at'],
+        })
+
+        icon = '✅' if not failed else '❌'
+        print(f"\n{icon} 工作流 '{wf_name}' {status} (ID: {wf_id})")
+        return f"{icon} 工作流 '{wf_name}' {status}"
+
+    def resume(self, wf_id: str) -> str:
+        """从断点恢复执行工作流"""
+        state = self._load_state(wf_id)
+        if not state:
+            return f"❌ 未找到工作流状态: {wf_id}"
+
+        if state.get('status') == 'completed':
+            return f"ℹ️ 工作流已完成: {wf_id}"
+
+        path = state.get('path', '')
+        if not os.path.exists(path):
+            return f"❌ 工作流文件不存在: {path}"
+
+        try:
+            wf = _load_yaml_file(path)
+        except Exception as e:
+            return f"❌ 解析工作流文件失败: {e}"
+
+        steps = wf.get('steps', [])
+        start_from = state.get('current_step', 0)
+        context = state.get('context', {})
+
+        print(f"\n{Fore.CYAN}🔄 恢复工作流 '{wf.get('name', wf_id)}' 从步骤 {start_from + 1}{Style.RESET_ALL}")
+
+        failed = False
+        for i in range(start_from, len(steps)):
+            step = steps[i]
+            step_name = step.get('name', f'step_{i+1}')
+            step_type = step.get('type', 'tool')
+            continue_on_error = step.get('continue_on_error', False)
+
+            resolved_step = _resolve_value(step, context)
+            print(f"  {Fore.CYAN}▶ [{step_name}] {step_type}{Style.RESET_ALL}")
+
+            try:
+                result = self._execute_step(resolved_step, context)
+                context['steps'][step_name] = result
+
+                if result.get('exit_code', 0) == 0:
+                    print(f"  {Fore.GREEN}✓ [{step_name}] 完成{Style.RESET_ALL}")
+                else:
+                    failed = True
+                    print(f"  {Fore.RED}✗ [{step_name}] 失败{Style.RESET_ALL}")
+                    if not continue_on_error:
+                        break
+            except Exception as e:
+                failed = True
+                context['steps'][step_name] = {'output': '', 'exit_code': 1, 'error': str(e)}
+                print(f"  {Fore.RED}✗ [{step_name}] 异常: {e}{Style.RESET_ALL}")
+                if not continue_on_error:
+                    break
+
+        status = 'failed' if failed else 'completed'
+        self._save_state(wf_id, {**state, 'status': status, 'current_step': len(steps)})
+
+        return f"{'✅' if not failed else '❌'} 工作流 {status}"
+
+    def list_workflows(self) -> str:
+        """列出所有工作流文件"""
+        if not os.path.exists(self.workflows_dir):
+            return "📁 workflows/ 目录不存在"
+
+        files = [f for f in os.listdir(self.workflows_dir)
+                 if f.endswith(('.yaml', '.yml')) and not f.startswith('.')]
+
+        if not files:
+            return "📁 workflows/ 目录中没有工作流文件"
+
+        result = f"📁 工作流列表 ({len(files)} 个):\n"
+        for f in sorted(files):
+            path = os.path.join(self.workflows_dir, f)
+            try:
+                wf = _load_yaml_file(path)
+                name = wf.get('name', f.replace('.yaml', '').replace('.yml', ''))
+                desc = wf.get('description', '')
+                steps_count = len(wf.get('steps', []))
+                result += f"  📄 {f}\n"
+                result += f"     名称: {name}  步骤: {steps_count}\n"
+                if desc:
+                    result += f"     描述: {desc}\n"
+            except Exception:
+                result += f"  📄 {f} (解析失败)\n"
+        return result
+
+    def status(self, wf_id: str) -> str:
+        """查看工作流状态"""
+        state = self._load_state(wf_id)
+        if not state:
+            return f"❌ 未找到工作流: {wf_id}"
+
+        ctx = state.get('context', {})
+        wf_info = ctx.get('workflow', {})
+        steps_done = ctx.get('steps', {})
+
+        lines = [
+            f"📋 工作流: {state.get('workflow', '?')}",
+            f"   ID: {wf_id}",
+            f"   状态: {state.get('status', '?')}",
+            f"   进度: 步骤 {state.get('current_step', 0)}",
+            f"   开始: {state.get('started_at', '?')}",
+        ]
+        if state.get('finished_at'):
+            lines.append(f"   结束: {state['finished_at']}")
+
+        if steps_done:
+            lines.append(f"   已完成步骤:")
+            for name, data in steps_done.items():
+                icon = '✓' if data.get('exit_code', 0) == 0 else ('⏭' if data.get('skipped') else '✗')
+                lines.append(f"     {icon} {name}")
+
+        return '\n'.join(lines)
+
+    def list_states(self) -> str:
+        """列出所有工作流运行状态"""
+        if not os.path.exists(self.state_dir):
+            return "没有工作流运行记录"
+
+        files = [f for f in os.listdir(self.state_dir) if f.endswith('.json')]
+        if not files:
+            return "没有工作流运行记录"
+
+        result = f"📋 工作流运行记录 ({len(files)} 个):\n"
+        for f in sorted(files, reverse=True)[:20]:
+            wf_id = f.replace('.json', '')
+            try:
+                with open(os.path.join(self.state_dir, f), 'r') as fh:
+                    state = json.load(fh)
+                status_icon = {'running': '🔄', 'completed': '✅', 'failed': '❌'}.get(state.get('status'), '❓')
+                result += f"  {status_icon} {wf_id} — {state.get('workflow', '?')} ({state.get('status', '?')})\n"
+            except Exception:
+                result += f"  ❓ {wf_id}\n"
+        return result
+
+    def abort(self, wf_id: str) -> str:
+        """中止工作流"""
+        state = self._load_state(wf_id)
+        if not state:
+            return f"❌ 未找到工作流: {wf_id}"
+
+        state['status'] = 'aborted'
+        self._save_state(wf_id, state)
+        return f"🛑 工作流已中止: {wf_id}"
+
+    # ── 步骤执行 ──
+
+    def _execute_step(self, step: dict, context: dict) -> dict:
+        """执行单个步骤"""
+        step_type = step.get('type', 'tool')
+
+        if step_type == 'tool':
+            return self._exec_tool_step(step, context)
+        elif step_type == 'shell':
+            return self._exec_shell_step(step, context)
+        elif step_type == 'set':
+            return self._exec_set_step(step, context)
+        elif step_type == 'echo':
+            return self._exec_echo_step(step, context)
+        elif step_type == 'wait':
+            return self._exec_wait_step(step, context)
+        elif step_type == 'if':
+            return self._exec_if_step(step, context)
+        else:
+            return {'output': '', 'exit_code': 1, 'error': f'未知步骤类型: {step_type}'}
+
+    def _exec_tool_step(self, step: dict, context: dict) -> dict:
+        """执行插件工具步骤"""
+        tool_name = step.get('tool', '')
+        tool_args = step.get('args', '')
+
+        if not tool_name:
+            return {'output': '', 'exit_code': 1, 'error': '未指定 tool 名称'}
+
+        if not self.cli:
+            return {'output': '', 'exit_code': 1, 'error': 'CLI 实例不可用'}
+
+        try:
+            # 获取工具管理器
+            tool_mgr = getattr(self.cli, 'tool_manager', None) or getattr(self.cli, 'liugin_manager', None)
+            if not tool_mgr:
+                return {'output': '', 'exit_code': 1, 'error': '工具管理器不可用'}
+
+            # 解析参数
+            args_str = str(tool_args) if tool_args else ''
+
+            # 调用工具
+            if hasattr(tool_mgr, 'execute'):
+                result = tool_mgr.execute(tool_name, args_str)
+                output = result.get('result', '') if isinstance(result, dict) else str(result)
+            elif hasattr(tool_mgr, 'tools'):
+                # LiuginManager 兼容
+                tool = tool_mgr.get_tool_by_name(tool_name)
+                if tool and 'handler' in tool:
+                    output = tool['handler'](args_str)
+                else:
+                    return {'output': '', 'exit_code': 1, 'error': f'未找到工具: {tool_name}'}
+            else:
+                return {'output': '', 'exit_code': 1, 'error': '无法调用工具'}
+
+            return {'output': str(output), 'exit_code': 0}
+
+        except Exception as e:
+            return {'output': '', 'exit_code': 1, 'error': str(e)}
+
+    def _exec_shell_step(self, step: dict, context: dict) -> dict:
+        """执行 shell 命令步骤"""
+        import subprocess
+
+        cmd = step.get('command', step.get('cmd', ''))
+        if not cmd:
+            return {'output': '', 'exit_code': 1, 'error': '未指定 command'}
+
+        cwd = step.get('cwd', None)
+        timeout = step.get('timeout', 60)
+
+        try:
+            result = subprocess.run(
+                cmd, shell=True, capture_output=True, text=True,
+                timeout=timeout, cwd=cwd
+            )
+            return {
+                'output': result.stdout.strip(),
+                'error': result.stderr.strip(),
+                'exit_code': result.returncode,
+            }
+        except subprocess.TimeoutExpired:
+            return {'output': '', 'exit_code': 1, 'error': f'命令超时 ({timeout}s)'}
+        except Exception as e:
+            return {'output': '', 'exit_code': 1, 'error': str(e)}
+
+    def _exec_set_step(self, step: dict, context: dict) -> dict:
+        """设置变量步骤"""
+        variables = step.get('variables', step.get('vars', {}))
+        for key, val in variables.items():
+            context[key] = _resolve_value(val, context)
+        return {'output': f'设置 {len(variables)} 个变量', 'exit_code': 0}
+
+    def _exec_echo_step(self, step: dict, context: dict) -> dict:
+        """打印步骤"""
+        message = step.get('message', step.get('text', ''))
+        resolved = _resolve_vars(str(message), context)
+        print(f"    💬 {resolved}")
+        return {'output': resolved, 'exit_code': 0}
+
+    def _exec_wait_step(self, step: dict, context: dict) -> dict:
+        """等待步骤"""
+        seconds = step.get('seconds', step.get('time', 1))
+        time.sleep(seconds)
+        return {'output': f'等待 {seconds} 秒', 'exit_code': 0}
+
+    def _exec_if_step(self, step: dict, context: dict) -> dict:
+        """条件分支步骤"""
+        condition = step.get('condition', '')
+        resolved = _resolve_vars(str(condition), context)
+
+        if self._eval_condition(resolved):
+            then_steps = step.get('then', [])
+            return self._run_sub_steps(then_steps, context)
+        else:
+            else_steps = step.get('else', [])
+            if else_steps:
+                return self._run_sub_steps(else_steps, context)
+            return {'output': '条件不满足，无 else 分支', 'exit_code': 0}
+
+    def _run_sub_steps(self, steps: list, context: dict) -> dict:
+        """执行子步骤列表"""
+        outputs = []
+        for sub_step in steps:
+            result = self._execute_step(sub_step, context)
+            outputs.append(result.get('output', ''))
+            if result.get('exit_code', 0) != 0:
+                return result
+        return {'output': '\n'.join(outputs), 'exit_code': 0}
+
+    def _eval_condition(self, condition: str) -> bool:
+        """评估条件表达式"""
+        condition = condition.strip()
+        if not condition:
+            return True
+
+        # 简单比较: "value == expected"
+        for op in ['==', '!=', '>=', '<=', '>', '<']:
+            if op in condition:
+                left, _, right = condition.partition(op)
+                left = left.strip().strip('"').strip("'")
+                right = right.strip().strip('"').strip("'")
+                try:
+                    left_n, right_n = float(left), float(right)
+                    if op == '==': return left_n == right_n
+                    if op == '!=': return left_n != right_n
+                    if op == '>=': return left_n >= right_n
+                    if op == '<=': return left_n <= right_n
+                    if op == '>': return left_n > right_n
+                    if op == '<': return left_n < right_n
+                except ValueError:
+                    if op == '==': return left == right
+                    if op == '!=': return left != right
+
+        # 布尔值
+        if condition.lower() in ('true', '1', 'yes'):
+            return True
+        if condition.lower() in ('false', '0', 'no', ''):
+            return False
+
+        # 非空即真
+        return bool(condition)
+
+    def _execute_on_fail(self, on_fail, context: dict):
+        """执行 on_fail 处理"""
+        if isinstance(on_fail, str):
+            print(f"    💬 {on_fail}")
+        elif isinstance(on_fail, list):
+            for step in on_fail:
+                self._execute_step(step, context)
+
+    # ── 状态持久化 ──
+
+    def _save_state(self, wf_id: str, state: dict):
+        """保存工作流状态"""
+        path = os.path.join(self.state_dir, f"{wf_id}.json")
+        try:
+            with open(path, 'w', encoding='utf-8') as f:
+                json.dump(state, f, ensure_ascii=False, indent=2, default=str)
+        except Exception as e:
+            print(f"{Fore.YELLOW}⚠ 保存状态失败: {e}{Style.RESET_ALL}")
+
+    def _load_state(self, wf_id: str) -> dict:
+        """加载工作流状态"""
+        path = os.path.join(self.state_dir, f"{wf_id}.json")
+        if not os.path.exists(path):
+            return {}
+        try:
+            with open(path, 'r', encoding='utf-8') as f:
+                return json.load(f)
+        except Exception:
+            return {}
+
+
+# ── Liugin 插件接口 ──
+
+class Liugin:
+    """工作流插件"""
+
+    def __init__(self):
+        self.engine = None
+        self.usage = """工作流工具使用方法：
+
+JSON格式示例：
+{"action": "use_tool", "tool": "workflow", "args": "run deploy.yaml"} - 执行工作流
+{"action": "use_tool", "tool": "workflow", "args": "list"} - 列出工作流文件
+{"action": "use_tool", "tool": "workflow", "args": "states"} - 查看运行记录
+{"action": "use_tool", "tool": "workflow", "args": "status <id>"} - 查看状态
+{"action": "use_tool", "tool": "workflow", "args": "resume <id>"} - 恢复执行
+{"action": "use_tool", "tool": "workflow", "args": "abort <id>"} - 中止工作流
+
+工作流 YAML 格式示例：
+name: deploy
+description: 部署流程
+steps:
+  - name: lint
+    type: shell
+    command: "npm run lint"
+  - name: test
+    type: shell
+    command: "npm test"
+    continue_on_error: true
+  - name: build
+    type: shell
+    command: "npm run build"
+    if: "{{test.exit_code}} == 0"
+  - name: notify
+    type: echo
+    message: "构建完成: {{build.output}}"
+
+支持的步骤类型:
+  - tool: 调用插件工具 (tool, args)
+  - shell: 执行系统命令 (command, cwd, timeout)
+  - echo: 打印消息 (message)
+  - wait: 等待 (seconds)
+  - set: 设置变量 (variables)
+  - if: 条件分支 (condition, then, else)
+
+变量引用:
+  {{steps.step_name.output}}  - 步骤输出
+  {{steps.step_name.exit_code}} - 步骤退出码
+  {{env.VAR_NAME}} - 环境变量
+  {{args.param}} - 工作流参数"""
+
+    def set_cli(self, cli):
+        self.cli = cli
+        self.engine = WorkflowEngine(cli)
+        cli.register_liugin_command('workflow', self.command_handler)
+        cli.register_liugin_command('wf', self.command_handler)
+
+    def command_handler(self, args):
+        return self.handle(args)
+
+    def get_tool_info(self):
+        return {
+            "name": "workflow",
+            "description": "工作流引擎 - YAML 定义的多步骤自动化流程，支持变量传递、条件分支、断点续跑",
+            "keywords": ["workflow", "工作流", "流程", "自动化", "pipeline", "deploy", "build", "run"],
+            "usage": self.usage
+        }
+
+    def get_mcp_definition(self):
+        return {
+            "name": "workflow",
+            "description": "工作流引擎 - 执行、管理 YAML 定义的自动化工作流",
+            "inputSchema": {
+                "type": "object",
+                "properties": {
+                    "operation": {
+                        "type": "string",
+                        "enum": ["run", "list", "states", "status", "resume", "abort"],
+                        "description": "操作类型"
+                    },
+                    "target": {
+                        "type": "string",
+                        "description": "工作流文件路径(run) 或工作流ID(status/resume/abort)"
+                    },
+                    "args": {
+                        "type": "object",
+                        "description": "工作流参数 (run 时使用)"
+                    }
+                },
+                "required": ["operation"]
+            }
+        }
+
+    def convert_mcp_args(self, arguments):
+        op = arguments.get("operation", "")
+        target = arguments.get("target", "")
+        args = arguments.get("args", "")
+        if args and isinstance(args, dict):
+            args_str = ' '.join(f'{k}={v}' for k, v in args.items())
+            return f"{op} {target} {args_str}".strip()
+        return f"{op} {target}".strip()
+
+    def handle(self, args):
+        try:
+            args = args.strip()
+            if not args:
+                return self.usage
+
+            parts = args.split(None, 2)
+            action = parts[0].lower()
+
+            if action == 'run':
+                if len(parts) < 2:
+                    return "❌ 请指定工作流文件: workflow run <file.yaml>"
+                path = parts[1]
+                # 解析内联参数
+                wf_args = {}
+                if len(parts) > 2:
+                    for part in parts[2].split():
+                        if '=' in part:
+                            k, v = part.split('=', 1)
+                            wf_args[k] = v
+                return self.engine.run(path, wf_args)
+
+            elif action == 'list':
+                return self.engine.list_workflows()
+
+            elif action == 'states':
+                return self.engine.list_states()
+
+            elif action == 'status':
+                if len(parts) < 2:
+                    return "❌ 请指定工作流 ID: workflow status <id>"
+                return self.engine.status(parts[1])
+
+            elif action == 'resume':
+                if len(parts) < 2:
+                    return "❌ 请指定工作流 ID: workflow resume <id>"
+                return self.engine.resume(parts[1])
+
+            elif action == 'abort':
+                if len(parts) < 2:
+                    return "❌ 请指定工作流 ID: workflow abort <id>"
+                return self.engine.abort(parts[1])
+
+            else:
+                # 默认当作 run
+                if os.path.exists(action):
+                    return self.engine.run(action)
+                return f"❌ 未知操作: {action}\n\n{self.usage}"
+
+        except Exception as e:
+            return f"❌ 工作流操作错误: {e}"

unified_tool_manager.py

@@ -1,8 +1,6 @@
 """
 统一工具管理器 - 协调 Liugin、Skill 和 MCP 三套协议
 自动判断并路由到正确的处理器
-
-集成插件内存管理: TTL 自动淘汰 + 懒加载
 """
 
 import os
@@ -29,24 +27,9 @@ class UnifiedToolManager:
     - 回退到 Liugin 协议
     - 支持强制指定协议
     - MCP 格式调用统一入口
-
-    内存管理:
-    - 空闲插件自动淘汰（TTL 机制）
-    - 调用时自动懒加载回来
-    - 核心插件可钉住不淘汰
     """
 
-    # 内存管理默认配置
+    def __init__(self, cli_instance=None):
-    DEFAULT_PLUGIN_TTL = 300       # 5分钟
-    DEFAULT_MAX_IDLE_PLUGINS = 10  # 最多保留10个空闲插件
-    DEFAULT_CHECK_INTERVAL = 60    # 60秒检查一次
-    DEFAULT_PINNED_PLUGINS = ['tool_search']  # 默认钉住的插件
-
-    def __init__(self, cli_instance=None,
-                 plugin_ttl: int = None,
-                 max_idle_plugins: int = None,
-                 check_interval: int = None,
-                 pinned_plugins: list = None):
         self.cli = cli_instance
         self._lock = threading.RLock()
 
@@ -63,62 +46,39 @@ class UnifiedToolManager:
         # MCP 工具定义缓存
         self._mcp_tools_cache: Optional[List[Dict]] = None
 
-        # ── 原始文件路径缓存（用于懒加载） ──
-        self._plugin_paths: Dict[str, str] = {}   # name -> file_path
-        self._skill_dirs: Dict[str, str] = {}     # name -> skill_dir
-
-        # ── 插件内存管理器 ──
-        from xcli_core.plugin_memory_manager import PluginMemoryManager
-        self._memory_mgr = PluginMemoryManager(
-            plugin_ttl=plugin_ttl or self.DEFAULT_PLUGIN_TTL,
-            max_idle_plugins=max_idle_plugins or self.DEFAULT_MAX_IDLE_PLUGINS,
-            check_interval=check_interval or self.DEFAULT_CHECK_INTERVAL,
-            pinned_plugins=pinned_plugins or self.DEFAULT_PINNED_PLUGINS,
-        )
-        self._memory_mgr.set_callbacks(
-            load_fn=self._lazy_load_plugin,
-            unload_fn=self._unload_plugin,
-        )
-    
     def initialize(self, plugins_dir: str = "plugins", skills_dir: str = "skills") -> None:
-        """
+        """初始化并加载所有工具"""
-        初始化并加载所有工具
-
-        Args:
-            plugins_dir: Liugin 目录
-            skills_dir: Skill 目录
-        """
-        # 加载 Liugins
         self._plugins_dir = plugins_dir
         self._skills_dir = skills_dir
         self._load_plugins(plugins_dir)
-
-        # 加载 Skills
         self._load_skills(skills_dir)
-
-        # 构建缓存
         self._build_mcp_cache()
 
-        # 启动内存管理器
-        self._memory_mgr.start()
-
         print(f"{Fore.GREEN}[统一工具管理器] 加载完成: {len(self._tools)} 个工具{Style.RESET_ALL}")
         print(f"  - Liugin 协议: {sum(1 for p in self._protocol_map.values() if p == 'liugin')}")
         print(f"  - Skill 协议: {sum(1 for p in self._protocol_map.values() if p == 'skill')}")
 
-        # 打印内存管理状态
+    def shutdown(self) -> None:
-        status = self._memory_mgr.get_status()
+        """关闭管理器，清理所有插件实例"""
-        print(f"  - 内存管理: TTL={status['ttl']}s, 最大空闲={status['max_idle']}, "
+        with self._lock:
-              f"钉住={status['pinned']}")
+            for name, tool in self._tools.items():
+                instance = tool.get('instance')
+                if instance and hasattr(instance, 'cleanup'):
+                    try:
+                        instance.cleanup()
+                    except Exception:
+                        pass
+            self._tools.clear()
+            self._protocol_map.clear()
+            self._mcp_tools_cache = None
 
     def _load_plugins(self, plugins_dir: str) -> None:
-        """加载 Liugin 协议工具（并注册到内存管理器）"""
+        """加载 Liugin 协议工具"""
         import importlib.util
 
         if not os.path.exists(plugins_dir):
             return
 
-        # 获取启用的插件列表
         enabled_plugins = self._get_enabled_plugins(plugins_dir)
 
         for filename in os.listdir(plugins_dir):
@@ -127,7 +87,6 @@ class UnifiedToolManager:
 
             plugin_name = filename[:-3]
 
-            # 检查是否启用
             if enabled_plugins is not None and plugin_name.lower() not in enabled_plugins:
                 continue
 
@@ -138,7 +97,6 @@ class UnifiedToolManager:
                 module = importlib.util.module_from_spec(spec)
                 spec.loader.exec_module(module)
 
-                # 兼容 Plugin 和 Liugin 两种类名
                 plugin_class = None
                 if hasattr(module, 'Liugin'):
                     plugin_class = module.Liugin
@@ -148,19 +106,16 @@ class UnifiedToolManager:
                 if plugin_class:
                     plugin_instance = plugin_class()
 
-                    # 设置 CLI 引用
                     if hasattr(plugin_instance, 'set_cli') and self.cli:
                         plugin_instance.set_cli(self.cli)
 
                     tool_info = plugin_instance.get_tool_info()
                     tool_name = tool_info.get('name', plugin_name)
 
-                    # 获取 MCP 工具定义（如果插件提供）
                     mcp_definition = None
                     if hasattr(plugin_instance, 'get_mcp_definition'):
                         mcp_definition = plugin_instance.get_mcp_definition()
 
-                    # 注册到统一工具表
                     self._tools[tool_name] = {
                         'name': tool_name,
                         'description': tool_info.get('description', ''),
@@ -173,22 +128,6 @@ class UnifiedToolManager:
                     }
                     self._protocol_map[tool_name] = 'liugin'
 
-                    # 记录路径（用于懒加载）
-                    self._plugin_paths[tool_name] = plugin_path
-
-                    # 注册到内存管理器
-                    self._memory_mgr.register(
-                        name=tool_name,
-                        path=plugin_path,
-                        protocol='liugin',
-                        info_cache={
-                            'name': tool_name,
-                            'description': tool_info.get('description', ''),
-                            'keywords': tool_info.get('keywords', []),
-                            'usage': tool_info.get('usage', ''),
-                        }
-                    )
-
             except Exception as e:
                 print(f"{Fore.RED}加载 Liugin '{plugin_name}' 失败: {e}{Style.RESET_ALL}")
 
@@ -203,11 +142,9 @@ class UnifiedToolManager:
 
             self._skill_loader = SkillLoader(skills_dir)
 
-            # 先发现所有技能
             discovered = self._skill_loader.discover_skills(skills_dir)
             print(f"{Fore.CYAN}发现 {len(discovered)} 个技能: {[s['name'] for s in discovered]}{Style.RESET_ALL}")
 
-            # 加载所有技能
             self._skill_loader.load_all_skills(skills_dir)
 
             loaded_count = len(self._skill_loader.skills)
@@ -216,24 +153,19 @@ class UnifiedToolManager:
                 return
 
             for skill_name, skill in self._skill_loader.skills.items():
-                # 设置 CLI 引用 (仅 Python 类技能)
                 if hasattr(skill, 'set_cli') and self.cli:
                     skill.set_cli(self.cli)
 
-                # 获取工具定义
                 tool_def = skill.get_tool_definition()
                 func_def = tool_def.get('function', {})
 
-                # 检查是否已有同名 Plugin，Skill 优先覆盖
                 if skill_name in self._tools:
                     old_protocol = self._protocol_map[skill_name]
                     print(f"{Fore.YELLOW}工具 '{skill_name}' 已存在({old_protocol})，使用 Skill 协议覆盖{Style.RESET_ALL}")
 
-                # 判断技能格式
                 is_markdown = isinstance(skill, MarkdownSkill)
                 skill_format = "markdown" if is_markdown else "python"
 
-                # 注册到统一工具表
                 self._tools[skill_name] = {
                     'name': skill_name,
                     'description': func_def.get('description', ''),
@@ -245,27 +177,10 @@ class UnifiedToolManager:
                     'instance': skill,
                     'tool_definition': tool_def,
                     'anthropic_definition': skill.get_anthropic_tool_definition(),
-                    'mcp_definition': tool_def  # Skill 的 OpenAI 格式就是 MCP 格式
+                    'mcp_definition': tool_def
                 }
                 self._protocol_map[skill_name] = 'skill'
 
-                # 记录路径（用于懒加载）
-                skill_path = os.path.join(skills_dir, skill_name)
-                self._skill_dirs[skill_name] = skill_path
-
-                # 注册到内存管理器
-                self._memory_mgr.register(
-                    name=skill_name,
-                    path=skill_path,
-                    protocol='skill',
-                    info_cache={
-                        'name': skill_name,
-                        'description': func_def.get('description', ''),
-                        'keywords': getattr(skill, 'tags', []),
-                        'format': skill_format,
-                    }
-                )
-
                 print(f"{Fore.GREEN}已加载 Skill ({skill_format}): {skill_name}{Style.RESET_ALL}")
 
         except ImportError as e:
@@ -274,19 +189,12 @@ class UnifiedToolManager:
             print(f"{Fore.RED}加载 Skills 失败: {e}{Style.RESET_ALL}")
 
     def _create_skill_handler(self, skill) -> callable:
-        """
+        """为 Skill 创建兼容 Plugin 的 handler"""
-        为 Skill 创建兼容 Plugin 的 handler
-        
-        支持 Markdown 和 Python 两种技能格式
-        """
         from skills.base import MarkdownSkill, Skill
 
         def handler(args: str) -> str:
-            import json
-            
             kwargs = {}
 
-            # 解析参数
             args = args.strip()
             if args.startswith('{'):
                 try:
@@ -294,7 +202,6 @@ class UnifiedToolManager:
                 except json.JSONDecodeError:
                     kwargs = {'operation': args}
             elif '=' in args:
-                # key=value 格式: "directory=. file_pattern=*.py"
                 for part in args.split():
                     if '=' in part:
                         k, v = part.split('=', 1)
@@ -304,11 +211,9 @@ class UnifiedToolManager:
             else:
                 kwargs = {'operation': args}
 
-            # Markdown 技能: 返回指令供 AI 执行
             if isinstance(skill, MarkdownSkill):
                 return f"[Skill: {skill.name}]\n{skill.instructions}"
 
-            # Python 技能: 直接执行
             if hasattr(skill, 'run'):
                 result = skill.run(**kwargs)
                 if result.success:
@@ -320,177 +225,14 @@ class UnifiedToolManager:
 
         return handler
 
-    # ==================== 懒加载/卸载 ====================
-
-    def _lazy_load_plugin(self, name: str) -> Optional[Dict[str, Any]]:
-        """
-        懒加载单个插件（由内存管理器回调）
-
-        Args:
-            name: 插件名
-
-        Returns:
-            tool_dict 如果成功，None 如果失败
-        """
-        import importlib.util
-
-        # ── 尝试 Liugin 插件 ──
-        plugin_path = self._plugin_paths.get(name)
-        if plugin_path and os.path.exists(plugin_path):
-            return self._load_single_plugin(name, plugin_path)
-
-        # ── 尝试 Skill ──
-        skill_dir = self._skill_dirs.get(name)
-        if skill_dir and os.path.exists(skill_dir):
-            return self._load_single_skill(name, skill_dir)
-
-        return None
-
-    def _load_single_plugin(self, name: str, plugin_path: str) -> Optional[Dict[str, Any]]:
-        """加载单个 Liugin 插件"""
-        import importlib.util
-
-        try:
-            plugin_name = os.path.splitext(os.path.basename(plugin_path))[0]
-            spec = importlib.util.spec_from_file_location(plugin_name, plugin_path)
-            module = importlib.util.module_from_spec(spec)
-            spec.loader.exec_module(module)
-
-            plugin_class = getattr(module, 'Liugin', None) or getattr(module, 'Plugin', None)
-            if not plugin_class:
-                return None
-
-            plugin_instance = plugin_class()
-            if hasattr(plugin_instance, 'set_cli') and self.cli:
-                plugin_instance.set_cli(self.cli)
-
-            tool_info = plugin_instance.get_tool_info()
-            tool_name = tool_info.get('name', name)
-
-            mcp_definition = None
-            if hasattr(plugin_instance, 'get_mcp_definition'):
-                mcp_definition = plugin_instance.get_mcp_definition()
-
-            tool_dict = {
-                'name': tool_name,
-                'description': tool_info.get('description', ''),
-                'keywords': tool_info.get('keywords', []),
-                'usage': tool_info.get('usage', ''),
-                'handler': plugin_instance.handle,
-                'protocol': 'liugin',
-                'instance': plugin_instance,
-                'mcp_definition': mcp_definition
-            }
-
-            with self._lock:
-                self._tools[tool_name] = tool_dict
-                self._protocol_map[tool_name] = 'liugin'
-
-            return tool_dict
-
-        except Exception as e:
-            logger.error(f"懒加载插件 '{name}' 失败: {e}")
-            return None
-
-    def _load_single_skill(self, name: str, skill_dir: str) -> Optional[Dict[str, Any]]:
-        """加载单个 Skill"""
-        try:
-            from skills.base import SkillLoader, MarkdownSkill, Skill
-
-            loader = SkillLoader(os.path.dirname(skill_dir))
-            skill = loader.load_skill(skill_dir)
-
-            if not skill:
-                return None
-
-            if hasattr(skill, 'set_cli') and self.cli:
-                skill.set_cli(self.cli)
-
-            tool_def = skill.get_tool_definition()
-            func_def = tool_def.get('function', {})
-
-            is_markdown = isinstance(skill, MarkdownSkill)
-            skill_format = "markdown" if is_markdown else "python"
-
-            tool_dict = {
-                'name': name,
-                'description': func_def.get('description', ''),
-                'keywords': getattr(skill, 'tags', []),
-                'usage': getattr(skill, 'instructions', '')[:500] if is_markdown else getattr(skill, 'usage', ''),
-                'handler': self._create_skill_handler(skill),
-                'protocol': 'skill',
-                'format': skill_format,
-                'instance': skill,
-                'tool_definition': tool_def,
-                'anthropic_definition': skill.get_anthropic_tool_definition(),
-                'mcp_definition': tool_def
-            }
-
-            with self._lock:
-                self._tools[name] = tool_dict
-                self._protocol_map[name] = 'skill'
-
-            return tool_dict
-
-        except Exception as e:
-            logger.error(f"懒加载 Skill '{name}' 失败: {e}")
-            return None
-
-    def _unload_plugin(self, name: str) -> None:
-        """
-        卸载插件实例（由内存管理器回调）
-
-        从工具表中移除，释放实例引用，但保留路径缓存以便重新加载
-        """
-        with self._lock:
-            if name in self._tools:
-                tool = self._tools[name]
-                # 清理实例引用
-                instance = tool.get('instance')
-                if instance and hasattr(instance, 'cleanup'):
-                    try:
-                        instance.cleanup()
-                    except Exception:
-                        pass
-                del self._tools[name]
-                # 不删除 _protocol_map 和路径缓存，懒加载时需要
-
-        # 重建 MCP 缓存
-        self._build_mcp_cache()
-
-    def _ensure_loaded(self, name: str) -> Optional[Dict[str, Any]]:
-        """
-        确保插件已加载（调用入口统一使用）
-
-        如果插件已加载，更新使用时间戳并返回 tool_dict
-        如果插件已卸载，自动懒加载后返回 tool_dict
-        如果插件不存在，返回 None
-        """
-        # 先检查是否在工具表中
-        with self._lock:
-            tool = self._tools.get(name)
-            if tool:
-                self._memory_mgr.mark_used(name)
-                return tool
-
-        # 不在工具表中，可能是已卸载，尝试懒加载
-        result = self._memory_mgr.ensure_loaded(name)
-        if result:
-            self._memory_mgr.mark_used(name)
-            return result
-
-        return None
-
     def _get_enabled_plugins(self, plugins_dir: str) -> Optional[List[str]]:
         """获取启用的插件列表"""
-        # 检查环境变量
         enabled_env = os.environ.get("XIAOLI_ENABLED_PLUGINS")
         if enabled_env is not None:
             if enabled_env == "":
                 return []
             return [p.lower() for p in enabled_env.split(",")]
 
-        # 检查配置文件
         config_file = os.path.join(os.path.dirname(plugins_dir), "plugins_config.py")
         if os.path.exists(config_file):
             try:
@@ -509,7 +251,7 @@ class UnifiedToolManager:
             except Exception:
                 pass
 
-        return None  # 加载全部
+        return None
 
     def _build_mcp_cache(self) -> None:
         """构建 MCP 工具定义缓存（线程安全）"""
@@ -518,56 +260,28 @@ class UnifiedToolManager:
             for tool in self._tools.values():
                 mcp_def = tool.get('mcp_definition')
                 if mcp_def:
-                    # 确保符合 MCP 标准
                     normalized = self._normalize_mcp_definition(mcp_def)
                     self._mcp_tools_cache.append(normalized)
                 else:
-                    # 为没有 MCP 定义的 Plugin 生成默认定义
                     self._mcp_tools_cache.append(self._generate_mcp_definition(tool))
 
     def _normalize_mcp_definition(self, mcp_def: Dict[str, Any]) -> Dict[str, Any]:
-        """
+        """标准化 MCP 工具定义"""
-        标准化 MCP 工具定义
-        
-        支持多种输入格式并转换为标准 MCP 格式
-        """
-        # 如果已经是 MCP 格式（有 name 或 inputSchema）
         if 'name' in mcp_def and ('inputSchema' in mcp_def or 'description' in mcp_def):
             return mcp_def
 
-        # 如果是 OpenAI Function Calling 格式，转换
         if mcp_def.get('type') == 'function' and 'function' in mcp_def:
             func = mcp_def['function']
             return {
                 "name": func.get('name', ''),
                 "description": func.get('description', ''),
-                "inputSchema": func.get('parameters', {
+                "inputSchema": func.get('parameters', {"type": "object", "properties": {}})
-                    "type": "object",
-                    "properties": {}
-                })
             }
 
-        # 其他情况，直接返回
         return mcp_def
 
     def _generate_mcp_definition(self, tool: Dict[str, Any]) -> Dict[str, Any]:
-        """
+        """为 Plugin 生成符合 MCP 标准的工具定义"""
-        为 Plugin 生成符合 MCP 标准的工具定义
-        
-        MCP 官方格式 (2025-03-26):
-        {
-            "name": "tool_name",
-            "title": "Human Readable Name",  // optional
-            "description": "Human-readable description",
-            "inputSchema": {
-                "type": "object",
-                "properties": {...},
-                "required": [...]
-            },
-            "outputSchema": {...},  // optional
-            "annotations": {...}    // optional
-        }
-        """
         return {
             "name": tool['name'],
             "description": tool.get('description', ''),
@@ -591,26 +305,16 @@ class UnifiedToolManager:
         return list(self._tools.values())
 
     def get_tool(self, name: str) -> Optional[Dict[str, Any]]:
-        """获取指定工具（支持懒加载）"""
+        """获取指定工具"""
-        return self._ensure_loaded(name)
+        return self._tools.get(name)
 
     def get_protocol(self, name: str) -> Optional[str]:
         """获取工具使用的协议"""
         return self._protocol_map.get(name)
 
     def execute(self, tool_name: str, args: str = "", **kwargs) -> Dict[str, Any]:
-        """
+        """执行工具调用"""
-        执行工具调用（支持懒加载）
+        tool = self._tools.get(tool_name)
-
-        Args:
-            tool_name: 工具名称
-            args: 参数字符串（Plugin 格式）
-            **kwargs: 关键字参数（Skill 格式或 MCP arguments）
-
-        Returns:
-            {"result": "..."} 格式的结果
-        """
-        tool = self._ensure_loaded(tool_name)
 
         if not tool:
             return {"result": f"未找到工具: {tool_name}"}
@@ -621,20 +325,16 @@ class UnifiedToolManager:
             instance = tool.get('instance')
 
             if protocol == 'skill' and kwargs:
-                # Skill 协议，使用 kwargs
                 skill = tool['instance']
                 result = skill.run(**kwargs)
                 return {"result": result.message or str(result.data) if result.success else f"错误: {result.error}"}
 
             elif protocol == 'liugin' and kwargs and not args:
-                # Liugin 协议但有 MCP 结构化参数，需要转换
-                # 检查插件是否有 convert_mcp_args 方法
                 if instance and hasattr(instance, 'convert_mcp_args'):
                     args = instance.convert_mcp_args(kwargs)
                 else:
                     args = self._convert_mcp_args_to_plugin_args(tool_name, kwargs, instance)
 
-            # 执行 handler
             result = handler(args)
             return {"result": result if result is not None else "无结果"}
 
@@ -644,43 +344,15 @@ class UnifiedToolManager:
     # ==================== MCP 协议支持 ====================
 
     def get_mcp_tools(self) -> List[Dict[str, Any]]:
-        """
+        """获取 MCP 格式的工具定义列表"""
-        获取 MCP 格式的工具定义列表
-        
-        MCP (Model Context Protocol) 使用 OpenAI Function Calling 格式
-        作为标准工具定义格式
-        
-        Returns:
-            List of MCP tool definitions
-        """
         with self._lock:
             if self._mcp_tools_cache is None:
                 self._build_mcp_cache()
-            return self._mcp_tools_cache.copy()  # 返回副本，避免外部修改
+            return self._mcp_tools_cache.copy()
 
     def mcp_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
-        """
+        """MCP 格式的工具调用入口"""
-        MCP 格式的工具调用入口（支持懒加载）
+        tool = self._tools.get(tool_name)
-
-        这是 MCP 协议的主要调用接口，接收结构化参数并返回结果
-
-        Args:
-            tool_name: 工具名称
-            arguments: 结构化参数 (JSON对象)
-
-        Returns:
-            MCP 标准响应格式:
-            {
-                "content": [
-                    {
-                        "type": "text",
-                        "text": "结果内容"
-                    }
-                ],
-                "isError": false
-            }
-        """
-        tool = self._ensure_loaded(tool_name)
 
         if not tool:
             return self._mcp_error(f"未找到工具: {tool_name}")
@@ -690,13 +362,11 @@ class UnifiedToolManager:
             instance = tool.get('instance')
 
             if protocol == 'liugin':
-                # Liugin 协议: 将 arguments 转换为 args 字符串
                 args = self._convert_mcp_args_to_plugin_args(tool_name, arguments, instance)
                 result = tool['handler'](args)
                 return self._mcp_success(result if result else "执行完成")
 
             elif protocol == 'skill':
-                # Skill 协议: 直接传递 arguments 作为 kwargs
                 from skills.base import MarkdownSkill
 
                 if isinstance(instance, MarkdownSkill):
@@ -716,36 +386,20 @@ class UnifiedToolManager:
 
     def _convert_mcp_args_to_plugin_args(self, tool_name: str, arguments: Dict[str, Any],
                                           instance: Any) -> str:
-        """
+        """将 MCP 结构化参数转换为 Plugin 的 args 字符串"""
-        将 MCP 结构化参数转换为 Plugin 的 args 字符串
-        
-        Args:
-            tool_name: 工具名称
-            arguments: MCP 参数对象
-            instance: 插件实例（可能有自定义转换方法）
-        
-        Returns:
-            args 字符串
-        """
-        # 检查插件是否有自定义的 MCP 参数转换方法
         if instance and hasattr(instance, 'convert_mcp_args'):
             return instance.convert_mcp_args(arguments)
 
-        # 默认转换策略
         if 'args' in arguments:
-            # 如果有 args 字段，直接使用
             return arguments['args']
 
-        # 将 arguments 转换为命令行格式
         parts = []
 
-        # 第一个参数通常是操作/命令
         if 'operation' in arguments:
             parts.append(arguments['operation'])
         elif 'action' in arguments:
             parts.append(arguments['action'])
 
-        # 添加其他参数
         for key, value in arguments.items():
             if key in ('operation', 'action'):
                 continue
@@ -764,51 +418,20 @@ class UnifiedToolManager:
         return ' '.join(parts)
 
     def _quote_if_needed(self, s: str) -> str:
-        """如果字符串包含空格，添加引号"""
         if ' ' in s or '"' in s:
             return f'"{s}"'
         return s
 
     def _mcp_success(self, text: str) -> Dict[str, Any]:
-        """生成 MCP 成功响应"""
+        return {"content": [{"type": "text", "text": text}], "isError": False}
-        return {
-            "content": [
-                {
-                    "type": "text",
-                    "text": text
-                }
-            ],
-            "isError": False
-        }
 
     def _mcp_error(self, error_msg: str) -> Dict[str, Any]:
-        """生成 MCP 错误响应"""
+        return {"content": [{"type": "text", "text": error_msg}], "isError": True}
-        return {
-            "content": [
-                {
-                    "type": "text",
-                    "text": error_msg
-                }
-            ],
-            "isError": True
-        }
 
     # ==================== 兼容性 API ====================
 
     def get_openai_tools(self) -> List[Dict[str, Any]]:
-        """
+        """获取 OpenAI Function Calling 格式的工具定义"""
-        获取 OpenAI Function Calling 格式的工具定义
-        
-        OpenAI 格式:
-        {
-            "type": "function",
-            "function": {
-                "name": "...",
-                "description": "...",
-                "parameters": {...}
-            }
-        }
-        """
         tools = []
         for mcp_def in self.get_mcp_tools():
             tools.append({
@@ -816,54 +439,28 @@ class UnifiedToolManager:
                 "function": {
                     "name": mcp_def.get("name", ""),
                     "description": mcp_def.get("description", ""),
-                    "parameters": mcp_def.get("inputSchema", {
+                    "parameters": mcp_def.get("inputSchema", {"type": "object", "properties": {}})
-                        "type": "object",
-                        "properties": {}
-                    })
                 }
             })
         return tools
 
     def get_anthropic_tools(self) -> List[Dict[str, Any]]:
-        """
+        """获取 Anthropic Tool Use 格式的工具定义"""
-        获取 Anthropic Tool Use 格式的工具定义
-        
-        Anthropic 格式:
-        {
-            "name": "tool_name",
-            "description": "...",
-            "input_schema": {...}
-        }
-        """
         tools = []
         for mcp_def in self.get_mcp_tools():
             tools.append({
                 "name": mcp_def.get("name", ""),
                 "description": mcp_def.get("description", ""),
-                "input_schema": mcp_def.get("inputSchema", {
+                "input_schema": mcp_def.get("inputSchema", {"type": "object", "properties": {}})
-                    "type": "object",
-                    "properties": {}
-                })
             })
         return tools
 
     def search_tools(self, query: str, limit: int = 5) -> List[Dict[str, Any]]:
-        """
+        """搜索工具"""
-        搜索工具（包括已卸载的插件，使用缓存信息）
-
-        Args:
-            query: 搜索关键词
-            limit: 返回数量限制
-
-        Returns:
-            匹配的工具列表
-        """
         query_lower = query.lower()
-        # 标准化：移除所有分隔符（连字符、下划线、空格）
         query_normalized = query_lower.replace('-', '').replace('_', '').replace(' ', '')
         matches = []
 
-        # 搜索已加载的工具
         for tool in self._tools.values():
             score = self._score_tool_match(tool, query_lower, query_normalized)
             if score > 0:
@@ -877,27 +474,6 @@ class UnifiedToolManager:
                     'score': score
                 })
 
-        # 搜索已卸载但注册的插件（使用缓存信息）
-        for name in self._memory_mgr.get_unloaded_names():
-            # 跳过已加载的（上面已搜索）
-            if name in self._tools:
-                continue
-            info = self._memory_mgr.get_info_cache(name)
-            if info:
-                score = self._score_tool_match(info, query_lower, query_normalized)
-                if score > 0:
-                    matches.append({
-                        'name': info.get('name', name),
-                        'description': info.get('description', ''),
-                        'usage': info.get('usage', ''),
-                        'keywords': info.get('keywords', []),
-                        'protocol': info.get('protocol', 'liugin'),
-                        'instance': None,
-                        'score': score,
-                        '_unloaded': True,  # 标记为已卸载
-                    })
-
-        # 按分数排序
         matches.sort(key=lambda x: x['score'], reverse=True)
         return matches[:limit]
 
@@ -909,16 +485,12 @@ class UnifiedToolManager:
         desc = tool.get('description', '').lower()
         keywords = [kw.lower() for kw in tool.get('keywords', [])]
 
-        # 名称完全匹配（忽略分隔符差异）
         if query_lower == name or query_normalized == name_normalized:
             score += 100
-        # 名称包含（标准化后匹配）
         elif query_normalized in name_normalized or query_lower in name:
             score += 50
-        # 描述包含
         if query_lower in desc:
             score += 30
-        # 关键词匹配
         for kw in keywords:
             kw_normalized = kw.replace('-', '').replace('_', '')
             if query_lower in kw or query_normalized in kw_normalized:
@@ -928,16 +500,7 @@ class UnifiedToolManager:
 
     def register_tool(self, name: str, handler: callable, description: str = "",
                       protocol: str = "custom", mcp_definition: Dict = None, **kwargs) -> None:
-        """
+        """动态注册工具"""
-        动态注册工具
-        
-        Args:
-            name: 工具名称
-            handler: 处理函数
-            description: 描述
-            protocol: 协议类型
-            mcp_definition: MCP 工具定义（可选）
-        """
         with self._lock:
             self._tools[name] = {
                 'name': name,
@@ -949,7 +512,6 @@ class UnifiedToolManager:
             }
             self._protocol_map[name] = protocol
 
-            # 更新缓存
             if self._mcp_tools_cache is not None:
                 if mcp_definition:
                     self._mcp_tools_cache.append(mcp_definition)
@@ -960,73 +522,42 @@ class UnifiedToolManager:
         """注销工具"""
         with self._lock:
             if name in self._tools:
+                instance = self._tools[name].get('instance')
+                if instance and hasattr(instance, 'cleanup'):
+                    try:
+                        instance.cleanup()
+                    except Exception:
+                        pass
                 del self._tools[name]
                 del self._protocol_map[name]
-                # 同时从内存管理器注销
-                if name in self._plugin_paths:
-                    del self._plugin_paths[name]
-                if name in self._skill_dirs:
-                    del self._skill_dirs[name]
-                # 重建缓存
                 self._build_mcp_cache()
                 return True
             return False
 
     def get_all_tools_info(self) -> List[Dict[str, Any]]:
-        """
+        """获取所有工具的简要信息，用于系统提示词"""
-        获取所有工具的简要信息（包括已卸载的）
-
-        用于系统提示词，让 AI 知道有哪些工具可用
-        已卸载的工具只有基本信息，调用时会自动加载
-        """
         result = []
-        # 已加载的
         for tool in self._tools.values():
             result.append({
                 'name': tool['name'],
                 'description': tool.get('description', ''),
                 'keywords': tool.get('keywords', []),
                 'protocol': tool.get('protocol', ''),
-                'loaded': True,
-            })
-        # 已卸载的（从内存管理器缓存读取）
-        for name in self._memory_mgr.get_unloaded_names():
-            if name in self._tools:
-                continue
-            info = self._memory_mgr.get_info_cache(name)
-            if info:
-                result.append({
-                    'name': info.get('name', name),
-                    'description': info.get('description', ''),
-                    'keywords': info.get('keywords', []),
-                    'protocol': info.get('protocol', 'liugin'),
-                    'loaded': False,
             })
         return result
 
     def get_tool_by_name(self, name: str) -> Optional[Dict[str, Any]]:
-        """根据名称获取工具（兼容 LiuginManager 接口，支持懒加载）"""
+        """根据名称获取工具（兼容 LiuginManager 接口）"""
-        return self._ensure_loaded(name)
+        return self._tools.get(name)
-
-    @property
-    def memory_manager(self):
-        """获取内存管理器实例"""
-        return self._memory_mgr
 
     def get_skills_system_prompt(self) -> str:
-        """
+        """获取所有技能的系统提示词"""
-        获取所有技能的系统提示词
-        
-        用于注入到 AI 的 system prompt 中，
-        让 AI 了解如何使用这些技能
-        """
         skill_prompts = []
 
         for name, tool in self._tools.items():
             if tool['protocol'] == 'skill':
                 instance = tool.get('instance')
                 if instance:
-                    # 获取技能指令
                     instructions = getattr(instance, 'instructions', '')
                     scripts_dir = getattr(instance, 'scripts_dir', '')
 
@@ -1034,7 +565,6 @@ class UnifiedToolManager:
                     prompt_parts.append(f"描述: {tool['description']}")
 
                     if instructions:
-                        # 截取关键部分，避免过长
                         prompt_parts.append(f"\n{instructions}")
 
                     if scripts_dir:
@@ -1049,14 +579,9 @@ class UnifiedToolManager:
         return ""
 
     def get_tools_prompt(self) -> str:
-        """
+        """获取工具使用提示词（兼容旧接口）"""
-        获取工具使用提示词（兼容旧接口）
-        
-        包含 Plugin 和 Skill 两种协议的工具说明
-        """
         prompts = []
 
-        # Liugin 协议工具
         plugin_tools = [t for t in self._tools.values() if t['protocol'] == 'liugin']
         if plugin_tools:
             prompts.append("## Liugin 工具")
@@ -1065,7 +590,6 @@ class UnifiedToolManager:
                 if usage:
                     prompts.append(f"### {tool['name']}\n{usage}\n")
 
-        # Skill 协议工具（使用系统提示词）
         skills_prompt = self.get_skills_system_prompt()
         if skills_prompt:
             prompts.append(skills_prompt)
@@ -1075,45 +599,16 @@ class UnifiedToolManager:
     # ==================== MCP Server 模式支持 ====================
 
     def get_mcp_server_info(self) -> Dict[str, Any]:
-        """
+        """获取 MCP Server 信息"""
-        获取 MCP Server 信息
-        
-        用于 MCP 客户端连接时的初始化响应
-        """
         return {
             "name": "xiaoli-tool-server",
             "version": "1.0.0",
             "protocolVersion": "2024-11-05",
-            "capabilities": {
+            "capabilities": {"tools": {"listChanged": True}}
-                "tools": {
-                    "listChanged": True
-                }
-            }
         }
 
     def handle_mcp_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
-        """
+        """处理 MCP 协议请求"""
-        处理 MCP 协议请求
-        
-        统一处理 MCP 客户端的各种请求
-        
-        Args:
-            request: MCP 请求对象
-            {
-                "jsonrpc": "2.0",
-                "id": 1,
-                "method": "tools/list" | "tools/call" | ...,
-                "params": {...}
-            }
-        
-        Returns:
-            MCP 响应对象
-            {
-                "jsonrpc": "2.0",
-                "id": 1,
-                "result": {...}
-            }
-        """
         method = request.get('method', '')
         params = request.get('params', {})
         req_id = request.get('id')

workflows/example.yaml

@@ -0,0 +1,31 @@
+# 示例工作流 - 代码质量检查与构建
+name: code-quality
+description: 代码质量检查、测试、构建的自动化流程
+
+steps:
+  - name: info
+    type: echo
+    message: "🚀 开始代码质量检查流程"
+
+  - name: lint
+    type: shell
+    command: "echo '✓ 代码风格检查通过'"
+    continue_on_error: true
+
+  - name: test
+    type: shell
+    command: "echo '✓ 单元测试通过 (42/42)'"
+    continue_on_error: true
+
+  - name: check_results
+    type: echo
+    message: "Lint: {{lint.output}} | Test: {{test.output}}"
+
+  - name: build
+    type: shell
+    command: "echo '✓ 构建完成 → dist/'"
+    if: "{{lint.exit_code}} == 0"
+
+  - name: done
+    type: echo
+    message: "✅ 流程完成，构建产物: {{build.output}}"

xcli_core/plugin_memory_manager.py

@@ -1,355 +0,0 @@
-"""
-插件内存管理器 - TTL 自动淘汰 + 懒加载
-
-机制:
-  1. 每个插件记录 last_used 时间戳
-  2. 后台线程定期扫描，超过 TTL 未使用的插件从内存卸载
-  3. AI 调用已卸载的插件时，自动重新加载
-  4. 核心插件（tool_search 等）可钉住，永不淘汰
-
-配置:
-  - plugin_ttl: 插件空闲多久后淘汰（秒），默认 300（5分钟）
-  - max_idle_plugins: 最多保留多少个空闲插件，超出后淘汰最久未用的，默认 10
-  - check_interval: 后台扫描间隔（秒），默认 60
-  - pinned_plugins: 钉住的插件名列表，永不淘汰
-"""
-
-import os
-import time
-import threading
-import importlib.util
-from typing import Dict, Optional, List, Any, Callable
-from colorama import Fore, Style
-
-from .config import logger
-
-
-class PluginMemoryManager:
-    """插件内存管理器 - 自动淘汰空闲插件，按需懒加载"""
-
-    # 默认钉住的插件（核心工具，不淘汰）
-    DEFAULT_PINNED = {'tool_search'}
-
-    def __init__(self,
-                 plugin_ttl: int = 300,
-                 max_idle_plugins: int = 10,
-                 check_interval: int = 60,
-                 pinned_plugins: Optional[List[str]] = None):
-        self.plugin_ttl = plugin_ttl
-        self.max_idle_plugins = max_idle_plugins
-        self.check_interval = check_interval
-        self.pinned = set(pinned_plugins or self.DEFAULT_PINNED)
-
-        # ── 插件元数据 ──
-        # name -> {path, protocol, last_used, info_cache, ...}
-        self._registry: Dict[str, Dict[str, Any]] = {}
-        self._lock = threading.RLock()
-
-        # ── 已加载的实例引用（由 UnifiedToolManager 提供注册/注销回调） ──
-        self._load_callback: Optional[Callable] = None   # (name) -> tool_dict or None
-        self._unload_callback: Optional[Callable] = None  # (name) -> None
-
-        # ── 后台清理线程 ──
-        self._cleanup_thread: Optional[threading.Thread] = None
-        self._running = False
-
-        # ── 统计 ──
-        self.stats = {
-            'evictions': 0,
-            'lazy_loads': 0,
-            'pins_hit': 0,
-        }
-
-    # ==================== 生命周期 ====================
-
-    def start(self) -> None:
-        """启动后台清理线程"""
-        if self._running:
-            return
-        self._running = True
-        self._cleanup_thread = threading.Thread(
-            target=self._cleanup_loop, daemon=True, name="plugin-gc"
-        )
-        self._cleanup_thread.start()
-        print(f"{Fore.CYAN}[插件内存管理] 已启动 (TTL={self.plugin_ttl}s, "
-              f"最大空闲={self.max_idle_plugins}, 检查间隔={self.check_interval}s){Style.RESET_ALL}")
-
-    def stop(self) -> None:
-        """停止后台清理线程"""
-        self._running = False
-        if self._cleanup_thread:
-            self._cleanup_thread.join(timeout=5)
-            self._cleanup_thread = None
-
-    # ==================== 回调注册 ====================
-
-    def set_callbacks(self,
-                      load_fn: Callable[[str], Optional[Dict]],
-                      unload_fn: Callable[[str], None]) -> None:
-        """
-        注册加载/卸载回调（由 UnifiedToolManager 提供）
-
-        Args:
-            load_fn: 加载插件函数，接收插件名，返回 tool_dict 或 None
-            unload_fn: 卸载插件函数，接收插件名
-        """
-        self._load_callback = load_fn
-        self._unload_callback = unload_fn
-
-    # ==================== 插件注册 ====================
-
-    def register(self, name: str, path: str, protocol: str = 'liugin',
-                 info_cache: Optional[Dict] = None) -> None:
-        """
-        注册一个插件的元数据（加载时调用）
-
-        Args:
-            name: 插件名
-            path: 插件文件路径
-            protocol: 协议类型 (liugin/skill)
-            info_cache: 缓存的工具信息（name, description, keywords 等）
-        """
-        with self._lock:
-            self._registry[name] = {
-                'path': path,
-                'protocol': protocol,
-                'last_used': time.time(),
-                'loaded': True,
-                'info_cache': info_cache or {},
-            }
-
-    def mark_used(self, name: str) -> None:
-        """标记插件被使用（每次工具调用时）"""
-        with self._lock:
-            if name in self._registry:
-                self._registry[name]['last_used'] = time.time()
-
-    def mark_loaded(self, name: str) -> None:
-        """标记插件已加载"""
-        with self._lock:
-            if name in self._registry:
-                self._registry[name]['loaded'] = True
-                self._registry[name]['last_used'] = time.time()
-
-    def mark_unloaded(self, name: str) -> None:
-        """标记插件已卸载"""
-        with self._lock:
-            if name in self._registry:
-                self._registry[name]['loaded'] = False
-
-    # ==================== 懒加载 ====================
-
-    def ensure_loaded(self, name: str) -> Optional[Dict]:
-        """
-        确保插件已加载，如果已卸载则自动重新加载
-
-        Args:
-            name: 插件名
-
-        Returns:
-            tool_dict 如果加载成功，None 如果失败
-        """
-        with self._lock:
-            entry = self._registry.get(name)
-            if not entry:
-                return None  # 未注册的插件，不管
-
-            if entry['loaded']:
-                # 已加载，更新时间戳
-                entry['last_used'] = time.time()
-                return None  # 返回 None 表示不需要重新加载
-
-        # 需要重新加载（锁外执行，避免死锁）
-        print(f"{Fore.CYAN}[插件内存管理] 懒加载插件: {name}{Style.RESET_ALL}")
-        result = self._do_load(name)
-        if result:
-            self.stats['lazy_loads'] += 1
-            print(f"{Fore.GREEN}[插件内存管理] 懒加载成功: {name}{Style.RESET_ALL}")
-        else:
-            print(f"{Fore.RED}[插件内存管理] 懒加载失败: {name}{Style.RESET_ALL}")
-        return result
-
-    def _do_load(self, name: str) -> Optional[Dict]:
-        """执行实际的插件加载"""
-        if not self._load_callback:
-            return None
-
-        with self._lock:
-            entry = self._registry.get(name)
-            if not entry:
-                return None
-
-        try:
-            result = self._load_callback(name)
-            if result:
-                with self._lock:
-                    if name in self._registry:
-                        self._registry[name]['loaded'] = True
-                        self._registry[name]['last_used'] = time.time()
-                return result
-        except Exception as e:
-            logger.error(f"懒加载插件 '{name}' 失败: {e}")
-        return None
-
-    # ==================== 淘汰 ====================
-
-    def evict(self, name: str) -> bool:
-        """
-        淘汰指定插件
-
-        Args:
-            name: 插件名
-
-        Returns:
-            是否成功淘汰
-        """
-        if name in self.pinned:
-            self.stats['pins_hit'] += 1
-            return False
-
-        with self._lock:
-            entry = self._registry.get(name)
-            if not entry or not entry['loaded']:
-                return False
-
-        # 执行卸载（锁外）
-        if self._unload_callback:
-            try:
-                self._unload_callback(name)
-            except Exception as e:
-                logger.error(f"卸载插件 '{name}' 失败: {e}")
-                return False
-
-        with self._lock:
-            if name in self._registry:
-                self._registry[name]['loaded'] = False
-
-        self.stats['evictions'] += 1
-        print(f"{Fore.YELLOW}[插件内存管理] 已淘汰: {name} "
-              f"(空闲 {self._idle_time(name):.0f}s){Style.RESET_ALL}")
-        return True
-
-    def _idle_time(self, name: str) -> float:
-        """获取插件空闲时间"""
-        with self._lock:
-            entry = self._registry.get(name)
-            if not entry:
-                return 0
-            return time.time() - entry['last_used']
-
-    # ==================== 后台清理 ====================
-
-    def _cleanup_loop(self) -> None:
-        """后台清理循环"""
-        while self._running:
-            try:
-                time.sleep(self.check_interval)
-                if not self._running:
-                    break
-                self._cleanup_once()
-            except Exception as e:
-                logger.error(f"插件清理线程异常: {e}")
-
-    def _cleanup_once(self) -> None:
-        """执行一次清理"""
-        with self._lock:
-            # 收集所有已加载且未钉住的插件
-            candidates = []
-            for name, entry in self._registry.items():
-                if entry['loaded'] and name not in self.pinned:
-                    idle = time.time() - entry['last_used']
-                    if idle >= self.plugin_ttl:
-                        candidates.append((name, idle))
-
-        if not candidates:
-            return
-
-        # 按空闲时间排序，最久未用的优先淘汰
-        candidates.sort(key=lambda x: x[1], reverse=True)
-
-        # 也检查总数限制
-        with self._lock:
-            loaded_count = sum(1 for e in self._registry.values()
-                              if e['loaded'] and e.get('name') not in self.pinned)
-
-        evicted = 0
-        for name, idle in candidates:
-            # 如果已超过最大空闲数限制 或 超过 TTL
-            if loaded_count - evicted > self.max_idle_plugins or idle >= self.plugin_ttl:
-                if self.evict(name):
-                    evicted += 1
-
-        if evicted:
-            print(f"{Fore.CYAN}[插件内存管理] 本轮淘汰 {evicted} 个插件, "
-                  f"累计淘汰 {self.stats['evictions']} 次, "
-                  f"懒加载 {self.stats['lazy_loads']} 次{Style.RESET_ALL}")
-
-    # ==================== 查询接口 ====================
-
-    def is_loaded(self, name: str) -> bool:
-        """检查插件是否已加载"""
-        with self._lock:
-            entry = self._registry.get(name)
-            return entry['loaded'] if entry else False
-
-    def is_registered(self, name: str) -> bool:
-        """检查插件是否已注册"""
-        with self._lock:
-            return name in self._registry
-
-    def get_info_cache(self, name: str) -> Optional[Dict]:
-        """获取插件的缓存信息（不需要加载实例）"""
-        with self._lock:
-            entry = self._registry.get(name)
-            return entry.get('info_cache') if entry else None
-
-    def get_all_registered(self) -> List[str]:
-        """获取所有已注册的插件名"""
-        with self._lock:
-            return list(self._registry.keys())
-
-    def get_loaded_names(self) -> List[str]:
-        """获取当前已加载的插件名"""
-        with self._lock:
-            return [n for n, e in self._registry.items() if e['loaded']]
-
-    def get_unloaded_names(self) -> List[str]:
-        """获取当前已卸载的插件名"""
-        with self._lock:
-            return [n for n, e in self._registry.items() if not e['loaded']]
-
-    def get_status(self) -> Dict[str, Any]:
-        """获取内存管理状态"""
-        with self._lock:
-            loaded = sum(1 for e in self._registry.values() if e['loaded'])
-            unloaded = sum(1 for e in self._registry.values() if not e['loaded'])
-            return {
-                'total': len(self._registry),
-                'loaded': loaded,
-                'unloaded': unloaded,
-                'pinned': list(self.pinned),
-                'ttl': self.plugin_ttl,
-                'max_idle': self.max_idle_plugins,
-                'check_interval': self.check_interval,
-                'stats': dict(self.stats),
-            }
-
-    def format_status(self) -> str:
-        """格式化状态信息（用于 /status 命令）"""
-        s = self.get_status()
-        lines = [
-            f"  插件内存管理:",
-            f"    总注册: {s['total']}  已加载: {s['loaded']}  已卸载: {s['unloaded']}",
-            f"    TTL: {s['ttl']}s  最大空闲: {s['max_idle']}  扫描间隔: {s['check_interval']}s",
-            f"    钉住: {', '.join(s['pinned']) or '无'}",
-            f"    统计: 淘汰 {s['stats']['evictions']} 次, "
-            f"懒加载 {s['stats']['lazy_loads']} 次, "
-            f"钉住保护 {s['stats']['pins_hit']} 次",
-        ]
-
-        # 显示已卸载的插件
-        unloaded = self.get_unloaded_names()
-        if unloaded:
-            lines.append(f"    已卸载: {', '.join(unloaded)}")
-
-        return '\n'.join(lines)

xcli_core/tui.py

@@ -562,13 +562,9 @@ if TEXTUAL_AVAILABLE:
             status_container.mount(Static(f"  工具: {len(self.bridge.tools())} 个", classes="tool-item"))
             status_container.mount(Static(f"  引擎: {len(self.bridge.engines())} 个", classes="tool-item"))
 
-            # 内存管理状态
+            # 工具数
-            if hasattr(self.cli, 'liugin_manager') and hasattr(self.cli.liugin_manager, 'memory_manager'):
+            if hasattr(self.cli, 'liugin_manager'):
-                mgr = self.cli.liugin_manager.memory_manager
+                status_container.mount(Static(f"  工具: {len(self.cli.liugin_manager.tools)} 个", classes="tool-item"))
-                s = mgr.get_status()
-                status_container.mount(Static(f"  已加载: {s['loaded']}", classes="tool-item"))
-                if s['unloaded'] > 0:
-                    status_container.mount(Static(f"  已卸载: {s['unloaded']}", classes="tool-item"))
 
         def _update_status(self, text: str):
             bar = self.query_one("#status-bar")
@@ -822,10 +818,7 @@ if TEXTUAL_AVAILABLE:
                 f"  工具数: {len(self.bridge.tools())}",
                 f"  对话数: {len(self.bridge.history())}",
             ]
-            # 内存管理状态
+
-            if hasattr(self.cli, 'liugin_manager') and hasattr(self.cli.liugin_manager, 'memory_manager'):
-                mgr = self.cli.liugin_manager.memory_manager
-                status_lines.append(mgr.format_status())
             self._system('\n'.join(status_lines))
 
         # ── AI 生成 ──