refactor: move admin management into session monitor, fix session_id enum format

- Remove standalone admins tab, replace with BotAdminsDialog in session monitor
- Add admin toggle button inline in chat header next to Active status
- Add BotAdminsDialog component with useBotAdmins hook
- Fix session_id written as LauncherTypes.PERSON_xxx instead of person_xxx
  (monitoring_helper.py x4, pipelinemgr.py x1 missing .value on launcher_type)
- Fix duplicate platform/person label in session list and chat header
- Migrate existing malformed session_id records in DB

README.md

@@ -5,7 +5,7 @@
 
 <div align="center">
 
-<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
+<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
 
 <h3>Production-grade platform for building agentic IM bots.</h3>
 <h4>Quickly build, debug, and ship AI bots to Slack, Discord, Telegram, WeChat, and more.</h4>
@@ -136,7 +136,7 @@ docker compose --profile all up -d
 | [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot)                             | GPU Platform | ✅     |
 | [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758)                                                    | GPU Platform | ✅     |
 | [接口 AI](https://jiekou.ai/)                                                                                     | Gateway      | ✅     |
-| [302.AI](https://share.302ai.cn/SuTG99)                                                                             | Gateway      | ✅     |
+| [302.AI](https://share.302.ai/SuTG99)                                                                             | Gateway      | ✅     |
 | [Qiniu](https://www.qiniu.com/ai/agent)                                                                           | Gateway      | ✅     |
 
 [→ View all integrations](https://link.langbot.app/en/docs/features)

README_CN.md

@@ -136,7 +136,7 @@ docker compose --profile all up -d
 | [优云智算](https://www.compshare.cn/?ytag=GPU_YY-gh_langbot) | GPU 平台 | ✅ |
 | [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | GPU 平台 | ✅ |
 | [接口 AI](https://jiekou.ai/) | 聚合平台 | ✅ |
-| [302.AI](https://share.302ai.cn/SuTG99) | 聚合平台 | ✅ |
+| [302.AI](https://share.302.ai/SuTG99) | 聚合平台 | ✅ |
 | [小马算力](https://www.tokenpony.cn/453z1) | 聚合平台 | ✅ |
 | [百宝箱Tbox](https://www.tbox.cn/open) | 智能体平台 | ✅ |
 | [七牛云Qiniu](https://www.qiniu.com/ai/agent) | 聚合平台 | ✅ |

README_ES.md

@@ -5,7 +5,7 @@
 
 <div align="center">
 
-<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
+<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
 
 <h3>Plataforma de grado de producción para construir bots de mensajería instantánea con agentes de IA.</h3>
 <h4>Construya, depure y despliegue bots de IA rápidamente en Slack, Discord, Telegram, WeChat y más.</h4>
@@ -135,7 +135,7 @@ docker compose --profile all up -d
 | [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | Plataforma GPU | ✅ |
 | [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758) | Plataforma GPU | ✅ |
 | [接口 AI](https://jiekou.ai/) | Pasarela | ✅ |
-| [302.AI](https://share.302ai.cn/SuTG99) | Pasarela | ✅ |
+| [302.AI](https://share.302.ai/SuTG99) | Pasarela | ✅ |
 | [Qiniu](https://www.qiniu.com/ai/agent) | Pasarela | ✅ |
 
 [→ Ver todas las integraciones](https://link.langbot.app/en/docs/features)

README_FR.md

@@ -5,7 +5,7 @@
 
 <div align="center">
 
-<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
+<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
 
 <h3>Plateforme de niveau production pour construire des bots de messagerie instantanée avec agents IA.</h3>
 <h4>Créez, déboguez et déployez rapidement des bots IA sur Slack, Discord, Telegram, WeChat et plus.</h4>
@@ -132,7 +132,7 @@ docker compose --profile all up -d
 | [ModelScope](https://modelscope.cn/docs/model-service/API-Inference/intro) | Passerelle | ✅ |
 | [GiteeAI](https://ai.gitee.com/) | Passerelle | ✅ |
 | [接口 AI](https://jiekou.ai/) | Passerelle | ✅ |
-| [302.AI](https://share.302ai.cn/SuTG99) | Passerelle | ✅ |
+| [302.AI](https://share.302.ai/SuTG99) | Passerelle | ✅ |
 | [CompShare](https://www.compshare.cn/?ytag=GPU_YY-gh_langbot) | Plateforme GPU | ✅ |
 | [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | Plateforme GPU | ✅ |
 | [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758) | Plateforme GPU | ✅ |

README_JP.md

@@ -5,7 +5,7 @@
 
 <div align="center">
 
-<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
+<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
 
 <h3>AIエージェント搭載IMボットを構築するための本番グレードプラットフォーム。</h3>
 <h4>Slack、Discord、Telegram、WeChat などに AI ボットを素早く構築、デバッグ、デプロイ。</h4>
@@ -135,7 +135,7 @@ docker compose --profile all up -d
 | [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | GPUプラットフォーム | ✅ |
 | [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758) | GPUプラットフォーム | ✅ |
 | [接口 AI](https://jiekou.ai/) | ゲートウェイ | ✅ |
-| [302.AI](https://share.302ai.cn/SuTG99) | ゲートウェイ | ✅ |
+| [302.AI](https://share.302.ai/SuTG99) | ゲートウェイ | ✅ |
 | [Qiniu](https://www.qiniu.com/ai/agent) | ゲートウェイ | ✅ |
 
 [→ すべての統合を表示](https://link.langbot.app/en/docs/features)

README_KO.md

@@ -5,7 +5,7 @@
 
 <div align="center">
 
-<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
+<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
 
 <h3>AI 에이전트 IM 봇 구축을 위한 프로덕션 등급 플랫폼.</h3>
 <h4>Slack, Discord, Telegram, WeChat 등에 AI 봇을 빠르게 구축, 디버그 및 배포.</h4>
@@ -135,7 +135,7 @@ docker compose --profile all up -d
 | [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | GPU 플랫폼 | ✅ |
 | [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758) | GPU 플랫폼 | ✅ |
 | [接口 AI](https://jiekou.ai/) | 게이트웨이 | ✅ |
-| [302.AI](https://share.302ai.cn/SuTG99) | 게이트웨이 | ✅ |
+| [302.AI](https://share.302.ai/SuTG99) | 게이트웨이 | ✅ |
 | [Qiniu](https://www.qiniu.com/ai/agent) | 게이트웨이 | ✅ |
 
 [→ 모든 통합 보기](https://link.langbot.app/en/docs/features)

README_RU.md

@@ -5,7 +5,7 @@
 
 <div align="center">
 
-<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
+<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
 
 <h3>Платформа производственного уровня для создания агентных IM-ботов.</h3>
 <h4>Быстро создавайте, отлаживайте и развертывайте ИИ-ботов в Slack, Discord, Telegram, WeChat и других платформах.</h4>
@@ -131,7 +131,7 @@ docker compose --profile all up -d
 | [Volc Engine Ark](https://console.volcengine.com/ark/region:ark+cn-beijing/model?vendor=Bytedance&view=LIST_VIEW) | Шлюз | ✅ |
 | [ModelScope](https://modelscope.cn/docs/model-service/API-Inference/intro) | Шлюз | ✅ |
 | [GiteeAI](https://ai.gitee.com/) | Шлюз | ✅ |
-| [302.AI](https://share.302ai.cn/SuTG99) | Шлюз | ✅ |
+| [302.AI](https://share.302.ai/SuTG99) | Шлюз | ✅ |
 | [接口 AI](https://jiekou.ai/) | Шлюз | ✅ |
 | [CompShare](https://www.compshare.cn/?ytag=GPU_YY-gh_langbot) | Платформа GPU | ✅ |
 | [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | Платформа GPU | ✅ |

README_TW.md

@@ -137,7 +137,7 @@ docker compose --profile all up -d
 | [優雲智算](https://www.compshare.cn/?ytag=GPU_YY-gh_langbot) | GPU 平台 | ✅ |
 | [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | GPU 平台 | ✅ |
 | [接口 AI](https://jiekou.ai/) | 聚合平台 | ✅ |
-| [302.AI](https://share.302ai.cn/SuTG99) | 聚合平台 | ✅ |
+| [302.AI](https://share.302.ai/SuTG99) | 聚合平台 | ✅ |
 | [Qiniu](https://www.qiniu.com/ai/agent) | 聚合平台 | ✅ |
 
 ### TTS（語音合成）

README_VI.md

@@ -5,7 +5,7 @@
 
 <div align="center">
 
-<a href="https://www.producthunt.com/products/langbot/launches/langbot?embed=true&amp;utm_source=badge-featured&amp;utm_medium=badge&amp;utm_campaign=badge-langbot" target="_blank" rel="noopener noreferrer"><img alt="LangBot - Easy-to-use global IM bot platform designed for the LLM era | Product Hunt" width="250" height="54" src="https://api.producthunt.com/widgets/embed-image/v1/featured.svg?post_id=979554&amp;theme=light&amp;t=1782822143403"></a>
+<a href="https://www.producthunt.com/products/langbot?utm_source=badge-follow&utm_medium=badge&utm_source=badge-langbot" target="_blank"><img src="https://api.producthunt.com/widgets/embed-image/v1/follow.svg?product_id=1077185&theme=light" alt="LangBot - Production&#0045;grade&#0032;IM&#0032;bot&#0032;made&#0032;easy&#0046; | Product Hunt" style="width: 250px; height: 54px;" width="250" height="54" /></a>
 
 <h3>Nền tảng cấp sản xuất để xây dựng bot IM với AI agent.</h3>
 <h4>Xây dựng, gỡ lỗi và triển khai bot AI nhanh chóng trên Slack, Discord, Telegram, WeChat và nhiều nền tảng khác.</h4>
@@ -135,7 +135,7 @@ docker compose --profile all up -d
 | [PPIO](https://ppinfra.com/user/register?invited_by=QJKFYD&utm_source=github_langbot) | Nền tảng GPU | ✅ |
 | [ShengSuanYun](https://www.shengsuanyun.com/?from=CH_KYIPP758) | Nền tảng GPU | ✅ |
 | [接口 AI](https://jiekou.ai/) | Cổng | ✅ |
-| [302.AI](https://share.302ai.cn/SuTG99) | Cổng | ✅ |
+| [302.AI](https://share.302.ai/SuTG99) | Cổng | ✅ |
 | [Qiniu](https://www.qiniu.com/ai/agent) | Cổng | ✅ |
 
 [→ Xem tất cả tích hợp](https://link.langbot.app/en/docs/features)

compare_nodes.py

@@ -1,163 +0,0 @@
-#!/usr/bin/env python3
-"""Compare YAML node definitions with frontend node-configs."""
-
-import yaml
-import os
-import re
-import json
-
-# 1. Parse YAML files
-yaml_dir = 'src/langbot/templates/metadata/nodes'
-yaml_nodes = {}
-
-for filename in sorted(os.listdir(yaml_dir)):
-    if filename.endswith('.yaml'):
-        filepath = os.path.join(yaml_dir, filename)
-        with open(filepath, 'r') as f:
-            data = yaml.safe_load(f)
-            node_name = data.get('name', filename.replace('.yaml', ''))
-            yaml_nodes[node_name] = {
-                'category': data.get('category', ''),
-                'inputs': [i['name'] for i in data.get('inputs', [])],
-                'outputs': [o['name'] for o in data.get('outputs', [])],
-                'config': [c['name'] for c in data.get('config', [])]
-            }
-
-# 2. Parse frontend node-configs TypeScript files
-node_configs_dir = 'web/src/app/home/workflows/components/workflow-editor/node-configs'
-
-frontend_nodes = {}
-
-def parse_ts_file(filepath):
-    """Parse a TypeScript file to extract node configurations."""
-    with open(filepath, 'r') as f:
-        content = f.read()
-    
-    # Find all node type definitions
-    # Pattern: nodeType: 'xxx'
-    node_type_pattern = r"nodeType:\s*'([^']+)'"
-    node_types = re.findall(node_type_pattern, content)
-    
-    # For each node type, extract inputs, outputs, and config
-    for node_type in node_types:
-        # Find the config object for this node type
-        # Look for the section between this nodeType and the next one or end of object
-        pattern = rf"nodeType:\s*'({re.escape(node_type)})'.*?(?=nodeType:|export\s+(const|function)|$)"
-        match = re.search(pattern, content, re.DOTALL)
-        
-        if match:
-            section = match.group(0)
-            
-            # Extract inputs
-            inputs = re.findall(r"createInput\('([^']+)'", section)
-            
-            # Extract outputs  
-            outputs = re.findall(r"createOutput\('([^']+)'", section)
-            
-            # Extract config names
-            config_names = re.findall(r"name:\s*'([^']+)'", section)
-            # Remove duplicates while preserving order
-            seen = set()
-            unique_config = []
-            for c in config_names:
-                if c not in seen:
-                    seen.add(c)
-                    unique_config.append(c)
-            
-            frontend_nodes[node_type] = {
-                'inputs': inputs,
-                'outputs': outputs,
-                'config': unique_config
-            }
-
-# Parse all config files
-for filename in os.listdir(node_configs_dir):
-    if filename.endswith('.ts') and filename != 'types.ts' and filename != 'index.ts':
-        filepath = os.path.join(node_configs_dir, filename)
-        parse_ts_file(filepath)
-
-# 3. Compare and report differences
-print("=" * 80)
-print("WORKFLOW NODE COMPARISON REPORT: YAML vs Frontend")
-print("=" * 80)
-
-all_node_types = sorted(set(list(yaml_nodes.keys()) + list(frontend_nodes.keys())))
-
-discrepancies = []
-
-for node_type in all_node_types:
-    yaml_def = yaml_nodes.get(node_type)
-    frontend_def = frontend_nodes.get(node_type)
-    
-    node_discrepancies = []
-    
-    if not yaml_def:
-        print(f"\n⚠️  {node_type}: ONLY in frontend (not in YAML)")
-        continue
-    if not frontend_def:
-        print(f"\n⚠️  {node_type}: ONLY in YAML (not in frontend)")
-        continue
-    
-    # Compare inputs
-    yaml_inputs = set(yaml_def['inputs'])
-    frontend_inputs = set(frontend_def['inputs'])
-    if yaml_inputs != frontend_inputs:
-        only_yaml = yaml_inputs - frontend_inputs
-        only_frontend = frontend_inputs - yaml_inputs
-        node_discrepancies.append({
-            'type': 'inputs',
-            'only_yaml': list(only_yaml),
-            'only_frontend': list(only_frontend)
-        })
-    
-    # Compare outputs
-    yaml_outputs = set(yaml_def['outputs'])
-    frontend_outputs = set(frontend_def['outputs'])
-    if yaml_outputs != frontend_outputs:
-        only_yaml = yaml_outputs - frontend_outputs
-        only_frontend = frontend_outputs - yaml_outputs
-        node_discrepancies.append({
-            'type': 'outputs',
-            'only_yaml': list(only_yaml),
-            'only_frontend': list(only_frontend)
-        })
-    
-    # Compare config
-    yaml_config = set(yaml_def['config'])
-    frontend_config = set(frontend_def['config'])
-    if yaml_config != frontend_config:
-        only_yaml = yaml_config - frontend_config
-        only_frontend = frontend_config - yaml_config
-        node_discrepancies.append({
-            'type': 'config',
-            'only_yaml': list(only_yaml),
-            'only_frontend': list(only_frontend)
-        })
-    
-    if node_discrepancies:
-        print(f"\n❌ {node_type} ({yaml_def['category']}): HAS DISCREPANCIES")
-        for d in node_discrepancies:
-            print(f"   {d['type']}:")
-            if d['only_yaml']:
-                print(f"     Only in YAML: {d['only_yaml']}")
-            if d['only_frontend']:
-                print(f"     Only in Frontend: {d['only_frontend']}")
-        discrepancies.append((node_type, node_discrepancies))
-    else:
-        print(f"\n✅ {node_type} ({yaml_def['category']}): OK")
-
-print(f"\n{'=' * 80}")
-print(f"SUMMARY: {len(discrepancies)} nodes with discrepancies out of {len(all_node_types)} total")
-print(f"{'=' * 80}")
-
-# Output as JSON for further processing
-output = {
-    'yaml_nodes': {k: v for k, v in yaml_nodes.items()},
-    'frontend_nodes': {k: v for k, v in frontend_nodes.items()},
-    'discrepancies': {k: v for k, v in discrepancies}
-}
-
-with open('node_comparison.json', 'w') as f:
-    json.dump(output, f, indent=2)
-
-print(f"\nDetailed comparison saved to node_comparison.json")

docs/development/workflow-system.md

@@ -1,713 +0,0 @@
-# Workflow 系统开发者文档
-
-本文档面向 LangBot 开发者，详细介绍 Workflow 系统的技术架构、核心组件和扩展方法。
-
-## 目录
-
-- [系统架构概述](#系统架构概述)
-- [目录结构](#目录结构)
-- [核心组件](#核心组件)
-  - [后端模块](#后端模块)
-  - [前端组件](#前端组件)
-- [数据库表结构](#数据库表结构)
-- [API 接口文档](#api-接口文档)
-- [如何添加新节点类型](#如何添加新节点类型)
-- [调试功能实现](#调试功能实现)
-
----
-
-## 系统架构概述
-
-Workflow 系统采用前后端分离架构，主要包含以下层次：
-
-```
-┌─────────────────────────────────────────────────────────────┐
-│                      前端层 (React)                          │
-│  ┌─────────────┬──────────────┬──────────────┬───────────┐  │
-│  │ 可视化编辑器 │   节点面板    │   属性面板   │  调试器   │  │
-│  │ ReactFlow   │  NodePalette │ PropertyPanel│ Debugger  │  │
-│  └─────────────┴──────────────┴──────────────┴───────────┘  │
-├─────────────────────────────────────────────────────────────┤
-│                      API 层 (Quart)                          │
-│  ┌─────────────┬──────────────┬──────────────────────────┐  │
-│  │ Workflow API│  Debug API   │     Node Types API       │  │
-│  └─────────────┴──────────────┴──────────────────────────┘  │
-├─────────────────────────────────────────────────────────────┤
-│                     核心引擎层 (Python)                       │
-│  ┌─────────────┬──────────────┬──────────────┬───────────┐  │
-│  │ Executor    │   Registry   │    Node      │ Entities  │  │
-│  │ 执行引擎     │  节点注册表   │   节点基类   │  数据结构  │  │
-│  └─────────────┴──────────────┴──────────────┴───────────┘  │
-├─────────────────────────────────────────────────────────────┤
-│                      存储层 (SQLAlchemy)                     │
-│  ┌─────────────┬──────────────┬──────────────────────────┐  │
-│  │  Workflow   │  Executions  │        Triggers          │  │
-│  └─────────────┴──────────────┴──────────────────────────┘  │
-└─────────────────────────────────────────────────────────────┘
-```
-
----
-
-## 目录结构
-
-### 后端代码结构
-
-```
-LangBot/src/langbot/pkg/
-├── workflow/                      # Workflow 核心模块
-│   ├── __init__.py               # 模块初始化，导出公共接口
-│   ├── entities.py               # 数据实体定义
-│   ├── executor.py               # 执行引擎
-│   ├── node.py                   # 节点基类和装饰器
-│   ├── registry.py               # 节点类型注册表
-│   └── nodes/                    # 内置节点实现
-│       ├── __init__.py           # 注册所有内置节点
-│       ├── trigger.py            # 触发节点
-│       ├── process.py            # 处理节点
-│       ├── control.py            # 控制节点
-│       └── action.py             # 动作节点
-├── entity/persistence/
-│   └── workflow.py               # 数据库模型
-├── api/http/
-│   ├── controller/groups/workflows/
-│   │   └── workflows.py          # API 路由控制器
-│   └── service/
-│       └── workflow.py           # 业务逻辑服务
-└── persistence/migrations/
-    └── dbm026_workflow_tables.py # 数据库迁移
-```
-
-### 前端代码结构
-
-```
-LangBot/web/src/app/home/workflows/
-├── page.tsx                      # Workflow 列表页
-├── WorkflowDetailContent.tsx     # 详情页内容
-├── store/
-│   └── useWorkflowStore.ts       # Zustand 状态管理
-└── components/
-    ├── workflow-editor/          # 可视化编辑器
-    │   ├── index.ts              # 导出
-    │   ├── WorkflowEditorComponent.tsx  # 主编辑器组件
-    │   ├── WorkflowNodeComponent.tsx    # 自定义节点组件
-    │   ├── NodePalette.tsx       # 节点面板
-    │   ├── PropertyPanel.tsx     # 属性面板
-    │   └── node-configs/         # 节点配置元数据
-    │       ├── types.ts          # 配置类型定义
-    │       ├── trigger-configs.ts
-    │       ├── ai-configs.ts
-    │       ├── process-configs.ts
-    │       ├── control-configs.ts
-    │       ├── action-configs.ts
-    │       ├── integration-configs.ts
-    │       └── index.ts          # 配置汇总
-    ├── workflow-debugger/        # 调试器组件
-    │   ├── index.ts
-    │   └── WorkflowDebugger.tsx
-    ├── workflow-form/            # 表单组件
-    │   └── WorkflowFormComponent.tsx
-    └── workflow-executions/      # 执行历史组件
-        └── WorkflowExecutionsTab.tsx
-```
-
----
-
-## 核心组件
-
-### 后端模块
-
-#### 1. 执行引擎 (WorkflowExecutor)
-
-位置：[`executor.py`](../../src/langbot/pkg/workflow/executor.py)
-
-执行引擎负责工作流的实际执行，包括：
-
-- **拓扑排序**：确定节点执行顺序
-- **节点执行**：调用各节点的 execute 方法
-- **控制流处理**：处理条件分支、循环、并行执行
-- **错误处理**：支持重试机制
-
-```python
-class WorkflowExecutor:
-    async def execute(
-        self, 
-        workflow: WorkflowDefinition, 
-        context: ExecutionContext,
-        start_node_id: Optional[str] = None
-    ) -> ExecutionContext:
-        """执行工作流"""
-        # 1. 构建执行图
-        # 2. 初始化节点状态
-        # 3. 找到起始节点
-        # 4. 按拓扑顺序执行
-```
-
-**调试执行器 (DebugWorkflowExecutor)**
-
-继承自 WorkflowExecutor，增加了调试支持：
-
-- 断点支持
-- 单步执行
-- 暂停/继续
-- 实时日志
-
-```python
-class DebugWorkflowExecutor(WorkflowExecutor):
-    async def execute_debug(
-        self,
-        workflow: WorkflowDefinition,
-        context: ExecutionContext,
-        debug_state: DebugExecutionState,
-    ) -> ExecutionContext:
-        """调试模式执行"""
-```
-
-#### 2. 节点注册表 (NodeTypeRegistry)
-
-位置：[`registry.py`](../../src/langbot/pkg/workflow/registry.py)
-
-单例模式管理所有节点类型：
-
-```python
-class NodeTypeRegistry:
-    _instance: Optional['NodeTypeRegistry'] = None
-    
-    def register(self, node_type: str, node_class: type[WorkflowNode]):
-        """注册节点类型"""
-        
-    def create_instance(self, node_type: str, node_id: str, config: dict) -> WorkflowNode:
-        """创建节点实例"""
-        
-    def list_all(self) -> list[dict]:
-        """获取所有节点类型的 Schema"""
-```
-
-#### 3. 节点基类 (WorkflowNode)
-
-位置：[`node.py`](../../src/langbot/pkg/workflow/node.py)
-
-所有节点必须继承此基类：
-
-```python
-class WorkflowNode(abc.ABC):
-    # 节点元数据
-    type_name: str = ""
-    name: str = ""
-    description: str = ""
-    category: str = "misc"
-    icon: str = ""
-    
-    # 端口定义
-    inputs: list[NodePort] = []
-    outputs: list[NodePort] = []
-    
-    # 配置 Schema
-    config_schema: list[NodeConfig] = []
-    
-    @abc.abstractmethod
-    async def execute(
-        self, 
-        inputs: dict[str, Any], 
-        context: ExecutionContext
-    ) -> dict[str, Any]:
-        """执行节点逻辑"""
-        pass
-```
-
-#### 4. 数据实体 (entities.py)
-
-主要数据结构：
-
-```python
-class WorkflowDefinition:
-    """工作流定义"""
-    uuid: str
-    name: str
-    nodes: list[NodeDefinition]
-    edges: list[EdgeDefinition]
-    settings: WorkflowSettings
-    
-class ExecutionContext:
-    """执行上下文"""
-    execution_id: str
-    workflow_id: str
-    status: ExecutionStatus
-    variables: dict
-    node_states: dict[str, NodeState]
-    history: list[ExecutionStep]
-```
-
-### 前端组件
-
-#### 1. WorkflowEditorComponent
-
-主编辑器组件，基于 React Flow 实现：
-
-- **画布交互**：拖拽、缩放、平移
-- **节点连接**：自动验证端口类型
-- **撤销/重做**：基于历史记录栈
-- **复制/粘贴**：支持多选复制
-
-关键功能：
-
-```tsx
-function WorkflowEditorInner() {
-  const { nodes, edges, onNodesChange, onEdgesChange, onConnect } = useWorkflowStore();
-  
-  // 拖放添加节点
-  const onDrop = useCallback((event: React.DragEvent) => {
-    const type = event.dataTransfer.getData('application/reactflow');
-    const position = screenToFlowPosition({ x: event.clientX, y: event.clientY });
-    addNode(type, position);
-  }, []);
-  
-  // 复制粘贴
-  const handleCopy = useCallback(() => { ... }, []);
-  const handlePaste = useCallback(() => { ... }, []);
-}
-```
-
-#### 2. NodePalette
-
-节点面板组件，展示可用节点类型：
-
-```tsx
-function NodePalette() {
-  // 按类别组织节点
-  const categories = [
-    { id: 'trigger', name: '触发节点', icon: Zap },
-    { id: 'ai', name: 'AI 节点', icon: Brain },
-    { id: 'process', name: '处理节点', icon: Cpu },
-    { id: 'control', name: '控制节点', icon: GitBranch },
-    { id: 'action', name: '动作节点', icon: Send },
-    { id: 'integration', name: '集成节点', icon: Plug },
-  ];
-  
-  // 拖拽开始
-  const onDragStart = (event: React.DragEvent, nodeType: string) => {
-    event.dataTransfer.setData('application/reactflow', nodeType);
-  };
-}
-```
-
-#### 3. PropertyPanel
-
-属性面板组件，动态渲染节点配置表单：
-
-```tsx
-function PropertyPanel() {
-  const { selectedNodeId, nodes, updateNodeData } = useWorkflowStore();
-  
-  // 根据节点类型获取配置元数据
-  const selectedNode = nodes.find(n => n.id === selectedNodeId);
-  const nodeConfig = getNodeConfig(selectedNode?.data?.nodeType);
-  
-  // 动态渲染配置字段
-  return (
-    <div>
-      {nodeConfig?.fields.map(field => (
-        <ConfigField key={field.name} field={field} />
-      ))}
-    </div>
-  );
-}
-```
-
-#### 4. WorkflowDebugger
-
-调试器组件，支持实时调试：
-
-```tsx
-function WorkflowDebugger({ workflowUuid, workflow }) {
-  const [debugState, setDebugState] = useState<DebugState>('idle');
-  const [executionId, setExecutionId] = useState<string>('');
-  const [logs, setLogs] = useState<ExecutionLog[]>([]);
-  
-  // 启动调试
-  const startDebug = async () => {
-    const result = await backendClient.post(
-      `/api/v1/workflows/${workflowUuid}/debug/start`,
-      { context, variables, breakpoints }
-    );
-    setExecutionId(result.execution_id);
-  };
-  
-  // 轮询状态
-  useEffect(() => {
-    if (debugState === 'running') {
-      const interval = setInterval(fetchState, 500);
-      return () => clearInterval(interval);
-    }
-  }, [debugState]);
-}
-```
-
-#### 5. useWorkflowStore
-
-Zustand 状态管理：
-
-```typescript
-interface WorkflowState {
-  nodes: WorkflowNode[];
-  edges: WorkflowEdge[];
-  selectedNodeId: string | null;
-  history: HistoryEntry[];
-  historyIndex: number;
-  isDirty: boolean;
-  
-  // Actions
-  addNode: (type: string, position: XYPosition) => void;
-  updateNodeData: (nodeId: string, data: Partial<NodeData>) => void;
-  deleteNode: (nodeId: string) => void;
-  undo: () => void;
-  redo: () => void;
-}
-
-export const useWorkflowStore = create<WorkflowState>((set, get) => ({
-  // ... state and actions
-}));
-```
-
----
-
-## 数据库表结构
-
-### workflows 表
-
-```sql
-CREATE TABLE workflows (
-    uuid VARCHAR(255) PRIMARY KEY,
-    name VARCHAR(255) NOT NULL,
-    description TEXT,
-    emoji VARCHAR(10) DEFAULT '🔄',
-    version INTEGER DEFAULT 1,
-    is_enabled BOOLEAN DEFAULT TRUE,
-    definition JSON NOT NULL,        -- 节点和边定义
-    global_config JSON DEFAULT '{}', -- 全局配置
-    extensions_preferences JSON,     -- 插件和 MCP 配置
-    created_at TIMESTAMP,
-    updated_at TIMESTAMP
-);
-```
-
-### workflow_versions 表
-
-```sql
-CREATE TABLE workflow_versions (
-    id INTEGER PRIMARY KEY AUTOINCREMENT,
-    workflow_uuid VARCHAR(255) NOT NULL,
-    version INTEGER NOT NULL,
-    definition JSON NOT NULL,
-    global_config JSON DEFAULT '{}',
-    created_at TIMESTAMP,
-    created_by VARCHAR(255),
-    UNIQUE(workflow_uuid, version)
-);
-```
-
-### workflow_executions 表
-
-```sql
-CREATE TABLE workflow_executions (
-    uuid VARCHAR(255) PRIMARY KEY,
-    workflow_uuid VARCHAR(255) NOT NULL,
-    workflow_version INTEGER NOT NULL,
-    status VARCHAR(20) NOT NULL,  -- pending/running/completed/failed/cancelled
-    trigger_type VARCHAR(50),
-    trigger_data JSON,
-    variables JSON,
-    start_time TIMESTAMP,
-    end_time TIMESTAMP,
-    error TEXT,
-    created_at TIMESTAMP
-);
-```
-
-### workflow_node_executions 表
-
-```sql
-CREATE TABLE workflow_node_executions (
-    id INTEGER PRIMARY KEY AUTOINCREMENT,
-    execution_uuid VARCHAR(255) NOT NULL,
-    node_id VARCHAR(100) NOT NULL,
-    node_type VARCHAR(50) NOT NULL,
-    status VARCHAR(20) NOT NULL,
-    inputs JSON,
-    outputs JSON,
-    start_time TIMESTAMP,
-    end_time TIMESTAMP,
-    error TEXT,
-    retry_count INTEGER DEFAULT 0
-);
-```
-
-### workflow_triggers 表
-
-```sql
-CREATE TABLE workflow_triggers (
-    uuid VARCHAR(255) PRIMARY KEY,
-    workflow_uuid VARCHAR(255) NOT NULL,
-    type VARCHAR(50) NOT NULL,  -- message/cron/event/webhook
-    config JSON NOT NULL,
-    is_enabled BOOLEAN DEFAULT TRUE,
-    priority INTEGER DEFAULT 0,
-    created_at TIMESTAMP,
-    updated_at TIMESTAMP
-);
-```
-
----
-
-## API 接口文档
-
-### Workflow CRUD
-
-| 方法 | 路径 | 描述 |
-|-----|------|------|
-| GET | `/api/v1/workflows` | 获取工作流列表 |
-| POST | `/api/v1/workflows` | 创建工作流 |
-| GET | `/api/v1/workflows/:uuid` | 获取单个工作流 |
-| PUT | `/api/v1/workflows/:uuid` | 更新工作流 |
-| DELETE | `/api/v1/workflows/:uuid` | 删除工作流 |
-| POST | `/api/v1/workflows/:uuid/copy` | 复制工作流 |
-
-### 执行相关
-
-| 方法 | 路径 | 描述 |
-|-----|------|------|
-| POST | `/api/v1/workflows/:uuid/execute` | 手动执行工作流 |
-| GET | `/api/v1/workflows/:uuid/executions` | 获取执行记录 |
-
-### 版本管理
-
-| 方法 | 路径 | 描述 |
-|-----|------|------|
-| GET | `/api/v1/workflows/:uuid/versions` | 获取版本列表 |
-| POST | `/api/v1/workflows/:uuid/rollback/:version` | 回滚到指定版本 |
-
-### 调试 API
-
-| 方法 | 路径 | 描述 |
-|-----|------|------|
-| POST | `/api/v1/workflows/:uuid/debug/start` | 启动调试 |
-| POST | `/api/v1/workflows/:uuid/debug/:exec_id/pause` | 暂停执行 |
-| POST | `/api/v1/workflows/:uuid/debug/:exec_id/resume` | 继续执行 |
-| POST | `/api/v1/workflows/:uuid/debug/:exec_id/stop` | 停止执行 |
-| POST | `/api/v1/workflows/:uuid/debug/:exec_id/step` | 单步执行 |
-| GET | `/api/v1/workflows/:uuid/debug/:exec_id/state` | 获取调试状态 |
-
-### 节点类型
-
-| 方法 | 路径 | 描述 |
-|-----|------|------|
-| GET | `/api/v1/workflows/_/node-types` | 获取所有节点类型 |
-| GET | `/api/v1/workflows/_/node-types/categories` | 按类别获取节点类型 |
-
----
-
-## 如何添加新节点类型
-
-### 步骤 1：创建节点类
-
-在 `LangBot/src/langbot/pkg/workflow/nodes/` 下创建或修改文件：
-
-```python
-from ..node import WorkflowNode, NodePort, NodeConfig, workflow_node
-from ..entities import ExecutionContext
-
-@workflow_node('my_custom_node')
-class MyCustomNode(WorkflowNode):
-    """自定义节点"""
-    
-    # 元数据
-    type_name = 'my_custom_node'
-    name = '我的自定义节点'
-    description = '这是一个自定义节点'
-    category = 'process'  # trigger/process/control/action/integration
-    icon = '🔧'
-    
-    # 输入端口
-    inputs = [
-        NodePort(name='input', type='string', description='输入数据', required=True),
-    ]
-    
-    # 输出端口
-    outputs = [
-        NodePort(name='output', type='string', description='输出数据'),
-    ]
-    
-    # 配置字段
-    config_schema = [
-        NodeConfig(
-            name='option',
-            type='select',
-            required=True,
-            options=['选项A', '选项B'],
-            description='选择一个选项'
-        ),
-        NodeConfig(
-            name='value',
-            type='string',
-            required=False,
-            default='默认值',
-            description='配置值'
-        ),
-    ]
-    
-    async def execute(
-        self, 
-        inputs: dict[str, Any], 
-        context: ExecutionContext
-    ) -> dict[str, Any]:
-        """执行节点逻辑"""
-        input_data = inputs.get('input', '')
-        option = self.get_config('option')
-        value = self.get_config('value', '')
-        
-        # 处理逻辑
-        result = f"处理: {input_data} with {option} and {value}"
-        
-        return {'output': result}
-```
-
-### 步骤 2：注册节点
-
-在 `LangBot/src/langbot/pkg/workflow/nodes/__init__.py` 中导入：
-
-```python
-from .process import (
-    CodeExecutorNode,
-    HttpRequestNode,
-    DataTransformNode,
-    MyCustomNode,  # 添加新节点
-)
-```
-
-### 步骤 3：添加前端配置
-
-在 `LangBot/web/src/app/home/workflows/components/workflow-editor/node-configs/` 目录下添加配置：
-
-```typescript
-// process-configs.ts
-export const processNodeConfigs: NodeConfigMap = {
-  // ... 其他配置
-  
-  my_custom_node: {
-    type: 'my_custom_node',
-    label: 'workflows.nodes.myCustomNode',
-    description: 'workflows.nodes.myCustomNodeDesc',
-    icon: 'Wrench',
-    category: 'process',
-    fields: [
-      {
-        name: 'option',
-        type: 'select',
-        label: 'workflows.fields.option',
-        required: true,
-        options: [
-          { value: '选项A', label: '选项 A' },
-          { value: '选项B', label: '选项 B' },
-        ],
-      },
-      {
-        name: 'value',
-        type: 'string',
-        label: 'workflows.fields.value',
-        required: false,
-        defaultValue: '默认值',
-      },
-    ],
-  },
-};
-```
-
-### 步骤 4：添加国际化
-
-在 `LangBot/web/src/i18n/locales/` 中添加翻译：
-
-```typescript
-// zh-Hans.ts
-workflows: {
-  nodes: {
-    myCustomNode: '我的自定义节点',
-    myCustomNodeDesc: '这是一个自定义节点',
-  },
-  fields: {
-    option: '选项',
-    value: '值',
-  },
-}
-```
-
----
-
-## 调试功能实现
-
-### 后端调试状态管理
-
-```python
-class DebugExecutionState:
-    """调试执行状态"""
-    
-    def __init__(self, execution_id: str, breakpoints: list[str] = None):
-        self.execution_id = execution_id
-        self.status: str = 'running'
-        self.is_paused: bool = False
-        self.is_stopped: bool = False
-        self.breakpoints: set[str] = set(breakpoints or [])
-        self.logs: list[ExecutionLog] = []
-        self._pause_event = asyncio.Event()
-    
-    def pause(self):
-        """暂停执行"""
-        self.is_paused = True
-        self._pause_event.clear()
-    
-    def resume(self):
-        """继续执行"""
-        self.is_paused = False
-        self._pause_event.set()
-    
-    async def wait_if_paused(self):
-        """如果暂停则等待"""
-        if self.is_paused:
-            await self._pause_event.wait()
-```
-
-### 前端调试流程
-
-1. **设置断点**：点击节点设置断点
-2. **启动调试**：调用 `/debug/start` 启动调试执行
-3. **轮询状态**：定期调用 `/debug/:id/state` 获取状态
-4. **控制执行**：调用 pause/resume/step/stop 控制执行
-5. **查看日志**：实时显示执行日志和节点状态
-
-```typescript
-// 调试状态轮询
-const fetchDebugState = async () => {
-  const state = await backendClient.get(
-    `/api/v1/workflows/${workflowUuid}/debug/${executionId}/state`
-  );
-  
-  // 更新节点状态
-  setNodeStates(state.node_states);
-  
-  // 追加新日志
-  if (state.new_logs.length > 0) {
-    setLogs(prev => [...prev, ...state.new_logs]);
-  }
-  
-  // 检查完成状态
-  if (state.status === 'completed' || state.status === 'error') {
-    setDebugState('idle');
-  }
-};
-```
-
----
-
-## 扩展阅读
-
-- [Workflow 功能设计文档](../../../plans/langbot-workflow-design.md)
-- [用户使用指南](../user-guide/workflow-guide.md)
-- [API 认证文档](../API_KEY_AUTH.md)

docs/review/mcp-resources-pr-2215-review.md

@@ -1,196 +0,0 @@
-# MCP Resources PR #2215 Review
-
-> 更新日期: 2026-06-29
-> 分支: `mcp_resources`
-> PR: langbot-app/LangBot#2215
-> 主题: MCP Resources 在 LangBot 中的产品价值、AgentRunner 集成方式与后续架构方向
-
-## 结论
-
-PR #2215 对 LangBot 有明确价值：它补齐了 MCP 协议中 Resources 这一重要能力，让 MCP server 不再只暴露 tools，也可以暴露文档、代码片段、配置、日志、图片等上下文资源。管理端可以发现和预览资源，Agent 也可以通过当前实现按需列出和读取资源。
-
-但当前 AgentRunner 层的接入方式更接近一个可用的第一阶段方案，而不是最终架构。现在 MCP Resources 被包装成两个 synthetic tools：
-
-- `langbot_mcp_list_resources`
-- `langbot_mcp_read_resource`
-
-这让模型可以通过 function calling 主动探索资源，落地成本低，也复用了已有 `ToolManager` / `LocalAgentRunner` 的工具调用链路。不过从 MCP 规范和主流实现来看，Resources 更适合作为一种一等上下文来源，而不是长期隐藏在工具列表里。
-
-建议保留当前 synthetic tools 作为探索能力，同时把后续主线设计调整为：MCP Resources 是 pipeline / conversation / message 级别可选择、可固定、可审计的上下文输入。
-
-## 当前实现判断
-
-当前 AgentRunner 集成路径如下：
-
-```text
-Pipeline 绑定 MCP server
-  -> query.variables['_pipeline_bound_mcp_servers']
-  -> Preproc 为 local-agent 加载工具
-  -> ToolManager.get_all_tools()
-  -> MCPLoader 注入 synthetic resource tools
-  -> LocalAgentRunner 将工具 schema 传给模型
-  -> 模型发起 list/read tool call
-  -> ToolManager.execute_func_call()
-  -> MCPLoader 调 MCP session.list_resources/read_resource
-  -> tool result 回灌给模型
-```
-
-这个路径的优点是：
-
-- 复用现有工具调用机制，改动范围小。
-- Agent 可以按需探索资源，不需要每轮预先读取所有资源。
-- 可以沿用 pipeline 绑定的 MCP server 范围，避免越权读取未绑定 server。
-- 对已有 MCP tools 行为影响较小。
-
-主要问题是：
-
-- Resources 在语义上被降级成 tools，和 MCP 规范里的 resource primitive 不完全一致。
-- 模型必须先理解并主动调用 `list/read`，资源不会自然成为上下文。
-- pipeline 不能配置“默认携带某些资源”或“本轮附加某些资源”。
-- UI 资源 tab 目前是管理端预览能力，和 Agent 上下文选择没有打通。
-- 对 blob、图片、大文件、结构化资源的处理还比较粗糙。
-- 缺少 resource templates、订阅更新、缓存、chunk、token budget、trace 与审计策略。
-
-## 主流项目做法
-
-### MCP 官方规范
-
-MCP Resources 是 server 暴露上下文数据的协议能力。规范没有要求 resources 必须以 tool call 形式给模型使用，而是把如何选择、过滤、读取和纳入上下文交给 Host application。
-
-这意味着比较正统的集成方式是：LangBot 作为 Host，在 pipeline、会话或消息层决定哪些 resources 进入模型上下文。
-
-参考: https://modelcontextprotocol.io/specification/2025-06-18/server/resources
-
-### VS Code Copilot
-
-VS Code 把 MCP Resources 做成 chat context 的一部分。用户可以通过 `Add Context > MCP Resources` 或命令浏览 MCP resources，并把选中的资源附加到一次 chat request。
-
-这是目前最值得 LangBot 参考的产品形态：资源不是模型工具，而是用户和 Host 可控的上下文附件。
-
-参考: https://code.visualstudio.com/docs/agent-customization/mcp-servers
-
-### Anthropic SDK
-
-Anthropic 的 client-side MCP helpers 提供资源读取和转换能力，例如把 MCP resource 转为 Claude message content 或 file。也就是说，应用先读取 resource，再显式放进模型消息。
-
-这同样是 application-owned context injection，而不是把 resource 伪装成模型工具。
-
-参考: https://platform.claude.com/docs/en/agents-and-tools/mcp-connector
-
-### LangChain MCP Adapters
-
-LangChain 把 MCP Resources 更像 data loader / document input 来处理，可以把资源加载成 `Blob`，再进入 LangChain 的文档、检索或上下文处理链路。
-
-这说明 Resources 很适合作为知识源、文档源或上下文源，而不只是即时工具调用。
-
-参考: https://docs.langchain.com/oss/python/langchain/mcp
-
-### OpenAI Agents SDK
-
-OpenAI Agents SDK 主路径仍偏向 MCP tools，但底层 MCP server API 已经有 `list_resources`、`list_resource_templates`、`read_resource` 等能力。当前形态说明 resources 是 client 能力，但并未默认变成 agent-visible tools。
-
-参考: https://openai.github.io/openai-agents-python/mcp/
-
-### Cline
-
-Cline 会拉取 MCP tools、resources、resourceTemplates、prompts，并通过类似 `access_mcp_resource` 的内置访问方式让模型读取资源。这个方向和 LangBot 当前 synthetic tools 比较接近。
-
-这种模式适合让 Agent 自主探索，但更像 Host 自定义的模型访问协议，不应成为唯一集成路径。
-
-参考: https://github.com/cline/cline/blob/main/src/services/mcp/McpHub.ts
-
-## 建议架构方向
-
-### 1. 保留探索型工具
-
-保留当前两个 synthetic tools：
-
-- `langbot_mcp_list_resources`
-- `langbot_mcp_read_resource`
-
-它们适合处理“用户没有显式选择资源，但 Agent 判断需要探索 MCP server 上下文”的场景。后续可以优化工具描述、返回格式、资源大小限制和错误信息。
-
-### 2. 增加一等 Resource Context
-
-新增一个 Host 层资源上下文概念，例如：
-
-```text
-PipelineResourceBinding
-ConversationResourceAttachment
-MessageResourceAttachment
-```
-
-Preproc 或独立的 `ResourceContextProvider` 在模型调用前读取这些资源，按 MIME 类型、大小、token budget 转为模型可消费的上下文。
-
-### 3. 打通 UI 与 Agent 上下文
-
-当前 MCP 详情页的 Resources tab 可以继续作为资源发现和预览入口。建议增加操作：
-
-- 添加到本轮上下文
-- 固定到当前 pipeline
-- 固定到当前 bot / conversation
-- 查看资源读取历史和错误
-
-这样 UI 资源管理能力才能真正影响 Agent 行为。
-
-### 4. 支持 resource templates
-
-MCP resource templates 允许 server 暴露参数化资源，例如：
-
-```text
-repo://{owner}/{repo}/file/{path}
-log://{service}/{date}
-```
-
-LangBot 后续应支持模板发现、参数填写、实例化和绑定。否则只能使用静态 resources，覆盖面会受限。
-
-### 5. 增加资源处理策略
-
-建议补齐：
-
-- 文本资源 token budget 与截断策略。
-- 大文件 chunk 与摘要策略。
-- 图片/blob 的模型能力判断与 fallback。
-- MIME 类型白名单与安全限制。
-- 缓存与过期策略。
-- `resources/listChanged` 或订阅更新。
-- resource read trace，便于审计 Agent 读取了什么上下文。
-
-## 推荐落地顺序
-
-### Phase 1: 完成当前 PR 可用性
-
-- 保留 synthetic tools。
-- 明确文档说明当前 Agent 集成是 tool-mediated。
-- 完善资源工具描述，降低模型误用概率。
-- 给 read/list 增加大小限制和更清晰的 MIME 处理。
-- 前端 Resources tab 与 Tools tab 分离，保持管理端清晰。
-
-### Phase 2: 做 Host-owned context attachments
-
-- 在 pipeline 或 conversation 层新增 resource attachment 配置。
-- Preproc 读取已绑定 resources，注入模型上下文。
-- UI 支持“添加到上下文 / 固定到 pipeline”。
-- 记录每轮实际注入的 resource URI 和 token 消耗。
-
-### Phase 3: 做完整 MCP Resources 能力
-
-- 支持 resource templates。
-- 支持资源订阅更新。
-- 支持 chunk、summary、RAG 化接入。
-- 为 DifyAgentRunner、LocalAgentRunner 等不同 runner 定义统一资源上下文接口。
-
-## 最终建议
-
-PR #2215 可以作为 MCP Resources 的第一阶段实现继续推进。它让 LangBot 快速拥有“资源发现、预览、按需读取”的闭环，也给 Agent 探索资源提供了可运行路径。
-
-但在正式设计上，不建议把 “Resources == Tools” 固化为长期抽象。LangBot 更应该把 MCP Resources 定位为上下文来源，与 tools、prompts、knowledge base 并列：
-
-```text
-Tools      -> Agent 可以执行的动作
-Resources  -> Host/用户/Agent 可以选择的上下文数据
-Prompts    -> 可复用的任务模板
-Knowledge  -> 可检索、可索引的长期知识
-```
-
-这样既尊重 MCP 协议语义，也能让 LangBot 在 Agent 工作流、企业知识接入和多 MCP server 管理上走得更稳。

docs/user-guide/workflow-guide.md

@@ -1,425 +0,0 @@
-# Workflow 用户指南
-
-本文档帮助您了解和使用 LangBot 的 Workflow（工作流）功能，通过可视化方式构建自动化的对话处理流程。
-
-## 目录
-
-- [功能介绍](#功能介绍)
-- [快速入门](#快速入门)
-- [节点类型说明](#节点类型说明)
-- [编辑器使用指南](#编辑器使用指南)
-- [调试功能](#调试功能)
-- [常见问题解答](#常见问题解答)
-
----
-
-## 功能介绍
-
-### 什么是 Workflow？
-
-Workflow（工作流）是 LangBot 提供的可视化自动化编排系统。通过拖拽节点、连接边的方式，您可以：
-
-- 📝 **构建复杂的对话流程**：使用条件分支、循环等控制节点
-- 🤖 **调用 AI 能力**：集成 LLM、知识库检索、参数提取
-- 🔗 **连接外部服务**：集成 Dify、n8n、Coze 等平台
-- ⚡ **自动化任务执行**：消息触发、定时触发、Webhook 触发
-
-### Workflow vs Pipeline
-
-| 对比项 | Pipeline | Workflow |
-|-------|----------|----------|
-| 配置方式 | 表单配置 | 可视化拖拽 |
-| 流程控制 | 线性执行 | 支持分支、循环、并行 |
-| 适用场景 | 简单对话 | 复杂流程 |
-| 学习曲线 | 低 | 中等 |
-
----
-
-## 快速入门
-
-### 第一步：创建 Workflow
-
-1. 在侧边栏点击 **Workflow** 进入工作流列表
-2. 点击右上角 **创建工作流** 按钮
-3. 填写基本信息：
-   - **名称**：给工作流起一个描述性的名字
-   - **描述**：可选，说明工作流的用途
-   - **图标**：选择一个 emoji 作为标识
-
-### 第二步：添加节点
-
-进入编辑器后，左侧是节点面板，中间是画布区域，右侧是属性面板。
-
-1. **添加触发节点**：从左侧面板拖拽一个"消息触发"节点到画布
-2. **添加 AI 节点**：拖拽一个"LLM 调用"节点
-3. **添加回复节点**：拖拽一个"回复消息"节点
-
-### 第三步：连接节点
-
-1. 将鼠标悬停在触发节点的输出端口（右侧小圆点）
-2. 按住鼠标拖拽到 LLM 节点的输入端口（左侧小圆点）
-3. 同样方式连接 LLM 节点和回复节点
-
-```
-[消息触发] ──▶ [LLM 调用] ──▶ [回复消息]
-```
-
-### 第四步：配置节点
-
-点击 LLM 调用节点，在右侧属性面板配置：
-
-- **运行方式**：选择"本地 Agent"
-- **系统提示词**：描述 AI 的角色和行为
-- **模型**：选择要使用的 LLM 模型
-
-点击回复消息节点配置：
-
-- **消息内容**：设置为 `{{nodes.llm_call.outputs.response}}`（引用 LLM 输出）
-
-### 第五步：保存并绑定
-
-1. 点击工具栏的 **保存** 按钮
-2. 返回 Bot 配置页面
-3. 在 Bot 的绑定设置中选择 **Workflow**，然后选择刚创建的工作流
-
-恭喜！您已经创建了第一个 Workflow。
-
----
-
-## 节点类型说明
-
-### 触发节点 (Trigger)
-
-触发节点是工作流的入口，定义何时启动执行。
-
-| 节点 | 说明 | 输出 |
-|-----|------|------|
-| 消息触发 | 收到消息时触发 | message, sender_id, platform |
-| 定时触发 | 按 Cron 表达式定时触发 | timestamp |
-| Webhook 触发 | 收到 HTTP 请求时触发 | request_body, headers |
-| 事件触发 | 系统事件触发 | event_type, event_data |
-
-**消息触发配置示例**：
-
-```yaml
-触发条件:
-  - 关键词匹配: ["帮助", "help"]
-  - 平台: ["wechat", "qq"]
-```
-
-### AI 节点
-
-AI 节点用于调用各种 AI 能力。
-
-| 节点 | 说明 | 典型用途 |
-|-----|------|---------|
-| LLM 调用 | 调用大语言模型 | 生成回复、理解意图 |
-| 问题分类器 | 对用户问题分类 | 路由到不同处理分支 |
-| 参数提取器 | 从文本提取结构化数据 | 提取订单号、日期等 |
-| 知识库检索 | 查询知识库 | RAG 增强回复 |
-
-**LLM 调用配置示例**：
-
-```yaml
-运行方式: 本地 Agent
-模型: gpt-4
-系统提示词: |
-  你是一个友好的客服助手。
-  请根据用户的问题提供帮助。
-温度: 0.7
-最大 Token 数: 2000
-```
-
-### 处理节点 (Process)
-
-处理节点用于数据处理和外部调用。
-
-| 节点 | 说明 | 典型用途 |
-|-----|------|---------|
-| 代码执行 | 执行 Python/JavaScript 代码 | 数据处理、格式转换 |
-| HTTP 请求 | 发送 HTTP 请求 | 调用外部 API |
-| 数据转换 | JSON/模板转换 | 数据格式化 |
-
-**HTTP 请求配置示例**：
-
-```yaml
-URL: https://api.example.com/data
-方法: POST
-请求头:
-  Content-Type: application/json
-  Authorization: Bearer {{variables.api_key}}
-请求体: |
-  {"query": "{{message.content}}"}
-```
-
-### 控制节点 (Control)
-
-控制节点用于流程控制。
-
-| 节点 | 说明 | 用途 |
-|-----|------|------|
-| 条件分支 | 二选一分支 | if-else 逻辑 |
-| 多路分支 | 多选一分支 | switch-case 逻辑 |
-| 循环 | 遍历数组 | 批量处理 |
-| 并行 | 同时执行多分支 | 并发处理 |
-| 等待 | 暂停执行 | 延时处理 |
-| 合并 | 合并多个分支 | 汇总结果 |
-
-**条件分支配置示例**：
-
-```yaml
-条件表达式: "{{nodes.classifier.outputs.category}}" == "complaint"
-真分支: 投诉处理
-假分支: 普通咨询
-```
-
-### 动作节点 (Action)
-
-动作节点执行具体操作。
-
-| 节点 | 说明 | 用途 |
-|-----|------|------|
-| 发送消息 | 主动发送消息 | 通知、推送 |
-| 回复消息 | 回复当前消息 | 对话回复 |
-| 存储数据 | 保存数据到存储 | 持久化 |
-| 调用 Pipeline | 调用现有 Pipeline | 复用现有流程 |
-
-**回复消息配置示例**：
-
-```yaml
-消息内容: |
-  感谢您的咨询！
-  
-  {{nodes.llm_call.outputs.response}}
-  
-  如有其他问题，随时联系我。
-```
-
-### 集成节点 (Integration)
-
-集成节点连接外部平台。
-
-| 节点 | 说明 | 平台 |
-|-----|------|------|
-| Dify 工作流 | 调用 Dify 应用 | Dify |
-| Dify 知识库 | 查询 Dify 知识库 | Dify |
-| n8n 工作流 | 调用 n8n 流程 | n8n |
-| Langflow | 调用 Langflow 流程 | Langflow |
-| Coze Bot | 调用扣子 Bot | Coze |
-
-**Dify 工作流配置示例**：
-
-```yaml
-API 地址: https://api.dify.ai/v1
-API Key: sk-xxxxx
-应用类型: workflow
-同步对话历史: true
-```
-
----
-
-## 编辑器使用指南
-
-### 画布操作
-
-| 操作 | 方式 |
-|-----|------|
-| 平移画布 | 按住鼠标中键/空格+左键 拖拽 |
-| 缩放画布 | 鼠标滚轮 / 工具栏按钮 |
-| 框选多个节点 | 按住 Shift + 拖拽框选 |
-| 适应视图 | 点击工具栏"适应"按钮 |
-
-### 节点操作
-
-| 操作 | 方式 |
-|-----|------|
-| 添加节点 | 从左侧面板拖拽到画布 |
-| 移动节点 | 点击节点拖拽 |
-| 删除节点 | 选中后按 Delete / 点击工具栏删除 |
-| 复制节点 | 选中后 Ctrl+C / 工具栏复制 |
-| 粘贴节点 | Ctrl+V / 工具栏粘贴 |
-
-### 连接操作
-
-| 操作 | 方式 |
-|-----|------|
-| 创建连接 | 从输出端口拖拽到输入端口 |
-| 删除连接 | 点击连接线后按 Delete |
-| 选中连接 | 点击连接线 |
-
-### 快捷键
-
-| 快捷键 | 功能 |
-|-------|------|
-| Ctrl + Z | 撤销 |
-| Ctrl + Shift + Z | 重做 |
-| Ctrl + C | 复制 |
-| Ctrl + V | 粘贴 |
-| Delete | 删除选中 |
-| Ctrl + S | 保存 |
-
-### 工具栏功能
-
-```
-[撤销] [重做] | [放大] [缩小] [适应] | [复制] [粘贴] [删除] | [保存] [调试]
-```
-
----
-
-## 调试功能
-
-### 启动调试
-
-1. 点击工具栏的 **调试** 按钮
-2. 在调试面板中配置初始数据：
-   - **输入消息**：模拟用户发送的消息
-   - **会话 ID**：可选，用于测试会话变量
-   - **变量**：设置初始变量值
-
-3. 点击 **开始调试** 按钮
-
-### 调试控制
-
-| 按钮 | 功能 |
-|-----|------|
-| ▶️ 开始/继续 | 开始或继续执行 |
-| ⏸️ 暂停 | 暂停执行 |
-| ⏹️ 停止 | 停止执行 |
-| ⏭️ 单步 | 执行下一个节点 |
-
-### 断点
-
-- **设置断点**：点击节点上的断点图标
-- **断点触发**：执行到断点时自动暂停
-- **查看状态**：在暂停时查看节点的输入输出
-
-### 执行日志
-
-调试面板下方显示实时日志：
-
-```
-[INFO] 2024-01-15 10:30:00 - Starting debug execution
-[INFO] 2024-01-15 10:30:00 - Executing node: message_trigger
-[DEBUG] 2024-01-15 10:30:00 - Node inputs: {"message": "你好"}
-[INFO] 2024-01-15 10:30:01 - Node completed in 50ms
-[INFO] 2024-01-15 10:30:01 - Executing node: llm_call
-...
-```
-
-### 节点状态颜色
-
-| 颜色 | 状态 |
-|-----|------|
-| 灰色 | 待执行 |
-| 蓝色 | 执行中 |
-| 绿色 | 已完成 |
-| 红色 | 失败 |
-| 黄色 | 已跳过 |
-
----
-
-## 常见问题解答
-
-### Q1：如何在节点间传递数据？
-
-使用表达式语法引用其他节点的输出：
-
-```
-{{nodes.节点ID.outputs.输出名称}}
-```
-
-例如：
-- `{{nodes.llm_call.outputs.response}}` - 引用 LLM 节点的响应
-- `{{nodes.http_request.outputs.body}}` - 引用 HTTP 请求的响应体
-
-### Q2：如何使用变量？
-
-Workflow 支持三种变量类型：
-
-1. **工作流变量**：`{{variables.变量名}}`
-2. **会话变量**：`{{conversation_variables.变量名}}`
-3. **消息上下文**：`{{message.content}}`、`{{message.sender_id}}`
-
-### Q3：条件分支如何写条件表达式？
-
-支持以下运算符：
-
-- 比较：`==`, `!=`, `>`, `<`, `>=`, `<=`
-- 逻辑：`and`, `or`, `not`
-- 包含：`in`
-
-示例：
-```python
-# 字符串比较
-"{{nodes.classifier.outputs.intent}}" == "purchase"
-
-# 数值比较
-{{nodes.extractor.outputs.amount}} > 1000
-
-# 包含检查
-"退款" in "{{message.content}}"
-```
-
-### Q4：如何处理错误？
-
-1. **节点级重试**：在节点配置中设置重试次数
-2. **全局错误处理**：在 Workflow 设置中配置错误处理策略
-3. **条件分支**：使用条件节点检查上一节点的状态
-
-### Q5：如何查看执行历史？
-
-1. 进入 Workflow 详情页
-2. 点击 **执行历史** 标签
-3. 查看每次执行的状态、耗时、输入输出
-
-### Q6：Workflow 可以被多个 Bot 使用吗？
-
-是的。一个 Workflow 可以被多个 Bot 绑定使用，但每个 Bot 只能绑定一个处理单元（Pipeline 或 Workflow）。
-
-### Q7：如何复制现有的 Workflow？
-
-在 Workflow 列表页，点击工作流卡片右上角的菜单，选择"复制"即可创建副本。
-
-### Q8：支持版本回滚吗？
-
-支持。每次保存都会创建新版本。在 Workflow 详情页可以查看版本历史并回滚到指定版本。
-
----
-
-## 最佳实践
-
-### 1. 合理命名
-
-- 为节点和 Workflow 使用描述性名称
-- 使用统一的命名规范
-
-### 2. 模块化设计
-
-- 将复杂流程拆分为多个小 Workflow
-- 使用"调用 Pipeline"节点复用现有流程
-
-### 3. 错误处理
-
-- 为关键节点设置重试机制
-- 使用条件分支处理异常情况
-- 添加日志记录便于排查问题
-
-### 4. 测试先行
-
-- 使用调试功能充分测试
-- 准备多种测试场景
-- 检查边界情况
-
-### 5. 性能优化
-
-- 避免不必要的节点
-- 使用并行节点提高效率
-- 合理设置超时时间
-
----
-
-## 更多资源
-
-- [开发者文档](../development/workflow-system.md)
-- [设计文档](../../../plans/langbot-workflow-design.md)
-- [API 文档](../service-api-openapi.json)

pyproject.toml

@@ -70,7 +70,7 @@ dependencies = [
     "chromadb>=1.0.0,<2.0.0",
     "qdrant-client (>=1.15.1,<2.0.0)",
     "pyseekdb==1.1.0.post3",
-    "langbot-plugin @ file:///home/qinjunyan/code/projects/langbot/langbot-plugin-sdk",
+    "langbot-plugin==0.4.6",
     "asyncpg>=0.30.0",
     "line-bot-sdk>=3.19.0",
     "matrix-nio>=0.25.2",

src/langbot/pkg/api/http/controller/groups/pipelines/embed.py

@@ -62,23 +62,15 @@ class EmbedRouterGroup(group.RouterGroup):
         """Resolve *bot_uuid* to ``(runtime_bot, pipeline_uuid)``.
 
         Returns ``(None, None)`` when the bot does not exist, is not a
-        ``web_page_bot``, is disabled, or has no pipeline/workflow bound.
+        ``web_page_bot``, is disabled, or has no pipeline bound.
         """
         for bot in self.ap.platform_mgr.bots:
             if (
                 bot.bot_entity.uuid == bot_uuid
                 and bot.bot_entity.adapter == 'web_page_bot'
                 and bot.bot_entity.enable
+                and bot.bot_entity.use_pipeline_uuid
             ):
-                # Check for workflow binding first
-                binding_type = getattr(bot.bot_entity, 'binding_type', 'pipeline') or 'pipeline'
-                binding_uuid = getattr(bot.bot_entity, 'binding_uuid', None)
-
-                if binding_type == 'workflow' and binding_uuid:
-                    # For workflow binding, return workflow UUID
-                    return bot, binding_uuid
-                elif bot.bot_entity.use_pipeline_uuid:
-                    # For pipeline binding, return pipeline UUID
                 return bot, bot.bot_entity.use_pipeline_uuid
         return None, None
 

src/langbot/pkg/api/http/controller/groups/pipelines/pipelines.py

@@ -86,10 +86,6 @@ class PipelinesRouterGroup(group.RouterGroup):
                         'available_plugins': plugins,
                         'bound_mcp_servers': extensions_prefs.get('mcp_servers', []),
                         'available_mcp_servers': mcp_servers,
-                        'bound_mcp_resources': extensions_prefs.get('mcp_resources', []),
-                        'mcp_resource_agent_read_enabled': extensions_prefs.get(
-                            'mcp_resource_agent_read_enabled', True
-                        ),
                         'bound_skills': extensions_prefs.get('skills', []),
                         'available_skills': available_skills,
                     }
@@ -103,8 +99,6 @@ class PipelinesRouterGroup(group.RouterGroup):
                 bound_plugins = json_data.get('bound_plugins', [])
                 bound_mcp_servers = json_data.get('bound_mcp_servers', [])
                 bound_skills = json_data.get('bound_skills', [])
-                bound_mcp_resources = json_data.get('bound_mcp_resources')
-                mcp_resource_agent_read_enabled = json_data.get('mcp_resource_agent_read_enabled')
 
                 await self.ap.pipeline_service.update_pipeline_extensions(
                     pipeline_uuid,
@@ -114,8 +108,6 @@ class PipelinesRouterGroup(group.RouterGroup):
                     enable_all_mcp_servers,
                     bound_skills=bound_skills,
                     enable_all_skills=enable_all_skills,
-                    bound_mcp_resources=bound_mcp_resources,
-                    mcp_resource_agent_read_enabled=mcp_resource_agent_read_enabled,
                 )
 
                 return self.success()

src/langbot/pkg/api/http/controller/groups/resources/mcp.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import quart
 import traceback
-from urllib.parse import unquote
 
 
 from ... import group
@@ -67,50 +66,3 @@ class MCPRouterGroup(group.RouterGroup):
             server_data = await quart.request.json
             task_id = await self.ap.mcp_service.test_mcp_server(server_name=server_name, server_data=server_data)
             return self.success(data={'task_id': task_id})
-
-        @self.route('/servers/<server_name>/resources', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
-        async def _(server_name: str) -> str:
-            """Get resources from an MCP server"""
-            server_name = unquote(server_name)
-            try:
-                resources = await self.ap.mcp_service.get_mcp_server_resources(server_name)
-                templates = await self.ap.mcp_service.get_mcp_server_resource_templates(server_name)
-                runtime_info = await self.ap.mcp_service.get_runtime_info(server_name)
-                return self.success(
-                    data={
-                        'resources': resources,
-                        'resource_templates': templates,
-                        'resource_capabilities': (runtime_info or {}).get('resource_capabilities', {}),
-                    }
-                )
-            except Exception as e:
-                return self.http_status(500, -1, f'Failed to get resources: {str(e)}')
-
-        @self.route('/servers/<server_name>/resource-templates', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
-        async def _(server_name: str) -> str:
-            """Get resource templates from an MCP server"""
-            server_name = unquote(server_name)
-            try:
-                templates = await self.ap.mcp_service.get_mcp_server_resource_templates(server_name)
-                return self.success(data={'resource_templates': templates})
-            except Exception as e:
-                return self.http_status(500, -1, f'Failed to get resource templates: {str(e)}')
-
-        @self.route('/servers/<server_name>/resources/read', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
-        async def _(server_name: str) -> str:
-            """Read a resource from an MCP server"""
-            server_name = unquote(server_name)
-            data = await quart.request.json
-            uri = data.get('uri')
-            if not uri:
-                return self.http_status(400, -1, 'URI is required')
-            try:
-                envelope = await self.ap.mcp_service.read_mcp_server_resource_envelope(
-                    server_name,
-                    uri,
-                    max_bytes=data.get('max_bytes'),
-                    include_blob=bool(data.get('include_blob', False)),
-                )
-                return self.success(data=envelope)
-            except Exception as e:
-                return self.http_status(500, -1, f'Failed to read resource: {str(e)}')

src/langbot/pkg/api/http/controller/groups/resources/tools.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import quart
-
 from ... import group
 
 
@@ -11,41 +9,25 @@ class ToolsRouterGroup(group.RouterGroup):
         @self.route('', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
         async def _() -> str:
             """获取所有可用工具列表"""
-            pipeline_uuid = quart.request.args.get('pipeline_uuid') or quart.request.args.get('pipeline_id')
+            tools = await self.ap.tool_mgr.get_all_tools()
-            bound_plugins: list[str] | None = None
-            bound_mcp_servers: list[str] | None = None
 
-            if pipeline_uuid:
+            tool_list = []
-                pipeline = await self.ap.pipeline_service.get_pipeline(pipeline_uuid)
+            for tool in tools:
-                if pipeline is None:
+                tool_list.append(
-                    return self.http_status(404, -1, 'pipeline not found')
+                    {
-
+                        'name': tool.name,
-                extensions_prefs = pipeline.get('extensions_preferences', {}) or {}
+                        'description': tool.description,
-                if not extensions_prefs.get('enable_all_plugins', True):
+                        'human_desc': tool.human_desc,
-                    bound_plugins = [
+                        'parameters': tool.parameters,
-                        f'{plugin.get("author", "")}/{plugin.get("name", "")}'
-                        for plugin in extensions_prefs.get('plugins', [])
-                        if isinstance(plugin, dict) and plugin.get('name')
-                    ]
-                if not extensions_prefs.get('enable_all_mcp_servers', True):
-                    bound_mcp_servers = [
-                        server for server in (extensions_prefs.get('mcp_servers', []) or []) if isinstance(server, str)
-                    ]
-
-            return self.success(
-                data={
-                    'tools': await self.ap.tool_mgr.get_tool_catalog(
-                        bound_plugins,
-                        bound_mcp_servers,
-                        include_skill_authoring=True,
-                    )
                     }
                 )
 
+            return self.success(data={'tools': tool_list})
+
         @self.route('/<tool_name>', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
         async def _(tool_name: str) -> str:
             """获取特定工具详情"""
-            tools = await self.ap.tool_mgr.get_all_tools(include_skill_authoring=True)
+            tools = await self.ap.tool_mgr.get_all_tools()
 
             for tool in tools:
                 if tool.name == tool_name:

src/langbot/pkg/api/http/controller/groups/workflows/__init__.py

@@ -1,5 +0,0 @@
-# Workflow router group
-from .workflows import WorkflowsRouterGroup, ExecutionsRouterGroup
-from .websocket_chat import WorkflowWebSocketChatRouterGroup
-
-__all__ = ['WorkflowsRouterGroup', 'ExecutionsRouterGroup', 'WorkflowWebSocketChatRouterGroup']

src/langbot/pkg/api/http/controller/groups/workflows/websocket_chat.py

@@ -1,260 +0,0 @@
-"""Workflow WebSocket聊天路由 - 支持工作流调试的双向实时通信"""
-
-import asyncio
-import datetime
-import json
-import logging
-
-import quart
-
-from ... import group
-from ......platform.sources.websocket_manager import ws_connection_manager
-
-logger = logging.getLogger(__name__)
-
-
-@group.group_class('workflow_websocket_chat', '/api/v1/workflows/<workflow_uuid>/ws')
-class WorkflowWebSocketChatRouterGroup(group.RouterGroup):
-    async def initialize(self) -> None:
-        @self.quart_app.websocket(self.path + '/connect')
-        async def workflow_websocket_connect(workflow_uuid: str):
-            """
-            建立工作流WebSocket连接
-
-            URL参数:
-                - workflow_uuid: 工作流UUID
-                - session_type: 会话类型 (person/group)
-            """
-            try:
-                session_type = quart.websocket.args.get('session_type', 'person')
-                logger.info(
-                    'Workflow WebSocket connect request received',
-                    extra={
-                        'workflow_uuid': workflow_uuid,
-                        'session_type': session_type,
-                        'path': quart.websocket.path,
-                        'query_string': quart.websocket.query_string.decode('utf-8', errors='ignore'),
-                        'remote_addr': getattr(quart.websocket, 'remote_addr', None),
-                        'user_agent': quart.websocket.headers.get('User-Agent', ''),
-                        'host': quart.websocket.headers.get('Host', ''),
-                        'origin': quart.websocket.headers.get('Origin', ''),
-                    },
-                )
-
-                if session_type not in ['person', 'group']:
-                    await quart.websocket.send(
-                        json.dumps({'type': 'error', 'message': 'session_type must be person or group'})
-                    )
-                    return
-
-                websocket_adapter = self.ap.platform_mgr.websocket_proxy_bot.adapter
-
-                if not websocket_adapter:
-                    logger.warning(
-                        'Workflow WebSocket adapter missing',
-                        extra={
-                            'workflow_uuid': workflow_uuid,
-                            'session_type': session_type,
-                        },
-                    )
-                    await quart.websocket.send(json.dumps({'type': 'error', 'message': 'WebSocket adapter not found'}))
-                    return
-
-                connection = await ws_connection_manager.add_connection(
-                    websocket=quart.websocket._get_current_object(),
-                    pipeline_uuid=workflow_uuid,
-                    session_type=session_type,
-                    metadata={'user_agent': quart.websocket.headers.get('User-Agent', ''), 'is_workflow': True},
-                )
-
-                await quart.websocket.send(
-                    json.dumps(
-                        {
-                            'type': 'connected',
-                            'connection_id': connection.connection_id,
-                            'workflow_uuid': workflow_uuid,
-                            'session_type': session_type,
-                            'timestamp': connection.created_at.isoformat(),
-                        }
-                    )
-                )
-
-                logger.debug(
-                    f'Workflow WebSocket connection established: {connection.connection_id} '
-                    f'(workflow={workflow_uuid}, session_type={session_type})'
-                )
-
-                receive_task = asyncio.create_task(self._handle_receive(connection, websocket_adapter))
-                send_task = asyncio.create_task(self._handle_send(connection))
-
-                try:
-                    await asyncio.gather(receive_task, send_task)
-                except Exception as e:
-                    logger.error(f'Workflow WebSocket task execution error: {e}')
-                finally:
-                    await ws_connection_manager.remove_connection(connection.connection_id)
-                    logger.debug(f'Workflow WebSocket connection cleaned: {connection.connection_id}')
-
-            except Exception as e:
-                logger.error(
-                    'Workflow WebSocket connection error',
-                    exc_info=True,
-                    extra={
-                        'workflow_uuid': workflow_uuid,
-                        'session_type': quart.websocket.args.get('session_type', 'person'),
-                        'path': quart.websocket.path,
-                        'query_string': quart.websocket.query_string.decode('utf-8', errors='ignore'),
-                        'remote_addr': getattr(quart.websocket, 'remote_addr', None),
-                    },
-                )
-                try:
-                    await quart.websocket.send(json.dumps({'type': 'error', 'message': str(e)}))
-                except Exception as send_error:
-                    logger.debug(
-                        'Failed to send error message to workflow websocket client',
-                        exc_info=True,
-                        extra={
-                            'workflow_uuid': workflow_uuid,
-                            'send_error': str(send_error),
-                        },
-                    )
-
-        @self.route('/messages/<session_type>', methods=['GET'])
-        async def get_messages(workflow_uuid: str, session_type: str) -> str:
-            """获取工作流消息历史"""
-            try:
-                if session_type not in ['person', 'group']:
-                    return self.http_status(400, -1, 'session_type must be person or group')
-
-                websocket_adapter = self.ap.platform_mgr.websocket_proxy_bot.adapter
-
-                if not websocket_adapter:
-                    return self.http_status(404, -1, 'WebSocket adapter not found')
-
-                messages = websocket_adapter.get_websocket_messages(workflow_uuid, session_type)
-
-                return self.success(data={'messages': messages})
-
-            except Exception as e:
-                return self.http_status(500, -1, f'Internal server error: {str(e)}')
-
-        @self.route('/reset/<session_type>', methods=['POST'])
-        async def reset_session(workflow_uuid: str, session_type: str) -> str:
-            """重置工作流会话"""
-            try:
-                if session_type not in ['person', 'group']:
-                    return self.http_status(400, -1, 'session_type must be person or group')
-
-                websocket_adapter = self.ap.platform_mgr.websocket_proxy_bot.adapter
-
-                if not websocket_adapter:
-                    return self.http_status(404, -1, 'WebSocket adapter not found')
-
-                websocket_adapter.reset_session(workflow_uuid, session_type)
-
-                return self.success(data={'message': 'Session reset successfully'})
-
-            except Exception as e:
-                return self.http_status(500, -1, f'Internal server error: {str(e)}')
-
-        @self.route('/connections', methods=['GET'])
-        async def get_connections(workflow_uuid: str) -> str:
-            """获取当前工作流连接统计"""
-            try:
-                stats = ws_connection_manager.get_stats()
-                connections = await ws_connection_manager.get_connections_by_pipeline(workflow_uuid)
-
-                return self.success(
-                    data={
-                        'stats': stats,
-                        'connections': [
-                            {
-                                'connection_id': conn.connection_id,
-                                'session_type': conn.session_type,
-                                'created_at': conn.created_at.isoformat(),
-                                'last_active': conn.last_active.isoformat(),
-                                'is_active': conn.is_active,
-                            }
-                            for conn in connections
-                        ],
-                    }
-                )
-
-            except Exception as e:
-                return self.http_status(500, -1, f'Internal server error: {str(e)}')
-
-        @self.route('/broadcast', methods=['POST'])
-        async def broadcast_message(workflow_uuid: str) -> str:
-            """向所有工作流连接广播消息"""
-            try:
-                data = await quart.request.get_json()
-                message = data.get('message')
-
-                if not message:
-                    return self.http_status(400, -1, 'message is required')
-
-                broadcast_data = {
-                    'type': 'broadcast',
-                    'message': message,
-                    'timestamp': datetime.datetime.now().isoformat(),
-                }
-
-                await ws_connection_manager.broadcast_to_pipeline(workflow_uuid, broadcast_data)
-
-                return self.success(data={'message': 'Broadcast sent successfully'})
-
-            except Exception as e:
-                return self.http_status(500, -1, f'Internal server error: {str(e)}')
-
-    async def _handle_receive(self, connection, websocket_adapter):
-        """处理接收消息的任务"""
-        try:
-            while connection.is_active:
-                message = await quart.websocket.receive()
-
-                await ws_connection_manager.update_activity(connection.connection_id)
-
-                try:
-                    data = json.loads(message)
-                    message_type = data.get('type', 'message')
-
-                    if message_type == 'ping':
-                        await connection.send_queue.put(
-                            {'type': 'pong', 'timestamp': datetime.datetime.now().isoformat()}
-                        )
-
-                    elif message_type == 'message':
-                        logger.debug(f'收到工作流消息: {data} from {connection.connection_id}')
-                        await websocket_adapter.handle_websocket_message(connection, data)
-
-                    elif message_type == 'disconnect':
-                        logger.debug(f'Client disconnected: {connection.connection_id}')
-                        break
-
-                    else:
-                        logger.warning(f'Unknown message type: {message_type}')
-
-                except json.JSONDecodeError:
-                    logger.error(f'Invalid JSON message: {message}')
-                    await connection.send_queue.put({'type': 'error', 'message': 'Invalid JSON format'})
-
-        except Exception as e:
-            logger.error(f'Receive message error: {e}', exc_info=True)
-        finally:
-            connection.is_active = False
-
-    async def _handle_send(self, connection):
-        """处理发送消息的任务"""
-        try:
-            while connection.is_active:
-                try:
-                    message = await asyncio.wait_for(connection.send_queue.get(), timeout=1.0)
-                    await quart.websocket.send(json.dumps(message))
-
-                except asyncio.TimeoutError:
-                    continue
-
-        except Exception as e:
-            logger.error(f'Send message error: {e}', exc_info=True)
-        finally:
-            connection.is_active = False

src/langbot/pkg/api/http/controller/groups/workflows/workflows.py

@@ -1,484 +0,0 @@
-from __future__ import annotations
-
-import quart
-
-from ... import group
-from ....service.workflow import WorkflowExecutionFailedError
-
-
-@group.group_class('workflows', '/api/v1/workflows')
-class WorkflowsRouterGroup(group.RouterGroup):
-    """Workflow API router group"""
-
-    async def initialize(self) -> None:
-        # Workflow CRUD
-        @self.route('', methods=['GET', 'POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _() -> str:
-            if quart.request.method == 'GET':
-                sort_by = quart.request.args.get('sort_by', 'created_at')
-                sort_order = quart.request.args.get('sort_order', 'DESC')
-                enabled_only = quart.request.args.get('enabled_only', 'false').lower() == 'true'
-                return self.success(
-                    data={'workflows': await self.ap.workflow_service.get_workflows(sort_by, sort_order, enabled_only)}
-                )
-            elif quart.request.method == 'POST':
-                json_data = await quart.request.json
-                workflow_uuid = await self.ap.workflow_service.create_workflow(json_data)
-                return self.success(data={'uuid': workflow_uuid})
-
-        # Get node types (available nodes for the editor)
-        @self.route('/_/node-types', methods=['GET'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _() -> str:
-            return self.success(
-                data={
-                    'node_types': await self.ap.workflow_service.get_node_types(),
-                    'categories': await self.ap.workflow_service.get_node_types_by_category_meta(),
-                }
-            )
-
-        # Get node types by category
-        @self.route('/_/node-types/categories', methods=['GET'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _() -> str:
-            return self.success(data={'categories': await self.ap.workflow_service.get_node_types_by_category()})
-
-        # Single workflow operations
-        @self.route(
-            '/<workflow_uuid>', methods=['GET', 'PUT', 'DELETE'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY
-        )
-        async def _(workflow_uuid: str) -> str:
-            if quart.request.method == 'GET':
-                workflow = await self.ap.workflow_service.get_workflow(workflow_uuid)
-                if workflow is None:
-                    return self.http_status(404, -1, 'workflow not found')
-                return self.success(data={'workflow': workflow})
-            elif quart.request.method == 'PUT':
-                json_data = await quart.request.json
-                try:
-                    await self.ap.workflow_service.update_workflow(workflow_uuid, json_data)
-                    return self.success()
-                except ValueError as e:
-                    return self.http_status(404, -1, str(e))
-            elif quart.request.method == 'DELETE':
-                await self.ap.workflow_service.delete_workflow(workflow_uuid)
-                return self.success()
-            return self.http_status(405, -1, 'method not allowed')
-
-        # Publish workflow (enable)
-        @self.route('/<workflow_uuid>/publish', methods=['POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _(workflow_uuid: str) -> str:
-            try:
-                await self.ap.workflow_service.publish_workflow(workflow_uuid)
-                return self.success()
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Unpublish workflow (disable)
-        @self.route('/<workflow_uuid>/unpublish', methods=['POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _(workflow_uuid: str) -> str:
-            try:
-                await self.ap.workflow_service.unpublish_workflow(workflow_uuid)
-                return self.success()
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Copy workflow
-        @self.route('/<workflow_uuid>/copy', methods=['POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _(workflow_uuid: str) -> str:
-            try:
-                new_uuid = await self.ap.workflow_service.copy_workflow(workflow_uuid)
-                return self.success(data={'uuid': new_uuid})
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Execute workflow manually
-        @self.route('/<workflow_uuid>/execute', methods=['POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _(workflow_uuid: str) -> str:
-            json_data = await quart.request.json or {}
-            trigger_data = json_data.get('trigger_data', {})
-            session_id = json_data.get('session_id')
-            user_id = json_data.get('user_id')
-            bot_id = json_data.get('bot_id')
-
-            try:
-                execution_id = await self.ap.workflow_service.execute_workflow(
-                    workflow_uuid,
-                    trigger_type='manual',
-                    trigger_data=trigger_data,
-                    session_id=session_id,
-                    user_id=user_id,
-                    bot_id=bot_id,
-                )
-                return self.success(data={'execution_id': execution_id})
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-            except WorkflowExecutionFailedError as e:
-                return self.http_status(500, -1, e.message)
-
-        # Get workflow executions
-        @self.route('/<workflow_uuid>/executions', methods=['GET'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _(workflow_uuid: str) -> str:
-            limit = int(quart.request.args.get('limit', 50))
-            offset = int(quart.request.args.get('offset', 0))
-            executions = await self.ap.workflow_service.get_executions(
-                workflow_uuid=workflow_uuid, limit=limit, offset=offset
-            )
-            return self.success(data=executions)
-
-        @self.route(
-            '/<workflow_uuid>/executions/<execution_uuid>',
-            methods=['GET'],
-            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
-        )
-        async def _(workflow_uuid: str, execution_uuid: str) -> str:
-            execution = await self.ap.workflow_service.get_execution(execution_uuid)
-            if execution is None:
-                return self.http_status(404, -1, 'execution not found')
-            if execution.get('workflow_uuid') != workflow_uuid:
-                return self.http_status(404, -1, 'execution not found in workflow')
-            return self.success(data={'execution': execution})
-
-        # Get workflow versions
-        @self.route('/<workflow_uuid>/versions', methods=['GET'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _(workflow_uuid: str) -> str:
-            versions = await self.ap.workflow_service.get_versions(workflow_uuid)
-            return self.success(data={'versions': versions})
-
-        # Rollback to a specific version
-        @self.route(
-            '/<workflow_uuid>/rollback/<int:version>', methods=['POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY
-        )
-        async def _(workflow_uuid: str, version: int) -> str:
-            try:
-                await self.ap.workflow_service.rollback_to_version(workflow_uuid, version)
-                return self.success()
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Workflow extensions (plugins and MCP servers)
-        @self.route(
-            '/<workflow_uuid>/extensions', methods=['GET', 'PUT'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY
-        )
-        async def _(workflow_uuid: str) -> str:
-            if quart.request.method == 'GET':
-                workflow = await self.ap.workflow_service.get_workflow(workflow_uuid)
-                if workflow is None:
-                    return self.http_status(404, -1, 'workflow not found')
-
-                # Get available plugins and MCP servers
-                pipeline_component_kinds = ['Command', 'EventListener', 'Tool']
-                plugins = await self.ap.plugin_connector.list_plugins(component_kinds=pipeline_component_kinds)
-                mcp_servers = await self.ap.mcp_service.get_mcp_servers(contain_runtime_info=True)
-
-                extensions_prefs = workflow.get('extensions_preferences', {})
-                return self.success(
-                    data={
-                        'enable_all_plugins': extensions_prefs.get('enable_all_plugins', True),
-                        'enable_all_mcp_servers': extensions_prefs.get('enable_all_mcp_servers', True),
-                        'bound_plugins': extensions_prefs.get('plugins', []),
-                        'available_plugins': plugins,
-                        'bound_mcp_servers': extensions_prefs.get('mcp_servers', []),
-                        'available_mcp_servers': mcp_servers,
-                    }
-                )
-            elif quart.request.method == 'PUT':
-                json_data = await quart.request.json
-                enable_all_plugins = json_data.get('enable_all_plugins', True)
-                enable_all_mcp_servers = json_data.get('enable_all_mcp_servers', True)
-                bound_plugins = json_data.get('bound_plugins', [])
-                bound_mcp_servers = json_data.get('bound_mcp_servers', [])
-
-                try:
-                    await self.ap.workflow_service.update_workflow_extensions(
-                        workflow_uuid, bound_plugins, bound_mcp_servers, enable_all_plugins, enable_all_mcp_servers
-                    )
-                    return self.success()
-                except ValueError as e:
-                    return self.http_status(404, -1, str(e))
-            return self.http_status(405, -1, 'method not allowed')
-
-        # Debug API - Start debug execution
-        @self.route('/<workflow_uuid>/debug/start', methods=['POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _(workflow_uuid: str) -> str:
-            json_data = await quart.request.json or {}
-            context = json_data.get('context', {})
-            variables = json_data.get('variables', {})
-            breakpoints = json_data.get('breakpoints', [])
-
-            try:
-                execution_id = await self.ap.workflow_service.start_debug_execution(
-                    workflow_uuid, context=context, variables=variables, breakpoints=breakpoints
-                )
-                return self.success(data={'execution_id': execution_id})
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Debug API - Pause execution
-        @self.route(
-            '/<workflow_uuid>/debug/<execution_uuid>/pause',
-            methods=['POST'],
-            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
-        )
-        async def _(workflow_uuid: str, execution_uuid: str) -> str:
-            try:
-                await self.ap.workflow_service.pause_debug_execution(workflow_uuid, execution_uuid)
-                return self.success()
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Debug API - Resume execution
-        @self.route(
-            '/<workflow_uuid>/debug/<execution_uuid>/resume',
-            methods=['POST'],
-            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
-        )
-        async def _(workflow_uuid: str, execution_uuid: str) -> str:
-            try:
-                await self.ap.workflow_service.resume_debug_execution(workflow_uuid, execution_uuid)
-                return self.success()
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Debug API - Step execution
-        @self.route(
-            '/<workflow_uuid>/debug/<execution_uuid>/step',
-            methods=['POST'],
-            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
-        )
-        async def _(workflow_uuid: str, execution_uuid: str) -> str:
-            try:
-                result = await self.ap.workflow_service.step_debug_execution(workflow_uuid, execution_uuid)
-                return self.success(data=result)
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Debug API - Stop execution
-        @self.route(
-            '/<workflow_uuid>/debug/<execution_uuid>/stop',
-            methods=['POST'],
-            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
-        )
-        async def _(workflow_uuid: str, execution_uuid: str) -> str:
-            try:
-                await self.ap.workflow_service.stop_debug_execution(workflow_uuid, execution_uuid)
-                return self.success()
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Debug API - Get debug state
-        @self.route(
-            '/<workflow_uuid>/debug/<execution_uuid>/state',
-            methods=['GET'],
-            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
-        )
-        async def _(workflow_uuid: str, execution_uuid: str) -> str:
-            try:
-                state = await self.ap.workflow_service.get_debug_state(workflow_uuid, execution_uuid)
-                return self.success(data=state)
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Get execution logs
-        @self.route(
-            '/<workflow_uuid>/executions/<execution_uuid>/logs',
-            methods=['GET'],
-            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
-        )
-        async def _(workflow_uuid: str, execution_uuid: str) -> str:
-            limit = int(quart.request.args.get('limit', 100))
-            offset = int(quart.request.args.get('offset', 0))
-            try:
-                result = await self.ap.workflow_service.get_execution_logs(workflow_uuid, execution_uuid, limit, offset)
-                return self.success(data=result)
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Rerun execution
-        @self.route(
-            '/<workflow_uuid>/executions/<execution_uuid>/rerun',
-            methods=['POST'],
-            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
-        )
-        async def _(workflow_uuid: str, execution_uuid: str) -> str:
-            try:
-                new_execution_id = await self.ap.workflow_service.rerun_execution(workflow_uuid, execution_uuid)
-                return self.success(data={'execution_uuid': new_execution_id})
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # Get workflow statistics
-        @self.route('/<workflow_uuid>/stats', methods=['GET'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _(workflow_uuid: str) -> str:
-            try:
-                stats = await self.ap.workflow_service.get_workflow_stats(workflow_uuid)
-                return self.success(data=stats)
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-
-        # LLM Node Performance Test Endpoint
-        # Tests each step of LLM node execution with detailed timing
-        @self.route('/_/test/llm-node', methods=['POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _() -> str:
-            """Test LLM node performance with detailed step-by-step timing.
-            
-            Request body:
-            {
-                "model_uuid": "uuid-of-model",
-                "system_prompt": "optional system prompt",
-                "user_prompt": "test message",
-                "temperature": 0.7,
-                "max_tokens": 100
-            }
-            
-            Response includes timing for each step:
-            - model_fetch: Time to get model from model_mgr
-            - prompt_build: Time to build messages
-            - llm_call: Time for actual LLM invocation
-            - total: Total time
-            - usage: Token usage information
-            """
-            import time
-            
-            json_data = await quart.request.json
-            if not json_data:
-                return self.http_status(400, -1, 'Request body is required')
-            
-            model_uuid = json_data.get('model_uuid', '')
-            if not model_uuid:
-                return self.http_status(400, -1, 'model_uuid is required')
-            
-            user_prompt = json_data.get('user_prompt', 'test')
-            system_prompt = json_data.get('system_prompt', '')
-            temperature = json_data.get('temperature')
-            max_tokens = json_data.get('max_tokens', 0)
-            
-            timings = {}
-            errors = []
-            
-            # Step 1: Model fetch
-            t_start = time.perf_counter()
-            try:
-                runtime_model = await self.ap.model_mgr.get_model_by_uuid(model_uuid)
-                timings['model_fetch_ms'] = round((time.perf_counter() - t_start) * 1000, 2)
-                timings['model_found'] = True
-                timings['model_name'] = runtime_model.model_entity.name if runtime_model else None
-            except Exception as e:
-                timings['model_fetch_ms'] = round((time.perf_counter() - t_start) * 1000, 2)
-                timings['model_found'] = False
-                errors.append(f'Model fetch failed: {str(e)}')
-                return self.http_status(400, -1, {
-                    'error': errors[0],
-                    'timings': timings,
-                })
-            
-            # Step 2: Build messages
-            t_start = time.perf_counter()
-            import langbot_plugin.api.entities.builtin.provider.message as provider_message
-            messages = []
-            if system_prompt:
-                messages.append(provider_message.Message(role='system', content=system_prompt))
-            messages.append(provider_message.Message(role='user', content=user_prompt))
-            timings['prompt_build_ms'] = round((time.perf_counter() - t_start) * 1000, 2)
-            
-            # Step 3: Build extra args
-            extra_args = {}
-            if temperature is not None:
-                extra_args['temperature'] = float(temperature)
-            if max_tokens and int(max_tokens) > 0:
-                extra_args['max_tokens'] = int(max_tokens)
-            
-            # Step 4: LLM call
-            t_start = time.perf_counter()
-            try:
-                result_message = await runtime_model.provider.invoke_llm(
-                    query=None,
-                    model=runtime_model,
-                    messages=messages,
-                    funcs=None,
-                    extra_args=extra_args,
-                )
-                timings['llm_call_ms'] = round((time.perf_counter() - t_start) * 1000, 2)
-                timings['llm_call_success'] = True
-                
-                # Extract response text
-                response_text = ''
-                if isinstance(result_message.content, str):
-                    response_text = result_message.content
-                elif isinstance(result_message.content, list):
-                    for elem in result_message.content:
-                        if hasattr(elem, 'text') and elem.text:
-                            response_text += elem.text
-                        elif isinstance(elem, str):
-                            response_text += elem
-                
-                timings['response_length'] = len(response_text)
-                timings['response_preview'] = response_text[:200]
-                
-                # Extract usage
-                usage = {'prompt_tokens': 0, 'completion_tokens': 0, 'total_tokens': 0}
-                if hasattr(result_message, 'usage') and result_message.usage:
-                    u = result_message.usage
-                    usage = {
-                        'prompt_tokens': getattr(u, 'prompt_tokens', 0) or 0,
-                        'completion_tokens': getattr(u, 'completion_tokens', 0) or 0,
-                        'total_tokens': getattr(u, 'total_tokens', 0) or 0,
-                    }
-                timings['usage'] = usage
-                
-            except Exception as e:
-                timings['llm_call_ms'] = round((time.perf_counter() - t_start) * 1000, 2)
-                timings['llm_call_success'] = False
-                errors.append(f'LLM call failed: {str(e)}')
-            
-            # Calculate total
-            timings['total_ms'] = round(sum([
-                timings.get('model_fetch_ms', 0),
-                timings.get('prompt_build_ms', 0),
-                timings.get('llm_call_ms', 0),
-            ]), 2)
-            
-            # Add breakdown percentage
-            if timings['total_ms'] > 0:
-                timings['breakdown'] = {
-                    'model_fetch_pct': round(timings.get('model_fetch_ms', 0) / timings['total_ms'] * 100, 1),
-                    'prompt_build_pct': round(timings.get('prompt_build_ms', 0) / timings['total_ms'] * 100, 1),
-                    'llm_call_pct': round(timings.get('llm_call_ms', 0) / timings['total_ms'] * 100, 1),
-                }
-            
-            if errors:
-                timings['errors'] = errors
-            
-            return self.success(data={'test_result': timings})
-
-
-@group.group_class('executions', '/api/v1/executions')
-class ExecutionsRouterGroup(group.RouterGroup):
-    """Workflow execution API router group"""
-
-    async def initialize(self) -> None:
-        # Get all executions (across all workflows)
-        @self.route('', methods=['GET'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _() -> str:
-            limit = int(quart.request.args.get('limit', 50))
-            offset = int(quart.request.args.get('offset', 0))
-            status = quart.request.args.get('status')
-            executions = await self.ap.workflow_service.get_executions(limit=limit, offset=offset, status=status)
-            return self.success(data=executions)
-
-        # Get single execution
-        @self.route('/<execution_uuid>', methods=['GET'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _(execution_uuid: str) -> str:
-            execution = await self.ap.workflow_service.get_execution(execution_uuid)
-            if execution is None:
-                return self.http_status(404, -1, 'execution not found')
-            return self.success(data={'execution': execution})
-
-        # Cancel execution
-        @self.route('/<execution_uuid>/cancel', methods=['POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
-        async def _(execution_uuid: str) -> str:
-            try:
-                await self.ap.workflow_service.cancel_execution(execution_uuid)
-                return self.success()
-            except ValueError as e:
-                return self.http_status(404, -1, str(e))
-            except RuntimeError as e:
-                return self.http_status(400, -1, str(e))

src/langbot/pkg/api/http/controller/main.py

@@ -17,7 +17,6 @@ from .groups import platform as groups_platform
 from .groups import pipelines as groups_pipelines
 from .groups import knowledge as groups_knowledge
 from .groups import resources as groups_resources
-from .groups import workflows as groups_workflows
 from ...mcp.mount import MCPMount
 
 importutil.import_modules_in_pkg(groups)
@@ -26,7 +25,6 @@ importutil.import_modules_in_pkg(groups_platform)
 importutil.import_modules_in_pkg(groups_pipelines)
 importutil.import_modules_in_pkg(groups_knowledge)
 importutil.import_modules_in_pkg(groups_resources)
-importutil.import_modules_in_pkg(groups_workflows)
 
 
 class HTTPController:

src/langbot/pkg/api/http/service/bot.py

@@ -99,23 +99,16 @@ class BotService:
         # TODO: 检查配置信息格式
         bot_data['uuid'] = str(uuid.uuid4())
 
-        # Set default binding_type if not provided
+        # bind the most recently updated pipeline if any exist
-        if 'binding_type' not in bot_data:
-            bot_data['binding_type'] = 'pipeline'
-
-        # checkout the default pipeline (for backward compatibility)
         result = await self.ap.persistence_mgr.execute_async(
-            sqlalchemy.select(persistence_pipeline.LegacyPipeline).where(
+            sqlalchemy.select(persistence_pipeline.LegacyPipeline)
-                persistence_pipeline.LegacyPipeline.is_default == True
+            .order_by(persistence_pipeline.LegacyPipeline.updated_at.desc())
-            )
+            .limit(1)
         )
         pipeline = result.first()
         if pipeline is not None:
             bot_data['use_pipeline_uuid'] = pipeline.uuid
             bot_data['use_pipeline_name'] = pipeline.name
-            # Also set binding_uuid for new unified binding model
-            if 'binding_uuid' not in bot_data:
-                bot_data['binding_uuid'] = pipeline.uuid
 
         await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_bot.Bot).values(bot_data))
 
@@ -127,45 +120,26 @@ class BotService:
 
     async def update_bot(self, bot_uuid: str, bot_data: dict) -> None:
         """Update bot"""
-        if 'uuid' in bot_data:
+        update_data = bot_data.copy()
-            del bot_data['uuid']
 
-        # Handle binding_type and binding_uuid for the new unified binding model
+        if 'uuid' in update_data:
-        # If binding_type is explicitly set to 'workflow', skip pipeline validation
+            del update_data['uuid']
-        binding_type = bot_data.get('binding_type')
 
-        # set use_pipeline_name (for backward compatibility with 'pipeline' binding_type)
+        # set use_pipeline_name
-        # Only validate pipeline when binding_type is 'pipeline' or not set (default to pipeline)
+        if 'use_pipeline_uuid' in update_data:
-        if 'use_pipeline_uuid' in bot_data and binding_type != 'workflow':
             result = await self.ap.persistence_mgr.execute_async(
                 sqlalchemy.select(persistence_pipeline.LegacyPipeline).where(
-                    persistence_pipeline.LegacyPipeline.uuid == bot_data['use_pipeline_uuid']
+                    persistence_pipeline.LegacyPipeline.uuid == update_data['use_pipeline_uuid']
                 )
             )
             pipeline = result.first()
             if pipeline is not None:
-                bot_data['use_pipeline_name'] = pipeline.name
+                update_data['use_pipeline_name'] = pipeline.name
-                # Also sync to binding_uuid if binding_type is 'pipeline' or not set
-                if binding_type is None or binding_type == 'pipeline':
-                    bot_data['binding_uuid'] = bot_data['use_pipeline_uuid']
-                    bot_data['binding_type'] = 'pipeline'
             else:
-                # Only raise error if binding_type is explicitly 'pipeline' or not set
-                if binding_type is None or binding_type == 'pipeline':
                 raise Exception('Pipeline not found')
-                # If binding_type is 'workflow', just clear the use_pipeline_uuid
-                bot_data['use_pipeline_uuid'] = None
-                bot_data['use_pipeline_name'] = None
-
-        # If binding_uuid is set directly (for workflow), clear pipeline fields
-        if 'binding_uuid' in bot_data and binding_type == 'workflow':
-            # For workflow binding, clear pipeline-related fields to avoid confusion
-            bot_data['binding_type'] = 'workflow'
-            bot_data['use_pipeline_uuid'] = None
-            bot_data['use_pipeline_name'] = None
 
         await self.ap.persistence_mgr.execute_async(
-            sqlalchemy.update(persistence_bot.Bot).values(bot_data).where(persistence_bot.Bot.uuid == bot_uuid)
+            sqlalchemy.update(persistence_bot.Bot).values(update_data).where(persistence_bot.Bot.uuid == bot_uuid)
         )
         await self.ap.platform_mgr.remove_bot(bot_uuid)
 

src/langbot/pkg/api/http/service/mcp.py

@@ -136,32 +136,6 @@ class MCPService:
             if server_name in self.ap.tool_mgr.mcp_tool_loader.sessions:
                 await self.ap.tool_mgr.mcp_tool_loader.remove_mcp_server(server_name)
 
-    async def get_mcp_server_resources(self, server_name: str) -> list[dict]:
-        """Get resources from a specific MCP server."""
-        return await self.ap.tool_mgr.mcp_tool_loader.get_resources(server_name)
-
-    async def get_mcp_server_resource_templates(self, server_name: str) -> list[dict]:
-        """Get resource templates from a specific MCP server."""
-        return await self.ap.tool_mgr.mcp_tool_loader.get_resource_templates(server_name)
-
-    async def read_mcp_server_resource_envelope(
-        self,
-        server_name: str,
-        uri: str,
-        *,
-        max_bytes: int | None = None,
-        include_blob: bool = False,
-    ) -> dict:
-        """Read a resource from a specific MCP server with metadata."""
-        kwargs = {'include_blob': include_blob, 'source': 'ui_preview'}
-        if max_bytes is not None:
-            kwargs['max_bytes'] = max_bytes
-        return await self.ap.tool_mgr.mcp_tool_loader.read_resource_envelope(server_name, uri, **kwargs)
-
-    async def read_mcp_server_resource(self, server_name: str, uri: str) -> list[dict]:
-        """Read a resource from a specific MCP server."""
-        return await self.ap.tool_mgr.mcp_tool_loader.read_resource(server_name, uri)
-
     async def test_mcp_server(self, server_name: str, server_data: dict) -> int:
         """测试 MCP 服务器连接并返回任务 ID"""
 

src/langbot/pkg/api/http/service/pipeline.py

@@ -73,20 +73,6 @@ class PipelineService:
 
         return self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
 
-    async def get_pipeline_by_name(self, pipeline_name: str) -> dict | None:
-        result = await self.ap.persistence_mgr.execute_async(
-            sqlalchemy.select(persistence_pipeline.LegacyPipeline).where(
-                persistence_pipeline.LegacyPipeline.name == pipeline_name
-            )
-        )
-
-        pipeline = result.first()
-
-        if pipeline is None:
-            return None
-
-        return self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
-
     async def create_pipeline(self, pipeline_data: dict, default: bool = False) -> str:
         from ....utils import paths as path_utils
 
@@ -114,8 +100,6 @@ class PipelineService:
                 'enable_all_mcp_servers': True,
                 'plugins': [],
                 'mcp_servers': [],
-                'mcp_resources': [],
-                'mcp_resource_agent_read_enabled': True,
             }
 
         await self.ap.persistence_mgr.execute_async(
@@ -209,8 +193,6 @@ class PipelineService:
                     'enable_all_mcp_servers': True,
                     'plugins': [],
                     'mcp_servers': [],
-                    'mcp_resources': [],
-                    'mcp_resource_agent_read_enabled': True,
                 }
             ),
         }
@@ -235,8 +217,6 @@ class PipelineService:
         enable_all_mcp_servers: bool = True,
         bound_skills: list[str] = None,
         enable_all_skills: bool = True,
-        bound_mcp_resources: list[dict] = None,
-        mcp_resource_agent_read_enabled: bool | None = None,
     ) -> None:
         """Update the bound plugins and MCP servers for a pipeline"""
         # Get current pipeline
@@ -256,14 +236,10 @@ class PipelineService:
         extensions_preferences['enable_all_mcp_servers'] = enable_all_mcp_servers
         extensions_preferences['enable_all_skills'] = enable_all_skills
         extensions_preferences['plugins'] = bound_plugins
-        if mcp_resource_agent_read_enabled is not None:
-            extensions_preferences['mcp_resource_agent_read_enabled'] = mcp_resource_agent_read_enabled
         if bound_mcp_servers is not None:
             extensions_preferences['mcp_servers'] = bound_mcp_servers
         if bound_skills is not None:
             extensions_preferences['skills'] = bound_skills
-        if bound_mcp_resources is not None:
-            extensions_preferences['mcp_resources'] = bound_mcp_resources
 
         await self.ap.persistence_mgr.execute_async(
             sqlalchemy.update(persistence_pipeline.LegacyPipeline)

src/langbot/pkg/core/app.py

@@ -32,7 +32,6 @@ from ..api.http.service import mcp as mcp_service
 from ..api.http.service import apikey as apikey_service
 from ..api.http.service import webhook as webhook_service
 from ..api.http.service import monitoring as monitoring_service
-from ..api.http.service import workflow as workflow_service
 from ..api.http.service import skill as skill_service
 from ..api.http.service import maintenance as maintenance_service
 from ..discover import engine as discover_engine
@@ -154,8 +153,6 @@ class Application:
 
     webhook_service: webhook_service.WebhookService = None
 
-    workflow_service: workflow_service.WorkflowService = None
-
     telemetry: telemetry_module.TelemetryManager = None
 
     survey: survey_module.SurveyManager = None
@@ -258,22 +255,6 @@ class Application:
                     scopes=[core_entities.LifecycleControlScope.APPLICATION],
                 )
 
-            async def workflow_execution_cleanup_loop():
-                check_interval_seconds = 60
-                while True:
-                    try:
-                        cancelled = await self.workflow_service.cleanup_stale_executions()
-                        if cancelled > 0:
-                            self.logger.info(f'Workflow execution auto-cleanup: cancelled {cancelled} stale executions')
-                    except Exception as e:
-                        self.logger.warning(f'Workflow execution auto-cleanup error: {e}')
-                    await asyncio.sleep(check_interval_seconds)
-
-            self.task_mgr.create_task(
-                workflow_execution_cleanup_loop(),
-                name='workflow-execution-cleanup',
-                scopes=[core_entities.LifecycleControlScope.APPLICATION],
-            )
             # Start storage/log maintenance task if enabled
             storage_cleanup_cfg = self.instance_config.data.get('storage', {}).get('cleanup', {})
             if storage_cleanup_cfg.get('enabled', True) and self.maintenance_service is not None:

src/langbot/pkg/core/stages/build_app.py

@@ -29,7 +29,6 @@ from ...api.http.service import mcp as mcp_service
 from ...api.http.service import apikey as apikey_service
 from ...api.http.service import webhook as webhook_service
 from ...api.http.service import monitoring as monitoring_service
-from ...api.http.service import workflow as workflow_service
 from ...api.http.service import skill as skill_service
 from ...skill import manager as skill_mgr
 from ...api.http.service import maintenance as maintenance_service
@@ -90,9 +89,6 @@ class BuildAppStage(stage.BootingStage):
         webhook_service_inst = webhook_service.WebhookService(ap)
         ap.webhook_service = webhook_service_inst
 
-        workflow_service_inst = workflow_service.WorkflowService(ap)
-        ap.workflow_service = workflow_service_inst
-
         skill_service_inst = skill_service.SkillService(ap)
         ap.skill_service = skill_service_inst
 

src/langbot/pkg/core/stages/load_config.py

@@ -231,34 +231,3 @@ class LoadConfigStage(stage.BootingStage):
         ap.pipeline_config_meta_safety = await load_resource_yaml_template_data('metadata/pipeline/safety.yaml')
         ap.pipeline_config_meta_ai = await load_resource_yaml_template_data('metadata/pipeline/ai.yaml')
         ap.pipeline_config_meta_output = await load_resource_yaml_template_data('metadata/pipeline/output.yaml')
-
-        # Load workflow node metadata from YAML files. YAML is the source of
-        # truth for workflow editor metadata; Python classes provide execution
-        # logic and are bound through the registry.
-        from langbot.pkg.workflow.metadata import NodeMetadataLoader
-        from langbot.pkg.workflow.registry import NodeTypeRegistry
-
-        workflow_metadata_loader = NodeMetadataLoader()
-        workflow_node_count = await workflow_metadata_loader.load_core_metadata()
-        ap.workflow_node_configs = workflow_metadata_loader.get_all_metadata()
-        ap.workflow_node_metadata_loader = workflow_metadata_loader
-
-        workflow_registry = NodeTypeRegistry.instance()
-        for node_config in ap.workflow_node_configs.values():
-            workflow_registry.register_metadata(node_config, source=node_config.get('_source', 'core'))
-
-        # Auto-discover and register workflow nodes using discovery engine
-        if hasattr(ap, 'discover') and ap.discover is not None:
-            workflow_registry.discover_nodes(ap.discover)
-
-        workflow_load_errors = workflow_metadata_loader.get_load_errors()
-        if workflow_load_errors:
-            print(f'Workflow node metadata load errors: {len(workflow_load_errors)}')
-            for error in workflow_load_errors:
-                print(f"  - {error.get('file')}: {error.get('error')}")
-
-        print(
-            f'Loaded {workflow_node_count} workflow node metadata files; '
-            f'registered {workflow_registry.metadata_count()} metadata definitions, '
-            f'{workflow_registry.count()} node types'
-        )

src/langbot/pkg/discover/engine.py

@@ -304,65 +304,3 @@ class ComponentDiscoveryEngine:
             if component.kind == kind:
                 result.append(component)
         return result
-
-    def discover_workflow_nodes(self, nodes_dir: str) -> typing.List[typing.Type]:
-        """Discover workflow node classes from a directory of Python modules.
-        
-        Scans all .py files in the given directory, imports them, and collects
-        classes that are subclasses of WorkflowNode.
-        
-        Args:
-            nodes_dir: Directory path like 'pkg/workflow/nodes/'
-            
-        Returns:
-            List of WorkflowNode subclasses found
-        """
-        from langbot.pkg.workflow.node import WorkflowNode
-        
-        node_classes: typing.List[typing.Type[WorkflowNode]] = []
-        
-        # Normalize path
-        if nodes_dir.endswith('/'):
-            nodes_dir = nodes_dir[:-1]
-        
-        # Import the nodes package to trigger all module imports
-        module_path = nodes_dir.replace('/', '.').replace('\\', '.')
-        package_path = module_path
-        
-        try:
-            # Import the package __init__ to trigger submodule imports
-            importlib.import_module(f'langbot.{package_path}')
-        except ImportError:
-            self.ap.logger.warning(f'Failed to import workflow nodes package: langbot.{package_path}')
-        
-        # Since workflow/__init__.py is empty, explicitly import all .py files in the nodes directory
-        import os
-        # engine.py is in langbot/pkg/discover/, nodes are in langbot/pkg/workflow/nodes/
-        nodes_abs_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'workflow', 'nodes'))
-        if os.path.isdir(nodes_abs_path):
-            for filename in os.listdir(nodes_abs_path):
-                if filename.endswith('.py') and not filename.startswith('_'):
-                    module_name = filename[:-3]
-                    try:
-                        importlib.import_module(f'langbot.{package_path}.{module_name}')
-                    except ImportError as e:
-                        self.ap.logger.warning(f'Failed to import workflow node module: {module_name}: {e}')
-        
-        # Now collect all WorkflowNode subclasses from sys.modules
-        import sys
-        prefix = f'langbot.{package_path}.'
-        for mod_name, mod in sys.modules.items():
-            if mod_name.startswith(prefix) and mod is not None:
-                for attr_name in dir(mod):
-                    attr = getattr(mod, attr_name)
-                    if (
-                        isinstance(attr, type)
-                        and issubclass(attr, WorkflowNode)
-                        and attr is not WorkflowNode
-                        and hasattr(attr, 'type_name')
-                        and attr.type_name
-                    ):
-                        if attr not in node_classes:
-                            node_classes.append(attr)
-        
-        return node_classes

src/langbot/pkg/entity/persistence/bot.py

@@ -31,13 +31,6 @@ class Bot(Base):
     use_pipeline_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
     use_pipeline_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
     pipeline_routing_rules = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, server_default='[]')
-
-    # New unified binding fields
-    # binding_type: 'pipeline' or 'workflow'
-    binding_type = sqlalchemy.Column(sqlalchemy.String(32), nullable=False, server_default='pipeline')
-    # binding_uuid: UUID of the bound Pipeline or Workflow
-    binding_uuid = sqlalchemy.Column(sqlalchemy.String(64), nullable=True)
-
     created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
     updated_at = sqlalchemy.Column(
         sqlalchemy.DateTime,

src/langbot/pkg/entity/persistence/pipeline.py

@@ -26,14 +26,7 @@ class LegacyPipeline(Base):
     extensions_preferences = sqlalchemy.Column(
         sqlalchemy.JSON,
         nullable=False,
-        default={
+        default={'enable_all_plugins': True, 'enable_all_mcp_servers': True, 'plugins': [], 'mcp_servers': []},
-            'enable_all_plugins': True,
-            'enable_all_mcp_servers': True,
-            'plugins': [],
-            'mcp_servers': [],
-            'mcp_resources': [],
-            'mcp_resource_agent_read_enabled': True,
-        },
     )
 
 

src/langbot/pkg/entity/persistence/workflow.py

@@ -1,126 +0,0 @@
-"""Workflow persistence entities"""
-
-import sqlalchemy
-
-from .base import Base
-
-
-class Workflow(Base):
-    """Workflow definition"""
-
-    __tablename__ = 'workflows'
-
-    uuid = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True, unique=True)
-    name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
-    description = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
-    emoji = sqlalchemy.Column(sqlalchemy.String(10), nullable=True, default='🔄')
-    version = sqlalchemy.Column(sqlalchemy.Integer, nullable=False, default=1)
-    is_enabled = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False, default=True)
-
-    # Workflow definition stored as JSON
-    # Contains: nodes, edges, variables, settings
-    definition = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, default={})
-
-    # Global config (inherited from Pipeline capabilities)
-    # Contains: safety, output configs
-    global_config = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, default={})
-
-    # Extensions preferences (same as Pipeline)
-    extensions_preferences = sqlalchemy.Column(
-        sqlalchemy.JSON,
-        nullable=False,
-        default={'enable_all_plugins': True, 'enable_all_mcp_servers': True, 'plugins': [], 'mcp_servers': []},
-    )
-
-    created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
-    updated_at = sqlalchemy.Column(
-        sqlalchemy.DateTime,
-        nullable=False,
-        server_default=sqlalchemy.func.now(),
-        onupdate=sqlalchemy.func.now(),
-    )
-
-
-class WorkflowVersion(Base):
-    """Workflow version history"""
-
-    __tablename__ = 'workflow_versions'
-
-    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True, autoincrement=True)
-    workflow_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True)
-    version = sqlalchemy.Column(sqlalchemy.Integer, nullable=False)
-    definition = sqlalchemy.Column(sqlalchemy.JSON, nullable=False)
-    global_config = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, default={})
-    created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
-    created_by = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
-
-    __table_args__ = (sqlalchemy.UniqueConstraint('workflow_uuid', 'version', name='uq_workflow_version'),)
-
-
-class WorkflowTrigger(Base):
-    """Workflow trigger configuration"""
-
-    __tablename__ = 'workflow_triggers'
-
-    uuid = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True, unique=True)
-    workflow_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True)
-    type = sqlalchemy.Column(sqlalchemy.String(50), nullable=False)  # message, cron, event, webhook
-    config = sqlalchemy.Column(sqlalchemy.JSON, nullable=False, default={})
-    is_enabled = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False, default=True)
-    priority = sqlalchemy.Column(sqlalchemy.Integer, nullable=False, default=0)
-    created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
-    updated_at = sqlalchemy.Column(
-        sqlalchemy.DateTime,
-        nullable=False,
-        server_default=sqlalchemy.func.now(),
-        onupdate=sqlalchemy.func.now(),
-    )
-
-
-class WorkflowExecution(Base):
-    """Workflow execution record"""
-
-    __tablename__ = 'workflow_executions'
-
-    uuid = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True, unique=True)
-    workflow_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True)
-    workflow_version = sqlalchemy.Column(sqlalchemy.Integer, nullable=False)
-    status = sqlalchemy.Column(sqlalchemy.String(20), nullable=False)  # pending, running, completed, failed, cancelled
-    trigger_type = sqlalchemy.Column(sqlalchemy.String(50), nullable=True)
-    trigger_data = sqlalchemy.Column(sqlalchemy.JSON, nullable=True)
-    variables = sqlalchemy.Column(sqlalchemy.JSON, nullable=True)
-    start_time = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
-    end_time = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
-    error = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
-    created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
-
-
-class WorkflowNodeExecution(Base):
-    """Workflow node execution record"""
-
-    __tablename__ = 'workflow_node_executions'
-
-    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True, autoincrement=True)
-    execution_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True)
-    node_id = sqlalchemy.Column(sqlalchemy.String(100), nullable=False)
-    node_type = sqlalchemy.Column(sqlalchemy.String(50), nullable=False)
-    status = sqlalchemy.Column(sqlalchemy.String(20), nullable=False)  # pending, running, completed, failed, skipped
-    inputs = sqlalchemy.Column(sqlalchemy.JSON, nullable=True)
-    outputs = sqlalchemy.Column(sqlalchemy.JSON, nullable=True)
-    start_time = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
-    end_time = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
-    error = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
-    retry_count = sqlalchemy.Column(sqlalchemy.Integer, nullable=False, default=0)
-
-
-class ScheduledJob(Base):
-    """Scheduled job for cron triggers"""
-
-    __tablename__ = 'workflow_scheduled_jobs'
-
-    uuid = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True, unique=True)
-    trigger_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True)
-    cron_expression = sqlalchemy.Column(sqlalchemy.String(100), nullable=True)
-    next_run_time = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
-    last_run_time = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
-    is_enabled = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False, default=True)

src/langbot/pkg/persistence/alembic/versions/0008_add_mcp_resource_preferences.py

@@ -1,95 +0,0 @@
-"""add mcp resource preferences to pipelines
-
-Revision ID: 0008_mcp_resource_prefs
-Revises: 0007_add_bot_admins
-Create Date: 2026-06-30
-"""
-
-from __future__ import annotations
-
-import json
-from typing import Any
-
-import sqlalchemy as sa
-from alembic import op
-
-revision = '0008_mcp_resource_prefs'
-down_revision = '0007_add_bot_admins'
-branch_labels = None
-depends_on = None
-
-
-_PIPELINE_TABLE = sa.table(
-    'legacy_pipelines',
-    sa.column('uuid', sa.String(255)),
-    sa.column('extensions_preferences', sa.JSON()),
-)
-
-
-def _has_extensions_preferences_table(conn: sa.Connection) -> bool:
-    inspector = sa.inspect(conn)
-    if 'legacy_pipelines' not in inspector.get_table_names():
-        return False
-    columns = {column['name'] for column in inspector.get_columns('legacy_pipelines')}
-    return 'extensions_preferences' in columns
-
-
-def _decode_preferences(value: Any) -> dict[str, Any]:
-    if value is None:
-        return {}
-    if isinstance(value, dict):
-        return dict(value)
-    if isinstance(value, str):
-        try:
-            decoded = json.loads(value)
-        except json.JSONDecodeError:
-            return {}
-        if isinstance(decoded, dict):
-            return decoded
-    return {}
-
-
-def _update_preferences(conn: sa.Connection, uuid: str, preferences: dict[str, Any]) -> None:
-    conn.execute(
-        _PIPELINE_TABLE.update().where(_PIPELINE_TABLE.c.uuid == uuid).values(extensions_preferences=preferences)
-    )
-
-
-def upgrade() -> None:
-    conn = op.get_bind()
-    if not _has_extensions_preferences_table(conn):
-        return
-
-    rows = conn.execute(sa.select(_PIPELINE_TABLE.c.uuid, _PIPELINE_TABLE.c.extensions_preferences)).all()
-    for uuid, raw_preferences in rows:
-        preferences = _decode_preferences(raw_preferences)
-        changed = False
-
-        if 'mcp_resources' not in preferences:
-            preferences['mcp_resources'] = []
-            changed = True
-        if 'mcp_resource_agent_read_enabled' not in preferences:
-            preferences['mcp_resource_agent_read_enabled'] = True
-            changed = True
-
-        if changed:
-            _update_preferences(conn, uuid, preferences)
-
-
-def downgrade() -> None:
-    conn = op.get_bind()
-    if not _has_extensions_preferences_table(conn):
-        return
-
-    rows = conn.execute(sa.select(_PIPELINE_TABLE.c.uuid, _PIPELINE_TABLE.c.extensions_preferences)).all()
-    for uuid, raw_preferences in rows:
-        preferences = _decode_preferences(raw_preferences)
-        changed = False
-
-        for key in ('mcp_resources', 'mcp_resource_agent_read_enabled'):
-            if key in preferences:
-                preferences.pop(key)
-                changed = True
-
-        if changed:
-            _update_preferences(conn, uuid, preferences)

src/langbot/pkg/persistence/alembic/versions/0009_add_workflow_tables.py

@@ -1,207 +0,0 @@
-"""add workflow tables and bot binding fields
-
-Revision ID: 0009_add_workflow_tables
-Revises: 0008_mcp_resource_prefs
-Create Date: 2026-07-01
-"""
-
-from __future__ import annotations
-
-import sqlalchemy as sa
-from alembic import op
-
-revision = '0009_add_workflow_tables'
-down_revision = '0008_mcp_resource_prefs'
-branch_labels = None
-depends_on = None
-
-
-def _table_exists(conn: sa.Connection, table_name: str) -> bool:
-    return table_name in sa.inspect(conn).get_table_names()
-
-
-def _has_column(conn: sa.Connection, table_name: str, column_name: str) -> bool:
-    if not _table_exists(conn, table_name):
-        return False
-    return column_name in {column['name'] for column in sa.inspect(conn).get_columns(table_name)}
-
-
-def _has_index_for_columns(conn: sa.Connection, table_name: str, columns: tuple[str, ...]) -> bool:
-    if not _table_exists(conn, table_name):
-        return False
-    for index in sa.inspect(conn).get_indexes(table_name):
-        if tuple(index.get('column_names') or ()) == columns:
-            return True
-    return False
-
-
-def _ensure_index(conn: sa.Connection, table_name: str, index_name: str, columns: list[str]) -> None:
-    if _has_index_for_columns(conn, table_name, tuple(columns)):
-        return
-    op.create_index(index_name, table_name, columns)
-
-
-def _create_workflow_tables(conn: sa.Connection) -> None:
-    if not _table_exists(conn, 'workflows'):
-        op.create_table(
-            'workflows',
-            sa.Column('uuid', sa.String(255), primary_key=True),
-            sa.Column('name', sa.String(255), nullable=False),
-            sa.Column('description', sa.Text(), nullable=True),
-            sa.Column('emoji', sa.String(10), nullable=True),
-            sa.Column('version', sa.Integer(), nullable=False, server_default='1'),
-            sa.Column('is_enabled', sa.Boolean(), nullable=False, server_default=sa.true()),
-            sa.Column('definition', sa.JSON(), nullable=False, server_default=sa.text("'{}'")),
-            sa.Column('global_config', sa.JSON(), nullable=False, server_default=sa.text("'{}'")),
-            sa.Column(
-                'extensions_preferences',
-                sa.JSON(),
-                nullable=False,
-                server_default=sa.text(
-                    '\'{"enable_all_plugins": true, "enable_all_mcp_servers": true, "plugins": [], "mcp_servers": []}\''
-                ),
-            ),
-            sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
-            sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
-        )
-
-    if not _table_exists(conn, 'workflow_versions'):
-        op.create_table(
-            'workflow_versions',
-            sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
-            sa.Column('workflow_uuid', sa.String(255), nullable=False),
-            sa.Column('version', sa.Integer(), nullable=False),
-            sa.Column('definition', sa.JSON(), nullable=False),
-            sa.Column('global_config', sa.JSON(), nullable=False, server_default=sa.text("'{}'")),
-            sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
-            sa.Column('created_by', sa.String(255), nullable=True),
-            sa.UniqueConstraint('workflow_uuid', 'version', name='uq_workflow_version'),
-        )
-
-    if not _table_exists(conn, 'workflow_triggers'):
-        op.create_table(
-            'workflow_triggers',
-            sa.Column('uuid', sa.String(255), primary_key=True),
-            sa.Column('workflow_uuid', sa.String(255), nullable=False),
-            sa.Column('type', sa.String(50), nullable=False),
-            sa.Column('config', sa.JSON(), nullable=False, server_default=sa.text("'{}'")),
-            sa.Column('is_enabled', sa.Boolean(), nullable=False, server_default=sa.true()),
-            sa.Column('priority', sa.Integer(), nullable=False, server_default='0'),
-            sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
-            sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
-        )
-
-    if not _table_exists(conn, 'workflow_executions'):
-        op.create_table(
-            'workflow_executions',
-            sa.Column('uuid', sa.String(255), primary_key=True),
-            sa.Column('workflow_uuid', sa.String(255), nullable=False),
-            sa.Column('workflow_version', sa.Integer(), nullable=False),
-            sa.Column('status', sa.String(20), nullable=False),
-            sa.Column('trigger_type', sa.String(50), nullable=True),
-            sa.Column('trigger_data', sa.JSON(), nullable=True),
-            sa.Column('variables', sa.JSON(), nullable=True),
-            sa.Column('start_time', sa.DateTime(), nullable=True),
-            sa.Column('end_time', sa.DateTime(), nullable=True),
-            sa.Column('error', sa.Text(), nullable=True),
-            sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
-        )
-
-    if not _table_exists(conn, 'workflow_node_executions'):
-        op.create_table(
-            'workflow_node_executions',
-            sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
-            sa.Column('execution_uuid', sa.String(255), nullable=False),
-            sa.Column('node_id', sa.String(100), nullable=False),
-            sa.Column('node_type', sa.String(50), nullable=False),
-            sa.Column('status', sa.String(20), nullable=False),
-            sa.Column('inputs', sa.JSON(), nullable=True),
-            sa.Column('outputs', sa.JSON(), nullable=True),
-            sa.Column('start_time', sa.DateTime(), nullable=True),
-            sa.Column('end_time', sa.DateTime(), nullable=True),
-            sa.Column('error', sa.Text(), nullable=True),
-            sa.Column('retry_count', sa.Integer(), nullable=False, server_default='0'),
-        )
-
-    if not _table_exists(conn, 'workflow_scheduled_jobs'):
-        op.create_table(
-            'workflow_scheduled_jobs',
-            sa.Column('uuid', sa.String(255), primary_key=True),
-            sa.Column('trigger_uuid', sa.String(255), nullable=False),
-            sa.Column('cron_expression', sa.String(100), nullable=True),
-            sa.Column('next_run_time', sa.DateTime(), nullable=True),
-            sa.Column('last_run_time', sa.DateTime(), nullable=True),
-            sa.Column('is_enabled', sa.Boolean(), nullable=False, server_default=sa.true()),
-        )
-
-    _ensure_index(conn, 'workflow_versions', 'ix_workflow_versions_workflow_uuid', ['workflow_uuid'])
-    _ensure_index(conn, 'workflow_triggers', 'ix_workflow_triggers_workflow_uuid', ['workflow_uuid'])
-    _ensure_index(conn, 'workflow_executions', 'ix_workflow_executions_workflow_uuid', ['workflow_uuid'])
-    _ensure_index(
-        conn,
-        'workflow_node_executions',
-        'ix_workflow_node_executions_execution_uuid',
-        ['execution_uuid'],
-    )
-    _ensure_index(conn, 'workflow_scheduled_jobs', 'ix_workflow_scheduled_jobs_trigger_uuid', ['trigger_uuid'])
-
-
-def _add_bot_binding_fields(conn: sa.Connection) -> None:
-    if not _table_exists(conn, 'bots'):
-        return
-
-    if not _has_column(conn, 'bots', 'binding_type'):
-        op.add_column(
-            'bots',
-            sa.Column('binding_type', sa.String(32), nullable=False, server_default='pipeline'),
-        )
-
-    if not _has_column(conn, 'bots', 'binding_uuid'):
-        op.add_column('bots', sa.Column('binding_uuid', sa.String(64), nullable=True))
-
-    conn.execute(
-        sa.text("""
-        UPDATE bots
-        SET binding_uuid = use_pipeline_uuid
-        WHERE use_pipeline_uuid IS NOT NULL
-          AND use_pipeline_uuid != ''
-          AND (binding_uuid IS NULL OR binding_uuid = '')
-        """)
-    )
-    conn.execute(
-        sa.text("""
-        UPDATE bots
-        SET binding_type = 'pipeline'
-        WHERE binding_uuid IS NOT NULL
-          AND binding_uuid != ''
-          AND (binding_type IS NULL OR binding_type = '')
-        """)
-    )
-
-
-def upgrade() -> None:
-    conn = op.get_bind()
-    _create_workflow_tables(conn)
-    _add_bot_binding_fields(conn)
-
-
-def downgrade() -> None:
-    conn = op.get_bind()
-
-    if _has_column(conn, 'bots', 'binding_uuid'):
-        with op.batch_alter_table('bots') as batch_op:
-            batch_op.drop_column('binding_uuid')
-    if _has_column(conn, 'bots', 'binding_type'):
-        with op.batch_alter_table('bots') as batch_op:
-            batch_op.drop_column('binding_type')
-
-    for table_name in (
-        'workflow_scheduled_jobs',
-        'workflow_node_executions',
-        'workflow_executions',
-        'workflow_triggers',
-        'workflow_versions',
-        'workflows',
-    ):
-        if _table_exists(conn, table_name):
-            op.drop_table(table_name)

src/langbot/pkg/pipeline/pipelinemgr.py

@@ -13,7 +13,7 @@ import langbot_plugin.api.entities.builtin.platform.message as platform_message
 import langbot_plugin.api.entities.builtin.platform.events as platform_events
 import langbot_plugin.api.entities.events as events
 from ..utils import importutil
-from .config import coerce_pipeline_config
+from .config_coercion import coerce_pipeline_config
 
 import langbot_plugin.api.entities.builtin.provider.session as provider_session
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -96,15 +96,6 @@ class RuntimePipeline:
         extensions_prefs = pipeline_entity.extensions_preferences or {}
         self.enable_all_plugins = extensions_prefs.get('enable_all_plugins', True)
         self.enable_all_mcp_servers = extensions_prefs.get('enable_all_mcp_servers', True)
-        local_agent_config = (pipeline_entity.config or {}).get('ai', {}).get('local-agent', {})
-        self.mcp_resource_attachments = local_agent_config.get(
-            'mcp-resources',
-            extensions_prefs.get('mcp_resources', []),
-        )
-        self.mcp_resource_agent_read_enabled = local_agent_config.get(
-            'mcp-resource-agent-read-enabled',
-            extensions_prefs.get('mcp_resource_agent_read_enabled', True),
-        )
 
         if self.enable_all_plugins:
             # None indicates to use all available plugins
@@ -125,8 +116,6 @@ class RuntimePipeline:
         # Store bound plugins and MCP servers in query for filtering
         query.variables['_pipeline_bound_plugins'] = self.bound_plugins
         query.variables['_pipeline_bound_mcp_servers'] = self.bound_mcp_servers
-        query.variables['_pipeline_mcp_resource_attachments'] = self.mcp_resource_attachments
-        query.variables['_pipeline_mcp_resource_agent_read_enabled'] = self.mcp_resource_agent_read_enabled
 
         # Record query start for monitoring
         try:
@@ -295,9 +284,9 @@ class RuntimePipeline:
         # Record query start and store message_id
         message_id = ''
         try:
-            from . import monitor
+            from . import monitoring_helper
 
-            message_id = await monitor.MonitoringHelper.record_query_start(
+            message_id = await monitoring_helper.MonitoringHelper.record_query_start(
                 ap=self.ap,
                 query=query,
                 bot_id=query.bot_uuid or 'unknown',
@@ -349,7 +338,7 @@ class RuntimePipeline:
             # Record query success only if no error occurred during processing
             if not query.variables.get('_monitoring_has_error', False):
                 try:
-                    await monitor.MonitoringHelper.record_query_success(
+                    await monitoring_helper.MonitoringHelper.record_query_success(
                         ap=self.ap,
                         message_id=message_id,
                         query=query,
@@ -359,7 +348,7 @@ class RuntimePipeline:
 
                 # Record bot response message
                 try:
-                    await monitor.MonitoringHelper.record_query_response(
+                    await monitoring_helper.MonitoringHelper.record_query_response(
                         ap=self.ap,
                         query=query,
                         bot_id=query.bot_uuid or 'unknown',
@@ -378,9 +367,9 @@ class RuntimePipeline:
 
             # Record query error
             try:
-                from . import monitor
+                from . import monitoring_helper
 
-                await monitor.MonitoringHelper.record_query_error(
+                await monitoring_helper.MonitoringHelper.record_query_error(
                     ap=self.ap,
                     query=query,
                     bot_id=query.bot_uuid or 'unknown',
@@ -395,8 +384,7 @@ class RuntimePipeline:
 
         finally:
             self.ap.logger.debug(f'Query {query.query_id} processed')
-            # Use pop with default to avoid KeyError if query was never cached
+            del self.ap.query_pool.cached_queries[query.query_id]
-            self.ap.query_pool.cached_queries.pop(query.query_id, None)
 
 
 class PipelineManager:

src/langbot/pkg/pipeline/preproc/preproc.py

@@ -25,21 +25,6 @@ class PreProcessor(stage.PipelineStage):
         - use_funcs
     """
 
-    @staticmethod
-    def _filter_selected_tools(
-        tools: list,
-        local_agent_config: dict,
-    ) -> list:
-        if local_agent_config.get('enable-all-tools', True) is not False:
-            return tools
-
-        selected_tools = local_agent_config.get('tools', [])
-        if not isinstance(selected_tools, list):
-            return []
-
-        selected_tool_names = {tool for tool in selected_tools if isinstance(tool, str)}
-        return [tool for tool in tools if tool.name in selected_tool_names]
-
     async def process(
         self,
         query: pipeline_query.Query,
@@ -47,7 +32,6 @@ class PreProcessor(stage.PipelineStage):
     ) -> entities.StageProcessResult:
         """Process"""
         selected_runner = query.pipeline_config['ai']['runner']['runner']
-        local_agent_config = query.pipeline_config.get('ai', {}).get('local-agent', {})
         include_skill_authoring = (
             selected_runner == 'local-agent' and getattr(self.ap, 'skill_service', None) is not None
         )
@@ -59,7 +43,7 @@ class PreProcessor(stage.PipelineStage):
         if selected_runner == 'local-agent':
             # Read model config — new format is { primary: str, fallbacks: [str] },
             # but handle legacy plain string for backward compatibility
-            model_config = local_agent_config.get('model', {})
+            model_config = query.pipeline_config['ai']['local-agent'].get('model', {})
             if isinstance(model_config, str):
                 # Legacy format: plain UUID string
                 primary_uuid = model_config
@@ -129,14 +113,11 @@ class PreProcessor(stage.PipelineStage):
                     # Get bound plugins and MCP servers for filtering tools
                     bound_plugins = query.variables.get('_pipeline_bound_plugins', None)
                     bound_mcp_servers = query.variables.get('_pipeline_bound_mcp_servers', None)
-                    include_mcp_resource_tools = query.variables.get('_pipeline_mcp_resource_agent_read_enabled', True)
+                    query.use_funcs = await self.ap.tool_mgr.get_all_tools(
-                    all_tools = await self.ap.tool_mgr.get_all_tools(
                         bound_plugins,
                         bound_mcp_servers,
                         include_skill_authoring=include_skill_authoring,
-                        include_mcp_resource_tools=include_mcp_resource_tools,
                     )
-                    query.use_funcs = self._filter_selected_tools(all_tools, local_agent_config)
 
                     self.ap.logger.debug(f'Bound plugins: {bound_plugins}')
                     self.ap.logger.debug(f'Bound MCP servers: {bound_mcp_servers}')
@@ -147,14 +128,11 @@ class PreProcessor(stage.PipelineStage):
             if not query.use_funcs and query.variables.get('_fallback_model_uuids'):
                 bound_plugins = query.variables.get('_pipeline_bound_plugins', None)
                 bound_mcp_servers = query.variables.get('_pipeline_bound_mcp_servers', None)
-                include_mcp_resource_tools = query.variables.get('_pipeline_mcp_resource_agent_read_enabled', True)
+                query.use_funcs = await self.ap.tool_mgr.get_all_tools(
-                all_tools = await self.ap.tool_mgr.get_all_tools(
                     bound_plugins,
                     bound_mcp_servers,
                     include_skill_authoring=include_skill_authoring,
-                    include_mcp_resource_tools=include_mcp_resource_tools,
                 )
-                query.use_funcs = self._filter_selected_tools(all_tools, local_agent_config)
 
         sender_name = ''
 

src/langbot/pkg/platform/botmgr.py

@@ -2,14 +2,12 @@ from __future__ import annotations
 
 import asyncio
 import json
-import logging
+import re
 import traceback
 import sqlalchemy
 
 from ..core import app, entities as core_entities, taskmgr
 
-logger = logging.getLogger(__name__)
-
 from ..discover import engine
 
 from ..entity.persistence import bot as persistence_bot
@@ -56,24 +54,29 @@ class RuntimeBot:
         self.task_context = taskmgr.TaskContext()
         self.logger = logger
 
+    @staticmethod
+    def _match_operator(actual: str, operator: str, expected: str) -> bool:
+        """Evaluate a single operator condition."""
+        if operator == 'eq':
+            return actual == expected
+        elif operator == 'neq':
+            return actual != expected
+        elif operator == 'contains':
+            return expected in actual
+        elif operator == 'not_contains':
+            return expected not in actual
+        elif operator == 'starts_with':
+            return actual.startswith(expected)
+        elif operator == 'regex':
+            try:
+                return bool(re.search(expected, actual))
+            except re.error:
+                return False
+        return False
+
     PIPELINE_DISCARD = '__discard__'
     PIPELINE_DISCARD_DISPLAY_NAME = 'Discarded'
 
-    def get_binding_info(self) -> tuple[str, str | None]:
-        """Get the binding type and UUID for this bot.
-
-        Returns:
-            tuple: (binding_type, binding_uuid) where binding_type is 'pipeline' or 'workflow'
-        """
-        binding_type = getattr(self.bot_entity, 'binding_type', 'pipeline') or 'pipeline'
-        binding_uuid = getattr(self.bot_entity, 'binding_uuid', None)
-
-        # Fallback to use_pipeline_uuid for backward compatibility
-        if not binding_uuid and binding_type == 'pipeline':
-            binding_uuid = self.bot_entity.use_pipeline_uuid
-
-        return binding_type, binding_uuid
-
     def resolve_pipeline_uuid(
         self,
         launcher_type: str,
@@ -81,94 +84,56 @@ class RuntimeBot:
         message_text: str,
         message_element_types: list[str] | None = None,
     ) -> tuple[str | None, bool]:
-        """Resolve pipeline UUID for message processing.
+        """Resolve pipeline UUID based on routing rules.
 
-        NOTE: Routing rules have been removed. Bot now directly binds to a
+        Rules are evaluated in order; first match wins.
-        Pipeline or Workflow. This method is kept for backward compatibility
+        Falls back to use_pipeline_uuid if no rule matches.
-        but only returns the direct binding.
+
+        Rule types:
+          - launcher_type: session type ("person" / "group")
+          - launcher_id: session / group id
+          - message_content: message text content
+          - message_has_element: message contains element of given type
+            (Image, Voice, File, Forward, Face, At, AtAll, Quote)
+            Operators: eq (has), neq (doesn't have)
+
+        Operators: eq, neq, contains, not_contains, starts_with, regex
+
+        When pipeline_uuid is ``__discard__``, the message should be
+        silently dropped by the caller.
 
         Returns:
-            tuple: (pipeline_uuid, routed_by_rule) - routed_by_rule is always False
+            tuple: (pipeline_uuid, routed_by_rule) - routed_by_rule is True
-            as routing rules are no longer used.
+            when a routing rule matched, False when falling back to default.
         """
-        binding_type, binding_uuid = self.get_binding_info()
+        rules = self.bot_entity.pipeline_routing_rules or []
+        element_type_set = set(message_element_types or [])
 
-        # If bound to workflow, return None for pipeline_uuid
+        for rule in rules:
-        # The caller should check binding_type and handle accordingly
+            rule_type = rule.get('type')
-        if binding_type == 'workflow':
+            operator = rule.get('operator', 'eq')
-            # For workflow binding, we still need to return something
+            rule_value = rule.get('value', '')
-            # The actual workflow handling should be done by the caller
+            target_uuid = rule.get('pipeline_uuid')
-            return None, False
+            if not rule_type or not target_uuid:
+                continue
 
-        return binding_uuid, False
+            if rule_type == 'launcher_type':
+                if self._match_operator(launcher_type, operator, rule_value):
+                    return target_uuid, True
+            elif rule_type == 'launcher_id':
+                if self._match_operator(str(launcher_id), operator, str(rule_value)):
+                    return target_uuid, True
+            elif rule_type == 'message_content':
+                if self._match_operator(message_text, operator, rule_value):
+                    return target_uuid, True
+            elif rule_type == 'message_has_element':
+                has_element = rule_value in element_type_set
+                if operator == 'eq' and has_element:
+                    return target_uuid, True
+                elif operator == 'neq' and not has_element:
+                    return target_uuid, True
 
-    async def _handle_workflow_message(
+        return self.bot_entity.use_pipeline_uuid, False
-        self,
-        event: platform_events.MessageEvent,
-        adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
-        workflow_uuid: str,
-        launcher_type: str,
-        launcher_id: str | int,
-        sender_id: str | int,
-    ) -> None:
-        """Handle message by executing the bound workflow directly."""
-        message_content = str(event.message_chain)
-        message_chain_obj = event.message_chain
-
-        # Build message context
-        sender_name = None
-        if hasattr(event, 'sender'):
-            sender = event.sender
-            if hasattr(sender, 'nickname'):
-                sender_name = sender.nickname
-            elif hasattr(sender, 'member_name'):
-                sender_name = sender.member_name
-
-        is_group = launcher_type == 'group'
-        message_context = {
-            'message_id': str(getattr(event, 'message_id', '')),
-            'message_content': message_content,
-            'sender_id': str(sender_id),
-            'sender_name': sender_name or 'User',
-            'platform': adapter.__class__.__name__,
-            'conversation_id': str(launcher_id),
-            'is_group': is_group,
-            'group_id': str(launcher_id) if is_group else None,
-            'mentions': [],
-            'reply_to': None,
-            'raw_message': {
-                'message': message_chain_obj.model_dump() if hasattr(message_chain_obj, 'model_dump') else str(message_chain_obj),
-                'launcher_id': launcher_id,
-                'session_type': launcher_type,
-            },
-        }
-
-        trigger_data = {
-            'message': message_content,
-            'message_chain': message_chain_obj.model_dump() if hasattr(message_chain_obj, 'model_dump') else str(message_chain_obj),
-            'session_type': launcher_type,
-            'connection_id': str(launcher_id),
-            'message_context': message_context,
-        }
-
-        session_id = f'{launcher_type}_{launcher_id}'
-        logger.info(f'Processing workflow message from {session_id}: {message_content}')
-
-        try:
-            from ..api.http.service.workflow import WorkflowExecutionFailedError
-
-            execution_id = await self.ap.workflow_service.execute_workflow(
-                workflow_uuid=workflow_uuid,
-                trigger_type='message',
-                trigger_data=trigger_data,
-                session_id=session_id,
-                user_id=str(sender_id),
-                bot_id=self.bot_entity.uuid,
-            )
-        except WorkflowExecutionFailedError as e:
-            await self.logger.error(f'Workflow execution failed: {e.message}')
-        except Exception as e:
-            await self.logger.error(f'Workflow execution error: {e}')
 
     async def _record_discarded_message(
         self,
@@ -264,20 +229,6 @@ class RuntimeBot:
 
                 message_text = str(event.message_chain)
                 element_types = [comp.type for comp in event.message_chain]
-                binding_type, binding_uuid = self.get_binding_info()
-
-                # Handle workflow binding separately from pipeline
-                if binding_type == 'workflow':
-                    await self._handle_workflow_message(
-                        event=event,
-                        adapter=adapter,
-                        workflow_uuid=binding_uuid,
-                        launcher_type='person',
-                        launcher_id=launcher_id,
-                        sender_id=event.sender.id,
-                    )
-                    return
-
                 pipeline_uuid, routed_by_rule = self.resolve_pipeline_uuid(
                     'person', launcher_id, message_text, element_types
                 )
@@ -339,20 +290,6 @@ class RuntimeBot:
 
                 message_text = str(event.message_chain)
                 element_types = [comp.type for comp in event.message_chain]
-                binding_type, binding_uuid = self.get_binding_info()
-
-                # Handle workflow binding separately from pipeline
-                if binding_type == 'workflow':
-                    await self._handle_workflow_message(
-                        event=event,
-                        adapter=adapter,
-                        workflow_uuid=binding_uuid,
-                        launcher_type='group',
-                        launcher_id=launcher_id,
-                        sender_id=event.sender.id,
-                    )
-                    return
-
                 pipeline_uuid, routed_by_rule = self.resolve_pipeline_uuid(
                     'group', launcher_id, message_text, element_types
                 )

src/langbot/pkg/platform/sources/aiocqhttp.py

@@ -4,7 +4,6 @@ import asyncio
 import traceback
 import datetime
 import json
-import time
 
 import aiocqhttp
 import pydantic
@@ -17,37 +16,6 @@ from ...utils import image
 import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
 
 
-_GROUP_NAME_CACHE_TTL_SECONDS = 3600
-_GROUP_NAME_NEGATIVE_CACHE_TTL_SECONDS = 60
-_GROUP_NAME_LOOKUP_TIMEOUT_SECONDS = 2
-_GROUP_MEMBER_INFO_CACHE_TTL_SECONDS = 86400
-_GROUP_MEMBER_INFO_NEGATIVE_CACHE_TTL_SECONDS = 600
-_GROUP_MEMBER_INFO_LOOKUP_TIMEOUT_SECONDS = 2
-
-
-def _normalize_base64_payload(value: str) -> str:
-    if value.startswith('base64://'):
-        return value.removeprefix('base64://')
-    if value.startswith('data:') and ';base64,' in value:
-        return value.split(';base64,', 1)[1]
-    return value
-
-
-def _get_field(data: dict, key: str, default: str = '') -> str:
-    value = data.get(key)
-    if value is None:
-        return default
-    return str(value)
-
-
-def _get_group_member_name(sender: dict) -> str:
-    return _get_field(sender, 'card') or _get_field(sender, 'nickname') or _get_field(sender, 'user_id')
-
-
-def _get_group_name_placeholder(group_id: typing.Union[int, str]) -> str:
-    return f'Group {group_id}'
-
-
 class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
     @staticmethod
     async def yiri2target(
@@ -67,7 +35,7 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
             elif type(msg) is platform_message.Image:
                 arg = ''
                 if msg.base64:
-                    arg = _normalize_base64_payload(msg.base64)
+                    arg = msg.base64
                     msg_list.append(aiocqhttp.MessageSegment.image(f'base64://{arg}'))
                 elif msg.url:
                     arg = msg.url
@@ -82,7 +50,7 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
             elif type(msg) is platform_message.Voice:
                 arg = ''
                 if msg.base64:
-                    arg = _normalize_base64_payload(msg.base64)
+                    arg = msg.base64
                     msg_list.append(aiocqhttp.MessageSegment.record(f'base64://{arg}'))
                 elif msg.url:
                     arg = msg.url
@@ -94,10 +62,7 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
                 for node in msg.node_list:
                     msg_list.extend((await AiocqhttpMessageConverter.yiri2target(node.message_chain))[0])
             elif isinstance(msg, platform_message.File):
-                file = msg.url or msg.path
+                msg_list.append({'type': 'file', 'data': {'file': msg.url, 'name': msg.name}})
-                if not file and msg.base64:
-                    file = f'base64://{_normalize_base64_payload(msg.base64)}'
-                msg_list.append({'type': 'file', 'data': {'file': file, 'name': msg.name}})
             elif isinstance(msg, platform_message.Face):
                 if msg.face_type == 'face':
                     msg_list.append(aiocqhttp.MessageSegment.face(msg.face_id))
@@ -359,96 +324,16 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
 
 
 class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
-    def __init__(self):
-        self._group_name_cache: dict[typing.Union[int, str], tuple[str, float]] = {}
-        self._group_name_negative_cache: dict[typing.Union[int, str], float] = {}
-        self._group_member_info_cache: dict[
-            tuple[typing.Union[int, str], typing.Union[int, str]], tuple[dict, float]
-        ] = {}
-        self._group_member_info_negative_cache: dict[tuple[typing.Union[int, str], typing.Union[int, str]], float] = {}
-
     @staticmethod
     async def yiri2target(event: platform_events.MessageEvent, bot_account_id: int):
         return event.source_platform_object
 
-    async def _get_group_name(self, group_id: typing.Union[int, str], bot=None) -> str:
+    @staticmethod
-        now = time.monotonic()
+    async def target2yiri(event: aiocqhttp.Event, bot=None):
-        if group_id in self._group_name_cache:
-            group_name, expires_at = self._group_name_cache[group_id]
-            if expires_at > now:
-                return group_name
-            del self._group_name_cache[group_id]
-        if group_id in self._group_name_negative_cache:
-            expires_at = self._group_name_negative_cache[group_id]
-            if expires_at > now:
-                return ''
-            del self._group_name_negative_cache[group_id]
-        if bot is None:
-            return ''
-        try:
-            group_info = await asyncio.wait_for(
-                bot.get_group_info(group_id=group_id),
-                timeout=_GROUP_NAME_LOOKUP_TIMEOUT_SECONDS,
-            )
-        except Exception:
-            self._group_name_negative_cache[group_id] = now + _GROUP_NAME_NEGATIVE_CACHE_TTL_SECONDS
-            return ''
-        group_name = _get_field(group_info, 'group_name') if isinstance(group_info, dict) else ''
-        if group_name:
-            self._group_name_cache[group_id] = (group_name, now + _GROUP_NAME_CACHE_TTL_SECONDS)
-            self._group_name_negative_cache.pop(group_id, None)
-        else:
-            self._group_name_negative_cache[group_id] = now + _GROUP_NAME_NEGATIVE_CACHE_TTL_SECONDS
-        return group_name
-
-    async def _get_group_member_info(
-        self,
-        group_id: typing.Union[int, str],
-        user_id: typing.Union[int, str],
-        bot=None,
-    ) -> dict:
-        now = time.monotonic()
-        cache_key = (group_id, user_id)
-        if cache_key in self._group_member_info_cache:
-            member_info, expires_at = self._group_member_info_cache[cache_key]
-            if expires_at > now:
-                return member_info
-            del self._group_member_info_cache[cache_key]
-        if cache_key in self._group_member_info_negative_cache:
-            expires_at = self._group_member_info_negative_cache[cache_key]
-            if expires_at > now:
-                return {}
-            del self._group_member_info_negative_cache[cache_key]
-        if bot is None:
-            return {}
-        try:
-            member_info = await asyncio.wait_for(
-                bot.get_group_member_info(group_id=group_id, user_id=user_id),
-                timeout=_GROUP_MEMBER_INFO_LOOKUP_TIMEOUT_SECONDS,
-            )
-        except Exception:
-            self._group_member_info_negative_cache[cache_key] = now + _GROUP_MEMBER_INFO_NEGATIVE_CACHE_TTL_SECONDS
-            return {}
-        if isinstance(member_info, dict) and member_info:
-            self._group_member_info_cache[cache_key] = (
-                member_info,
-                now + _GROUP_MEMBER_INFO_CACHE_TTL_SECONDS,
-            )
-            self._group_member_info_negative_cache.pop(cache_key, None)
-            return member_info
-        self._group_member_info_negative_cache[cache_key] = now + _GROUP_MEMBER_INFO_NEGATIVE_CACHE_TTL_SECONDS
-        return {}
-
-    async def target2yiri(self, event: aiocqhttp.Event, bot=None):
         yiri_chain = await AiocqhttpMessageConverter.target2yiri(event.message, event.message_id, bot)
 
         if event.message_type == 'group':
             permission = 'MEMBER'
-            group_name = await self._get_group_name(event.group_id, bot) or _get_group_name_placeholder(event.group_id)
-            special_title = _get_field(event.sender, 'title')
-            if not special_title:
-                member_info = await self._get_group_member_info(event.group_id, event.sender['user_id'], bot)
-                special_title = _get_field(member_info, 'title')
 
             if 'role' in event.sender:
                 if event.sender['role'] == 'admin':
@@ -458,14 +343,14 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
             converted_event = platform_events.GroupMessage(
                 sender=platform_entities.GroupMember(
                     id=event.sender['user_id'],  # message_seq 放哪？
-                    member_name=_get_group_member_name(event.sender),
+                    member_name=event.sender['nickname'],
                     permission=permission,
                     group=platform_entities.Group(
                         id=event.group_id,
-                        name=group_name,
+                        name=event.sender['nickname'],
                         permission=platform_entities.Permission.Member,
                     ),
-                    special_title=special_title,
+                    special_title=event.sender['title'] if 'title' in event.sender else '',
                 ),
                 message_chain=yiri_chain,
                 time=event.time,
@@ -489,7 +374,7 @@ class AiocqhttpAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
     bot: aiocqhttp.CQHttp = pydantic.Field(exclude=True, default_factory=aiocqhttp.CQHttp)
 
     message_converter: AiocqhttpMessageConverter = AiocqhttpMessageConverter()
-    event_converter: AiocqhttpEventConverter = pydantic.Field(default_factory=AiocqhttpEventConverter)
+    event_converter: AiocqhttpEventConverter = AiocqhttpEventConverter()
 
     on_websocket_connection_event_cache: typing.List[typing.Callable[[aiocqhttp.Event], None]] = []
 
@@ -548,7 +433,9 @@ class AiocqhttpAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
                     elif isinstance(component, platform_message.Image):
                         img_data = {}
                         if component.base64:
-                            b64 = _normalize_base64_payload(component.base64)
+                            b64 = component.base64
+                            if b64.startswith('data:'):
+                                b64 = b64.split(',', 1)[-1] if ',' in b64 else b64
                             img_data['file'] = f'base64://{b64}'
                         elif component.url:
                             img_data['file'] = component.url

src/langbot/pkg/platform/sources/websocket_adapter.py

@@ -422,64 +422,6 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
             session_type=session_type,
         )
 
-        # Determine if pipeline_uuid is a workflow or a legacy pipeline by querying both services
-        workflow_dict = await self.ap.workflow_service.get_workflow(pipeline_uuid)
-        pipeline_dict = await self.ap.pipeline_service.get_pipeline(pipeline_uuid)
-
-        if workflow_dict is not None:
-            # UUID exists in workflow table - execute as workflow
-            # Set pipeline_uuid for workflow nodes to broadcast messages correctly
-            self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid = pipeline_uuid
-
-            message_content = str(message_chain)
-            message_context = {
-                'message_id': str(message_id),
-                'message_content': message_content,
-                'sender_id': f'websocket_{connection.connection_id}',
-                'sender_name': 'User',
-                'platform': 'websocket',
-                'conversation_id': connection.connection_id,
-                'is_group': session_type == 'group',
-                'group_id': 'websocketgroup' if session_type == 'group' else None,
-                'mentions': [],
-                'reply_to': None,
-                'raw_message': {
-                    'message': message_chain_obj,
-                    'connection_id': connection.connection_id,
-                    'session_type': session_type,
-                },
-            }
-
-            trigger_data = {
-                'message': message_content,
-                'message_chain': message_chain_obj,
-                'session_type': session_type,
-                'connection_id': connection.connection_id,
-                'message_context': message_context,
-            }
-
-            try:
-                from ...api.http.service.workflow import WorkflowExecutionFailedError
-
-                # Log workflow execution start (matching pipeline logging)
-                session_id = f'{session_type}_{connection.connection_id}'
-                logger.info(f'Processing workflow message from {session_id}: {message_content}')
-
-                execution_id = await self.ap.workflow_service.execute_workflow(
-                    pipeline_uuid,  # This is actually a workflow UUID
-                    trigger_type='message',
-                    trigger_data=trigger_data,
-                    session_id=session_id,
-                    user_id=message_context['sender_id'],
-                    bot_id=self.ap.platform_mgr.websocket_proxy_bot.bot_entity.uuid,
-                )
-            except WorkflowExecutionFailedError as e:
-                await connection.send_queue.put({'type': 'error', 'message': e.message})
-            except Exception as e:
-                logger.error(f'Workflow websocket execution error: {e}', exc_info=True)
-                await connection.send_queue.put({'type': 'error', 'message': str(e)})
-            return
-
         # 添加消息源
         message_chain.insert(0, platform_message.Source(id=message_id, time=datetime.now().timestamp()))
 

src/langbot/pkg/plugin/handler.py

@@ -363,19 +363,9 @@ class RuntimeConnectionHandler(handler.Handler):
                 extra_args=extra_args,
             )
 
-            # invoke_llm returns (message, usage_info) tuple
-            if isinstance(result, tuple) and len(result) == 2:
-                msg, usage_info = result
-                msg_dump = msg.model_dump()
-                # Attach usage info to message dump
-                if usage_info:
-                    msg_dump['usage'] = usage_info
-            else:
-                msg_dump = result.model_dump()
-
             return handler.ActionResponse.success(
                 data={
-                    'message': msg_dump,
+                    'message': result.model_dump(),
                 },
             )
 

src/langbot/pkg/provider/modelmgr/requester.py

@@ -81,13 +81,8 @@ class RuntimeProvider:
                 msg, usage_info = result
                 if usage_info:
                     _store_llm_usage(query, usage_info)
-                    input_tokens = usage_info.get('prompt_tokens', usage_info.get('input_tokens', 0))
+                    input_tokens = usage_info.get('prompt_tokens', 0)
-                    output_tokens = usage_info.get('completion_tokens', usage_info.get('output_tokens', 0))
+                    output_tokens = usage_info.get('completion_tokens', 0)
-                    # Attach usage info to message using object.__setattr__ to bypass pydantic validation
-                    try:
-                        object.__setattr__(msg, 'usage', usage_info)
-                    except (AttributeError, TypeError):
-                        pass  # If we can't set it, just skip it
                 return msg
             else:
                 return result
@@ -103,7 +98,7 @@ class RuntimeProvider:
 
                 # Import monitoring helper
                 try:
-                    from ...pipeline import monitor
+                    from ...pipeline import monitoring_helper
 
                     # Get monitoring metadata from query variables
                     if query.variables:
@@ -115,7 +110,7 @@ class RuntimeProvider:
                         pipeline_name = 'Unknown'
                         message_id = None
 
-                    await monitor.MonitoringHelper.record_llm_call(
+                    await monitoring_helper.MonitoringHelper.record_llm_call(
                         ap=self.requester.ap,
                         query=query,
                         bot_id=query.bot_uuid or 'unknown',
@@ -182,7 +177,7 @@ class RuntimeProvider:
 
                 # Import monitoring helper
                 try:
-                    from ...pipeline import monitor
+                    from ...pipeline import monitoring_helper
 
                     # Get monitoring metadata from query variables
                     if query.variables:
@@ -194,7 +189,7 @@ class RuntimeProvider:
                         pipeline_name = 'Unknown'
                         message_id = None
 
-                    await monitor.MonitoringHelper.record_llm_call(
+                    await monitoring_helper.MonitoringHelper.record_llm_call(
                         ap=self.requester.ap,
                         query=query,
                         bot_id=query.bot_uuid or 'unknown',

src/langbot/pkg/provider/runners/localagent.py

@@ -417,30 +417,6 @@ class LocalAgentRunner(runner.RequestRunner):
                     ce.text = final_user_message_text
                     break
 
-        mcp_loader = getattr(getattr(self.ap, 'tool_mgr', None), 'mcp_tool_loader', None)
-        if mcp_loader is not None:
-            resource_context = await mcp_loader.build_resource_context_for_query(query)
-            if resource_context:
-                resource_addition = (
-                    '\n\nMCP resource context selected by LangBot host:\n'
-                    f'{resource_context}\n\n'
-                    'Use this context as read-only reference material. If it conflicts with the user message, '
-                    'ask for clarification before taking external actions.'
-                )
-                if isinstance(user_message.content, str):
-                    user_message.content += resource_addition
-                elif isinstance(user_message.content, list):
-                    appended = False
-                    for ce in user_message.content:
-                        if ce.type == 'text':
-                            ce.text = (ce.text or '') + resource_addition
-                            appended = True
-                            break
-                    if not appended:
-                        user_message.content.append(
-                            provider_message.ContentElement.from_text(resource_addition.strip())
-                        )
-
         req_messages = self._build_request_messages(query, user_message)
 
         try:

src/langbot/pkg/provider/tools/loaders/plugin.py

@@ -33,24 +33,6 @@ class PluginToolLoader(loader.ToolLoader):
 
         return all_functions
 
-    async def get_tool_catalog(self, bound_plugins: list[str] | None = None) -> list[dict[str, typing.Any]]:
-        catalog: list[dict[str, typing.Any]] = []
-
-        for tool in await self.ap.plugin_connector.list_tools(bound_plugins):
-            catalog.append(
-                {
-                    'name': tool.metadata.name,
-                    'description': tool.spec['llm_prompt'],
-                    'human_desc': tool.metadata.description.en_US,
-                    'parameters': tool.spec['parameters'],
-                    'source': 'plugin',
-                    'source_name': tool.owner,
-                    'source_id': tool.owner,
-                }
-            )
-
-        return catalog
-
     async def has_tool(self, name: str) -> bool:
         """检查工具是否存在"""
         for tool in await self.ap.plugin_connector.list_tools():

src/langbot/pkg/provider/tools/toolmgr.py

@@ -59,7 +59,6 @@ class ToolManager:
         bound_plugins: list[str] | None = None,
         bound_mcp_servers: list[str] | None = None,
         include_skill_authoring: bool = False,
-        include_mcp_resource_tools: bool = True,
     ) -> list[resource_tool.LLMTool]:
         all_functions: list[resource_tool.LLMTool] = []
 
@@ -67,51 +66,10 @@ class ToolManager:
         if include_skill_authoring:
             all_functions.extend(await self.skill_tool_loader.get_tools())
         all_functions.extend(await self.plugin_tool_loader.get_tools(bound_plugins))
-        all_functions.extend(
+        all_functions.extend(await self.mcp_tool_loader.get_tools(bound_mcp_servers))
-            await self.mcp_tool_loader.get_tools(
-                bound_mcp_servers,
-                include_resource_tools=include_mcp_resource_tools,
-            )
-        )
 
         return all_functions
 
-    async def get_tool_catalog(
-        self,
-        bound_plugins: list[str] | None = None,
-        bound_mcp_servers: list[str] | None = None,
-        include_skill_authoring: bool = False,
-        include_mcp_resource_tools: bool = False,
-    ) -> list[dict[str, typing.Any]]:
-        catalog: list[dict[str, typing.Any]] = []
-
-        def append_tools(source: str, source_name: str, tools: list[resource_tool.LLMTool]) -> None:
-            for tool in tools:
-                catalog.append(
-                    {
-                        'name': tool.name,
-                        'description': tool.description,
-                        'human_desc': tool.human_desc,
-                        'parameters': tool.parameters,
-                        'source': source,
-                        'source_name': source_name,
-                    }
-                )
-
-        append_tools('builtin', 'LangBot', await self.native_tool_loader.get_tools())
-        if include_skill_authoring:
-            append_tools('skill', 'LangBot', await self.skill_tool_loader.get_tools())
-        catalog.extend(await self.plugin_tool_loader.get_tool_catalog(bound_plugins))
-
-        if self.mcp_tool_loader:
-            for item in await self.mcp_tool_loader.get_tool_catalog(
-                bound_mcp_servers,
-                include_resource_tools=include_mcp_resource_tools,
-            ):
-                catalog.append(item)
-
-        return catalog
-
     async def get_tool_by_name(self, name: str) -> tool_loader.ToolLookupResult | None:
         """Get tool by name from any active loader."""
         for active_loader in (

src/langbot/pkg/workflow/adapters.py

@@ -1,204 +0,0 @@
-"""Workflow-Pipeline通信适配器
-
-这个模块提供了Workflow和Pipeline之间的通信适配，使用SDK标准的MessageEnvelope格式。
-"""
-
-from __future__ import annotations
-
-import logging
-from typing import Any, Optional
-
-logger = logging.getLogger(__name__)
-
-
-class _WorkflowPipelineCaptureAdapter:
-    """Workflow-Pipeline通信适配器
-    
-    用于在Workflow节点和Pipeline之间进行标准化的消息传递。
-    支持MessageEnvelope格式的双向转换。
-    """
-    
-    def __init__(self, context: Any):
-        """初始化适配器
-        
-        Args:
-            context: ExecutionContext - Workflow执行上下文
-        """
-        self.context = context
-        self.responses: list[dict[str, Any]] = []
-        self.bot_account_id: Optional[str] = None
-        self._logger = logging.getLogger(__name__)
-    
-    async def call_pipeline_with_envelope(
-        self,
-        envelope: Any,
-        pipeline_executor: Any
-    ) -> Any:
-        """使用MessageEnvelope调用Pipeline
-        
-        Args:
-            envelope: MessageEnvelope - 标准消息信封
-            pipeline_executor: Pipeline执行器实例
-            
-        Returns:
-            MessageEnvelope - 执行结果信封
-        """
-        try:
-            # 动态导入以避免循环依赖
-            from langbot_plugin_sdk.workflow import envelope_to_query, query_to_envelope
-            
-            # 1. 转换为Query
-            query = envelope_to_query(envelope)
-            
-            # 2. 调用Pipeline
-            result_query = await pipeline_executor.execute(query)
-            
-            # 3. 转换回Envelope
-            result_envelope = query_to_envelope(result_query, envelope)
-            
-            self._logger.debug(
-                f'Pipeline execution completed for workflow {envelope.workflow_id}',
-                extra={
-                    'workflow_id': envelope.workflow_id,
-                    'execution_id': envelope.execution_id,
-                    'node_id': envelope.node_id,
-                }
-            )
-            
-            return result_envelope
-            
-        except Exception as e:
-            self._logger.error(
-                f'Pipeline execution failed: {e}',
-                exc_info=True,
-                extra={
-                    'workflow_id': envelope.workflow_id,
-                    'execution_id': envelope.execution_id,
-                    'node_id': envelope.node_id,
-                }
-            )
-            raise
-    
-    def validate_envelope(self, envelope: Any) -> bool:
-        """验证MessageEnvelope的有效性
-        
-        Args:
-            envelope: MessageEnvelope - 要验证的消息信封
-            
-        Returns:
-            bool - 验证是否通过
-        """
-        required_fields = [
-            'message_id',
-            'workflow_id',
-            'node_id',
-            'execution_id',
-            'payload',
-            'launcher_type',
-        ]
-        
-        for field in required_fields:
-            if not hasattr(envelope, field):
-                self._logger.warning(
-                    f'MessageEnvelope missing required field: {field}'
-                )
-                return False
-        
-        return True
-    
-    def get_responses(self) -> list[dict[str, Any]]:
-        """获取所有响应
-        
-        Returns:
-            list - 响应列表
-        """
-        return self.responses.copy()
-    
-    def add_response(self, response: dict[str, Any]) -> None:
-        """添加响应
-        
-        Args:
-            response: dict - 响应数据
-        """
-        self.responses.append(response)
-    
-    def get_last_text_response(self) -> str:
-        """获取最后一个文本响应
-        
-        Returns:
-            str - 最后一个响应的文本内容
-        """
-        if not self.responses:
-            return ''
-        
-        last_response = self.responses[-1]
-        return str(last_response.get('content', '') or '')
-    
-    def clear_responses(self) -> None:
-        """清空所有响应"""
-        self.responses.clear()
-
-
-class WorkflowPipelineCompatibilityLayer:
-    """Workflow-Pipeline兼容性层
-    
-    提供向后兼容性，支持旧的Pipeline Query格式和新的MessageEnvelope格式。
-    """
-    
-    def __init__(self):
-        """初始化兼容性层"""
-        self._logger = logging.getLogger(__name__)
-    
-    def is_workflow_context(self, query: Any) -> bool:
-        """检查Query是否包含Workflow上下文
-        
-        Args:
-            query: Query - Pipeline Query对象
-            
-        Returns:
-            bool - 是否来自Workflow
-        """
-        if hasattr(query, 'is_from_workflow'):
-            return query.is_from_workflow()
-        
-        if hasattr(query, 'get_workflow_context'):
-            context = query.get_workflow_context()
-            return bool(context and context.get('workflow_id'))
-        
-        return False
-    
-    def get_workflow_id(self, query: Any) -> Optional[str]:
-        """从Query获取Workflow ID
-        
-        Args:
-            query: Query - Pipeline Query对象
-            
-        Returns:
-            str - Workflow ID，如果不存在则返回None
-        """
-        if hasattr(query, 'get_workflow_id'):
-            return query.get_workflow_id()
-        
-        if hasattr(query, 'get_workflow_context'):
-            context = query.get_workflow_context()
-            return context.get('workflow_id') if context else None
-        
-        return None
-    
-    def get_execution_id(self, query: Any) -> Optional[str]:
-        """从Query获取执行ID
-        
-        Args:
-            query: Query - Pipeline Query对象
-            
-        Returns:
-            str - 执行ID，如果不存在则返回None
-        """
-        if hasattr(query, 'get_execution_id'):
-            return query.get_execution_id()
-        
-        if hasattr(query, 'get_workflow_context'):
-            context = query.get_workflow_context()
-            return context.get('execution_id') if context else None
-        
-        return None

src/langbot/pkg/workflow/debug.py

@@ -1,504 +0,0 @@
-"""Workflow debug execution support.
-
-This module provides debugging capabilities for workflow execution, including:
-- ExecutionLog: Structured log entries for execution tracking
-- DebugExecutionState: State management for debug sessions (pause, resume, breakpoints)
-- DebugWorkflowExecutor: Extended executor with step-by-step debugging support
-"""
-
-from __future__ import annotations
-
-import asyncio
-import logging
-import traceback
-import uuid
-from datetime import datetime
-from typing import Any, Optional, TYPE_CHECKING
-
-from .entities import (
-    WorkflowDefinition,
-    NodeDefinition,
-    EdgeDefinition,
-    ExecutionContext,
-    ExecutionStatus,
-    NodeState,
-    NodeStatus,
-)
-from .executor import WorkflowExecutor
-
-if TYPE_CHECKING:
-    from ..core import app
-
-logger = logging.getLogger(__name__)
-
-
-class ExecutionLog:
-    """Execution log entry"""
-
-    def __init__(self, level: str, message: str, node_id: Optional[str] = None, data: Optional[dict] = None):
-        self.id = str(uuid.uuid4())
-        self.timestamp = datetime.now().isoformat()
-        self.level = level
-        self.message = message
-        self.node_id = node_id
-        self.data = data or {}
-
-    def to_dict(self) -> dict:
-        return {
-            'id': self.id,
-            'timestamp': self.timestamp,
-            'level': self.level,
-            'message': self.message,
-            'node_id': self.node_id,
-            'data': self.data,
-        }
-
-
-class DebugExecutionState:
-    """State for a debug execution"""
-
-    def __init__(self, execution_id: str, breakpoints: list[str] = None):
-        self.execution_id = execution_id
-        self.status: str = 'running'
-        self.is_paused: bool = False
-        self.is_stopped: bool = False
-        self.current_node_id: Optional[str] = None
-        self.breakpoints: set[str] = set(breakpoints or [])
-        self.logs: list[ExecutionLog] = []
-        self.pending_logs: list[ExecutionLog] = []
-        self._pause_event = asyncio.Event()
-        self._pause_event.set()  # Initially not paused
-        self._stop_event = asyncio.Event()
-
-    def add_log(self, level: str, message: str, node_id: str = None, data: dict = None):
-        """Add a log entry"""
-        log = ExecutionLog(level, message, node_id, data)
-        self.logs.append(log)
-        self.pending_logs.append(log)
-        logger.log(
-            getattr(logging, level.upper(), logging.INFO),
-            f'[Workflow Debug] {message}',
-            extra={'node_id': node_id, 'data': data},
-        )
-
-    def get_pending_logs(self) -> list[dict]:
-        """Get and clear pending logs"""
-        logs = [log.to_dict() for log in self.pending_logs]
-        self.pending_logs = []
-        return logs
-
-    def pause(self):
-        """Pause execution"""
-        self.is_paused = True
-        self._pause_event.clear()
-        self.add_log('info', 'Execution paused')
-
-    def resume(self):
-        """Resume execution"""
-        self.is_paused = False
-        self._pause_event.set()
-        self.add_log('info', 'Execution resumed')
-
-    def stop(self):
-        """Stop execution"""
-        self.is_stopped = True
-        self.status = 'cancelled'
-        self._stop_event.set()
-        self._pause_event.set()  # Release any pause
-        self.add_log('info', 'Execution stopped')
-
-    async def wait_if_paused(self):
-        """Wait if execution is paused"""
-        if self.is_paused:
-            self.add_log('info', 'Waiting for resume...')
-            await self._pause_event.wait()
-
-    def check_breakpoint(self, node_id: str) -> bool:
-        """Check if there's a breakpoint at the given node"""
-        return node_id in self.breakpoints
-
-
-class DebugWorkflowExecutor(WorkflowExecutor):
-    """
-    Debug-enabled workflow executor with step-by-step execution support.
-    Extends WorkflowExecutor with debugging capabilities.
-    """
-
-    # Class-level storage for active debug sessions
-    _debug_states: dict[str, DebugExecutionState] = {}
-
-    def __init__(self, ap: Optional['app.Application'] = None):
-        super().__init__(ap)
-
-    @classmethod
-    def get_debug_state(cls, execution_id: str) -> Optional[DebugExecutionState]:
-        """Get debug state for an execution"""
-        return cls._debug_states.get(execution_id)
-
-    @classmethod
-    def create_debug_state(cls, execution_id: str, breakpoints: list[str] = None) -> DebugExecutionState:
-        """Create a new debug state"""
-        state = DebugExecutionState(execution_id, breakpoints)
-        cls._debug_states[execution_id] = state
-        return state
-
-    @classmethod
-    def remove_debug_state(cls, execution_id: str):
-        """Remove debug state for an execution"""
-        cls._debug_states.pop(execution_id, None)
-
-    async def execute_debug(
-        self,
-        workflow: WorkflowDefinition,
-        context: ExecutionContext,
-        debug_state: DebugExecutionState,
-    ) -> ExecutionContext:
-        """
-        Execute a workflow in debug mode.
-
-        Args:
-            workflow: Workflow definition
-            context: Execution context
-            debug_state: Debug execution state
-
-        Returns:
-            Updated execution context
-        """
-        context.status = ExecutionStatus.RUNNING
-        context.start_time = datetime.now()
-        debug_state.add_log('info', f'Starting debug execution for workflow: {workflow.name}')
-
-        try:
-            # Build execution graph
-            node_map = {node.id: node for node in workflow.nodes}
-            edge_map = self._build_edge_map(workflow.edges)
-            self._edges = workflow.edges
-
-            # Initialize node states
-            for node in workflow.nodes:
-                if node.id not in context.node_states:
-                    context.node_states[node.id] = NodeState(node_id=node.id)
-
-            # Find start node(s)
-            start_nodes = self._find_start_nodes(workflow.nodes, workflow.edges)
-
-            if not start_nodes:
-                raise ValueError('No start nodes found in workflow')
-
-            debug_state.add_log('info', f'Found {len(start_nodes)} start node(s)')
-
-            # Execute from start nodes
-            for start_node in start_nodes:
-                if debug_state.is_stopped:
-                    break
-
-                await self._execute_debug_from_node(
-                    start_node, node_map, edge_map, context, debug_state, workflow.settings.max_retries
-                )
-
-            # Set final status
-            if debug_state.is_stopped:
-                context.status = ExecutionStatus.CANCELLED
-                debug_state.status = 'cancelled'
-            else:
-                all_completed = all(
-                    state.status in (NodeStatus.COMPLETED, NodeStatus.SKIPPED) for state in context.node_states.values()
-                )
-
-                if all_completed:
-                    context.status = ExecutionStatus.COMPLETED
-                    debug_state.status = 'completed'
-                    debug_state.add_log('info', 'Workflow execution completed successfully')
-                else:
-                    has_failed = any(state.status == NodeStatus.FAILED for state in context.node_states.values())
-                    if has_failed:
-                        context.status = ExecutionStatus.FAILED
-                        debug_state.status = 'error'
-
-        except Exception as e:
-            context.status = ExecutionStatus.FAILED
-            context.error = str(e)
-            debug_state.status = 'error'
-            debug_state.add_log('error', f'Workflow execution failed: {e}', data={'traceback': traceback.format_exc()})
-            logger.error(f'Debug workflow execution failed: {e}\n{traceback.format_exc()}')
-
-        finally:
-            context.end_time = datetime.now()
-
-        return context
-
-    async def _execute_debug_from_node(
-        self,
-        node: NodeDefinition,
-        node_map: dict[str, NodeDefinition],
-        edge_map: dict[str, list[EdgeDefinition]],
-        context: ExecutionContext,
-        debug_state: DebugExecutionState,
-        max_retries: int = 3,
-    ):
-        """Execute workflow from a node with debug support"""
-
-        # Check if stopped
-        if debug_state.is_stopped:
-            return
-
-        # Wait if paused
-        await debug_state.wait_if_paused()
-
-        # Check if should skip
-        if await self._should_skip_node(node, context):
-            if context.node_states[node.id].status == NodeStatus.SKIPPED:
-                debug_state.add_log('info', f'Skipping node: {node.id}', node_id=node.id)
-            return
-
-        # Check breakpoint
-        if debug_state.check_breakpoint(node.id):
-            debug_state.add_log('info', f'Hit breakpoint at node: {node.id}', node_id=node.id)
-            debug_state.pause()
-            await debug_state.wait_if_paused()
-
-        # Update current node
-        debug_state.current_node_id = node.id
-        debug_state.add_log('info', f'Executing node: {node.id} ({node.type})', node_id=node.id)
-
-        # Execute node
-        await self._execute_debug_node(node, context, debug_state, max_retries)
-
-        # Check if stopped or failed
-        if debug_state.is_stopped:
-            return
-        if context.node_states[node.id].status == NodeStatus.FAILED:
-            return
-
-        # Get outgoing edges
-        outgoing_edges = edge_map.get(node.id, [])
-
-        # Execute next nodes
-        for edge in outgoing_edges:
-            if debug_state.is_stopped:
-                break
-
-            target_node = node_map.get(edge.target_node)
-            if not target_node:
-                continue
-
-            # Check edge condition
-            if edge.condition:
-                condition_met = await self._evaluate_condition(edge.condition, context)
-                if not condition_met:
-                    debug_state.add_log('debug', f'Edge condition not met: {edge.condition}', node_id=node.id)
-                    continue
-
-            # Check if all inputs are ready
-            if await self._inputs_ready(target_node, edge_map, context):
-                await self._execute_debug_from_node(target_node, node_map, edge_map, context, debug_state, max_retries)
-
-    async def _execute_debug_node(
-        self, node: NodeDefinition, context: ExecutionContext, debug_state: DebugExecutionState, max_retries: int = 3
-    ):
-        """Execute a single node with debug logging"""
-
-        node_state = context.node_states[node.id]
-        node_state.status = NodeStatus.RUNNING
-        node_state.start_time = datetime.now()
-
-        # Get node instance (pass ap for access to services)
-        node_instance = self.registry.create_instance(node.type, node.id, node.config, ap=self.ap)
-
-        if not node_instance:
-            node_state.status = NodeStatus.FAILED
-            node_state.error = f'Unknown node type: {node.type}'
-            node_state.end_time = datetime.now()
-            debug_state.add_log('error', f'Unknown node type: {node.type}', node_id=node.id)
-            self._record_execution_step(node, node_state, context)
-            await self._persist_node_execution(node, node_state, context)
-            return
-
-        # Resolve inputs
-        inputs = await self._resolve_inputs(node, context)
-        node_state.inputs = inputs
-        debug_state.add_log(
-            'debug', 'Node inputs resolved', node_id=node.id, data={'inputs': self._safe_serialize(inputs)}
-        )
-
-        # Validate inputs
-        validation_errors = await node_instance.validate_inputs(inputs)
-        if validation_errors:
-            node_state.status = NodeStatus.FAILED
-            node_state.error = '; '.join(validation_errors)
-            node_state.end_time = datetime.now()
-            debug_state.add_log('error', f'Input validation failed: {node_state.error}', node_id=node.id)
-            self._record_execution_step(node, node_state, context)
-            await self._persist_node_execution(node, node_state, context)
-            return
-
-        # Execute with retries
-        for attempt in range(max_retries + 1):
-            if debug_state.is_stopped:
-                node_state.status = NodeStatus.FAILED
-                node_state.error = 'Execution stopped'
-                node_state.end_time = datetime.now()
-                break
-
-            try:
-                outputs = await node_instance.execute(inputs, context)
-                node_state.outputs = outputs
-                node_state.status = NodeStatus.COMPLETED
-                node_state.end_time = datetime.now()
-
-                duration_ms = int((node_state.end_time - node_state.start_time).total_seconds() * 1000)
-                debug_state.add_log(
-                    'info',
-                    f'Node completed in {duration_ms}ms',
-                    node_id=node.id,
-                    data={'outputs': self._safe_serialize(outputs), 'duration_ms': duration_ms},
-                )
-                break
-
-            except Exception as e:
-                node_state.retry_count = attempt + 1
-                debug_state.add_log(
-                    'warning', f'Node execution failed (attempt {attempt + 1}/{max_retries + 1}): {e}', node_id=node.id
-                )
-
-                if attempt < max_retries:
-                    await asyncio.sleep(1)
-                else:
-                    node_state.status = NodeStatus.FAILED
-                    node_state.error = str(e)
-                    node_state.end_time = datetime.now()
-                    debug_state.add_log(
-                        'error',
-                        f'Node failed after {max_retries + 1} attempts: {e}',
-                        node_id=node.id,
-                        data={'error': str(e), 'traceback': traceback.format_exc()},
-                    )
-
-        self._record_execution_step(node, node_state, context)
-        await self._persist_node_execution(node, node_state, context)
-
-    async def step_execute(
-        self,
-        workflow: WorkflowDefinition,
-        context: ExecutionContext,
-        debug_state: DebugExecutionState,
-    ) -> dict:
-        """
-        Execute one step (one node) in debug mode.
-
-        Returns:
-            Dict with node_id, node_state, and completed status
-        """
-        # Find next node to execute
-        next_node = self._find_next_executable_node(workflow, context)
-
-        if not next_node:
-            debug_state.status = 'completed'
-            return {'completed': True}
-
-        # Execute single node
-        debug_state.current_node_id = next_node.id
-        await self._execute_debug_node(next_node, context, debug_state, workflow.settings.max_retries)
-
-        node_state = context.node_states.get(next_node.id)
-
-        # Check if workflow is complete
-        all_done = all(
-            state.status in (NodeStatus.COMPLETED, NodeStatus.SKIPPED, NodeStatus.FAILED)
-            for state in context.node_states.values()
-        )
-
-        if all_done:
-            debug_state.status = 'completed'
-            context.status = ExecutionStatus.COMPLETED
-
-        return {
-            'node_id': next_node.id,
-            'node_state': {
-                'status': node_state.status.value if node_state else 'unknown',
-                'inputs': self._safe_serialize(node_state.inputs) if node_state else {},
-                'outputs': self._safe_serialize(node_state.outputs) if node_state else {},
-                'error': node_state.error if node_state else None,
-            },
-            'completed': all_done,
-        }
-
-    def _find_next_executable_node(
-        self, workflow: WorkflowDefinition, context: ExecutionContext
-    ) -> Optional[NodeDefinition]:
-        """Find the next node that can be executed"""
-        edge_map = self._build_edge_map(workflow.edges)
-
-        for node in workflow.nodes:
-            state = context.node_states.get(node.id)
-
-            # Skip completed, running, or failed nodes
-            if state and state.status in (
-                NodeStatus.COMPLETED,
-                NodeStatus.RUNNING,
-                NodeStatus.FAILED,
-                NodeStatus.SKIPPED,
-            ):
-                continue
-
-            incoming_nodes = self._incoming_dependency_nodes(node.id, edge_map)
-
-            # If no incoming nodes, it's a start node
-            if not incoming_nodes:
-                return node
-
-            # Check if all incoming nodes are done
-            all_incoming_done = True
-            for source_id in incoming_nodes:
-                source_state = context.node_states.get(source_id)
-                if not source_state or source_state.status not in (NodeStatus.COMPLETED, NodeStatus.SKIPPED):
-                    all_incoming_done = False
-                    break
-
-            if all_incoming_done:
-                return node
-
-        return None
-
-    def _safe_serialize(self, data: Any) -> Any:
-        """Safely serialize data for logging"""
-        if data is None:
-            return None
-        if isinstance(data, (str, int, float, bool)):
-            return data
-        if isinstance(data, (list, tuple)):
-            return [self._safe_serialize(item) for item in data[:100]]  # Limit list size
-        if isinstance(data, dict):
-            result = {}
-            for key, value in list(data.items())[:50]:  # Limit dict size
-                result[str(key)] = self._safe_serialize(value)
-            return result
-        # For complex objects, try to convert to string
-        try:
-            return str(data)[:1000]  # Limit string length
-        except Exception:
-            return '<non-serializable>'
-
-    def get_execution_state(self, context: ExecutionContext, debug_state: DebugExecutionState) -> dict:
-        """Get current execution state for API response"""
-        node_states = {}
-        for node_id, state in context.node_states.items():
-            node_states[node_id] = {
-                'status': state.status.value,
-                'inputs': self._safe_serialize(state.inputs),
-                'outputs': self._safe_serialize(state.outputs),
-                'error': state.error,
-                'startTime': state.start_time.isoformat() if state.start_time else None,
-                'endTime': state.end_time.isoformat() if state.end_time else None,
-                'duration': int((state.end_time - state.start_time).total_seconds() * 1000)
-                if state.start_time and state.end_time
-                else None,
-            }
-
-        return {
-            'status': debug_state.status,
-            'current_node_id': debug_state.current_node_id,
-            'node_states': node_states,
-            'new_logs': debug_state.get_pending_logs(),
-            'error': context.error,
-        }

src/langbot/pkg/workflow/entities.py

@@ -1,168 +0,0 @@
-"""Workflow entities and data models
-
-This module defines workflow entities using SDK standard entities where available,
-and local-specific entities for LangBot_copy-specific functionality.
-"""
-
-from __future__ import annotations
-
-from datetime import datetime
-from typing import Any, Optional
-import pydantic
-
-# Import SDK entities for standard workflow protocol types
-# These are re-exported for use by other modules in the workflow package.
-from langbot_plugin.api.entities.builtin.workflow.entities import (
-    ExecutionContext as ExecutionContext,
-    ExecutionStep as ExecutionStep,
-    MessageContext as MessageContext,
-    NodeDefinition,
-    NodeState as NodeState,
-    PortDefinition as PortDefinition,
-)
-from langbot_plugin.api.entities.builtin.workflow.enums import (
-    ExecutionStatus as ExecutionStatus,
-    NodeStatus as NodeStatus,
-)
-
-__all__ = [
-    "ExecutionContext",
-    "ExecutionStep",
-    "MessageContext",
-    "NodeDefinition",
-    "NodeState",
-    "PortDefinition",
-    "ExecutionStatus",
-    "NodeStatus",
-]
-
-
-class Position(pydantic.BaseModel):
-    """Node position on canvas"""
-
-    x: float = 0
-    y: float = 0
-
-
-class EdgeDefinition(pydantic.BaseModel):
-    """Workflow edge definition (connection between nodes)"""
-
-    id: str
-    source_node: str
-    source_port: str = 'output'
-    target_node: str
-    target_port: str = 'input'
-    edge_type: str = 'legacy'  # control, data, or legacy (old mixed semantics)
-    condition: Optional[str] = None  # Optional condition expression
-
-
-class TriggerDefinition(pydantic.BaseModel):
-    """Workflow trigger definition"""
-
-    id: str
-    type: str  # message, cron, event, webhook
-    config: dict[str, Any] = {}
-    enabled: bool = True
-
-
-class WorkflowSettings(pydantic.BaseModel):
-    """Workflow settings"""
-
-    # Execution settings
-    max_execution_time: int = 300  # seconds
-    max_retries: int = 3
-    retry_delay: int = 5  # seconds
-
-    # Error handling
-    error_handling: str = 'stop'  # stop, continue, retry
-
-    # Logging
-    log_level: str = 'info'
-    save_execution_history: bool = True
-
-    # Concurrency
-    max_concurrent_executions: int = 10
-
-
-class SafetyConfig(pydantic.BaseModel):
-    """Safety configuration (inherited from Pipeline)"""
-
-    content_filter: dict[str, Any] = {'enable': False, 'sensitive_words': [], 'replace_with': '***'}
-    rate_limit: dict[str, Any] = {'enable': False, 'requests_per_minute': 60, 'burst_limit': 10}
-
-
-class OutputConfig(pydantic.BaseModel):
-    """Output configuration (inherited from Pipeline)"""
-
-    long_text_processing: dict[str, Any] = {
-        'strategy': 'split',  # split, truncate, file
-        'max_length': 4000,
-        'split_separator': '\n\n',
-    }
-    force_delay: dict[str, Any] = {'enable': False, 'min_delay_ms': 0, 'max_delay_ms': 0}
-    misc: dict[str, Any] = {}
-
-
-class WorkflowGlobalConfig(pydantic.BaseModel):
-    """Workflow global configuration (inherited from Pipeline capabilities)"""
-
-    safety: SafetyConfig = SafetyConfig()
-    output: OutputConfig = OutputConfig()
-
-
-class ExtensionsPreferences(pydantic.BaseModel):
-    """Extensions preferences (same as Pipeline)"""
-
-    enable_all_plugins: bool = True
-    enable_all_mcp_servers: bool = True
-    plugins: list[str] = []
-    mcp_servers: list[str] = []
-
-
-class ConversationVariable(pydantic.BaseModel):
-    """Conversation-level variable definition"""
-
-    name: str
-    type: str = 'string'  # string, number, boolean, object, array
-    description: str = ''
-    default_value: Any = None
-    max_length: Optional[int] = None  # For strings
-
-
-class WorkflowDefinition(pydantic.BaseModel):
-    """Complete workflow definition"""
-
-    uuid: str
-    name: str
-    description: str = ''
-    emoji: str = '💼'
-    version: int = 1
-
-    # Workflow graph
-    nodes: list[NodeDefinition] = []
-    edges: list[EdgeDefinition] = []
-
-    # Variables
-    variables: dict[str, Any] = {}  # Global variables
-    conversation_variables: list[ConversationVariable] = []  # Session-level variables
-
-    # Settings
-    settings: WorkflowSettings = WorkflowSettings()
-
-    # Triggers (for automation)
-    triggers: list[TriggerDefinition] = []
-
-    # Global configuration (inherited from Pipeline)
-    global_config: WorkflowGlobalConfig = WorkflowGlobalConfig()
-
-    # Extensions
-    extensions_preferences: ExtensionsPreferences = ExtensionsPreferences()
-
-    # Metadata
-    is_enabled: bool = True
-    created_at: Optional[datetime] = None
-    updated_at: Optional[datetime] = None
-
-    # Source tracking (for imported workflows)
-    source: Optional[str] = None  # dify, n8n, langflow, etc.
-    source_id: Optional[str] = None

src/langbot/pkg/workflow/executor.py

@@ -1,759 +0,0 @@
-"""Workflow execution engine.
-
-This module contains the core workflow execution logic:
-- WorkflowExecutor: Main execution engine with control flow handling
-- ParallelExecutor: Parallel branch execution
-- LoopExecutor: Loop/iterator execution
-
-Debug execution support has been moved to the ``debug`` module.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import logging
-import re
-import uuid
-from datetime import datetime
-from typing import Any, Optional, TYPE_CHECKING
-
-import sqlalchemy
-
-from .entities import (
-    WorkflowDefinition,
-    NodeDefinition,
-    EdgeDefinition,
-    ExecutionContext,
-    ExecutionStatus,
-    NodeState,
-    NodeStatus,
-    ExecutionStep,
-)
-from ..entity.persistence import workflow as persistence_workflow
-from .registry import NodeTypeRegistry
-from .safe_eval import safe_eval_with_vars
-
-if TYPE_CHECKING:
-    from ..core import app
-
-logger = logging.getLogger(__name__)
-
-
-class WorkflowExecutor:
-    """
-    Workflow execution engine.
-    Handles the execution of workflow definitions with proper control flow.
-    """
-
-    def __init__(self, ap: Optional['app.Application'] = None):
-        self.ap = ap
-        self.registry = NodeTypeRegistry.instance()
-        self._edges: list[EdgeDefinition] = []
-
-    async def execute(
-        self, workflow: WorkflowDefinition, context: ExecutionContext, start_node_id: Optional[str] = None
-    ) -> ExecutionContext:
-        """
-        Execute a workflow.
-
-        Args:
-            workflow: Workflow definition
-            context: Execution context
-            start_node_id: Optional starting node (for resumption)
-
-        Returns:
-            Updated execution context
-        """
-        context.status = ExecutionStatus.RUNNING
-        context.start_time = datetime.now()
-
-        try:
-            # Build execution graph
-            node_map = {node.id: node for node in workflow.nodes}
-            edge_map = self._build_edge_map(workflow.edges)
-            self._edges = workflow.edges
-
-            # Initialize node states
-            for node in workflow.nodes:
-                if node.id not in context.node_states:
-                    context.node_states[node.id] = NodeState(node_id=node.id, node_type=node.type, status=NodeStatus.PENDING)
-
-            # Find start node(s)
-            if start_node_id:
-                start_nodes = [node_map[start_node_id]]
-            else:
-                start_nodes = self._find_start_nodes(workflow.nodes, workflow.edges)
-
-            if not start_nodes:
-                raise ValueError('No start nodes found in workflow')
-
-            # Execute from start nodes
-            for start_node in start_nodes:
-                await self._execute_from_node(
-                    start_node, node_map, edge_map, context, workflow.settings.max_retries, path=set()
-                )
-
-            # Check final status
-            all_completed = all(
-                state.status in (NodeStatus.COMPLETED, NodeStatus.SKIPPED) for state in context.node_states.values()
-            )
-
-            if all_completed:
-                context.status = ExecutionStatus.COMPLETED
-            else:
-                # Some nodes might still be waiting
-                has_failed = any(state.status == NodeStatus.FAILED for state in context.node_states.values())
-                if has_failed:
-                    context.status = ExecutionStatus.FAILED
-
-        except Exception as e:
-            context.status = ExecutionStatus.FAILED
-            context.error = str(e)
-            logger.error(
-                'Workflow execution failed',
-                exc_info=True,
-                extra={
-                    'workflow_id': workflow.uuid,
-                    'execution_id': context.execution_id,
-                    'node_states': {
-                        node_id: {
-                            'status': state.status.value if state.status else None,
-                            'error': state.error,
-                        }
-                        for node_id, state in context.node_states.items()
-                    },
-                },
-            )
-
-            # Note: Frontend panel logging has been removed.
-            # A new solution will be implemented separately.
-
-        finally:
-            context.end_time = datetime.now()
-
-            # Note: Frontend panel logging has been removed.
-            # A new solution will be implemented separately.
-
-        return context
-
-    async def _execute_from_node(
-        self,
-        node: NodeDefinition,
-        node_map: dict[str, NodeDefinition],
-        edge_map: dict[str, list[EdgeDefinition]],
-        context: ExecutionContext,
-        max_retries: int = 3,
-        path: set[str] | None = None,
-    ):
-        """Execute workflow starting from a specific node"""
-
-        # Initialize path set for cycle detection (path-based, not global visited)
-        if path is None:
-            path = set()
-
-        # Check for circular dependency on the *current path* only
-        # This correctly allows diamond shapes (A→B, A→C, B→D, C→D)
-        if node.id in path:
-            logger.warning(f'Circular dependency detected at node: {node.id}')
-            context.node_states[node.id].status = NodeStatus.SKIPPED
-            context.node_states[node.id].error = 'Circular dependency detected'
-            context.node_states[node.id].end_time = datetime.now()
-            await self._persist_node_execution(node, context.node_states[node.id], context)
-            return
-
-        # Add node to current path
-        path.add(node.id)
-
-        # Check if node should be skipped
-        if await self._should_skip_node(node, context):
-            existing_state = context.node_states[node.id]
-            if existing_state.status == NodeStatus.SKIPPED:
-                existing_state.end_time = existing_state.end_time or datetime.now()
-                await self._persist_node_execution(node, existing_state, context)
-            path.discard(node.id)
-            return
-
-        # Execute current node
-        await self._execute_node(node, context, max_retries)
-
-        # If node failed and we should stop on error, return
-        if context.node_states[node.id].status == NodeStatus.FAILED:
-            path.discard(node.id)
-            return
-
-        node_state = context.node_states[node.id]
-        node_type_name = node.type.split('.')[-1] if '.' in node.type else node.type
-
-        # ── Control flow integration ────────────────────────────────
-        # For loop / iterator nodes: run the LoopExecutor over
-        # downstream body nodes for each item, then continue to the
-        # "completed" output edge.
-        if node_type_name in ('loop', 'iterator'):
-            items = node_state.outputs.get('_items') or []
-            if not items:
-                # iterator: items come from inputs
-                items = node_state.inputs.get('items', node_state.inputs.get('array', []))
-                if not isinstance(items, list):
-                    items = [items] if items else []
-            max_iter = int(node.config.get('max_iterations', 100))
-            items = items[:max_iter]
-
-            # Collect downstream "body" nodes (connected via edges)
-            outgoing_edges = edge_map.get(node.id, [])
-            body_nodes = []
-            for edge in outgoing_edges:
-                target = node_map.get(edge.target_node)
-                if target:
-                    body_nodes.append(target)
-
-            if body_nodes and items:
-                loop_exec = LoopExecutor(self)
-                results = await loop_exec.execute_loop(items, body_nodes, context, max_iter)
-                node_state.outputs['results'] = results
-                node_state.outputs['completed'] = True
-            else:
-                node_state.outputs['results'] = []
-                node_state.outputs['completed'] = True
-
-            path.discard(node.id)
-            return  # body nodes already executed by LoopExecutor
-
-        # For parallel nodes: run downstream branches concurrently
-        if node_type_name == 'parallel':
-            outgoing_edges = edge_map.get(node.id, [])
-            branch_nodes = []
-            for edge in outgoing_edges:
-                target = node_map.get(edge.target_node)
-                if target:
-                    branch_nodes.append([target])
-
-            if branch_nodes:
-                par_exec = ParallelExecutor(self)
-                results = await par_exec.execute_parallel(branch_nodes, context)
-                node_state.outputs['results'] = results
-
-            path.discard(node.id)
-            return  # branch nodes already executed by ParallelExecutor
-
-        # ── Standard edge-based continuation ────────────────────────
-        # Get outgoing edges
-        outgoing_edges = edge_map.get(node.id, [])
-
-        # Execute next nodes based on edge conditions
-        for edge in outgoing_edges:
-            target_node = node_map.get(edge.target_node)
-            if not target_node:
-                continue
-
-            # Check edge condition
-            if edge.condition:
-                condition_met = await self._evaluate_condition(edge.condition, context)
-                if not condition_met:
-                    continue
-
-            # Check if all inputs are ready
-            if await self._inputs_ready(target_node, edge_map, context):
-                await self._execute_from_node(target_node, node_map, edge_map, context, max_retries, path)
-
-        # Remove node from path when backtracking (allows diamond revisit)
-        path.discard(node.id)
-
-    async def _execute_node(self, node: NodeDefinition, context: ExecutionContext, max_retries: int = 3):
-        """Execute a single node with retry logic"""
-
-        node_state = context.node_states[node.id]
-        node_state.status = NodeStatus.RUNNING
-        node_state.start_time = datetime.now()
-
-        # Get node instance (pass ap for access to services)
-        node_instance = self.registry.create_instance(node.type, node.id, node.config, ap=self.ap)
-
-        if not node_instance:
-            node_state.status = NodeStatus.FAILED
-            node_state.error = f'Unknown node type: {node.type}'
-            node_state.end_time = datetime.now()
-            self._record_execution_step(node, node_state, context)
-            await self._persist_node_execution(node, node_state, context)
-            return
-
-        # Resolve inputs
-        inputs = await self._resolve_inputs(node, context)
-        node_state.inputs = inputs
-
-        # Validate inputs
-        validation_errors = await node_instance.validate_inputs(inputs)
-        if validation_errors:
-            node_state.status = NodeStatus.FAILED
-            node_state.error = '; '.join(validation_errors)
-            node_state.end_time = datetime.now()
-            self._record_execution_step(node, node_state, context)
-            await self._persist_node_execution(node, node_state, context)
-            return
-
-        # Check if node supports streaming (has execute_stream method and stream config is enabled)
-        use_streaming = hasattr(node_instance, 'execute_stream') and node.config.get('stream', False)
-
-        # Execute with retries
-        for attempt in range(max_retries + 1):
-            try:
-                if use_streaming:
-                    # Streaming execution with aggregation and timeout
-                    aggregated_response = ''
-                    try:
-                        async with asyncio.timeout(300):  # 5 minute timeout for streaming
-                            async for chunk in node_instance.execute_stream(inputs, context):
-                                if chunk:
-                                    aggregated_response += chunk
-                    except asyncio.TimeoutError:
-                        logger.warning(f'Node {node.id} ({node.type}) streaming timed out, falling back to non-streaming')
-                        use_streaming = False
-                        outputs = await node_instance.execute(inputs, context)
-                    else:
-                        # Get response from context if set by execute_stream, otherwise use aggregated
-                        final_response = context.variables.pop('_last_llm_response', aggregated_response)
-                        outputs = {'response': final_response, 'usage': {'prompt_tokens': 0, 'completion_tokens': 0, 'total_tokens': 0}}
-                        logger.info(f'Node {node.id} ({node.type}) streaming completed, response length: {len(final_response)}')
-                else:
-                    outputs = await node_instance.execute(inputs, context)
-                node_state.outputs = outputs
-                node_state.status = NodeStatus.COMPLETED
-                node_state.end_time = datetime.now()
-                break
-            except Exception as e:
-                node_state.retry_count = attempt + 1
-                logger.error(
-                    f'Node {node.id} ({node.type}) execution failed (attempt {attempt + 1}/{max_retries + 1}): {e}',
-                    exc_info=True,
-                    extra={
-                        'node_id': node.id,
-                        'node_type': node.type,
-                        'attempt': attempt + 1,
-                        'max_retries': max_retries,
-                        'execution_id': context.execution_id,
-                    },
-                )
-
-                if attempt < max_retries:
-                    await asyncio.sleep(1)  # Brief delay before retry
-                else:
-                    node_state.status = NodeStatus.FAILED
-                    node_state.error = str(e)
-                    node_state.end_time = datetime.now()
-                    logger.error(
-                        f'Node {node.id} ({node.type}) permanently failed after {max_retries + 1} attempts',
-                        extra={
-                            'node_id': node.id,
-                            'node_type': node.type,
-                            'error': str(e),
-                            'execution_id': context.execution_id,
-                        },
-                    )
-
-        self._record_execution_step(node, node_state, context)
-        await self._persist_node_execution(node, node_state, context)
-
-    async def _resolve_inputs(self, node: NodeDefinition, context: ExecutionContext) -> dict[str, Any]:
-        """Resolve input values for a node from connected nodes and context"""
-        inputs = {}
-
-        # Get inputs from context variables
-        if 'message' in context.variables:
-            inputs['message'] = context.variables['message']
-
-        # Get inputs from message context
-        if context.message_context:
-            inputs['message'] = context.message_context.message_content
-            inputs['message_content'] = context.message_context.message_content
-            inputs['sender_id'] = context.message_context.sender_id
-            inputs['platform'] = context.message_context.platform
-        else:
-            logger.warning(
-                f'[_resolve_inputs] node={node.id} ({node.type}): message_context is None!',
-                extra={
-                    'node_id': node.id,
-                    'node_type': node.type,
-                    'execution_id': context.execution_id,
-                    'variables_keys': list(context.variables.keys()) if context.variables else [],
-                },
-            )
-
-        # Log current inputs state after message_context processing
-        logger.debug(
-            f'[_resolve_inputs] node={node.id} after message_context: {list(inputs.keys())}',
-        )
-
-        # Get inputs from node config that reference other nodes
-        for key, value in node.config.items():
-            if isinstance(value, str) and value.startswith('{{') and value.endswith('}}'):
-                resolved = await self._resolve_expression(value[2:-2], context)
-                inputs[key] = resolved
-            else:
-                inputs[key] = value
-
-        # Get inputs from connected upstream nodes via data edges.
-        # Build a reverse map: for each incoming edge to this node, find the
-        # source node and the specific source/target port.
-        for edge in self._edges:
-            if not self._is_data_edge(edge):
-                continue
-            if edge.target_node != node.id:
-                continue
-            source_state = context.node_states.get(edge.source_node)
-            if not source_state or source_state.status != NodeStatus.COMPLETED:
-                continue
-            target_port = edge.target_port or 'input'
-            source_port = edge.source_port or 'output'
-            # Map the source node's output port value to this node's input port
-            if source_port in source_state.outputs:
-                inputs[target_port] = source_state.outputs[source_port]
-            elif 'output' in source_state.outputs:
-                # Fallback: if exact port not found, try generic 'output'
-                inputs[target_port] = source_state.outputs['output']
-            elif source_state.outputs:
-                # Last resort: use the first available output
-                inputs[target_port] = next(iter(source_state.outputs.values()))
-
-        # Smart input mapping: if a node needs 'message' but received a different
-        # port name (e.g., 'content' from llm_call), copy the value to 'message'.
-        # This handles edge connection mismatches where the sender uses a different
-        # port name than what the receiver expects.
-        if 'message' not in inputs or inputs.get('message') is None:
-            for fallback_key in ('content', 'response', 'input', 'output', 'result', 'text'):
-                if fallback_key in inputs and inputs[fallback_key] is not None:
-                    inputs['message'] = inputs[fallback_key]
-                    logger.debug(
-                        f'[_resolve_inputs] node={node.id}: mapped {fallback_key} -> message',
-                    )
-                    break
-
-        logger.debug(
-            f'[_resolve_inputs] node={node.id} final inputs keys: {list(inputs.keys())}, message={repr(inputs.get("message", "<missing>")[:100] if isinstance(inputs.get("message"), str) else inputs.get("message"))}',
-        )
-        return inputs
-
-    async def _resolve_expression(self, expression: str, context: ExecutionContext) -> Any:
-        """Resolve a variable expression like 'nodes.node1.outputs.text'"""
-        parts = expression.strip().split('.')
-
-        if not parts:
-            return None
-
-        if parts[0] == 'nodes' and len(parts) >= 4:
-            # nodes.node_id.outputs.output_name
-            node_id = parts[1]
-            if parts[2] == 'outputs' and node_id in context.node_states:
-                output_name = '.'.join(parts[3:])
-                return context.node_states[node_id].outputs.get(output_name)
-
-        elif parts[0] == 'variables':
-            # variables.var_name
-            var_name = '.'.join(parts[1:])
-            return context.variables.get(var_name)
-
-        elif parts[0] == 'conversation_variables':
-            # conversation_variables.var_name
-            var_name = '.'.join(parts[1:])
-            return context.conversation_variables.get(var_name)
-
-        elif parts[0] == 'message':
-            # message.content, message.sender_id, etc.
-            if context.message_context:
-                attr = parts[1] if len(parts) > 1 else None
-                if attr == 'content':
-                    return context.message_context.message_content
-                elif attr == 'sender_id':
-                    return context.message_context.sender_id
-                elif attr == 'platform':
-                    return context.message_context.platform
-                elif attr == 'conversation_id':
-                    return context.message_context.conversation_id
-
-        return None
-
-    async def _evaluate_condition(self, condition: str, context: ExecutionContext) -> bool:
-        """Evaluate a condition expression safely.
-
-        Any ``{{ ... }}`` references are resolved against the execution context
-        and bound as **variables** that are passed to :func:`safe_eval_with_vars`.
-        Values are never string-concatenated into the expression, which avoids
-        broken parsing (e.g. values containing quotes) and any injection risk
-        from non-literal value types (lists, dicts, etc.).
-        """
-        variables: dict[str, Any] = {}
-        try:
-            # Resolve variable references in condition into bound variables.
-            if '{{' in condition:
-                pattern = r'\{\{([^}]+)\}\}'
-
-                placeholders: dict[str, str] = {}
-                placeholder_idx = 0
-
-                def replace_with_placeholder(match: re.Match[str]) -> str:
-                    nonlocal placeholder_idx
-                    var_expr = match.group(1)
-                    placeholder = f'__ph{placeholder_idx}__'
-                    placeholders[placeholder] = var_expr
-                    placeholder_idx += 1
-                    return placeholder
-
-                condition = re.sub(pattern, replace_with_placeholder, condition)
-
-                # Resolve each placeholder and bind it as a variable, so the
-                # actual value (of any type) is passed through unchanged.
-                for placeholder, var_expr in placeholders.items():
-                    variables[placeholder] = await self._resolve_expression(var_expr, context)
-
-            # Safe expression evaluation with bound variables (AST whitelist).
-            result = safe_eval_with_vars(condition, variables)
-            return bool(result)
-
-        except Exception as e:
-            logger.warning(f'Condition evaluation failed: {condition} - {e}')
-            return False
-
-    async def _should_skip_node(self, node: NodeDefinition, context: ExecutionContext) -> bool:
-        """Check if a node should be skipped"""
-        state = context.node_states.get(node.id)
-        if state and state.status in (NodeStatus.COMPLETED, NodeStatus.RUNNING, NodeStatus.SKIPPED):
-            return True
-        return False
-
-    async def _inputs_ready(
-        self, node: NodeDefinition, edge_map: dict[str, list[EdgeDefinition]], context: ExecutionContext
-    ) -> bool:
-        """Check if all control predecessors and data providers are ready."""
-        incoming_nodes = self._incoming_dependency_nodes(node.id, edge_map)
-
-        # Check if all incoming nodes have completed
-        for source_id in incoming_nodes:
-            state = context.node_states.get(source_id)
-            if not state or state.status not in (NodeStatus.COMPLETED, NodeStatus.SKIPPED):
-                return False
-
-        return True
-
-    def _find_start_nodes(self, nodes: list[NodeDefinition], edges: list[EdgeDefinition]) -> list[NodeDefinition]:
-        """Find nodes that have no incoming edges (start nodes)"""
-        target_nodes = {edge.target_node for edge in edges if self._is_control_edge(edge) or self._is_data_edge(edge)}
-        start_nodes = [node for node in nodes if node.id not in target_nodes]
-
-        # Also check for trigger nodes
-        trigger_types = {'message_trigger', 'cron_trigger', 'webhook_trigger', 'event_trigger'}
-        for node in nodes:
-            if node.type in trigger_types and node not in start_nodes:
-                start_nodes.insert(0, node)
-
-        return start_nodes
-
-    def _build_edge_map(self, edges: list[EdgeDefinition]) -> dict[str, list[EdgeDefinition]]:
-        """Build a map of source node ID to outgoing control edges."""
-        edge_map: dict[str, list[EdgeDefinition]] = {}
-        for edge in edges:
-            if not self._is_control_edge(edge):
-                continue
-            if edge.source_node not in edge_map:
-                edge_map[edge.source_node] = []
-            edge_map[edge.source_node].append(edge)
-        return edge_map
-
-    def _edge_type(self, edge: EdgeDefinition) -> str:
-        edge_type = (getattr(edge, 'edge_type', None) or 'legacy').strip().lower()
-        if edge_type not in {'control', 'data', 'legacy'}:
-            return 'legacy'
-        return edge_type
-
-    def _is_control_edge(self, edge: EdgeDefinition) -> bool:
-        return self._edge_type(edge) in {'control', 'legacy'}
-
-    def _is_data_edge(self, edge: EdgeDefinition) -> bool:
-        return self._edge_type(edge) in {'data', 'legacy'}
-
-    def _incoming_dependency_nodes(
-        self, node_id: str, edge_map: dict[str, list[EdgeDefinition]]
-    ) -> set[str]:
-        incoming_nodes: set[str] = set()
-        for source_id, edges in edge_map.items():
-            for edge in edges:
-                if edge.target_node == node_id:
-                    incoming_nodes.add(source_id)
-
-        for edge in self._edges:
-            if self._is_data_edge(edge) and edge.target_node == node_id:
-                incoming_nodes.add(edge.source_node)
-
-        return incoming_nodes
-
-    def _record_execution_step(self, node: NodeDefinition, node_state: NodeState, context: ExecutionContext):
-        """Record an execution step in the history"""
-        duration_ms = 0
-        if node_state.start_time and node_state.end_time:
-            duration_ms = int((node_state.end_time - node_state.start_time).total_seconds() * 1000)
-
-        step = ExecutionStep(
-            step_id=f"step_{uuid.uuid4().hex[:8]}",
-            timestamp=datetime.now(),
-            node_id=node.id,
-            node_type=node.type,
-            status=node_state.status,
-            duration_ms=duration_ms,
-            error=node_state.error,
-            inputs=node_state.inputs,
-            outputs=node_state.outputs,
-        )
-        context.history.append(step)
-
-    async def _persist_node_execution(
-        self,
-        node: NodeDefinition,
-        node_state: NodeState,
-        context: ExecutionContext,
-    ):
-        """Persist node execution state for execution detail and logs."""
-        if not self.ap:
-            return
-
-        values = {
-            'execution_uuid': context.execution_id,
-            'node_id': node.id,
-            'node_type': node.type,
-            'status': node_state.status.value,
-            'inputs': node_state.inputs,
-            'outputs': node_state.outputs,
-            'start_time': node_state.start_time,
-            'end_time': node_state.end_time,
-            'error': node_state.error,
-            'retry_count': node_state.retry_count,
-        }
-
-        existing_query = sqlalchemy.select(persistence_workflow.WorkflowNodeExecution).where(
-            persistence_workflow.WorkflowNodeExecution.execution_uuid == context.execution_id,
-            persistence_workflow.WorkflowNodeExecution.node_id == node.id,
-        )
-        existing_result = await self.ap.persistence_mgr.execute_async(existing_query)
-        existing = existing_result.first()
-
-        if existing is None:
-            await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.insert(persistence_workflow.WorkflowNodeExecution).values(**values)
-            )
-        else:
-            await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.update(persistence_workflow.WorkflowNodeExecution)
-                .where(persistence_workflow.WorkflowNodeExecution.id == existing.id)
-                .values(**values)
-            )
-
-
-class ParallelExecutor:
-    """Execute multiple branches in parallel"""
-
-    def __init__(self, executor: WorkflowExecutor):
-        self.executor = executor
-
-    async def execute_parallel(
-        self, branches: list[list[NodeDefinition]], context: ExecutionContext
-    ) -> list[dict[str, Any]]:
-        """
-        Execute multiple branches in parallel.
-
-        Args:
-            branches: List of node sequences to execute in parallel
-            context: Execution context
-
-        Returns:
-            List of results from each branch
-        """
-        tasks = []
-        for branch in branches:
-            task = self._execute_branch(branch, context)
-            tasks.append(task)
-
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-
-        processed_results = []
-        for index, result in enumerate(results):
-            if isinstance(result, Exception):
-                logger.error(
-                    f'Parallel branch {index} failed: {result}',
-                    exc_info=result,
-                    extra={'branch_index': index, 'execution_id': context.execution_id},
-                )
-                processed_results.append({'error': str(result)})
-            else:
-                processed_results.append(result)
-
-        return processed_results
-
-    async def _execute_branch(self, nodes: list[NodeDefinition], context: ExecutionContext) -> dict[str, Any]:
-        """Execute a single branch"""
-        # Create a copy of context for this branch
-        branch_outputs = {}
-
-        for node in nodes:
-            await self.executor._execute_node(node, context, max_retries=3)
-            state = context.node_states.get(node.id)
-            if state and state.status == NodeStatus.COMPLETED:
-                branch_outputs[node.id] = state.outputs
-            elif state and state.status == NodeStatus.FAILED:
-                branch_outputs['error'] = state.error
-                break
-
-        return branch_outputs
-
-
-class LoopExecutor:
-    """Execute loop iterations"""
-
-    def __init__(self, executor: WorkflowExecutor):
-        self.executor = executor
-
-    async def execute_loop(
-        self, items: list[Any], loop_body: list[NodeDefinition], context: ExecutionContext, max_iterations: int = 100
-    ) -> list[dict[str, Any]]:
-        """
-        Execute a loop over items.
-
-        Args:
-            items: Items to iterate over
-            loop_body: Nodes to execute for each item
-            context: Execution context
-            max_iterations: Maximum number of iterations
-
-        Returns:
-            List of results from each iteration
-        """
-        results = []
-
-        for i, item in enumerate(items[:max_iterations]):
-            # Set loop variables
-            context.variables['loop_item'] = item
-            context.variables['loop_index'] = i
-            context.variables['loop_is_first'] = i == 0
-            context.variables['loop_is_last'] = i == len(items) - 1
-
-            iteration_result = {}
-
-            for node in loop_body:
-                # Reset node state for this iteration
-                context.node_states[node.id] = NodeState(node_id=node.id, node_type=node.type, status=NodeStatus.PENDING)
-
-                await self.executor._execute_node(node, context, max_retries=3)
-
-                state = context.node_states.get(node.id)
-                if state:
-                    iteration_result[node.id] = state.outputs
-
-                    # Check for break condition
-                    if state.outputs.get('break', False):
-                        results.append(iteration_result)
-                        return results
-
-            results.append(iteration_result)
-
-        # Clean up loop variables
-        context.variables.pop('loop_item', None)
-        context.variables.pop('loop_index', None)
-        context.variables.pop('loop_is_first', None)
-        context.variables.pop('loop_is_last', None)
-
-        return results

src/langbot/pkg/workflow/metadata.py

@@ -1,284 +0,0 @@
-"""Workflow node metadata loading and validation.
-
-This module makes YAML files under ``templates/metadata/nodes`` the backend
-source of truth for workflow node metadata. Python node classes still provide
-execution logic, but UI-facing metadata is loaded from YAML.
-"""
-
-from __future__ import annotations
-
-import copy
-import logging
-from importlib import resources
-from pathlib import Path
-from typing import Any, Iterable, Optional
-
-import yaml
-
-logger = logging.getLogger(__name__)
-
-
-class MetadataLoadError(Exception):
-    """Raised when a workflow node metadata file cannot be loaded."""
-
-
-class MetadataValidationError(Exception):
-    """Raised when workflow node metadata does not match the expected shape."""
-
-
-class NodeMetadataValidator:
-    """Validate workflow node metadata loaded from YAML files.
-
-    The validator is intentionally strict about the structural fields that the
-    editor needs, but tolerant of legacy YAML details such as missing top-level
-    ``label`` or additional frontend field types.
-    """
-
-    REQUIRED_FIELDS = ('name', 'category', 'inputs', 'outputs', 'config')
-    VALID_CATEGORIES = {'trigger', 'process', 'control', 'action', 'integration', 'misc'}
-    VALID_PORT_TYPES = {'any', 'string', 'number', 'integer', 'boolean', 'object', 'array', 'datetime', 'null'}
-    VALID_CONFIG_TYPES = {
-        'string',
-        'integer',
-        'number',
-        'float',
-        'boolean',
-        'select',
-        'json',
-        'textarea',
-        'text',
-        'secret',
-        'array[string]',
-        'file',
-        'array[file]',
-        'llm-model-selector',
-        'embedding-model-selector',
-        'rerank-model-selector',
-        'pipeline-selector',
-        'knowledge-base-selector',
-        'knowledge-base-multi-selector',
-        'bot-selector',
-        'tools-selector',
-        'model-fallback-selector',
-        'prompt-editor',
-        'plugin-selector',
-        'webhook-url',
-        'embed-code',
-        'workflow-selector',
-    }
-
-    def validate(self, metadata: dict[str, Any]) -> list[str]:
-        """Return validation errors. An empty list means the metadata is valid."""
-        errors: list[str] = []
-
-        if not isinstance(metadata, dict):
-            return ['metadata root must be a mapping']
-
-        for field in self.REQUIRED_FIELDS:
-            if field not in metadata:
-                errors.append(f'missing required field: {field}')
-
-        if errors:
-            return errors
-
-        name = metadata.get('name')
-        if not isinstance(name, str) or not name.strip():
-            errors.append('field "name" must be a non-empty string')
-
-        category = metadata.get('category')
-        if category not in self.VALID_CATEGORIES:
-            errors.append(f'invalid category: {category}')
-
-        errors.extend(self._validate_ports(metadata.get('inputs'), 'inputs'))
-        errors.extend(self._validate_ports(metadata.get('outputs'), 'outputs'))
-        errors.extend(self._validate_config(metadata.get('config')))
-
-        return errors
-
-    def validate_or_raise(self, metadata: dict[str, Any]) -> dict[str, Any]:
-        """Validate metadata and raise ``MetadataValidationError`` on failure."""
-        errors = self.validate(metadata)
-        if errors:
-            node_name = metadata.get('name', 'unknown') if isinstance(metadata, dict) else 'unknown'
-            raise MetadataValidationError(f'invalid metadata for {node_name}: {errors}')
-        return metadata
-
-    def _validate_ports(self, ports: Any, field_name: str) -> list[str]:
-        errors: list[str] = []
-        if not isinstance(ports, list):
-            return [f'{field_name} must be a list']
-
-        seen_names: set[str] = set()
-        for index, port in enumerate(ports):
-            path = f'{field_name}[{index}]'
-            if not isinstance(port, dict):
-                errors.append(f'{path} must be a mapping')
-                continue
-
-            name = port.get('name')
-            if not isinstance(name, str) or not name:
-                errors.append(f'{path}.name must be a non-empty string')
-                continue
-
-            if name in seen_names:
-                errors.append(f'{path}.name duplicates "{name}"')
-            seen_names.add(name)
-
-            port_type = port.get('type', 'any')
-            if port_type not in self.VALID_PORT_TYPES:
-                errors.append(f'{path}.type has unsupported value "{port_type}"')
-
-        return errors
-
-    def _validate_config(self, config: Any) -> list[str]:
-        errors: list[str] = []
-        if not isinstance(config, list):
-            return ['config must be a list']
-
-        seen_names: set[str] = set()
-        for index, item in enumerate(config):
-            path = f'config[{index}]'
-            if not isinstance(item, dict):
-                errors.append(f'{path} must be a mapping')
-                continue
-
-            name = item.get('name')
-            if not isinstance(name, str) or not name:
-                errors.append(f'{path}.name must be a non-empty string')
-                continue
-
-            if name in seen_names:
-                errors.append(f'{path}.name duplicates "{name}"')
-            seen_names.add(name)
-
-            item_type = item.get('type', 'string')
-            if item_type not in self.VALID_CONFIG_TYPES:
-                errors.append(f'{path}.type has unsupported value "{item_type}"')
-
-            min_value = item.get('min_value')
-            max_value = item.get('max_value')
-            if isinstance(min_value, (int, float)) and isinstance(max_value, (int, float)) and min_value > max_value:
-                errors.append(f'{path}.min_value must be <= max_value')
-
-        return errors
-
-
-class NodeMetadataLoader:
-    """Load and cache workflow node metadata from YAML files."""
-
-    def __init__(self, validator: Optional[NodeMetadataValidator] = None) -> None:
-        self._validator = validator or NodeMetadataValidator()
-        self._metadata: dict[str, dict[str, Any]] = {}
-        self._sources: dict[str, str] = {}
-        self._load_errors: list[dict[str, str]] = []
-
-    async def load_core_metadata(self, resource_dir: str = 'metadata/nodes') -> int:
-        """Load all core node metadata from the ``langbot.templates`` package."""
-        return await self.load_package_directory('langbot.templates', resource_dir, source='core')
-
-    async def load_package_directory(self, package: str, resource_dir: str, source: str = 'core') -> int:
-        """Load YAML files from a package resource directory."""
-        try:
-            root = resources.files(package).joinpath(resource_dir)
-            yaml_files = sorted(
-                (item for item in root.iterdir() if item.is_file() and item.name.endswith(('.yaml', '.yml'))),
-                key=lambda item: item.name,
-            )
-        except Exception as exc:
-            raise MetadataLoadError(f'failed to scan package directory {package}:{resource_dir}: {exc}') from exc
-
-        return self._load_files(yaml_files, source=source)
-
-    async def load_directory(self, directory: str | Path, source: str) -> int:
-        """Load YAML files from an external filesystem directory, e.g. a plugin."""
-        directory_path = Path(directory)
-        if not directory_path.exists():
-            logger.warning('Workflow metadata directory does not exist: %s', directory_path)
-            return 0
-        if not directory_path.is_dir():
-            raise MetadataLoadError(f'workflow metadata path is not a directory: {directory_path}')
-
-        yaml_files = sorted(directory_path.glob('*.yml')) + sorted(directory_path.glob('*.yaml'))
-        return self._load_files(yaml_files, source=source)
-
-    def get_metadata(self, node_type: str) -> Optional[dict[str, Any]]:
-        """Return metadata by full type or short node name."""
-        if node_type in self._metadata:
-            return copy.deepcopy(self._metadata[node_type])
-
-        short_name = node_type.split('.')[-1]
-        for registered_type, metadata in self._metadata.items():
-            if registered_type.split('.')[-1] == short_name or metadata.get('name') == short_name:
-                return copy.deepcopy(metadata)
-
-        return None
-
-    def get_all_metadata(self) -> dict[str, dict[str, Any]]:
-        """Return a deep copy of all loaded metadata keyed by canonical node type."""
-        return copy.deepcopy(self._metadata)
-
-    def get_load_errors(self) -> list[dict[str, str]]:
-        """Return metadata files that failed to load or validate."""
-        return copy.deepcopy(self._load_errors)
-
-    def clear(self) -> None:
-        """Clear all cached metadata and errors."""
-        self._metadata.clear()
-        self._sources.clear()
-        self._load_errors.clear()
-
-    def _load_files(self, yaml_files: Iterable[Any], source: str) -> int:
-        count = 0
-        for yaml_file in yaml_files:
-            file_name = getattr(yaml_file, 'name', str(yaml_file))
-            try:
-                metadata = self._load_yaml(yaml_file)
-                self._validator.validate_or_raise(metadata)
-                node_type = build_node_type(metadata)
-
-                if node_type in self._metadata:
-                    existing_source = self._sources.get(node_type, 'unknown')
-                    if existing_source == 'core' and source != 'core':
-                        raise MetadataLoadError(
-                            f'plugin source "{source}" attempted to override core node "{node_type}"'
-                        )
-                    logger.warning(
-                        'Workflow node metadata %s from %s overrides previous source %s',
-                        node_type,
-                        source,
-                        existing_source,
-                    )
-
-                cached_metadata = copy.deepcopy(metadata)
-                cached_metadata['_source'] = source
-                cached_metadata['_file'] = file_name
-                self._metadata[node_type] = cached_metadata
-                self._sources[node_type] = source
-                count += 1
-            except Exception as exc:
-                self._load_errors.append({'file': file_name, 'source': source, 'error': str(exc)})
-                logger.error('Failed to load workflow node metadata %s: %s', file_name, exc)
-
-        return count
-
-    def _load_yaml(self, yaml_file: Any) -> dict[str, Any]:
-        try:
-            if hasattr(yaml_file, 'open'):
-                with yaml_file.open('r', encoding='utf-8') as file:
-                    data = yaml.load(file, Loader=yaml.FullLoader)
-            else:
-                with open(yaml_file, 'r', encoding='utf-8') as file:
-                    data = yaml.load(file, Loader=yaml.FullLoader)
-        except Exception as exc:
-            raise MetadataLoadError(f'failed to parse YAML: {exc}') from exc
-
-        if not isinstance(data, dict):
-            raise MetadataLoadError('YAML root must be a mapping')
-        return data
-
-
-def build_node_type(metadata: dict[str, Any]) -> str:
-    """Build canonical ``category.name`` node type from metadata."""
-    category = metadata.get('category') or 'misc'
-    name = metadata.get('name') or ''
-    return f'{category}.{name}'

src/langbot/pkg/workflow/monitor.py

@@ -1,61 +0,0 @@
-"""
-Monitoring helper for recording events during workflow execution.
-This module provides convenient methods to record monitoring data
-without cluttering the main workflow code.
-
-NOTE: All frontend panel logging functionality has been removed.
-A new solution will be implemented separately.
-"""
-
-from __future__ import annotations
-
-import typing
-import time
-
-if typing.TYPE_CHECKING:
-    from ..core import app
-    from langbot_plugin.api.entities.builtin.workflow.query import WorkflowQuery
-
-
-class WorkflowMonitoringHelper:
-    """Helper class for workflow monitoring operations"""
-
-    # All frontend panel logging methods have been removed.
-    # A new solution will be implemented separately.
-    pass
-
-
-class LLMCallMonitor:
-    """Context manager for monitoring LLM calls in workflow"""
-
-    def __init__(
-        self,
-        ap: app.Application,
-        query: WorkflowQuery,
-        bot_id: str,
-        bot_name: str,
-        workflow_id: str,
-        workflow_name: str,
-        node_name: str,
-        model_name: str,
-    ):
-        self.ap = ap
-        self.query = query
-        self.bot_id = bot_id
-        self.bot_name = bot_name
-        self.workflow_id = workflow_id
-        self.workflow_name = workflow_name
-        self.node_name = node_name
-        self.model_name = model_name
-        self.start_time = None
-        self.input_tokens = 0
-        self.output_tokens = 0
-
-    async def __aenter__(self):
-        self.start_time = time.time()
-        return self
-
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        # LLM call monitoring has been removed.
-        # A new solution will be implemented separately.
-        return False

src/langbot/pkg/workflow/monitoring_helper.py

@@ -1,363 +0,0 @@
-"""
-Monitoring helper for recording events during workflow execution.
-This module provides convenient methods to record monitoring data
-without cluttering the main workflow code.
-
-Logging scheme (aligned with pipeline monitoring):
-- Trigger log: stores original user message content directly
-- LLM call log: uses record_llm_call only (no additional message record)
-- LLM response log: stores response message content directly
-- Reply log: stores reply content directly
-
-Fields are extracted from WorkflowQuery object when available, with fallback to context_vars.
-"""
-
-from __future__ import annotations
-
-import typing
-import time
-import json
-
-if typing.TYPE_CHECKING:
-    from ..core import app
-
-
-class WorkflowMonitoringHelper:
-    """Helper class for workflow monitoring operations"""
-
-    @staticmethod
-    def _is_workflow_query(query) -> bool:
-        """Check if query is a WorkflowQuery object"""
-        if query is None or isinstance(query, str):
-            return False
-        # Check for WorkflowQuery attributes
-        return hasattr(query, 'launcher_type') or hasattr(query, 'workflow_uuid')
-
-    @staticmethod
-    def _get_session_id(query, context_vars: dict | None = None) -> str:
-        """Build session_id from query or context_vars"""
-        # Try to get from WorkflowQuery first
-        if WorkflowMonitoringHelper._is_workflow_query(query) and query.launcher_type:
-            launcher_type = query.launcher_type.value if hasattr(query.launcher_type, 'value') else str(query.launcher_type)
-            launcher_id = query.launcher_id or 'unknown'
-            return f'{launcher_type}_{launcher_id}'
-        
-        # Fallback to context_vars
-        if context_vars and context_vars.get('_launcher_type') and context_vars.get('_launcher_id'):
-            return f"{context_vars['_launcher_type']}_{context_vars['_launcher_id']}"
-        
-        return 'workflow_session'
-
-    @staticmethod
-    def _get_platform(query, context_vars: dict | None = None) -> str:
-        """Get platform name from query or context_vars"""
-        # Try WorkflowQuery first
-        if WorkflowMonitoringHelper._is_workflow_query(query) and query.launcher_type:
-            if hasattr(query.launcher_type, 'value'):
-                return query.launcher_type.value
-            return str(query.launcher_type)
-        
-        # Fallback to context_vars for launcher_type (person/group)
-        if context_vars and context_vars.get('_launcher_type'):
-            return context_vars['_launcher_type']
-        
-        return 'workflow'
-
-    @staticmethod
-    def _get_sender_name(query, context_vars: dict | None = None) -> str | None:
-        """Get sender name from query or context_vars"""
-        # Try WorkflowQuery first
-        if WorkflowMonitoringHelper._is_workflow_query(query):
-            if query.sender_name:
-                return query.sender_name
-            if query.message_event and hasattr(query.message_event, 'sender'):
-                sender = query.message_event.sender
-                if hasattr(sender, 'nickname'):
-                    return sender.nickname
-                if hasattr(sender, 'member_name'):
-                    return sender.member_name
-        
-        # Fallback to context_vars
-        if context_vars:
-            return context_vars.get('_sender_name')
-        
-        return None
-
-    @staticmethod
-    async def record_trigger_log(
-        ap: app.Application,
-        query,
-        workflow_id: str,
-        workflow_name: str,
-        bot_name: str = 'Workflow',
-        context_vars: dict | None = None,
-    ) -> str:
-        """Record trigger node log (stores original user message content directly)
-        
-        Aligned with pipeline monitoring: record_query_start
-        """
-        try:
-            session_id = WorkflowMonitoringHelper._get_session_id(query, context_vars)
-            platform = WorkflowMonitoringHelper._get_platform(query, context_vars)
-            sender_name = WorkflowMonitoringHelper._get_sender_name(query, context_vars)
-            
-            # Get message content - store original content directly
-            message_content = ''
-            if isinstance(query, str):
-                message_content = query
-            elif not isinstance(query, str) and query.message_context:
-                message_content = query.message_context.message_content
-            elif not isinstance(query, str) and query.message_chain and hasattr(query.message_chain, 'model_dump'):
-                message_content = json.dumps(query.message_chain.model_dump(), ensure_ascii=False)
-            elif not isinstance(query, str) and query.user_message:
-                message_content = str(query.user_message)
-            
-            # Get bot_id and user_id
-            bot_id = ''
-            user_id = None
-            if not isinstance(query, str):
-                bot_id = query.bot_uuid or ''
-                user_id = query.sender_id
-            elif context_vars:
-                bot_id = context_vars.get('_bot_id', '') or ''
-                user_id = context_vars.get('_user_id')
-            
-            message_id = await ap.monitoring_service.record_message(
-                bot_id=bot_id,
-                bot_name=bot_name,
-                pipeline_id=workflow_id,
-                pipeline_name=workflow_name or 'Workflow',
-                message_content=message_content,
-                session_id=session_id,
-                status='success',
-                level='info',
-                platform=platform,
-                user_id=user_id,
-                user_name=sender_name,
-                role='user',
-                runner_name='local-workflow',
-            )
-            
-            return message_id
-        except Exception as e:
-            ap.logger.error(f'Failed to record trigger log: {e}')
-            return ''
-
-    @staticmethod
-    async def record_llm_call_log(
-        ap: app.Application,
-        query,
-        workflow_id: str,
-        workflow_name: str,
-        node_name: str,
-        model_name: str,
-        input_tokens: int,
-        output_tokens: int,
-        duration_ms: int,
-        status: str = 'success',
-        error_message: str | None = None,
-        bot_name: str = 'Workflow',
-        context_vars: dict | None = None,
-        input_message: str | None = None,
-        message_id: str | None = None,
-    ):
-        """Record LLM call log with message_id association
-        
-        Aligned with pipeline monitoring: record_llm_call with message_id
-        LLM calls are aggregated under the trigger log via message_id.
-        """
-        try:
-            session_id = WorkflowMonitoringHelper._get_session_id(query, context_vars)
-            
-            # Get bot_id
-            bot_id = ''
-            if not isinstance(query, str):
-                bot_id = query.bot_uuid or ''
-            elif context_vars:
-                bot_id = context_vars.get('_bot_id', '') or ''
-            
-            # Record LLM call with message_id for association
-            await ap.monitoring_service.record_llm_call(
-                bot_id=bot_id,
-                bot_name=bot_name,
-                pipeline_id=workflow_id,
-                pipeline_name=workflow_name or 'Workflow',
-                session_id=session_id,
-                model_name=model_name,
-                input_tokens=input_tokens,
-                output_tokens=output_tokens,
-                duration=duration_ms,
-                status=status,
-                error_message=error_message,
-                message_id=message_id,
-            )
-        except Exception as e:
-            ap.logger.error(f'Failed to record LLM call log: {e}')
-
-    @staticmethod
-    async def record_llm_response_log(
-        ap: app.Application,
-        query,
-        workflow_id: str,
-        workflow_name: str,
-        node_name: str,
-        response_content: str,
-        bot_name: str = 'Workflow',
-        context_vars: dict | None = None,
-    ):
-        """Record LLM response log (stores response content directly)
-        
-        Aligned with pipeline monitoring: record_query_response
-        """
-        try:
-            session_id = WorkflowMonitoringHelper._get_session_id(query, context_vars)
-            platform = WorkflowMonitoringHelper._get_platform(query, context_vars)
-            sender_name = WorkflowMonitoringHelper._get_sender_name(query, context_vars)
-            
-            # Get bot_id and user_id
-            bot_id = ''
-            user_id = None
-            if not isinstance(query, str):
-                bot_id = query.bot_uuid or ''
-                user_id = query.sender_id
-            elif context_vars:
-                bot_id = context_vars.get('_bot_id', '') or ''
-                user_id = context_vars.get('_user_id')
-            
-            # Store response content directly, no prefix
-            await ap.monitoring_service.record_message(
-                bot_id=bot_id,
-                bot_name=bot_name,
-                pipeline_id=workflow_id,
-                pipeline_name=workflow_name or 'Workflow',
-                message_content=response_content[:2000],  # Limit length
-                session_id=session_id,
-                status='success',
-                level='info',
-                platform=platform,
-                user_id=user_id,
-                user_name=sender_name,
-                role='assistant',
-                runner_name='local-workflow',
-            )
-        except Exception as e:
-            ap.logger.error(f'Failed to record LLM response log: {e}')
-
-    @staticmethod
-    async def record_reply_log(
-        ap: app.Application,
-        query,
-        workflow_id: str,
-        workflow_name: str,
-        node_name: str,
-        reply_content: str,
-        bot_name: str = 'Workflow',
-        context_vars: dict | None = None,
-    ):
-        """Record reply message log (stores reply content directly)
-        
-        Aligned with pipeline monitoring: record_query_response
-        """
-        try:
-            session_id = WorkflowMonitoringHelper._get_session_id(query, context_vars)
-            platform = WorkflowMonitoringHelper._get_platform(query, context_vars)
-            sender_name = WorkflowMonitoringHelper._get_sender_name(query, context_vars)
-            
-            # Get bot_id and user_id
-            bot_id = ''
-            user_id = None
-            if not isinstance(query, str):
-                bot_id = query.bot_uuid or ''
-                user_id = query.sender_id
-            elif context_vars:
-                bot_id = context_vars.get('_bot_id', '') or ''
-                user_id = context_vars.get('_user_id')
-            
-            # Store reply content directly, no prefix
-            await ap.monitoring_service.record_message(
-                bot_id=bot_id,
-                bot_name=bot_name,
-                pipeline_id=workflow_id,
-                pipeline_name=workflow_name or 'Workflow',
-                message_content=reply_content[:2000],  # Limit length
-                session_id=session_id,
-                status='success',
-                level='info',
-                platform=platform,
-                user_id=user_id,
-                user_name=sender_name,
-                role='assistant',
-                runner_name='local-workflow',
-            )
-        except Exception as e:
-            ap.logger.error(f'Failed to record reply log: {e}')
-
-
-class LLMCallMonitor:
-    """Context manager for monitoring LLM calls in workflow"""
-
-    def __init__(
-        self,
-        ap: app.Application,
-        query,
-        bot_id: str,
-        bot_name: str,
-        workflow_id: str,
-        workflow_name: str,
-        node_name: str,
-        model_name: str,
-        context_vars: dict | None = None,
-    ):
-        self.ap = ap
-        self.query = query
-        self.bot_id = bot_id
-        self.bot_name = bot_name
-        self.workflow_id = workflow_id
-        self.workflow_name = workflow_name
-        self.node_name = node_name
-        self.model_name = model_name
-        self.context_vars = context_vars
-        self.start_time = None
-        self.input_tokens = 0
-        self.output_tokens = 0
-
-    async def __aenter__(self):
-        self.start_time = time.time()
-        return self
-
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        duration_ms = int((time.time() - self.start_time) * 1000) if self.start_time else 0
-
-        if exc_type is not None:
-            await WorkflowMonitoringHelper.record_llm_call_log(
-                ap=self.ap,
-                query=self.query,
-                workflow_id=self.workflow_id,
-                workflow_name=self.workflow_name,
-                node_name=self.node_name,
-                model_name=self.model_name,
-                input_tokens=self.input_tokens,
-                output_tokens=self.output_tokens,
-                duration_ms=duration_ms,
-                status='error',
-                error_message=str(exc_val) if exc_val else None,
-                bot_name=self.bot_name,
-                context_vars=self.context_vars,
-            )
-        else:
-            await WorkflowMonitoringHelper.record_llm_call_log(
-                ap=self.ap,
-                query=self.query,
-                workflow_id=self.workflow_id,
-                workflow_name=self.workflow_name,
-                node_name=self.node_name,
-                model_name=self.model_name,
-                input_tokens=self.input_tokens,
-                output_tokens=self.output_tokens,
-                duration_ms=duration_ms,
-                status='success',
-                bot_name=self.bot_name,
-                context_vars=self.context_vars,
-            )
-
-        return False

src/langbot/pkg/workflow/node.py

@@ -1,164 +0,0 @@
-"""Workflow node base class and decorators"""
-
-from __future__ import annotations
-
-import abc
-from typing import Any, Callable, Optional, TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from .entities import ExecutionContext
-    from ..core import app
-
-
-class WorkflowNode(abc.ABC):
-    """Base class for all workflow nodes.
-
-    Node metadata (inputs, outputs, config schema, label, icon, etc.) is
-    defined exclusively in YAML files under templates/metadata/nodes/.
-    Python subclasses only provide execution logic and runtime behaviour.
-    """
-
-    # Set by @workflow_node decorator
-    type_name: str = ''
-
-    # Category is kept as a fallback for registry when YAML is missing
-    category: str = 'misc'
-
-    # Pipeline config reuse (referenced by registry merge logic)
-    config_schema_source: Optional[str] = None
-    config_stages: list[str] = []
-
-    def __init__(self, node_id: str, config: dict[str, Any], ap: Optional['app.Application'] = None):
-        """Initialize node with ID and configuration"""
-        self.node_id = node_id
-        self.config = config
-        self.ap = ap
-
-    @abc.abstractmethod
-    async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        """Execute the node logic.
-
-        Args:
-            inputs: Input data from connected nodes
-            context: Execution context with workflow state
-
-        Returns:
-            Dictionary of output values
-        """
-        pass
-
-    # ------------------------------------------------------------------
-    # Validation helpers — metadata is resolved from the registry at
-    # runtime so that YAML remains the single source of truth.
-    # ------------------------------------------------------------------
-
-    async def validate_inputs(self, inputs: dict[str, Any]) -> list[str]:
-        """Validate input data against YAML port definitions.
-
-        Returns:
-            List of validation error messages (empty if valid)
-        """
-        metadata = self._get_metadata()
-        if metadata is None:
-            return []
-
-        errors: list[str] = []
-        for port in metadata.get('inputs', []):
-            if port.get('required', True) and port.get('name') and port['name'] not in inputs:
-                errors.append(f"Missing required input: {port['name']}")
-        return errors
-
-    async def validate_config(self) -> list[str]:
-        """Validate node configuration against YAML config schema.
-
-        Returns:
-            List of validation error messages (empty if valid)
-        """
-        metadata = self._get_metadata()
-        if metadata is None:
-            return []
-
-        errors: list[str] = []
-        for cfg in metadata.get('config', []):
-            name = cfg.get('name', '')
-            if not name:
-                continue
-            required = cfg.get('required', False)
-            cfg_type = cfg.get('type', 'string')
-
-            if required and name not in self.config:
-                errors.append(f'Missing required config: {name}')
-            elif name in self.config:
-                value = self.config[name]
-                # Type validation
-                if cfg_type == 'integer' and not isinstance(value, int):
-                    errors.append(f'Config {name} must be an integer')
-                elif cfg_type == 'number' and not isinstance(value, (int, float)):
-                    errors.append(f'Config {name} must be a number')
-                elif cfg_type == 'boolean' and not isinstance(value, bool):
-                    errors.append(f'Config {name} must be a boolean')
-                # Range validation
-                min_val = cfg.get('min_value')
-                max_val = cfg.get('max_value')
-                if min_val is not None and isinstance(value, (int, float)):
-                    if value < min_val:
-                        errors.append(f'Config {name} must be >= {min_val}')
-                if max_val is not None and isinstance(value, (int, float)):
-                    if value > max_val:
-                        errors.append(f'Config {name} must be <= {max_val}')
-        return errors
-
-    def get_config(self, key: str, default: Any = None) -> Any:
-        """Get configuration value with default"""
-        return self.config.get(key, default)
-
-    def _get_metadata(self) -> Optional[dict[str, Any]]:
-        """Retrieve YAML metadata for this node from the registry."""
-        from .registry import NodeTypeRegistry
-        registry = NodeTypeRegistry.instance()
-        return registry.get_metadata(self.type_name)
-
-    @classmethod
-    def to_schema(cls) -> dict[str, Any]:
-        """Return a schema dict for this node type.
-
-        This is used by tests and tooling to inspect node capabilities.
-        """
-        from .registry import NodeTypeRegistry
-        registry = NodeTypeRegistry.instance()
-        metadata = registry.get_metadata(cls.type_name)
-        if metadata:
-            return registry._metadata_to_schema(metadata)
-        # Fallback: build a minimal schema from class attributes
-        return {
-            'type': f'{cls.category}.{cls.type_name}' if cls.type_name else cls.type_name,
-            'category': cls.category,
-            'label': getattr(cls, 'name', cls.type_name),
-            'description': getattr(cls, 'description', ''),
-            'inputs': [],
-            'outputs': [],
-            'config_schema': [],
-        }
-
-
-# ------------------------------------------------------------------
-# Decorator for setting type_name attribute
-# ------------------------------------------------------------------
-
-
-def workflow_node(type_name: str) -> Callable[[type[WorkflowNode]], type[WorkflowNode]]:
-    """Decorator to set the type_name attribute on a workflow node class.
-
-    Usage:
-        @workflow_node('llm_call')
-        class LLMCallNode(WorkflowNode):
-            ...
-
-    The actual registration is now handled by the discovery engine.
-    """
-
-    def decorator(cls: type[WorkflowNode]) -> type[WorkflowNode]:
-        cls.type_name = type_name
-        return cls
-
-    return decorator

src/langbot/pkg/workflow/nodes/__init__.py

@@ -1 +0,0 @@
-

src/langbot/pkg/workflow/nodes/call_pipeline.py

@@ -1,263 +0,0 @@
-"""Call Pipeline Node - invoke an existing pipeline
-
-Node metadata is loaded from: ../../templates/metadata/nodes/call_pipeline.yaml
-"""
-
-from __future__ import annotations
-
-from typing import Any, Optional
-
-import pydantic
-
-import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter
-import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_event_logger
-import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
-import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
-import langbot_plugin.api.entities.builtin.platform.events as platform_events
-import langbot_plugin.api.entities.builtin.platform.message as platform_message
-import langbot_plugin.api.entities.builtin.provider.session as provider_session
-
-from langbot_plugin.api.entities.builtin.workflow.entities import ExecutionContext
-from ..node import WorkflowNode, workflow_node
-
-
-class _NoOpEventLogger(abstract_event_logger.AbstractEventLogger):
-    """No-op event logger for workflow pipeline adapter."""
-
-    async def info(
-        self,
-        text: str,
-        images: Optional[list[platform_message.Image]] = None,
-        message_session_id: Optional[str] = None,
-        no_throw: bool = True,
-    ):
-        pass
-
-    async def debug(
-        self,
-        text: str,
-        images: Optional[list[platform_message.Image]] = None,
-        message_session_id: Optional[str] = None,
-        no_throw: bool = True,
-    ):
-        pass
-
-    async def warning(
-        self,
-        text: str,
-        images: Optional[list[platform_message.Image]] = None,
-        message_session_id: Optional[str] = None,
-        no_throw: bool = True,
-    ):
-        pass
-
-    async def error(
-        self,
-        text: str,
-        images: Optional[list[platform_message.Image]] = None,
-        message_session_id: Optional[str] = None,
-        no_throw: bool = True,
-    ):
-        pass
-
-@workflow_node('call_pipeline')
-class CallPipelineNode(WorkflowNode):
-    """Call pipeline node - invoke an existing pipeline"""
-
-    category = 'action'
-
-    async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        if not self.ap:
-            raise RuntimeError('Application instance not available — cannot call pipeline')
-
-        raw_query = inputs.get('query', '')
-        query_text = str(raw_query or inputs.get('input') or '')
-        pipeline_ref = str(self.get_config('pipeline_uuid', '') or '').strip()
-
-        if not pipeline_ref:
-            raise ValueError('No pipeline configured for call pipeline node')
-
-        pipeline_data = await self.ap.pipeline_service.get_pipeline(pipeline_ref)
-        if pipeline_data is None:
-            pipeline_data = await self.ap.pipeline_service.get_pipeline_by_name(pipeline_ref)
-        if pipeline_data is None:
-            raise ValueError(f'Pipeline not found: {pipeline_ref}')
-
-        pipeline_uuid = str(pipeline_data.get('uuid', '') or '')
-        if not pipeline_uuid:
-            raise ValueError(f'Pipeline UUID missing for: {pipeline_ref}')
-
-        runtime_pipeline = await self.ap.pipeline_mgr.get_pipeline_by_uuid(pipeline_uuid)
-        if runtime_pipeline is None:
-            raise ValueError(f'Runtime pipeline not loaded: {pipeline_uuid}')
-
-        adapter = _WorkflowPipelineCaptureAdapter(context=context)
-        adapter.bot_account_id = 'workflow-call-pipeline'
-
-        message_event = self._build_message_event(query_text, context)
-        message_chain = message_event.message_chain
-        launcher_type = (
-            provider_session.LauncherTypes.GROUP
-            if context.message_context and context.message_context.is_group
-            else provider_session.LauncherTypes.PERSON
-        )
-        launcher_id = context.session_id or context.execution_id
-        sender_id = (
-            context.message_context.sender_id
-            if context.message_context and context.message_context.sender_id
-            else context.user_id or f'workflow_{context.execution_id}'
-        )
-
-        query = pipeline_query.Query(
-            bot_uuid=context.bot_id,
-            query_id=-1,
-            launcher_type=launcher_type,
-            launcher_id=launcher_id,
-            sender_id=sender_id,
-            message_event=message_event,
-            message_chain=message_chain,
-            variables={
-                '_called_from_workflow': True,
-                '_workflow_execution_id': context.execution_id,
-                '_workflow_id': context.workflow_id,
-                **dict(context.variables or {}),
-            },
-            resp_messages=[],
-            resp_message_chain=[],
-            adapter=adapter,
-            pipeline_uuid=pipeline_uuid,
-        )
-
-        await runtime_pipeline.run(query)
-
-        response_text = adapter.get_last_text_response()
-        result = {
-            'pipeline_uuid': pipeline_uuid,
-            'pipeline_name': pipeline_data.get('name', ''),
-            'responses': adapter.responses,
-            'query_text': query_text,
-        }
-
-        return {'response': response_text, 'result': result}
-
-    def _build_message_event(
-        self,
-        query_text: str,
-        context: ExecutionContext,
-    ) -> platform_events.MessageEvent:
-        message_chain_data = context.trigger_data.get('message_chain') or context.trigger_data.get('message', [])
-        if isinstance(message_chain_data, list) and message_chain_data:
-            message_chain = platform_message.MessageChain.model_validate(message_chain_data)
-        else:
-            message_chain = platform_message.MessageChain([platform_message.Plain(text=query_text)])
-
-        if context.message_context and context.message_context.is_group:
-            group = platform_entities.Group(
-                id=context.message_context.group_id or context.session_id or 'workflow_group',
-                name=context.message_context.raw_message.get('group_name', 'Workflow Group') if context.message_context.raw_message else 'Workflow Group',
-                permission=platform_entities.Permission.Member,
-            )
-            sender = platform_entities.GroupMember(
-                id=context.message_context.sender_id,
-                member_name=context.message_context.sender_name or 'Workflow User',
-                permission=platform_entities.Permission.Member,
-                group=group,
-            )
-            return platform_events.GroupMessage(
-                sender=sender,
-                message_chain=message_chain,
-                time=context.message_context.raw_message.get('time') if context.message_context.raw_message else None,
-            )
-
-        sender = platform_entities.Friend(
-            id=context.message_context.sender_id if context.message_context else context.user_id or 'workflow_user',
-            nickname=context.message_context.sender_name if context.message_context else 'Workflow User',
-            remark=context.message_context.sender_name if context.message_context else 'Workflow User',
-        )
-        return platform_events.FriendMessage(
-            sender=sender,
-            message_chain=message_chain,
-            time=context.message_context.raw_message.get('time')
-            if context.message_context and context.message_context.raw_message
-            else None,
-        )
-
-class _WorkflowPipelineCaptureAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
-    """Adapter to capture pipeline responses for workflow execution."""
-
-    class Config:
-        arbitrary_types_allowed = True
-
-    responses: list[dict[str, Any]] = []
-    context: Optional[ExecutionContext] = pydantic.Field(default=None, exclude=True)
-
-    def __init__(self, context: ExecutionContext):
-        super().__init__(config={}, logger=_NoOpEventLogger(), context=context)
-        self.responses = []
-
-    async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
-        payload = {
-            'type': 'send',
-            'target_type': target_type,
-            'target_id': target_id,
-            'content': str(message),
-            'message_chain': message.model_dump(),
-        }
-        self.responses.append(payload)
-        return payload
-
-    async def reply_message(
-        self,
-        message_source: platform_events.MessageEvent,
-        message: platform_message.MessageChain,
-        quote_origin: bool = False,
-    ):
-        payload = {
-            'type': 'reply',
-            'content': str(message),
-            'message_chain': message.model_dump(),
-            'quote_origin': quote_origin,
-        }
-        self.responses.append(payload)
-        return payload
-
-    async def reply_message_chunk(
-        self,
-        message_source: platform_events.MessageEvent,
-        bot_message: dict,
-        message: platform_message.MessageChain,
-        quote_origin: bool = False,
-        is_final: bool = False,
-    ):
-        payload = {
-            'type': 'reply_chunk',
-            'content': str(message),
-            'message_chain': message.model_dump(),
-            'quote_origin': quote_origin,
-            'is_final': is_final,
-        }
-        self.responses.append(payload)
-        return payload
-
-    async def create_message_card(self, message_id, event: platform_events.MessageEvent) -> bool:
-        return False
-
-    def register_listener(self, event_type, callback):
-        return None
-
-    def unregister_listener(self, event_type, callback):
-        return None
-
-    async def run_async(self):
-        return None
-
-    async def is_stream_output_supported(self) -> bool:
-        return False
-
-    async def kill(self) -> bool:
-        return True
-
-    def get_last_text_response(self) -> str:
-        if not self.responses:
-            return ''
-        return str(self.responses[-1].get('content', '') or '')

src/langbot/pkg/workflow/nodes/call_workflow.py

@@ -1,85 +0,0 @@
-"""Call Workflow Node - invoke an existing workflow
-
-Node metadata is loaded from: ../../templates/metadata/nodes/call_workflow.yaml
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-from langbot_plugin.api.entities.builtin.workflow.entities import ExecutionContext
-
-from ..node import WorkflowNode, workflow_node
-
-
-@workflow_node('call_workflow')
-class CallWorkflowNode(WorkflowNode):
-    """Call workflow node - invoke an existing workflow"""
-
-    category = 'action'
-
-    async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        if not self.ap:
-            raise RuntimeError('Application instance not available — cannot call workflow')
-
-        # Get workflow reference from config
-        workflow_ref = str(self.get_config('workflow_uuid', '') or '').strip()
-        if not workflow_ref:
-            raise ValueError('No workflow configured for call workflow node')
-
-        # Get workflow definition from service
-        workflow_data = await self.ap.workflow_service.get_workflow(workflow_ref)
-        if workflow_data is None:
-            raise ValueError(f'Workflow not found: {workflow_ref}')
-
-        workflow_uuid = str(workflow_data.get('uuid', '') or '')
-        if not workflow_uuid:
-            raise ValueError(f'Workflow UUID missing for: {workflow_ref}')
-
-        # Build variables to pass to the called workflow
-        variables = dict(inputs.get('variables', {}) or {})
-
-        # Inherit current workflow variables if configured
-        if self.get_config('inherit_variables', True):
-            for key, value in (context.variables or {}).items():
-                if key not in variables:
-                    variables[key] = value
-
-        # Add context markers for debugging
-        variables['_called_from_workflow'] = True
-        variables['_parent_workflow_id'] = context.workflow_id
-        variables['_parent_execution_id'] = context.execution_id
-
-        # Execute the workflow
-        execution_id = await self.ap.workflow_service.execute_workflow(
-            workflow_uuid=workflow_uuid,
-            trigger_type='workflow_call',
-            trigger_data={
-                'variables': variables,
-                'parent_execution_id': context.execution_id,
-            },
-            session_id=context.session_id,
-            user_id=context.user_id,
-            bot_id=context.bot_id,
-        )
-
-        # Get execution result
-        execution = await self.ap.workflow_service.get_execution(execution_id)
-        if execution is None:
-            raise ValueError(f'Execution result not found: {execution_id}')
-
-        # Build result
-        result = {
-            'workflow_uuid': workflow_uuid,
-            'workflow_name': workflow_data.get('name', ''),
-            'execution_id': execution_id,
-            'status': execution.get('status', 'unknown'),
-            'variables': execution.get('variables', {}),
-            'error': execution.get('error'),
-        }
-
-        return {
-            'result': result,
-            'status': execution.get('status', 'unknown'),
-            'error': execution.get('error'),
-        }

src/langbot/pkg/workflow/nodes/code_executor.py

@@ -1,156 +0,0 @@
-"""Code Executor Node - run Python or JavaScript code
-
-Node metadata is loaded from: ../../templates/metadata/nodes/code_executor.yaml
-"""
-
-from __future__ import annotations
-
-import ast
-import io
-import logging
-import sys
-import threading
-from typing import Any
-
-from langbot_plugin.api.entities.builtin.workflow.entities import ExecutionContext
-from ..node import WorkflowNode, workflow_node
-
-logger = logging.getLogger(__name__)
-
-# 危险的内置函数和模块黑名单
-_DANGEROUS_BUILTINS = {
-    '__import__', 'eval', 'exec', 'compile', 'open', 'file',
-    'input', 'exit', 'quit', 'globals', 'locals', 'vars',
-    'dir', 'help', 'breakpoint',
-}
-
-# 允许的安全内置函数
-_SAFE_BUILTINS = {
-    'abs': abs, 'all': all, 'any': any, 'bin': bin, 'bool': bool,
-    'bytearray': bytearray, 'bytes': bytes, 'callable': callable,
-    'chr': chr, 'complex': complex, 'dict': dict, 'divmod': divmod,
-    'enumerate': enumerate, 'filter': filter, 'float': float,
-    'format': format, 'frozenset': frozenset, 'hash': hash,
-    'hex': hex, 'int': int, 'isinstance': isinstance, 'issubclass': issubclass,
-    'iter': iter, 'len': len, 'list': list, 'map': map, 'max': max,
-    'min': min, 'next': next, 'object': object, 'oct': oct, 'ord': ord,
-    'pow': pow, 'print': print, 'range': range, 'repr': repr,
-    'reversed': reversed, 'round': round, 'set': set, 'slice': slice,
-    'sorted': sorted, 'str': str, 'sum': sum, 'tuple': tuple,
-    'type': type, 'zip': zip,
-}
-
-
-def _check_code_safety(code: str) -> list[str]:
-    """检查代码中是否包含危险操作"""
-    warnings = []
-    try:
-        tree = ast.parse(code)
-        for node in ast.walk(tree):
-            # 检查 import 语句
-            if isinstance(node, (ast.Import, ast.ImportFrom)):
-                warnings.append('Import statements are not allowed')
-            # 检查危险函数调用
-            if isinstance(node, ast.Call):
-                if isinstance(node.func, ast.Name) and node.func.id in _DANGEROUS_BUILTINS:
-                    warnings.append(f'Dangerous function call: {node.func.id}')
-                # 检查 __import__ 通过 getattr 调用
-                if isinstance(node.func, ast.Attribute):
-                    if node.func.attr in ('__import__', 'eval', 'exec', 'open', 'file'):
-                        warnings.append(f'Dangerous attribute access: {node.func.attr}')
-    except SyntaxError as e:
-        warnings.append(f'Syntax error in code: {e}')
-    return warnings
-
-
-class _ExecutionTimeoutError(Exception):
-    """执行超时错误"""
-    pass
-
-
-def _run_with_timeout(func, timeout: float = 10.0):
-    """带超时限制的函数执行"""
-    result = [None]
-    error = [None]
-
-    def _target():
-        try:
-            result[0] = func()
-        except Exception as e:
-            error[0] = e
-
-    thread = threading.Thread(target=_target)
-    thread.daemon = True
-    thread.start()
-    thread.join(timeout)
-
-    if thread.is_alive():
-        raise _ExecutionTimeoutError(f'Code execution timed out after {timeout} seconds')
-
-    if error[0]:
-        raise error[0]
-
-    return result[0]
-
-
-@workflow_node('code_executor')
-class CodeExecutorNode(WorkflowNode):
-    """Code executor node - run Python or JavaScript code"""
-
-    category = 'process'
-
-    async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        code = self.get_config('code', '')
-        language = self.get_config('language', 'python')
-        timeout = self.get_config('timeout', 10)
-
-        # 限制最大超时时间
-        timeout = min(max(timeout, 1), 30)
-
-        if not code:
-            return {'output': None, 'console': '', 'error': 'No code provided'}
-
-        if language == 'python':
-            return await self._execute_python(code, inputs, context, timeout)
-        else:
-            return await self._execute_javascript(code, inputs, context)
-
-    async def _execute_python(self, code: str, inputs: dict[str, Any], context: ExecutionContext, timeout: float) -> dict[str, Any]:
-        # 安全检查
-        warnings = _check_code_safety(code)
-        if warnings:
-            logger.warning('Code safety warnings: %s', warnings)
-            return {'output': None, 'console': '', 'error': '; '.join(warnings)}
-
-        stdout_capture = io.StringIO()
-        old_stdout = sys.stdout
-
-        def _exec_code():
-            nonlocal stdout_capture
-            sys.stdout = stdout_capture
-            try:
-                # 使用更安全的执行方式
-                compiled = compile(code, '<workflow>', 'exec')
-                safe_globals = {
-                    '__builtins__': _SAFE_BUILTINS,
-                    '__name__': '__workflow_sandbox__',
-                }
-                local_vars = {'inputs': inputs, 'output': None}
-                exec(compiled, safe_globals, local_vars)
-                return local_vars.get('output')
-            finally:
-                sys.stdout = old_stdout
-
-        try:
-            output = _run_with_timeout(_exec_code, timeout)
-            console_output = stdout_capture.getvalue()
-            return {'output': output, 'console': console_output, 'error': None}
-        except _ExecutionTimeoutError as e:
-            logger.error('Code execution timeout: %s', e)
-            return {'output': None, 'console': stdout_capture.getvalue(), 'error': str(e)}
-        except Exception as e:
-            logger.error('Code execution error: %s', e)
-            return {'output': None, 'console': stdout_capture.getvalue(), 'error': f'{type(e).__name__}: {e}'}
-
-    async def _execute_javascript(self, code: str, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        return {'output': None, 'console': '', 'error': 'JavaScript execution is not implemented'}

src/langbot/pkg/workflow/nodes/condition.py

@@ -1,125 +0,0 @@
-"""Condition Node - branch based on condition
-
-Node metadata is loaded from: ../../templates/metadata/nodes/condition.yaml
-"""
-
-from __future__ import annotations
-
-import logging
-import re
-import signal
-from typing import Any
-
-from langbot_plugin.api.entities.builtin.workflow.entities import ExecutionContext
-from ..node import WorkflowNode, workflow_node
-from ..safe_eval import safe_eval_with_vars
-
-logger = logging.getLogger(__name__)
-
-# 正则表达式超时限制（秒）
-_REGEX_TIMEOUT = 2
-
-
-class _RegexTimeoutError(Exception):
-    """正则表达式超时错误"""
-    pass
-
-
-def _handle_timeout(signum, frame):
-    """超时信号处理"""
-    raise _RegexTimeoutError('Regex match timed out')
-
-
-def _safe_regex_match(pattern: str, text: str) -> tuple[bool, str]:
-    """安全地执行正则表达式匹配，带有超时限制"""
-    # 设置超时信号
-    old_handler = signal.signal(signal.SIGALRM, _handle_timeout)
-    signal.setitimer(signal.ITIMER_REAL, _REGEX_TIMEOUT)
-
-    try:
-        result = bool(re.match(pattern, str(text)))
-        return result, ''
-    except _RegexTimeoutError:
-        logger.warning('Regex match timed out for pattern: %s', pattern[:50])
-        return False, 'Regex match timed out'
-    except re.error as e:
-        logger.warning('Invalid regex pattern: %s', e)
-        return False, f'Invalid regex: {e}'
-    finally:
-        signal.setitimer(signal.ITIMER_REAL, 0)
-        signal.signal(signal.SIGALRM, old_handler)
-
-
-@workflow_node('condition')
-class ConditionNode(WorkflowNode):
-    """Condition node - branch based on condition"""
-
-    category = 'control'
-
-    async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        condition_type = self.get_config('condition_type', 'expression')
-        input_data = inputs.get('input')
-
-        result = False
-
-        if condition_type == 'expression':
-            expression = self.get_config('expression', 'false')
-            result = await self._evaluate_expression(expression, input_data, context)
-        elif condition_type == 'comparison':
-            result = await self._evaluate_comparison(input_data, context)
-        elif condition_type == 'contains':
-            left = self.get_config('left_value', '')
-            right = self.get_config('right_value', '')
-            result = right in left
-        elif condition_type == 'empty':
-            result = not bool(input_data)
-        elif condition_type == 'regex':
-            left = self.get_config('left_value', '')
-            pattern = self.get_config('right_value', '')
-            result, error = _safe_regex_match(pattern, left)
-            if error:
-                return {'true': None, 'false': input_data, 'error': error}
-
-        if result:
-            return {'true': input_data, 'false': None}
-        else:
-            return {'true': None, 'false': input_data}
-
-    async def _evaluate_expression(self, expression: str, data: Any, context: ExecutionContext) -> bool:
-        try:
-            local_vars = {'input': data, 'data': data, 'variables': context.variables}
-            return bool(safe_eval_with_vars(expression, local_vars))
-        except Exception as e:
-            logger.warning('Expression evaluation error: %s', e)
-            return False
-
-    async def _evaluate_comparison(self, data: Any, context: ExecutionContext) -> bool:
-        left = self.get_config('left_value', '')
-        right = self.get_config('right_value', '')
-        operator = self.get_config('operator', '==')
-
-        try:
-            left_num = float(left)
-            right_num = float(right)
-
-            if operator == '==':
-                return left_num == right_num
-            elif operator == '!=':
-                return left_num != right_num
-            elif operator == '>':
-                return left_num > right_num
-            elif operator == '<':
-                return left_num < right_num
-            elif operator == '>=':
-                return left_num >= right_num
-            elif operator == '<=':
-                return left_num <= right_num
-        except ValueError:
-            if operator == '==':
-                return left == right
-            elif operator == '!=':
-                return left != right
-            elif operator in ('>', '<', '>=', '<='):
-                return False
-
-        return False

src/langbot/pkg/workflow/nodes/coze_bot.py

@@ -1,39 +0,0 @@
-"""Coze Bot Node - call Coze API bot
-
-Node metadata is loaded from: ../../templates/metadata/nodes/coze_bot.yaml
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-from langbot_plugin.api.entities.builtin.workflow.entities import ExecutionContext
-from ..node import WorkflowNode, workflow_node
-
-@workflow_node('coze_bot')
-class CozeBotNode(WorkflowNode):
-    """Coze bot node - call Coze API bot"""
-
-    category = 'integration'
-
-    async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        api_key = self.get_config('api_key', '')
-        bot_id = self.get_config('bot_id', '')
-        api_base = self.get_config('api_base', 'https://api.coze.cn')
-        query = inputs.get('query', '')
-        conversation_id = inputs.get('conversation_id')
-
-        # Safe API key truncation
-        masked_key = f'{api_key[:4]}...{api_key[-4:]}' if len(api_key) > 8 else '***' if api_key else ''
-
-        return {
-            'answer': '',
-            'conversation_id': conversation_id,
-            'success': False,
-            '_debug': {
-                'api_key': masked_key,
-                'bot_id': bot_id,
-                'api_base': api_base,
-                'query': query,
-            },
-        }

src/langbot/pkg/workflow/nodes/cron_trigger.py

@@ -1,26 +0,0 @@
-"""Cron Trigger Node - triggers workflow on schedule
-
-Node metadata is loaded from: ../../templates/metadata/nodes/cron_trigger.yaml
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-from langbot_plugin.api.entities.builtin.workflow.entities import ExecutionContext
-from ..node import WorkflowNode, workflow_node
-
-@workflow_node('cron_trigger')
-class CronTriggerNode(WorkflowNode):
-    """Cron trigger node - triggers workflow on schedule"""
-
-    category = 'trigger'
-
-    async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        from datetime import datetime
-
-        return {
-            'timestamp': datetime.now().isoformat(),
-            'schedule': self.get_config('cron', ''),
-            'context': context.trigger_data,
-        }

src/langbot/pkg/workflow/nodes/data_transform.py

@@ -1,68 +0,0 @@
-"""Data Transform Node - transform data using templates or JSONPath
-
-Node metadata is loaded from: ../../templates/metadata/nodes/data_transform.yaml
-"""
-
-from __future__ import annotations
-
-from typing import Any
-
-from langbot_plugin.api.entities.builtin.workflow.entities import ExecutionContext
-from ..node import WorkflowNode, workflow_node
-from ..safe_eval import safe_eval_with_vars
-
-@workflow_node('data_transform')
-class DataTransformNode(WorkflowNode):
-    """Data transform node - transform data using templates or JSONPath"""
-
-    category = 'process'
-
-    async def execute(self, inputs: dict[str, Any], context: ExecutionContext) -> dict[str, Any]:
-        data = inputs.get('data')
-        transform_type = self.get_config('transform_type', 'template')
-
-        if transform_type == 'template':
-            template = self.get_config('template', '')
-            result = self._apply_template(template, data, context)
-        elif transform_type == 'jsonpath':
-            expression = self.get_config('expression', '$')
-            result = self._apply_jsonpath(expression, data)
-        elif transform_type == 'expression':
-            expression = self.get_config('expression', '')
-            result = self._evaluate_expression(expression, data, context)
-        else:
-            result = data
-
-        return {'result': result}
-
-    def _apply_template(self, template: str, data: Any, context: ExecutionContext) -> str:
-        result = template
-        if isinstance(data, dict):
-            for key, value in data.items():
-                result = result.replace(f'{{{{data.{key}}}}}', str(value))
-        for key, value in context.variables.items():
-            result = result.replace(f'{{{{variables.{key}}}}}', str(value))
-        return result
-
-    def _apply_jsonpath(self, expression: str, data: Any) -> Any:
-        if expression == '$':
-            return data
-        if expression.startswith('$.'):
-            parts = expression[2:].split('.')
-            result = data
-            for part in parts:
-                if isinstance(result, dict):
-                    result = result.get(part)
-                elif isinstance(result, list) and part.isdigit():
-                    result = result[int(part)]
-                else:
-                    return None
-            return result
-        return data
-
-    def _evaluate_expression(self, expression: str, data: Any, context: ExecutionContext) -> Any:
-        local_vars = {'data': data, 'variables': context.variables}
-        try:
-            return safe_eval_with_vars(expression, local_vars)
-        except Exception:
-            return None