fix(mcp): use uniform session memory_mb=1024 to avoid BoxSessionConflictError

All MCPs share one Box session (mcp-shared). When session memory_mb differed by command type (512 for python, 1024 for node), the second MCP to call create_session raised BoxSessionConflictError. Fix: always use 1024 MB for the shared session so python and node MCPs coexist without conflict.
fix(deps): pin langbot-plugin 0.4.12 (BoxManagedProcessSpec.memory_mb fix) (#2315 )
2026-07-16 09:26:07 +00:00 · 2026-07-03 21:46:46 -04:00 · 2026-07-04 09:31:09 +08:00 · 2026-07-04 09:03:43 +08:00 · 2026-07-04 08:31:08 +08:00 · 2026-07-03 20:23:12 +08:00
142 changed files with 10599 additions and 1195 deletions
@@ -5,7 +5,7 @@

 <div align="center">

-<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>
+<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>

 <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.302.ai/SuTG99)                                                                             | Gateway      | ✅     |
+| [302.AI](https://share.302ai.cn/SuTG99)                                                                             | Gateway      | ✅     |
 | [Qiniu](https://www.qiniu.com/ai/agent)                                                                           | Gateway      | ✅     |

 [→ View all integrations](https://link.langbot.app/en/docs/features)
@@ -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.302.ai/SuTG99) | 聚合平台 | ✅ |
+| [302.AI](https://share.302ai.cn/SuTG99) | 聚合平台 | ✅ |
 | [小马算力](https://www.tokenpony.cn/453z1) | 聚合平台 | ✅ |
 | [百宝箱Tbox](https://www.tbox.cn/open) | 智能体平台 | ✅ |
 | [七牛云Qiniu](https://www.qiniu.com/ai/agent) | 聚合平台 | ✅ |
@@ -5,7 +5,7 @@

 <div align="center">

-<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>
+<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>

 <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.302.ai/SuTG99) | Pasarela | ✅ |
+| [302.AI](https://share.302ai.cn/SuTG99) | Pasarela | ✅ |
 | [Qiniu](https://www.qiniu.com/ai/agent) | Pasarela | ✅ |

 [→ Ver todas las integraciones](https://link.langbot.app/en/docs/features)
@@ -5,7 +5,7 @@

 <div align="center">

-<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>
+<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>

 <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.302.ai/SuTG99) | Passerelle | ✅ |
+| [302.AI](https://share.302ai.cn/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 | ✅ |
@@ -5,7 +5,7 @@

 <div align="center">

-<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>
+<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>

 <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.302.ai/SuTG99) | ゲートウェイ | ✅ |
+| [302.AI](https://share.302ai.cn/SuTG99) | ゲートウェイ | ✅ |
 | [Qiniu](https://www.qiniu.com/ai/agent) | ゲートウェイ | ✅ |

 [→ すべての統合を表示](https://link.langbot.app/en/docs/features)
@@ -5,7 +5,7 @@

 <div align="center">

-<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>
+<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>

 <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.302.ai/SuTG99) | 게이트웨이 | ✅ |
+| [302.AI](https://share.302ai.cn/SuTG99) | 게이트웨이 | ✅ |
 | [Qiniu](https://www.qiniu.com/ai/agent) | 게이트웨이 | ✅ |

 [→ 모든 통합 보기](https://link.langbot.app/en/docs/features)
@@ -5,7 +5,7 @@

 <div align="center">

-<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>
+<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>

 <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.302.ai/SuTG99) | Шлюз | ✅ |
+| [302.AI](https://share.302ai.cn/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 | ✅ |
@@ -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.302.ai/SuTG99) | 聚合平台 | ✅ |
+| [302.AI](https://share.302ai.cn/SuTG99) | 聚合平台 | ✅ |
 | [Qiniu](https://www.qiniu.com/ai/agent) | 聚合平台 | ✅ |

 ### TTS（語音合成）
@@ -5,7 +5,7 @@

 <div align="center">

-<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>
+<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>

 <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.302.ai/SuTG99) | Cổng | ✅ |
+| [302.AI](https://share.302ai.cn/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)
@@ -0,0 +1,196 @@
+# 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 管理上走得更稳。
@@ -1,6 +1,6 @@
 [project]
 name = "langbot"
-version = "4.10.4"
+version = "4.10.5"
 description = "Production-grade platform for building agentic IM bots"
 readme = "README.md"
 license-files = ["LICENSE"]
@@ -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==0.4.6",
+    "langbot-plugin==0.4.12",
    "asyncpg>=0.30.0",
    "line-bot-sdk>=3.19.0",
    "matrix-nio>=0.25.2",
@@ -138,6 +138,39 @@ class MonitoringRouterGroup(group.RouterGroup):
                }
            )

+        @self.route('/tool-calls', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
+        async def get_tool_calls() -> str:
+            """Get tool call records"""
+            bot_ids = quart.request.args.getlist('botId')
+            pipeline_ids = quart.request.args.getlist('pipelineId')
+            session_ids = quart.request.args.getlist('sessionId')
+            start_time_str = quart.request.args.get('startTime')
+            end_time_str = quart.request.args.get('endTime')
+            limit = int(quart.request.args.get('limit', 100))
+            offset = int(quart.request.args.get('offset', 0))
+
+            start_time = parse_iso_datetime(start_time_str)
+            end_time = parse_iso_datetime(end_time_str)
+
+            tool_calls, total = await self.ap.monitoring_service.get_tool_calls(
+                bot_ids=bot_ids if bot_ids else None,
+                pipeline_ids=pipeline_ids if pipeline_ids else None,
+                session_ids=session_ids if session_ids else None,
+                start_time=start_time,
+                end_time=end_time,
+                limit=limit,
+                offset=offset,
+            )
+
+            return self.success(
+                data={
+                    'tool_calls': tool_calls,
+                    'total': total,
+                    'limit': limit,
+                    'offset': offset,
+                }
+            )
+
        @self.route('/embedding-calls', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
        async def get_embedding_calls() -> str:
            """Get embedding call records"""
@@ -284,6 +317,16 @@ class MonitoringRouterGroup(group.RouterGroup):
                offset=0,
            )

+            # Get tool calls
+            tool_calls, tool_calls_total = await self.ap.monitoring_service.get_tool_calls(
+                bot_ids=bot_ids if bot_ids else None,
+                pipeline_ids=pipeline_ids if pipeline_ids else None,
+                start_time=start_time,
+                end_time=end_time,
+                limit=limit,
+                offset=0,
+            )
+
            # Get sessions
            sessions, sessions_total = await self.ap.monitoring_service.get_sessions(
                bot_ids=bot_ids if bot_ids else None,
@@ -318,12 +361,14 @@ class MonitoringRouterGroup(group.RouterGroup):
                    'overview': overview,
                    'messages': messages,
                    'llmCalls': llm_calls,
+                    'toolCalls': tool_calls,
                    'embeddingCalls': embedding_calls,
                    'sessions': sessions,
                    'errors': errors,
                    'totalCount': {
                        'messages': messages_total,
                        'llmCalls': llm_calls_total,
+                        'toolCalls': tool_calls_total,
                        'embeddingCalls': embedding_calls_total,
                        'sessions': sessions_total,
                        'errors': errors_total,
@@ -86,6 +86,10 @@ 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,
                    }
@@ -99,6 +103,8 @@ 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,
@@ -108,6 +114,8 @@ 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()
@@ -18,7 +18,6 @@ class BotsRouterGroup(group.RouterGroup):
        @self.route('/<bot_uuid>', methods=['GET', 'PUT', 'DELETE'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
        async def _(bot_uuid: str) -> str:
            if quart.request.method == 'GET':
-                # 返回运行时信息，包括webhook地址等
                bot = await self.ap.bot_service.get_runtime_bot_info(bot_uuid)
                if bot is None:
                    return self.http_status(404, -1, 'bot not found')
@@ -37,30 +36,21 @@ class BotsRouterGroup(group.RouterGroup):
            from_index = json_data.get('from_index', -1)
            max_count = json_data.get('max_count', 10)
            logs, total_count = await self.ap.bot_service.list_event_logs(bot_uuid, from_index, max_count)
-            return self.success(
-                data={
-                    'logs': logs,
-                    'total_count': total_count,
-                }
-            )
+            return self.success(data={'logs': logs, 'total_count': total_count})

        @self.route('/<bot_uuid>/send_message', methods=['POST'], auth_type=group.AuthType.API_KEY)
        async def _(bot_uuid: str) -> str:
-            """Send message to a specific target via bot"""
            json_data = await quart.request.json
            target_type = json_data.get('target_type')
            target_id = json_data.get('target_id')
            message_chain_data = json_data.get('message_chain')

-            # Validate required fields
            if not target_type:
                return self.http_status(400, -1, 'target_type is required')
            if not target_id:
                return self.http_status(400, -1, 'target_id is required')
            if not message_chain_data:
                return self.http_status(400, -1, 'message_chain is required')
-
-            # Validate target_type
            if target_type not in ['person', 'group']:
                return self.http_status(400, -1, 'target_type must be either "person" or "group"')

@@ -72,3 +62,29 @@ class BotsRouterGroup(group.RouterGroup):

                traceback.print_exc()
                return self.http_status(500, -1, f'Failed to send message: {str(e)}')
+
+        # ============ Bot Admins ============
+
+        @self.route('/<bot_uuid>/admins', methods=['GET', 'POST'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY)
+        async def _(bot_uuid: str) -> str:
+            if quart.request.method == 'GET':
+                admins = await self.ap.bot_service.get_bot_admins(bot_uuid)
+                return self.success(data={'admins': admins})
+            elif quart.request.method == 'POST':
+                json_data = await quart.request.json
+                launcher_type = json_data.get('launcher_type', '').strip()
+                launcher_id = str(json_data.get('launcher_id', '')).strip()
+                if not launcher_type or not launcher_id:
+                    return self.http_status(400, -1, 'launcher_type and launcher_id are required')
+                try:
+                    admin_id = await self.ap.bot_service.add_bot_admin(bot_uuid, launcher_type, launcher_id)
+                    return self.success(data={'id': admin_id})
+                except Exception as e:
+                    return self.http_status(409, -1, str(e))
+
+        @self.route(
+            '/<bot_uuid>/admins/<int:admin_id>', methods=['DELETE'], auth_type=group.AuthType.USER_TOKEN_OR_API_KEY
+        )
+        async def _(bot_uuid: str, admin_id: int) -> str:
+            await self.ap.bot_service.delete_bot_admin(bot_uuid, admin_id)
+            return self.success()
@@ -2,6 +2,7 @@ from __future__ import annotations

 import quart
 import traceback
+from urllib.parse import unquote


 from ... import group
@@ -28,11 +29,11 @@ class MCPRouterGroup(group.RouterGroup):
                    traceback.print_exc()
                    return self.http_status(500, -1, f'Failed to create MCP server: {str(e)}')

-        @self.route('/servers/<server_name>', methods=['GET', 'PUT', 'DELETE'], auth_type=group.AuthType.USER_TOKEN)
+        @self.route(
+            '/servers/<path:server_name>', methods=['GET', 'PUT', 'DELETE'], auth_type=group.AuthType.USER_TOKEN
+        )
        async def _(server_name: str) -> str:
            """获取、更新或删除MCP服务器配置"""
-            from urllib.parse import unquote
-
            server_name = unquote(server_name)

            server_data = await self.ap.mcp_service.get_mcp_server_by_name(server_name)
@@ -57,12 +58,72 @@ class MCPRouterGroup(group.RouterGroup):
                except Exception as e:
                    return self.http_status(500, -1, f'Failed to delete MCP server: {str(e)}')

-        @self.route('/servers/<server_name>/test', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
+        @self.route('/servers/<path:server_name>/test', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
        async def _(server_name: str) -> str:
            """测试MCP服务器连接"""
-            from urllib.parse import unquote
-
            server_name = unquote(server_name)
            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/<path: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/<path: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/<path:server_name>/logs', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
+        async def _(server_name: str) -> str:
+            """Get logs from an MCP server"""
+            server_name = unquote(server_name)
+            try:
+                limit = int(quart.request.args.get('limit', 200))
+            except (TypeError, ValueError):
+                limit = 200
+            limit = min(limit, 500)
+            level = quart.request.args.get('level') or None
+            logs = await self.ap.mcp_service.get_mcp_server_logs(server_name, limit=limit, level=level)
+            return self.success(data={'logs': logs})
+
+        @self.route('/servers/<path: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)}')
@@ -1,5 +1,7 @@
 from __future__ import annotations

+import quart
+
 from ... import group


@@ -9,25 +11,41 @@ class ToolsRouterGroup(group.RouterGroup):
        @self.route('', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
        async def _() -> str:
            """获取所有可用工具列表"""
-            tools = await self.ap.tool_mgr.get_all_tools()
+            pipeline_uuid = quart.request.args.get('pipeline_uuid') or quart.request.args.get('pipeline_id')
+            bound_plugins: list[str] | None = None
+            bound_mcp_servers: list[str] | None = None

-            tool_list = []
-            for tool in tools:
-                tool_list.append(
-                    {
-                        'name': tool.name,
-                        'description': tool.description,
-                        'human_desc': tool.human_desc,
-                        'parameters': tool.parameters,
-                    }
-                )
+            if pipeline_uuid:
+                pipeline = await self.ap.pipeline_service.get_pipeline(pipeline_uuid)
+                if pipeline is None:
+                    return self.http_status(404, -1, 'pipeline not found')

-            return self.success(data={'tools': tool_list})
+                extensions_prefs = pipeline.get('extensions_preferences', {}) or {}
+                if not extensions_prefs.get('enable_all_plugins', True):
+                    bound_plugins = [
+                        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,
+                    )
+                }
+            )

        @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()
+            tools = await self.ap.tool_mgr.get_all_tools(include_skill_authoring=True)

            for tool in tools:
                if tool.name == tool_name:
@@ -195,6 +195,13 @@ class UserRouterGroup(group.RouterGroup):
        @self.route('/set-password', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
        async def _(user_email: str) -> str:
            """Set password for Space account (first time) or change password"""
+            # Check if modifying login info is allowed
+            allow_modify_login_info = self.ap.instance_config.data.get('system', {}).get(
+                'allow_modify_login_info', True
+            )
+            if not allow_modify_login_info:
+                return self.http_status(403, -1, 'Modifying login info is disabled')
+
            json_data = await quart.request.json
            new_password = json_data.get('new_password')
            current_password = json_data.get('current_password')
@@ -199,3 +199,35 @@ class BotService:

        # Send message via adapter
        await runtime_bot.adapter.send_message(target_type, str(target_id), message_chain)
+
+    # ============ Bot Admins ============
+
+    async def get_bot_admins(self, bot_uuid: str) -> list[dict]:
+        from ....entity.persistence import bot as persistence_bot
+
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.select(persistence_bot.BotAdmin).where(persistence_bot.BotAdmin.bot_uuid == bot_uuid)
+        )
+        return [{'id': r.id, 'launcher_type': r.launcher_type, 'launcher_id': r.launcher_id} for r in result.all()]
+
+    async def add_bot_admin(self, bot_uuid: str, launcher_type: str, launcher_id: str) -> int:
+        from ....entity.persistence import bot as persistence_bot
+
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.insert(persistence_bot.BotAdmin).values(
+                bot_uuid=bot_uuid,
+                launcher_type=launcher_type,
+                launcher_id=launcher_id,
+            )
+        )
+        return result.inserted_primary_key[0]
+
+    async def delete_bot_admin(self, bot_uuid: str, admin_id: int) -> None:
+        from ....entity.persistence import bot as persistence_bot
+
+        await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.delete(persistence_bot.BotAdmin).where(
+                persistence_bot.BotAdmin.bot_uuid == bot_uuid,
+                persistence_bot.BotAdmin.id == admin_id,
+            )
+        )
@@ -243,6 +243,7 @@ class MaintenanceService:
        tables = {
            'messages': persistence_monitoring.MonitoringMessage.id,
            'llm_calls': persistence_monitoring.MonitoringLLMCall.id,
+            'tool_calls': persistence_monitoring.MonitoringToolCall.id,
            'embedding_calls': persistence_monitoring.MonitoringEmbeddingCall.id,
            'errors': persistence_monitoring.MonitoringError.id,
            'sessions': persistence_monitoring.MonitoringSession.session_id,
@@ -48,6 +48,17 @@ class MCPService:
            if total_extensions >= max_extensions:
                raise ValueError(f'Maximum number of extensions ({max_extensions}) reached')

+        server_name = str(server_data.get('name') or '').strip()
+        if not server_name:
+            raise ValueError('MCP server name is required')
+        server_data['name'] = server_name
+
+        existing_result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.select(persistence_mcp.MCPServer).where(persistence_mcp.MCPServer.name == server_name)
+        )
+        if existing_result.first() is not None:
+            raise ValueError(f'MCP server already exists: {server_name}')
+
        server_data['uuid'] = str(uuid.uuid4())
        await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_mcp.MCPServer).values(server_data))

@@ -136,6 +147,32 @@ 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"""

@@ -151,10 +188,22 @@ class MCPService:
            persisted_session = runtime_mcp_session

            async def _refresh_and_report() -> None:
-                if persisted_session.status == MCPSessionStatus.ERROR:
+                # Testing a persisted server should REUSE its live shared-session
+                # process, not rebuild it. Try a lightweight refresh (a real
+                # list_tools probe over the existing connection) first; only fall
+                # back to a full start() when the session has no live connection
+                # to probe (never connected, or the process is actually gone).
+                needs_start = persisted_session.status == MCPSessionStatus.ERROR or persisted_session.session is None
+                if needs_start:
                    await persisted_session.start()
                else:
-                    await persisted_session.refresh()
+                    try:
+                        await persisted_session.refresh()
+                    except Exception:
+                        # The live connection was stale/dropped: reconnect once
+                        # (reusing the live managed process where possible) and
+                        # re-probe, instead of reporting a false failure.
+                        await persisted_session.start()
                # Surface the discovered tools so the config page can render them
                # even for an already-hosted server.
                ctx.metadata['runtime_info'] = persisted_session.get_runtime_info_dict()
@@ -195,3 +244,19 @@ class MCPService:
            context=ctx,
        )
        return wrapper.id
+
+    async def get_mcp_server_logs(self, server_name: str, limit: int = 200, level: str | None = None) -> list[dict]:
+        """Get recent log lines captured from the MCP server's stderr."""
+        session = self.ap.tool_mgr.mcp_tool_loader.get_session(server_name)
+        if not session:
+            return []
+
+        # Get logs from the session's buffer
+        logs = list(session._log_buffer)
+
+        # Filter by level if specified
+        if level:
+            logs = [log for log in logs if log.get('level') == level]
+
+        # Return the most recent 'limit' logs
+        return logs[-limit:]
@@ -2,6 +2,7 @@ from __future__ import annotations

 import uuid
 import datetime
+import json
 import sqlalchemy

 from ....core import app
@@ -50,6 +51,12 @@ class MonitoringService:
                persistence_monitoring.MonitoringLLMCall.timestamp,
                persistence_monitoring.MonitoringLLMCall.id,
            ),
+            (
+                'monitoring_tool_calls',
+                persistence_monitoring.MonitoringToolCall,
+                persistence_monitoring.MonitoringToolCall.timestamp,
+                persistence_monitoring.MonitoringToolCall.id,
+            ),
            (
                'monitoring_embedding_calls',
                persistence_monitoring.MonitoringEmbeddingCall,
@@ -131,6 +138,68 @@ class MonitoringService:
            await autocommit_conn.execute(sqlalchemy.text('PRAGMA wal_checkpoint(TRUNCATE)'))
            await autocommit_conn.execute(sqlalchemy.text('VACUUM'))

+    def _serialize_tool_payload(self, payload: object, max_length: int = 20000) -> str | None:
+        """Serialize tool arguments/results for monitoring storage."""
+        if payload is None:
+            return None
+
+        if isinstance(payload, str):
+            text = payload
+        else:
+            try:
+                text = json.dumps(payload, ensure_ascii=False, default=str)
+            except Exception:
+                text = str(payload)
+
+        if len(text) <= max_length:
+            return text
+
+        return f'{text[:max_length]}... [truncated {len(text) - max_length} chars]'
+
+    async def _get_message_for_tool_context(
+        self,
+        message_id: str | None = None,
+        session_id: str | None = None,
+    ):
+        if message_id:
+            result = await self.ap.persistence_mgr.execute_async(
+                sqlalchemy.select(persistence_monitoring.MonitoringMessage).where(
+                    persistence_monitoring.MonitoringMessage.id == message_id
+                )
+            )
+            row = result.first()
+            if row:
+                return row[0]
+
+        if not session_id:
+            return None
+
+        user_query = (
+            sqlalchemy.select(persistence_monitoring.MonitoringMessage)
+            .where(
+                sqlalchemy.and_(
+                    persistence_monitoring.MonitoringMessage.session_id == session_id,
+                    persistence_monitoring.MonitoringMessage.role == 'user',
+                )
+            )
+            .order_by(persistence_monitoring.MonitoringMessage.timestamp.desc())
+            .limit(1)
+        )
+        result = await self.ap.persistence_mgr.execute_async(user_query)
+        row = result.first()
+        if row:
+            return row[0]
+
+        any_query = (
+            sqlalchemy.select(persistence_monitoring.MonitoringMessage)
+            .where(persistence_monitoring.MonitoringMessage.session_id == session_id)
+            .order_by(persistence_monitoring.MonitoringMessage.timestamp.desc())
+            .limit(1)
+        )
+        result = await self.ap.persistence_mgr.execute_async(any_query)
+        row = result.first()
+        return row[0] if row else None
+
    # ========== Recording Methods ==========

    async def record_message(
@@ -220,6 +289,57 @@ class MonitoringService:

        return call_id

+    async def record_tool_call(
+        self,
+        tool_name: str,
+        tool_source: str,
+        duration: int,
+        status: str = 'success',
+        bot_id: str | None = None,
+        bot_name: str | None = None,
+        pipeline_id: str | None = None,
+        pipeline_name: str | None = None,
+        session_id: str | None = None,
+        message_id: str | None = None,
+        arguments: object | None = None,
+        result: object | None = None,
+        error_message: str | None = None,
+    ) -> str:
+        """Record a tool call."""
+        context_message = await self._get_message_for_tool_context(message_id=message_id, session_id=session_id)
+        if context_message:
+            bot_id = bot_id or context_message.bot_id
+            bot_name = bot_name or context_message.bot_name
+            pipeline_id = pipeline_id or context_message.pipeline_id
+            pipeline_name = pipeline_name or context_message.pipeline_name
+            session_id = session_id or context_message.session_id
+            message_id = message_id or context_message.id
+
+        call_id = str(uuid.uuid4())
+        call_data = {
+            'id': call_id,
+            'timestamp': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None),
+            'tool_name': tool_name,
+            'tool_source': tool_source,
+            'duration': max(0, duration),
+            'status': status,
+            'bot_id': bot_id or 'unknown',
+            'bot_name': bot_name or 'Unknown',
+            'pipeline_id': pipeline_id or 'unknown',
+            'pipeline_name': pipeline_name or 'Unknown',
+            'session_id': session_id,
+            'message_id': message_id,
+            'arguments': self._serialize_tool_payload(arguments),
+            'result': self._serialize_tool_payload(result),
+            'error_message': self._serialize_tool_payload(error_message),
+        }
+
+        await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.insert(persistence_monitoring.MonitoringToolCall).values(call_data)
+        )
+
+        return call_id
+
    async def record_embedding_call(
        self,
        model_name: str,
@@ -749,6 +869,58 @@ class MonitoringService:
            total,
        )

+    async def get_tool_calls(
+        self,
+        bot_ids: list[str] | None = None,
+        pipeline_ids: list[str] | None = None,
+        session_ids: list[str] | None = None,
+        start_time: datetime.datetime | None = None,
+        end_time: datetime.datetime | None = None,
+        limit: int = 100,
+        offset: int = 0,
+    ) -> tuple[list[dict], int]:
+        """Get tool calls with filters"""
+        conditions = []
+
+        if bot_ids:
+            conditions.append(persistence_monitoring.MonitoringToolCall.bot_id.in_(bot_ids))
+        if pipeline_ids:
+            conditions.append(persistence_monitoring.MonitoringToolCall.pipeline_id.in_(pipeline_ids))
+        if session_ids:
+            conditions.append(persistence_monitoring.MonitoringToolCall.session_id.in_(session_ids))
+        if start_time:
+            conditions.append(persistence_monitoring.MonitoringToolCall.timestamp >= start_time)
+        if end_time:
+            conditions.append(persistence_monitoring.MonitoringToolCall.timestamp <= end_time)
+
+        count_query = sqlalchemy.select(sqlalchemy.func.count(persistence_monitoring.MonitoringToolCall.id))
+        if conditions:
+            count_query = count_query.where(sqlalchemy.and_(*conditions))
+
+        count_result = await self.ap.persistence_mgr.execute_async(count_query)
+        total = count_result.scalar() or 0
+
+        query = sqlalchemy.select(persistence_monitoring.MonitoringToolCall).order_by(
+            persistence_monitoring.MonitoringToolCall.timestamp.desc()
+        )
+        if conditions:
+            query = query.where(sqlalchemy.and_(*conditions))
+
+        query = query.limit(limit).offset(offset)
+
+        result = await self.ap.persistence_mgr.execute_async(query)
+        tool_calls_rows = result.all()
+
+        return (
+            [
+                self.ap.persistence_mgr.serialize_model(
+                    persistence_monitoring.MonitoringToolCall, row[0] if isinstance(row, tuple) else row
+                )
+                for row in tool_calls_rows
+            ],
+            total,
+        )
+
    async def get_embedding_calls(
        self,
        start_time: datetime.datetime | None = None,
@@ -971,6 +1143,34 @@ class MonitoringService:
            else:
                error_llm_calls += 1

+        # Get tool calls for this session
+        tool_query = (
+            sqlalchemy.select(persistence_monitoring.MonitoringToolCall)
+            .where(persistence_monitoring.MonitoringToolCall.session_id == session_id)
+            .order_by(persistence_monitoring.MonitoringToolCall.timestamp.asc())
+        )
+        tool_result = await self.ap.persistence_mgr.execute_async(tool_query)
+        tool_rows = tool_result.all()
+
+        tool_calls = [
+            self.ap.persistence_mgr.serialize_model(
+                persistence_monitoring.MonitoringToolCall, row[0] if isinstance(row, tuple) else row
+            )
+            for row in tool_rows
+        ]
+
+        total_tool_calls = len(tool_rows)
+        success_tool_calls = 0
+        error_tool_calls = 0
+        total_tool_duration = 0
+        for row in tool_rows:
+            tool_call = row[0] if isinstance(row, tuple) else row
+            total_tool_duration += tool_call.duration
+            if tool_call.status == 'success':
+                success_tool_calls += 1
+            else:
+                error_tool_calls += 1
+
        # Get errors for this session
        error_query = (
            sqlalchemy.select(persistence_monitoring.MonitoringError)
@@ -1014,6 +1214,14 @@ class MonitoringService:
                'total_tokens': total_tokens,
                'average_duration_ms': int(total_duration / total_llm_calls) if total_llm_calls > 0 else 0,
            },
+            'tool_calls': tool_calls,
+            'tool_stats': {
+                'total_calls': total_tool_calls,
+                'success_calls': success_tool_calls,
+                'error_calls': error_tool_calls,
+                'total_duration_ms': total_tool_duration,
+                'average_duration_ms': int(total_tool_duration / total_tool_calls) if total_tool_calls > 0 else 0,
+            },
            'errors': errors,
            'session_duration_seconds': session_duration_seconds,
        }
@@ -100,6 +100,8 @@ 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(
@@ -193,6 +195,8 @@ class PipelineService:
                    'enable_all_mcp_servers': True,
                    'plugins': [],
                    'mcp_servers': [],
+                    'mcp_resources': [],
+                    'mcp_resource_agent_read_enabled': True,
                }
            ),
        }
@@ -217,6 +221,8 @@ 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
@@ -236,10 +242,14 @@ 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)
@@ -84,7 +84,17 @@ class CommandManager:

        privilege = 1

-        if f'{query.launcher_type.value}_{query.launcher_id}' in self.ap.instance_config.data['admins']:
+        import sqlalchemy as _sa
+        from ..entity.persistence.bot import BotAdmin as _BotAdmin
+
+        _admins = await self.ap.persistence_mgr.execute_async(
+            _sa.select(_BotAdmin).where(
+                _BotAdmin.bot_uuid == (query.bot_uuid or ''),
+                _BotAdmin.launcher_type == query.launcher_type.value,
+                _BotAdmin.launcher_id == str(query.launcher_id),
+            )
+        )
+        if _admins.first() is not None:
            privilege = 2

        ctx = command_context.ExecuteContext(
@@ -3,6 +3,20 @@ import sqlalchemy
 from .base import Base


+class BotAdmin(Base):
+    """Bot admin — a launcher that has admin privilege for a specific bot's commands"""
+
+    __tablename__ = 'bot_admins'
+
+    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True, autoincrement=True)
+    bot_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
+    launcher_type = sqlalchemy.Column(sqlalchemy.String(64), nullable=False)
+    launcher_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
+    created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
+
+    __table_args__ = (sqlalchemy.UniqueConstraint('bot_uuid', 'launcher_type', 'launcher_id', name='uq_bot_admin'),)
+
+
 class Bot(Base):
    """Bot"""

@@ -49,6 +49,28 @@ class MonitoringLLMCall(Base):
    message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)  # Associated message ID


+class MonitoringToolCall(Base):
+    """Tool call records"""
+
+    __tablename__ = 'monitoring_tool_calls'
+
+    id = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True)
+    timestamp = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, index=True)
+    tool_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
+    tool_source = sqlalchemy.Column(sqlalchemy.String(50), nullable=False)  # native, plugin, mcp, skill
+    duration = sqlalchemy.Column(sqlalchemy.Integer, nullable=False)  # milliseconds
+    status = sqlalchemy.Column(sqlalchemy.String(50), nullable=False)  # success, error
+    bot_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True)
+    bot_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
+    pipeline_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=False, index=True)
+    pipeline_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
+    session_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)
+    message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)
+    arguments = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
+    result = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
+    error_message = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
+
+
 class MonitoringSession(Base):
    """Session tracking records"""

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


@@ -0,0 +1,84 @@
+"""add bot_admins table and migrate config admins
+
+Revision ID: 0007_add_bot_admins
+Revises: 0006_normalize_mcp_remote_mode
+Create Date: 2026-06-26
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = '0007_add_bot_admins'
+down_revision = '0006_normalize_mcp_remote_mode'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if 'bot_admins' in sa.inspect(conn).get_table_names():
+        return
+    op.create_table(
+        'bot_admins',
+        sa.Column('id', sa.Integer, primary_key=True, autoincrement=True),
+        sa.Column('bot_uuid', sa.String(255), nullable=False),
+        sa.Column('launcher_type', sa.String(64), nullable=False),
+        sa.Column('launcher_id', sa.String(255), nullable=False),
+        sa.Column('created_at', sa.DateTime, nullable=False, server_default=sa.func.now()),
+        sa.UniqueConstraint('bot_uuid', 'launcher_type', 'launcher_id', name='uq_bot_admin'),
+    )
+
+    # Migrate old config-based admins into the first bot (best-effort)
+    inspector = sa.inspect(conn)
+    tables = inspector.get_table_names()
+
+    if 'bots' not in tables:
+        return
+
+    # Read the first bot uuid
+    row = conn.execute(sa.text('SELECT uuid FROM bots ORDER BY created_at LIMIT 1')).first()
+    if row is None:
+        return
+    first_bot_uuid = row[0]
+
+    # Read instance_config metadata key that holds the admins list
+    if 'metadata' not in tables:
+        return
+    meta_row = conn.execute(sa.text("SELECT value FROM metadata WHERE key = 'instance_config'")).first()
+    if meta_row is None:
+        return
+
+    import json
+
+    try:
+        cfg = json.loads(meta_row[0])
+    except Exception:
+        return
+
+    admins = cfg.get('admins', [])
+    for entry in admins:
+        parts = entry.split('_', 1)
+        if len(parts) != 2:
+            continue
+        launcher_type, launcher_id = parts
+        try:
+            conn.execute(
+                sa.text(
+                    'INSERT OR IGNORE INTO bot_admins (bot_uuid, launcher_type, launcher_id) VALUES (:bu, :lt, :li)'
+                ),
+                {'bu': first_bot_uuid, 'lt': launcher_type, 'li': launcher_id},
+            )
+        except Exception:
+            pass
+
+    # Remove admins key from stored config
+    if 'admins' in cfg:
+        del cfg['admins']
+        conn.execute(
+            sa.text("UPDATE metadata SET value = :v WHERE key = 'instance_config'"),
+            {'v': json.dumps(cfg)},
+        )
+
+
+def downgrade() -> None:
+    op.drop_table('bot_admins')
@@ -0,0 +1,95 @@
+"""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)
@@ -0,0 +1,17 @@
+from langbot.pkg.entity.persistence import monitoring as persistence_monitoring
+from .. import migration
+
+
+@migration.migration_class(26)
+class DBMigrateMonitoringToolCalls(migration.DBMigration):
+    """Add monitoring_tool_calls table"""
+
+    async def upgrade(self):
+        """Upgrade"""
+        async with self.ap.persistence_mgr.get_db_engine().begin() as conn:
+            await conn.run_sync(persistence_monitoring.MonitoringToolCall.__table__.create, checkfirst=True)
+
+    async def downgrade(self):
+        """Downgrade"""
+        async with self.ap.persistence_mgr.get_db_engine().begin() as conn:
+            await conn.run_sync(persistence_monitoring.MonitoringToolCall.__table__.drop, checkfirst=True)
@@ -32,7 +32,7 @@ class MonitoringHelper:
        """Record the start of query processing, returns message_id"""
        try:
            # Check if session exists, if not, record session start
-            session_id = f'{query.launcher_type}_{query.launcher_id}'
+            session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'

            # Get sender name from message event
            sender_name = None
@@ -137,7 +137,7 @@ class MonitoringHelper:
    ):
        """Record bot response message to monitoring"""
        try:
-            session_id = f'{query.launcher_type}_{query.launcher_id}'
+            session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'

            # Get sender name from message event
            sender_name = None
@@ -202,7 +202,7 @@ class MonitoringHelper:
    ) -> str:
        """Record query processing error, returns message_id"""
        try:
-            session_id = f'{query.launcher_type}_{query.launcher_id}'
+            session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'

            # Get sender name from message event
            sender_name = None
@@ -268,7 +268,7 @@ class MonitoringHelper:
    ):
        """Record LLM call"""
        try:
-            session_id = f'{query.launcher_type}_{query.launcher_id}'
+            session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'

            await ap.monitoring_service.record_llm_call(
                bot_id=bot_id,
@@ -96,6 +96,15 @@ 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
@@ -116,6 +125,8 @@ 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:
@@ -178,7 +189,7 @@ class RuntimePipeline:
                bot_name = query.variables.get('_monitoring_bot_name', 'Unknown')
                pipeline_name = query.variables.get('_monitoring_pipeline_name', 'Unknown')
                message_id = query.variables.get('_monitoring_message_id', '')
-                session_id = f'{query.launcher_type}_{query.launcher_id}'
+                session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}'

                # Update message status to error
                if message_id:
@@ -0,0 +1,274 @@
+from __future__ import annotations
+
+import traceback
+import weakref
+from dataclasses import dataclass, field
+from typing import Any
+
+import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
+import langbot_plugin.api.entities.builtin.platform.message as platform_message
+
+
+@dataclass(frozen=True)
+class PluginResponseSource:
+    plugin: dict[str, str]
+    event_name: str | None = None
+    is_approximate: bool = False
+
+
+@dataclass
+class QueryDiagnosticState:
+    pending_by_chain_id: dict[int, list[PluginResponseSource]] = field(default_factory=dict)
+    by_response_index: dict[int, list[PluginResponseSource]] = field(default_factory=dict)
+    finalizer: weakref.finalize | None = None
+
+
+_QUERY_STATES: dict[int, QueryDiagnosticState] = {}
+
+
+def record_plugin_response_source(
+    query: pipeline_query.Query,
+    response_index: int,
+    response_sources: list[dict[str, Any]] | None,
+    emitted_plugins: list[dict[str, Any]] | None = None,
+    event_name: str | None = None,
+) -> None:
+    plugin_sources = _build_plugin_sources(response_sources, emitted_plugins, event_name)
+    if not plugin_sources:
+        return
+    state = _get_or_create_query_state(query)
+    state.by_response_index[response_index] = plugin_sources
+
+
+def record_last_plugin_response_source(
+    query: pipeline_query.Query,
+    response_sources: list[dict[str, Any]] | None,
+    emitted_plugins: list[dict[str, Any]] | None = None,
+    event_name: str | None = None,
+) -> None:
+    record_plugin_response_source(
+        query,
+        len(query.resp_message_chain) - 1,
+        response_sources,
+        emitted_plugins,
+        event_name,
+    )
+
+
+def record_pending_plugin_response_source(
+    query: pipeline_query.Query,
+    message_chain: platform_message.MessageChain,
+    response_sources: list[dict[str, Any]] | None,
+    emitted_plugins: list[dict[str, Any]] | None = None,
+    event_name: str | None = None,
+) -> None:
+    plugin_sources = _build_plugin_sources(response_sources, emitted_plugins, event_name)
+    if not plugin_sources:
+        return
+    state = _get_or_create_query_state(query)
+    state.pending_by_chain_id[id(message_chain)] = plugin_sources
+
+
+def consume_pending_plugin_response_source(
+    query: pipeline_query.Query,
+    message_chain: platform_message.MessageChain,
+    response_index: int,
+) -> None:
+    state = _get_query_state(query)
+    if state is None:
+        return
+    source = state.pending_by_chain_id.pop(id(message_chain), None)
+    if source is None:
+        return
+    state.by_response_index[response_index] = source
+
+
+def clear_response_source(query: pipeline_query.Query, response_index: int) -> None:
+    state = _get_query_state(query)
+    if state is None:
+        return
+    state.by_response_index.pop(response_index, None)
+    _discard_query_state_if_empty(query)
+
+
+async def notify_response_delivery_failure(
+    ap: Any,
+    query: pipeline_query.Query,
+    response_index: int,
+    message_chain: platform_message.MessageChain,
+    error: Exception,
+) -> None:
+    try:
+        plugin_refs = _get_response_sources(query, response_index)
+        if not plugin_refs:
+            return
+        connector = getattr(ap, 'plugin_connector', None)
+        if connector is None or not hasattr(connector, 'notify_plugin_diagnostic'):
+            return
+        for source in plugin_refs:
+            payload = _build_delivery_failure_payload(
+                plugin_ref=source.plugin,
+                event_name=source.event_name,
+                is_approximate=source.is_approximate,
+                query=query,
+                response_index=response_index,
+                message_chain=message_chain,
+                error=error,
+            )
+            try:
+                await connector.notify_plugin_diagnostic(payload)
+            except Exception as diag_error:
+                _debug(ap, f'Plugin diagnostic forwarding failed: {diag_error}')
+    except Exception as diag_error:
+        _debug(ap, f'Plugin diagnostic forwarding skipped: {diag_error}')
+
+
+def get_emitted_plugins(event_ctx: Any) -> list[dict[str, Any]]:
+    emitted_plugins = getattr(event_ctx, '_emitted_plugins', [])
+    return emitted_plugins if isinstance(emitted_plugins, list) else []
+
+
+def get_response_sources(event_ctx: Any) -> list[dict[str, Any]] | None:
+    event_attrs = vars(event_ctx)
+    if '_response_sources' not in event_attrs:
+        return None
+    response_sources = event_attrs['_response_sources']
+    return response_sources if isinstance(response_sources, list) else []
+
+
+def _get_or_create_query_state(query: pipeline_query.Query) -> QueryDiagnosticState:
+    query_key = id(query)
+    state = _QUERY_STATES.get(query_key)
+    if state is not None:
+        return state
+
+    state = QueryDiagnosticState()
+    try:
+        state.finalizer = weakref.finalize(query, _discard_query_state, query_key)
+    except TypeError:
+        state.finalizer = None
+    _QUERY_STATES[query_key] = state
+    return state
+
+
+def _get_query_state(query: pipeline_query.Query) -> QueryDiagnosticState | None:
+    return _QUERY_STATES.get(id(query))
+
+
+def _discard_query_state(query_key: int) -> None:
+    _QUERY_STATES.pop(query_key, None)
+
+
+def _discard_query_state_if_empty(query: pipeline_query.Query) -> None:
+    query_key = id(query)
+    state = _QUERY_STATES.get(query_key)
+    if state is None:
+        return
+    if state.pending_by_chain_id or state.by_response_index:
+        return
+    if state.finalizer is not None:
+        state.finalizer.detach()
+    _discard_query_state(query_key)
+
+
+def _get_response_sources(
+    query: pipeline_query.Query,
+    response_index: int,
+) -> list[PluginResponseSource]:
+    state = _get_query_state(query)
+    if state is None:
+        return []
+    return state.by_response_index.get(response_index, [])
+
+
+def _extract_plugin_ref(plugin: Any) -> dict[str, str] | None:
+    manifest = plugin.get('manifest') if isinstance(plugin, dict) else None
+    metadata = manifest.get('metadata') if isinstance(manifest, dict) else None
+    if not isinstance(metadata, dict):
+        return None
+    author = metadata.get('author')
+    name = metadata.get('name')
+    if not author or not name:
+        return None
+    return {'author': str(author), 'name': str(name)}
+
+
+def _extract_response_source_plugin_ref(source: Any) -> dict[str, str] | None:
+    if not isinstance(source, dict):
+        return None
+    if source.get('kind') != 'reply_message_chain':
+        return None
+    plugin_ref = source.get('plugin')
+    if not isinstance(plugin_ref, dict):
+        return None
+    author = plugin_ref.get('author')
+    name = plugin_ref.get('name')
+    if not author or not name:
+        return None
+    return {'author': str(author), 'name': str(name)}
+
+
+def _build_plugin_sources(
+    response_sources: list[dict[str, Any]] | None,
+    emitted_plugins: list[dict[str, Any]] | None,
+    event_name: str | None,
+) -> list[PluginResponseSource]:
+    if response_sources is not None:
+        plugin_refs = [_extract_response_source_plugin_ref(source) for source in response_sources]
+        return [
+            PluginResponseSource(plugin=plugin, event_name=event_name) for plugin in plugin_refs if plugin is not None
+        ]
+
+    if emitted_plugins:
+        plugin_refs = [_extract_plugin_ref(plugin) for plugin in emitted_plugins]
+        return [
+            PluginResponseSource(plugin=plugin, event_name=event_name, is_approximate=True)
+            for plugin in plugin_refs
+            if plugin is not None
+        ]
+    return []
+
+
+def _debug(ap: Any, message: str) -> None:
+    logger = getattr(ap, 'logger', None)
+    if logger is not None:
+        logger.debug(message)
+
+
+def _build_delivery_failure_payload(
+    plugin_ref: dict[str, str],
+    event_name: str | None,
+    is_approximate: bool,
+    query: pipeline_query.Query,
+    response_index: int,
+    message_chain: platform_message.MessageChain,
+    error: Exception,
+) -> dict[str, Any]:
+    details: dict[str, Any] = {
+        'message_component_types': [component.__class__.__name__ for component in message_chain],
+        'message_preview': str(message_chain)[:200],
+    }
+    if is_approximate:
+        details['attribution_warning'] = (
+            'This diagnostic was delivered to all plugins that handled the event because the '
+            'plugin runtime did not report the exact reply_message_chain source.'
+        )
+
+    return {
+        'level': 'ERROR',
+        'code': 'response_delivery_failed',
+        'message': 'Failed to deliver a plugin-provided response message.',
+        'plugin': plugin_ref,
+        'query': {
+            'query_id': query.query_id,
+            'event_name': event_name or query.message_event.__class__.__name__,
+            'stage': query.current_stage_name or 'SendResponseBackStage',
+            'response_index': response_index,
+        },
+        'details': details,
+        'delivery': {
+            'error_type': error.__class__.__name__,
+            'error_message': str(error),
+            'traceback': traceback.format_exception_only(type(error), error)[-1].strip(),
+        },
+    }
@@ -25,6 +25,21 @@ 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,
@@ -32,6 +47,7 @@ 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
        )
@@ -43,7 +59,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 = query.pipeline_config['ai']['local-agent'].get('model', {})
+            model_config = local_agent_config.get('model', {})
            if isinstance(model_config, str):
                # Legacy format: plain UUID string
                primary_uuid = model_config
@@ -113,11 +129,14 @@ 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)
-                    query.use_funcs = await self.ap.tool_mgr.get_all_tools(
+                    include_mcp_resource_tools = query.variables.get('_pipeline_mcp_resource_agent_read_enabled', True)
+                    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}')
@@ -128,11 +147,14 @@ 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)
-                query.use_funcs = await self.ap.tool_mgr.get_all_tools(
+                include_mcp_resource_tools = query.variables.get('_pipeline_mcp_resource_agent_read_enabled', True)
+                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 = ''

@@ -9,6 +9,7 @@ from datetime import datetime

 from .. import handler
 from ... import entities
+from ... import plugin_diagnostics
 from ....provider import runner as runner_module

 import langbot_plugin.api.entities.events as events
@@ -58,6 +59,13 @@ class ChatMessageHandler(handler.MessageHandler):
        if event_ctx.is_prevented_default():
            if event_ctx.event.reply_message_chain is not None:
                mc = event_ctx.event.reply_message_chain
+                plugin_diagnostics.record_pending_plugin_response_source(
+                    query,
+                    mc,
+                    plugin_diagnostics.get_response_sources(event_ctx),
+                    plugin_diagnostics.get_emitted_plugins(event_ctx),
+                    event.event_name,
+                )
                query.resp_messages.append(mc)

                yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
@@ -4,6 +4,7 @@ import typing

 from .. import handler
 from ... import entities
+from ... import plugin_diagnostics
 import langbot_plugin.api.entities.builtin.provider.message as provider_message
 import langbot_plugin.api.entities.builtin.provider.session as provider_session
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -52,6 +53,13 @@ class CommandHandler(handler.MessageHandler):
        if event_ctx.is_prevented_default():
            if event_ctx.event.reply_message_chain is not None:
                mc = event_ctx.event.reply_message_chain
+                plugin_diagnostics.record_pending_plugin_response_source(
+                    query,
+                    mc,
+                    plugin_diagnostics.get_response_sources(event_ctx),
+                    plugin_diagnostics.get_emitted_plugins(event_ctx),
+                    event.event_name,
+                )

                query.resp_messages.append(mc)

@@ -9,6 +9,7 @@ import langbot_plugin.api.entities.builtin.platform.message as platform_message
 import langbot_plugin.api.entities.builtin.provider.message as provider_message

 from .. import stage, entities
+from .. import plugin_diagnostics
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query


@@ -39,20 +40,35 @@ class SendResponseBackStage(stage.PipelineStage):

        has_chunks = any(isinstance(msg, provider_message.MessageChunk) for msg in query.resp_messages)
        # TODO 命令与流式的兼容性问题
-        if await query.adapter.is_stream_output_supported() and has_chunks:
-            is_final = [msg.is_final for msg in query.resp_messages][0]
-            await query.adapter.reply_message_chunk(
-                message_source=query.message_event,
-                bot_message=query.resp_messages[-1],
-                message=query.resp_message_chain[-1],
-                quote_origin=quote_origin,
-                is_final=is_final,
-            )
-        else:
-            await query.adapter.reply_message(
-                message_source=query.message_event,
-                message=query.resp_message_chain[-1],
-                quote_origin=quote_origin,
+        response_index = len(query.resp_message_chain) - 1
+        message_chain = query.resp_message_chain[-1]
+
+        try:
+            if await query.adapter.is_stream_output_supported() and has_chunks:
+                is_final = [msg.is_final for msg in query.resp_messages][0]
+                await query.adapter.reply_message_chunk(
+                    message_source=query.message_event,
+                    bot_message=query.resp_messages[-1],
+                    message=message_chain,
+                    quote_origin=quote_origin,
+                    is_final=is_final,
+                )
+            else:
+                await query.adapter.reply_message(
+                    message_source=query.message_event,
+                    message=message_chain,
+                    quote_origin=quote_origin,
+                )
+        except Exception as e:
+            await plugin_diagnostics.notify_response_delivery_failure(
+                self.ap,
+                query,
+                response_index,
+                message_chain,
+                e,
            )
+            plugin_diagnostics.clear_response_source(query, response_index)
+            raise
+        plugin_diagnostics.clear_response_source(query, response_index)

        return entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)
@@ -3,6 +3,7 @@ from __future__ import annotations
 import typing

 from .. import entities
+from .. import plugin_diagnostics
 from .. import stage

 import langbot_plugin.api.entities.builtin.platform.message as platform_message
@@ -78,6 +79,11 @@ class ResponseWrapper(stage.PipelineStage):
        # 如果 resp_messages[-1] 已经是 MessageChain 了
        if isinstance(query.resp_messages[-1], platform_message.MessageChain):
            query.resp_message_chain.append(query.resp_messages[-1])
+            plugin_diagnostics.consume_pending_plugin_response_source(
+                query,
+                query.resp_messages[-1],
+                len(query.resp_message_chain) - 1,
+            )

            yield entities.StageProcessResult(result_type=entities.ResultType.CONTINUE, new_query=query)

@@ -129,8 +135,10 @@ class ResponseWrapper(stage.PipelineStage):
                        else:
                            if event_ctx.event.reply_message_chain is not None:
                                reply_chain = event_ctx.event.reply_message_chain
+                                is_plugin_reply = True
                            else:
                                reply_chain = result.get_content_platform_message_chain()
+                                is_plugin_reply = False

                            # Attach files the agent produced in the sandbox
                            # outbox, but only on the terminal assistant message.
@@ -138,6 +146,13 @@ class ResponseWrapper(stage.PipelineStage):
                                await self._append_outbound_attachments(query, reply_chain)

                            query.resp_message_chain.append(reply_chain)
+                            if is_plugin_reply:
+                                plugin_diagnostics.record_last_plugin_response_source(
+                                    query,
+                                    plugin_diagnostics.get_response_sources(event_ctx),
+                                    plugin_diagnostics.get_emitted_plugins(event_ctx),
+                                    event.event_name,
+                                )

                            yield entities.StageProcessResult(
                                result_type=entities.ResultType.CONTINUE,
@@ -180,6 +195,12 @@ class ResponseWrapper(stage.PipelineStage):
                            else:
                                if event_ctx.event.reply_message_chain is not None:
                                    query.resp_message_chain.append(event_ctx.event.reply_message_chain)
+                                    plugin_diagnostics.record_last_plugin_response_source(
+                                        query,
+                                        plugin_diagnostics.get_response_sources(event_ctx),
+                                        plugin_diagnostics.get_emitted_plugins(event_ctx),
+                                        event.event_name,
+                                    )

                                else:
                                    query.resp_message_chain.append(
@@ -4,6 +4,7 @@ import asyncio
 import traceback
 import datetime
 import json
+import time

 import aiocqhttp
 import pydantic
@@ -16,6 +17,37 @@ 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(
@@ -35,7 +67,7 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
            elif type(msg) is platform_message.Image:
                arg = ''
                if msg.base64:
-                    arg = msg.base64
+                    arg = _normalize_base64_payload(msg.base64)
                    msg_list.append(aiocqhttp.MessageSegment.image(f'base64://{arg}'))
                elif msg.url:
                    arg = msg.url
@@ -50,7 +82,7 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
            elif type(msg) is platform_message.Voice:
                arg = ''
                if msg.base64:
-                    arg = msg.base64
+                    arg = _normalize_base64_payload(msg.base64)
                    msg_list.append(aiocqhttp.MessageSegment.record(f'base64://{arg}'))
                elif msg.url:
                    arg = msg.url
@@ -62,7 +94,10 @@ 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):
-                msg_list.append({'type': 'file', 'data': {'file': msg.url, 'name': msg.name}})
+                file = msg.url or msg.path
+                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))
@@ -324,16 +359,96 @@ 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

-    @staticmethod
-    async def target2yiri(event: aiocqhttp.Event, bot=None):
+    async def _get_group_name(self, group_id: typing.Union[int, str], bot=None) -> str:
+        now = time.monotonic()
+        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':
@@ -343,14 +458,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=event.sender['nickname'],
+                    member_name=_get_group_member_name(event.sender),
                    permission=permission,
                    group=platform_entities.Group(
                        id=event.group_id,
-                        name=event.sender['nickname'],
+                        name=group_name,
                        permission=platform_entities.Permission.Member,
                    ),
-                    special_title=event.sender['title'] if 'title' in event.sender else '',
+                    special_title=special_title,
                ),
                message_chain=yiri_chain,
                time=event.time,
@@ -374,7 +489,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 = AiocqhttpEventConverter()
+    event_converter: AiocqhttpEventConverter = pydantic.Field(default_factory=AiocqhttpEventConverter)

    on_websocket_connection_event_cache: typing.List[typing.Callable[[aiocqhttp.Event], None]] = []

@@ -433,9 +548,7 @@ class AiocqhttpAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
                    elif isinstance(component, platform_message.Image):
                        img_data = {}
                        if component.base64:
-                            b64 = component.base64
-                            if b64.startswith('data:'):
-                                b64 = b64.split(',', 1)[-1] if ',' in b64 else b64
+                            b64 = _normalize_base64_payload(component.base64)
                            img_data['file'] = f'base64://{b64}'
                        elif component.url:
                            img_data['file'] = component.url
@@ -2,6 +2,7 @@ from __future__ import annotations
 import typing
 import asyncio
 import traceback
+import uuid

 import datetime
 import pydantic
@@ -182,7 +183,28 @@ class WecomCSAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                )

    async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
-        pass
+        if target_type != 'person':
+            raise ValueError('WeCom customer service only supports sending messages to person targets')
+
+        open_kfid = self.bot_account_id
+        external_userid = target_id
+        if '|' in target_id:
+            open_kfid, external_userid = target_id.split('|', 1)
+        if external_userid.startswith('u'):
+            external_userid = external_userid[1:]
+        if not open_kfid:
+            raise ValueError('WeCom customer service open_kfid is required before sending messages')
+
+        content_list = await WecomMessageConverter.yiri2target(message, self.bot)
+        for content in content_list:
+            msgid = f'langbot_{uuid.uuid4().hex}'
+            if content['type'] == 'text':
+                await self.bot.send_text_msg(
+                    open_kfid=open_kfid,
+                    external_userid=external_userid,
+                    msgid=msgid,
+                    content=content['content'],
+                )

    def set_bot_uuid(self, bot_uuid: str):
        """设置 bot UUID（用于生成 webhook URL）"""
@@ -737,6 +737,8 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
        event_ctx = context.EventContext.from_event(event)

        if not self.is_enable_plugin:
+            event_ctx._emitted_plugins = []
+            event_ctx._response_sources = []
            return event_ctx

        # Pass include_plugins to runtime for filtering
@@ -745,9 +747,21 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
        )

        event_ctx = context.EventContext.model_validate(event_ctx_result['event_context'])
+        event_ctx._emitted_plugins = event_ctx_result.get('emitted_plugins', [])
+        if 'response_sources' in event_ctx_result:
+            event_ctx._response_sources = event_ctx_result['response_sources']

        return event_ctx

+    async def notify_plugin_diagnostic(self, diagnostic: dict[str, Any]) -> None:
+        """Best-effort diagnostic forwarding to the plugin runtime."""
+        if not self.is_enable_plugin:
+            return
+        try:
+            await self.handler.notify_plugin_diagnostic(diagnostic)
+        except Exception as e:
+            self.ap.logger.debug(f'Plugin diagnostic forwarding skipped: {e}')
+
    async def list_tools(self, bound_plugins: list[str] | None = None) -> list[ComponentManifest]:
        if not self.is_enable_plugin:
            return []
@@ -26,6 +26,15 @@ from ..core import app
 from ..utils import constants


+class _RawAction:
+    def __init__(self, value: str):
+        self.value = value
+
+
+def _langbot_to_runtime_action(enum_name: str, fallback_value: str) -> Any:
+    return getattr(LangBotToRuntimeAction, enum_name, _RawAction(fallback_value))
+
+
 def _make_rag_error_response(error: Exception, error_type: str, **extra_context) -> handler.ActionResponse:
    """Create a clean error response for RAG operations.

@@ -923,6 +932,18 @@ class RuntimeConnectionHandler(handler.Handler):

        return result

+    async def notify_plugin_diagnostic(self, diagnostic: dict[str, Any]) -> dict[str, Any]:
+        """Notify the plugin runtime about a best-effort plugin diagnostic.
+
+        This intentionally uses the raw protocol string instead of a SDK enum so
+        LangBot can keep running with older langbot-plugin versions.
+        """
+        return await self.call_action(
+            _langbot_to_runtime_action('PLUGIN_DIAGNOSTIC', 'plugin_diagnostic'),
+            diagnostic,
+            timeout=5,
+        )
+
    async def list_tools(self, include_plugins: list[str] | None = None) -> list[dict[str, Any]]:
        """List tools"""
        result = await self.call_action(
@@ -417,6 +417,30 @@ 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:
@@ -6,7 +6,7 @@ import os
 import shutil
 import shlex
 import threading
-from contextlib import suppress
+from contextlib import suppress, AsyncExitStack
 from typing import TYPE_CHECKING, Any

 import pydantic
@@ -74,6 +74,35 @@ class MCPServerBoxConfig(pydantic.BaseModel):
    model_config = pydantic.ConfigDict(extra='ignore')


+_HANDSHAKE_ATTEMPT_TIMEOUT_SEC = 10.0
+
+
+class _TransferredStack:
+    """Adapts an already-populated AsyncExitStack into an async context manager
+    so ownership of its resources can be transferred into another exit stack.
+    Entering is a no-op; exiting closes the wrapped stack (and thus the live WS
+    transport + ClientSession) when the owning session shuts down."""
+
+    def __init__(self, stack: AsyncExitStack):
+        self._stack = stack
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self._stack.aclose()
+        return False
+
+
+class _ColdStartRetry(Exception):
+    """Signal: the managed process is alive but not yet answering the MCP
+    handshake because it is still cold-starting (e.g. `npx -y <pkg>` is still
+    installing). The outer lifecycle retry treats this like a transient
+    reconnect: it reuses the live process and does not count toward the fatal
+    retry budget, so a slow cold start is waited out rather than failing.
+    """
+
+
 class BoxStdioSessionRuntime:
    """Encapsulate Box-backed stdio MCP session orchestration."""

@@ -113,7 +142,11 @@ class BoxStdioSessionRuntime:
            read_only_rootfs=self.config.read_only_rootfs if self.config.read_only_rootfs is not None else False,
            image=self.config.image,
            cpus=self.config.cpus,
-            memory_mb=self.config.memory_mb,
+            # Node.js runtimes (npx/bunx) reserve large virtual address space and
+            # load WebAssembly modules (llhttp) on startup; the default 512 MB
+            # cgroup_mem_max is too small and causes OOM kills (return_code=137).
+            # Auto-bump to 1024 MB when the runner is npx/bunx/pnpm dlx.
+            memory_mb=self.config.memory_mb or 1024,
            pids_limit=self.config.pids_limit,
            persistent=True,
        )
@@ -173,28 +206,55 @@ class BoxStdioSessionRuntime:
                    stderr_preview = (result.stderr or '')[:500]
                    raise Exception(f'Dependency install failed (exit code {result.exit_code}): {stderr_preview}')

-        try:
-            process_workspace = (
-                self._build_workspace(host_path=host_path, workdir=process_cwd, mount_path=process_cwd)
-                if host_path
-                else workspace
+        # Reuse an already-running managed process instead of rebuilding it.
+        # The Box runtime keeps the managed process alive across a transient
+        # WebSocket transport drop, so on a reconnect we only need to re-attach
+        # the WS below. Rebuilding here would needlessly stop a healthy process
+        # and re-run the (slow, network-touching) dependency bootstrap.
+        if not await self._managed_process_is_running():
+            try:
+                process_workspace = (
+                    self._build_workspace(host_path=host_path, workdir=process_cwd, mount_path=process_cwd)
+                    if host_path
+                    else workspace
+                )
+                payload = process_workspace.build_process_payload(
+                    self.server_config['command'],
+                    self.server_config.get('args', []),
+                    env=self.server_config.get('env', {}),
+                    cwd=process_cwd,
+                )
+                if install_cmd:
+                    payload = self._wrap_process_payload_with_python_env(payload, process_cwd)
+                payload['process_id'] = self.process_id
+                await workspace.box_service.start_managed_process(workspace.session_id, payload)
+            except Exception:
+                self.owner.error_phase = MCPSessionErrorPhase.PROCESS_START
+                raise
+        else:
+            self.ap.logger.info(
+                f'MCP server {self.server_name}: reusing live managed process '
+                f'process_id={self.process_id} (transport reconnect)'
            )
-            payload = process_workspace.build_process_payload(
-                self.server_config['command'],
-                self.server_config.get('args', []),
-                env=self.server_config.get('env', {}),
-                cwd=process_cwd,
-            )
-            if install_cmd:
-                payload = self._wrap_process_payload_with_python_env(payload, process_cwd)
-            payload['process_id'] = self.process_id
-            await workspace.box_service.start_managed_process(workspace.session_id, payload)
-        except Exception:
-            self.owner.error_phase = MCPSessionErrorPhase.PROCESS_START
-            raise

+        websocket_url = workspace.get_managed_process_websocket_url(self.process_id)
+
+        # Attach the WS transport + MCP session ONCE, on the owner's exit stack,
+        # in the same task as the serve loop that follows. websocket_client and
+        # ClientSession use anyio task groups whose cancel scope is bound to the
+        # frame/stack that entered them, so they must live on the owner exit
+        # stack (not a deferred/transferred one) or the streams close the moment
+        # initialize() returns and the next request fails with "Connection
+        # closed".
+        #
+        # A slow (`npx -y <pkg>`) cold start makes this single attempt fail
+        # while the process is still alive — the package is still installing and
+        # cannot answer the handshake. We surface that to the outer retry loop
+        # as a _ColdStartRetry: it must NOT stop the process (it is healthy and
+        # will be reused) and must NOT consume the fatal retry budget. The next
+        # attempt re-attaches to the same live process; once it has finished
+        # cold start the handshake succeeds and stays healthy.
        try:
-            websocket_url = workspace.get_managed_process_websocket_url(self.process_id)
            transport = await self.owner.exit_stack.enter_async_context(websocket_client(websocket_url))
            read_stream, write_stream = transport
            self.owner.session = await self.owner.exit_stack.enter_async_context(
@@ -202,12 +262,19 @@ class BoxStdioSessionRuntime:
            )
        except Exception:
            self.owner.error_phase = MCPSessionErrorPhase.RELAY_CONNECT
+            if not await self._managed_process_has_exited():
+                # Process is alive but not yet serving (cold start) — reconnect.
+                raise _ColdStartRetry(f'{self.server_name}: transport not ready during cold start')
            raise

        try:
-            await self.owner.session.initialize()
-        except Exception:
+            await asyncio.wait_for(self.owner.session.initialize(), timeout=_HANDSHAKE_ATTEMPT_TIMEOUT_SEC)
+        except Exception as exc:
            self.owner.error_phase = MCPSessionErrorPhase.MCP_INIT
+            if not await self._managed_process_has_exited():
+                raise _ColdStartRetry(
+                    f'{self.server_name}: handshake not ready during cold start ({type(exc).__name__})'
+                )
            raise

    async def monitor_process_health(self) -> None:
@@ -234,8 +301,74 @@ class BoxStdioSessionRuntime:
                )
                if consecutive_errors >= self.owner._MONITOR_MAX_CONSECUTIVE_ERRORS:
                    return
+
+            # Capture stderr logs from the managed process
+            if isinstance(info, dict):
+                stderr_text = info.get('stderr', '') or info.get('stderr_preview', '')
+            else:
+                stderr_text = getattr(info, 'stderr', '') or getattr(info, 'stderr_preview', '')
+
+            if stderr_text and stderr_text != self.owner._last_stderr_text:
+                # Find new lines not in the previous snapshot
+                old_lines = set(self.owner._last_stderr_text.splitlines()) if self.owner._last_stderr_text else set()
+                new_lines = [l for l in stderr_text.splitlines() if l and l not in old_lines]
+                self.owner._last_stderr_text = stderr_text
+
+                import time as _time
+
+                for line in new_lines:
+                    level = (
+                        'error'
+                        if any(k in line.upper() for k in ('ERROR', 'CRITICAL'))
+                        else 'warning'
+                        if 'WARNING' in line.upper()
+                        else 'debug'
+                        if 'DEBUG' in line.upper()
+                        else 'info'
+                    )
+                    self.owner._log_buffer.append({'ts': _time.time(), 'level': level, 'text': line})
+
            await asyncio.sleep(self.owner._MONITOR_POLL_INTERVAL)

+    async def _managed_process_is_running(self) -> bool:
+        """Return True if this server's managed process exists and is running.
+
+        Used to decide whether initialize() must (re)start the process or can
+        simply re-attach the WebSocket transport to a process the Box runtime
+        kept alive across a transient transport drop.
+        """
+        from langbot_plugin.box.models import BoxManagedProcessStatus
+
+        workspace = self._build_workspace()
+        try:
+            info = await workspace.get_managed_process(self.process_id)
+        except Exception:
+            return False
+        status = info.get('status', '') if isinstance(info, dict) else getattr(info, 'status', '')
+        return status in (BoxManagedProcessStatus.RUNNING.value, BoxManagedProcessStatus.RUNNING)
+
+    async def _managed_process_has_exited(self) -> bool:
+        """Return True only if the process is DEFINITIVELY gone (reports EXITED).
+
+        Distinct from ``not _managed_process_is_running()``: a process that has
+        just been spawned may not yet report RUNNING, and a transient query
+        error is not proof of exit. During the cold-start handshake retry we
+        must NOT treat 'not yet running' or 'query failed' as a terminal
+        failure, or we bail out to the outer rebuild path and churn the
+        process (relay then rejects the early re-attach with HTTP 400). Only a
+        successful query that reports EXITED stops the retry loop.
+        """
+        from langbot_plugin.box.models import BoxManagedProcessStatus
+
+        workspace = self._build_workspace()
+        try:
+            info = await workspace.get_managed_process(self.process_id)
+        except Exception:
+            # Unknown — treat as 'still coming up', not exited.
+            return False
+        status = info.get('status', '') if isinstance(info, dict) else getattr(info, 'status', '')
+        return status in (BoxManagedProcessStatus.EXITED.value, BoxManagedProcessStatus.EXITED)
+
    async def _stage_host_path_to_shared_workspace(self, host_path: str) -> str:
        source_path = normalize_host_path(host_path)
        if not source_path:
@@ -342,16 +475,20 @@ class BoxStdioSessionRuntime:

        workspace = self._build_workspace(host_path=None)

-        # Transient test sessions own their isolated Box session, so tear the
-        # whole session down rather than leaking it. This cannot affect live
-        # servers because they live in the separate shared session.
+        # Transient config-page tests now share the same 'mcp-shared' Box
+        # session as live servers, so we must NOT tear the session down here —
+        # that would kill every other MCP server in the container. A test is
+        # isolated at the process level: it ran under its own process_id, so we
+        # stop only that process, exactly like a live server does below. The
+        # shared session and all other servers' live processes are untouched.
+        # (Staged per-test workspace files are still cleaned up.)
        if getattr(self.owner, 'is_transient', False):
            try:
-                await workspace.cleanup()
+                await workspace.stop_managed_process(self.process_id)
            except Exception as exc:
                self.ap.logger.warning(
-                    f'MCP server {self.server_name}: failed to delete transient test session '
-                    f'{self.owner._build_box_session_id()}: {type(exc).__name__}: {exc}'
+                    f'MCP server {self.server_name}: failed to stop transient test process '
+                    f'process_id={self.process_id}: {type(exc).__name__}: {exc}'
                )
            await self._cleanup_staged_workspace()
            return
@@ -33,6 +33,24 @@ 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():
@@ -1,6 +1,7 @@
 from __future__ import annotations

 import typing
+import time
 from typing import TYPE_CHECKING

 import langbot_plugin.api.entities.builtin.resource.tool as resource_tool
@@ -59,6 +60,7 @@ 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] = []

@@ -66,10 +68,51 @@ 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(await self.mcp_tool_loader.get_tools(bound_mcp_servers))
+        all_functions.extend(
+            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 (
@@ -100,21 +143,130 @@ class ToolManager:

        return tools

+    def _get_query_session_id(self, query: pipeline_query.Query) -> str | None:
+        launcher_type = getattr(query, 'launcher_type', None)
+        launcher_id = getattr(query, 'launcher_id', None)
+        if launcher_type is None or launcher_id is None:
+            return None
+
+        launcher_type_value = launcher_type.value if hasattr(launcher_type, 'value') else launcher_type
+        return f'{launcher_type_value}_{launcher_id}'
+
+    async def _record_tool_call(
+        self,
+        *,
+        name: str,
+        source: str,
+        parameters: dict,
+        query: pipeline_query.Query,
+        duration_ms: int,
+        status: str,
+        result: typing.Any = None,
+        error_message: str | None = None,
+    ) -> None:
+        monitoring_service = getattr(self.ap, 'monitoring_service', None)
+        if not monitoring_service:
+            return
+
+        variables = getattr(query, 'variables', {}) or {}
+        message_id = variables.get('_monitoring_message_id') if isinstance(variables, dict) else None
+        bot_name = variables.get('_monitoring_bot_name') if isinstance(variables, dict) else None
+        pipeline_name = variables.get('_monitoring_pipeline_name') if isinstance(variables, dict) else None
+
+        try:
+            await monitoring_service.record_tool_call(
+                tool_name=name,
+                tool_source=source,
+                duration=duration_ms,
+                status=status,
+                bot_id=getattr(query, 'bot_uuid', None),
+                bot_name=bot_name,
+                pipeline_name=pipeline_name,
+                session_id=self._get_query_session_id(query),
+                message_id=message_id,
+                arguments=parameters,
+                result=result,
+                error_message=error_message,
+            )
+        except Exception as e:
+            self.ap.logger.warning(f'Failed to record tool call: {e}')
+
+    async def _invoke_tool_with_monitoring(
+        self,
+        *,
+        source: str,
+        name: str,
+        parameters: dict,
+        query: pipeline_query.Query,
+        invoke: typing.Callable[[], typing.Awaitable[typing.Any]],
+    ) -> typing.Any:
+        start_time = time.perf_counter()
+        try:
+            result = await invoke()
+        except Exception as e:
+            duration_ms = int((time.perf_counter() - start_time) * 1000)
+            await self._record_tool_call(
+                name=name,
+                source=source,
+                parameters=parameters,
+                query=query,
+                duration_ms=duration_ms,
+                status='error',
+                error_message=str(e),
+            )
+            raise
+
+        duration_ms = int((time.perf_counter() - start_time) * 1000)
+        await self._record_tool_call(
+            name=name,
+            source=source,
+            parameters=parameters,
+            query=query,
+            duration_ms=duration_ms,
+            status='success',
+            result=result,
+        )
+        return result
+
    async def execute_func_call(self, name: str, parameters: dict, query: pipeline_query.Query) -> typing.Any:
        from langbot.pkg.telemetry import features as telemetry_features

        if await self.native_tool_loader.has_tool(name):
            telemetry_features.increment(query, 'tool_calls', 'native')
-            return await self.native_tool_loader.invoke_tool(name, parameters, query)
+            return await self._invoke_tool_with_monitoring(
+                source='native',
+                name=name,
+                parameters=parameters,
+                query=query,
+                invoke=lambda: self.native_tool_loader.invoke_tool(name, parameters, query),
+            )
        if await self.plugin_tool_loader.has_tool(name):
            telemetry_features.increment(query, 'tool_calls', 'plugin')
-            return await self.plugin_tool_loader.invoke_tool(name, parameters, query)
+            return await self._invoke_tool_with_monitoring(
+                source='plugin',
+                name=name,
+                parameters=parameters,
+                query=query,
+                invoke=lambda: self.plugin_tool_loader.invoke_tool(name, parameters, query),
+            )
        if await self.mcp_tool_loader.has_tool(name):
            telemetry_features.increment(query, 'tool_calls', 'mcp')
-            return await self.mcp_tool_loader.invoke_tool(name, parameters, query)
+            return await self._invoke_tool_with_monitoring(
+                source='mcp',
+                name=name,
+                parameters=parameters,
+                query=query,
+                invoke=lambda: self.mcp_tool_loader.invoke_tool(name, parameters, query),
+            )
        if await self.skill_tool_loader.has_tool(name):
            telemetry_features.increment(query, 'tool_calls', 'skill')
-            return await self.skill_tool_loader.invoke_tool(name, parameters, query)
+            return await self._invoke_tool_with_monitoring(
+                source='skill',
+                name=name,
+                parameters=parameters,
+                query=query,
+                invoke=lambda: self.skill_tool_loader.invoke_tool(name, parameters, query),
+            )
        raise ToolNotFoundError(name)

    async def shutdown(self):
@@ -1,4 +1,3 @@
-admins: []
 api:
    port: 5300
    webhook_prefix: 'http://127.0.0.1:5300'
@@ -118,20 +118,6 @@ stages:
        default:
          - role: system
            content: "You are a helpful assistant."
-      - name: knowledge-bases
-        label:
-          en_US: Knowledge Bases
-          zh_Hans: 知识库
-        description:
-          en_US: Configure the knowledge bases to use for the agent, if not selected, the agent will directly use the LLM to reply
-          zh_Hans: 配置用于提升回复质量的知识库，若不选择，则直接使用大模型回复
-        type: knowledge-base-multi-selector
-        required: false
-        default: []
-        show_if:
-          field: __system.is_wizard
-          operator: neq
-          value: true
      - name: box-session-id-template
        label:
          en_US: Sandbox Scope
@@ -254,6 +240,34 @@ stages:
          field: rerank-model
          operator: neq
          value: ''
+      - name: tools
+        label:
+          en_US: Tools
+          zh_Hans: 工具
+        description:
+          en_US: Select plugin, MCP, skill, and built-in tools available to this Local Agent.
+          zh_Hans: 选择此内置 Agent 可以调用的插件、MCP、技能和内置工具。
+        type: rich-tools-selector
+        required: false
+        default: []
+        show_if:
+          field: __system.is_wizard
+          operator: neq
+          value: true
+      - name: knowledge-bases
+        label:
+          en_US: Resources
+          zh_Hans: 资源
+        description:
+          en_US: Select MCP resources and knowledge bases available to this Local Agent.
+          zh_Hans: 选择此内置 Agent 可以读取的 MCP 资源和知识库。
+        type: resources-selector
+        required: false
+        default: []
+        show_if:
+          field: __system.is_wizard
+          operator: neq
+          value: true
  - name: dify-service-api
    label:
      en_US: Dify Service API
@@ -81,6 +81,7 @@ def fake_monitoring_app():
    )
    app.monitoring_service.get_messages = AsyncMock(return_value=([{'id': 'msg-1', 'content': 'test'}], 100))
    app.monitoring_service.get_llm_calls = AsyncMock(return_value=([{'id': 'llm-1'}], 50))
+    app.monitoring_service.get_tool_calls = AsyncMock(return_value=([{'id': 'tool-1'}], 5))
    app.monitoring_service.get_embedding_calls = AsyncMock(return_value=([{'id': 'emb-1'}], 10))
    app.monitoring_service.get_sessions = AsyncMock(return_value=([{'session_id': 'sess-1'}], 20))
    app.monitoring_service.get_errors = AsyncMock(return_value=([{'id': 'err-1'}], 2))
@@ -662,6 +662,100 @@ class TestSendResponseBackStage:
        assert len(outbound) == 1
        assert outbound[0]['type'] == 'reply'

+    @pytest.mark.asyncio
+    async def test_send_response_failure_notifies_plugin_diagnostic(self, pipeline_app):
+        """Plugin-provided deferred replies should report delivery failures."""
+        from langbot.pkg.pipeline import plugin_diagnostics
+        from langbot.pkg.pipeline.respback import respback
+        from tests.factories.message import text_chain
+        from langbot_plugin.api.entities.builtin.provider.message import Message
+
+        query = text_query('hello')
+        query.adapter.reply_message.side_effect = RuntimeError('send failed')
+        query.pipeline_config = create_minimal_pipeline_config()
+        query.current_stage_name = 'SendResponseBackStage'
+        query.resp_messages = [Message(role='assistant', content='test response')]
+        query.resp_message_chain = [text_chain('test response')]
+        plugin_diagnostics.record_plugin_response_source(
+            query,
+            0,
+            [
+                {
+                    'kind': 'reply_message_chain',
+                    'plugin': {'author': 'tester', 'name': 'demo'},
+                }
+            ],
+            [{'manifest': {'metadata': {'author': 'observer', 'name': 'not-reply-source'}}}],
+            'NormalMessageResponded',
+        )
+        pipeline_app.plugin_connector.notify_plugin_diagnostic = AsyncMock()
+
+        respback_stage = respback.SendResponseBackStage(pipeline_app)
+
+        with pytest.raises(RuntimeError, match='send failed'):
+            await respback_stage.process(query, 'SendResponseBackStage')
+
+        pipeline_app.plugin_connector.notify_plugin_diagnostic.assert_awaited_once()
+        payload = pipeline_app.plugin_connector.notify_plugin_diagnostic.await_args.args[0]
+        assert payload['code'] == 'response_delivery_failed'
+        assert payload['plugin'] == {'author': 'tester', 'name': 'demo'}
+        assert payload['query']['event_name'] == 'NormalMessageResponded'
+        assert payload['delivery']['error_type'] == 'RuntimeError'
+        assert 'attribution_warning' not in payload['details']
+
+    @pytest.mark.asyncio
+    async def test_send_response_failure_warns_for_old_runtime_attribution(self, pipeline_app):
+        """Older plugin runtimes without response_sources should get approximate diagnostics."""
+        from langbot.pkg.pipeline import plugin_diagnostics
+        from langbot.pkg.pipeline.respback import respback
+        from tests.factories.message import text_chain
+        from langbot_plugin.api.entities.builtin.provider.message import Message
+
+        query = text_query('hello')
+        query.adapter.reply_message.side_effect = RuntimeError('send failed')
+        query.pipeline_config = create_minimal_pipeline_config()
+        query.resp_messages = [Message(role='assistant', content='test response')]
+        query.resp_message_chain = [text_chain('test response')]
+        plugin_diagnostics.record_plugin_response_source(
+            query,
+            0,
+            None,
+            [{'manifest': {'metadata': {'author': 'tester', 'name': 'demo'}}}],
+            'NormalMessageResponded',
+        )
+        pipeline_app.plugin_connector.notify_plugin_diagnostic = AsyncMock()
+
+        respback_stage = respback.SendResponseBackStage(pipeline_app)
+
+        with pytest.raises(RuntimeError, match='send failed'):
+            await respback_stage.process(query, 'SendResponseBackStage')
+
+        payload = pipeline_app.plugin_connector.notify_plugin_diagnostic.await_args.args[0]
+        assert payload['plugin'] == {'author': 'tester', 'name': 'demo'}
+        assert 'attribution_warning' in payload['details']
+
+    @pytest.mark.asyncio
+    async def test_send_response_failure_ignores_query_variable_spoofing(self, pipeline_app):
+        """Plugin-controlled query variables must not mask delivery failures."""
+        from langbot.pkg.pipeline.respback import respback
+        from tests.factories.message import text_chain
+        from langbot_plugin.api.entities.builtin.provider.message import Message
+
+        query = text_query('hello')
+        query.adapter.reply_message.side_effect = RuntimeError('send failed')
+        query.pipeline_config = create_minimal_pipeline_config()
+        query.resp_messages = [Message(role='assistant', content='test response')]
+        query.resp_message_chain = [text_chain('test response')]
+        query.variables['_plugin_response_sources'] = {0: ['malformed']}
+        pipeline_app.plugin_connector.notify_plugin_diagnostic = AsyncMock()
+
+        respback_stage = respback.SendResponseBackStage(pipeline_app)
+
+        with pytest.raises(RuntimeError, match='send failed'):
+            await respback_stage.process(query, 'SendResponseBackStage')
+
+        pipeline_app.plugin_connector.notify_plugin_diagnostic.assert_not_called()
+

@pytest.mark.usefixtures('mock_circular_import_chain')
 class TestStageChainIntegration:
@@ -90,6 +90,56 @@ class TestMCPServiceGetRuntimeInfo:
        assert result is None


+class TestMCPServiceResources:
+    """Tests for MCP resource helpers."""
+
+    async def test_get_resource_templates_delegates_to_loader(self):
+        ap = SimpleNamespace()
+        ap.tool_mgr = SimpleNamespace()
+        ap.tool_mgr.mcp_tool_loader = SimpleNamespace()
+        ap.tool_mgr.mcp_tool_loader.get_resource_templates = AsyncMock(
+            return_value=[{'uri_template': 'file:///{path}', 'name': 'files'}]
+        )
+
+        service = MCPService(ap)
+
+        result = await service.get_mcp_server_resource_templates('docs')
+
+        assert result == [{'uri_template': 'file:///{path}', 'name': 'files'}]
+        ap.tool_mgr.mcp_tool_loader.get_resource_templates.assert_awaited_once_with('docs')
+
+    async def test_read_resource_envelope_uses_ui_preview_source(self):
+        ap = SimpleNamespace()
+        ap.tool_mgr = SimpleNamespace()
+        ap.tool_mgr.mcp_tool_loader = SimpleNamespace()
+        ap.tool_mgr.mcp_tool_loader.read_resource_envelope = AsyncMock(
+            return_value={
+                'server_name': 'docs',
+                'uri': 'file:///README.md',
+                'contents': [],
+                'source': 'ui_preview',
+            }
+        )
+
+        service = MCPService(ap)
+
+        result = await service.read_mcp_server_resource_envelope(
+            'docs',
+            'file:///README.md',
+            max_bytes=4096,
+            include_blob=True,
+        )
+
+        assert result['source'] == 'ui_preview'
+        ap.tool_mgr.mcp_tool_loader.read_resource_envelope.assert_awaited_once_with(
+            'docs',
+            'file:///README.md',
+            include_blob=True,
+            source='ui_preview',
+            max_bytes=4096,
+        )
+
+
 class TestMCPServiceGetMCPServers:
    """Tests for get_mcp_servers method."""

@@ -230,6 +280,25 @@ class TestMCPServiceCreateMCPServer:
        assert server_uuid is not None
        assert len(server_uuid) == 36  # UUID format

+    async def test_create_mcp_server_duplicate_name_raises(self):
+        """Rejects duplicate MCP server names."""
+        # Setup
+        ap = SimpleNamespace()
+        ap.persistence_mgr = SimpleNamespace()
+        ap.instance_config = SimpleNamespace()
+        ap.instance_config.data = {'system': {'limitation': {'max_extensions': -1}}}
+        ap.tool_mgr = None
+
+        existing_server = _create_mock_mcp_server(name='Existing Server')
+        ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_mock_result(first_item=existing_server))
+        ap.persistence_mgr.serialize_model = Mock(return_value={})
+
+        service = MCPService(ap)
+
+        # Execute & Verify
+        with pytest.raises(ValueError, match='MCP server already exists: Existing Server'):
+            await service.create_mcp_server({'name': 'Existing Server'})
+
    async def test_create_mcp_server_loads_server(self):
        """Loads server into tool_mgr when enabled."""
        # Setup
@@ -251,7 +320,7 @@ class TestMCPServiceCreateMCPServer:
            nonlocal call_count
            call_count += 1
            if call_count == 1:
-                return _create_mock_result([])  # Empty list for limit check
+                return _create_mock_result([])  # Empty result for duplicate-name check
            elif call_count == 2:
                return Mock()  # Insert
            return _create_mock_result(first_item=server_entity)  # Select created
@@ -348,6 +348,8 @@ class TestPipelineServiceCreatePipeline:
            'enable_all_mcp_servers': True,
            'plugins': [],
            'mcp_servers': [],
+            'mcp_resources': [],
+            'mcp_resource_agent_read_enabled': True,
        }


@@ -814,6 +816,47 @@ class TestPipelineServiceUpdatePipelineExtensions:
        # Verify - persistence was called
        ap.persistence_mgr.execute_async.assert_called()

+    async def test_update_extensions_preserves_mcp_resource_agent_read_when_omitted(self):
+        """Does not reset mcp_resource_agent_read_enabled when omitted by older clients."""
+        ap = SimpleNamespace()
+        ap.persistence_mgr = SimpleNamespace()
+        ap.pipeline_mgr = SimpleNamespace()
+        ap.pipeline_mgr.remove_pipeline = AsyncMock()
+        ap.pipeline_mgr.load_pipeline = AsyncMock()
+
+        original_pipeline = _create_mock_pipeline(
+            extensions_preferences={
+                'enable_all_plugins': True,
+                'enable_all_mcp_servers': True,
+                'plugins': [],
+                'mcp_servers': [],
+                'mcp_resources': [{'server_uuid': 'srv-1', 'uri': 'file:///README.md'}],
+                'mcp_resource_agent_read_enabled': False,
+            }
+        )
+
+        call_count = 0
+
+        async def mock_execute(query):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                return _create_mock_result(first_item=original_pipeline)
+            return Mock()
+
+        ap.persistence_mgr.execute_async = AsyncMock(side_effect=mock_execute)
+        ap.persistence_mgr.serialize_model = Mock(return_value={'uuid': 'test-uuid'})
+
+        service = PipelineService(ap)
+        service.get_pipeline = AsyncMock(return_value={'uuid': 'test-uuid'})
+
+        await service.update_pipeline_extensions('test-uuid', bound_plugins=[])
+
+        assert original_pipeline.extensions_preferences['mcp_resource_agent_read_enabled'] is False
+        assert original_pipeline.extensions_preferences['mcp_resources'] == [
+            {'server_uuid': 'srv-1', 'uri': 'file:///README.md'}
+        ]
+

 class TestDefaultStageOrder:
    """Tests for default_stage_order constant."""
@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import sys
+import types
+from importlib import import_module
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+import quart
+
+core_app_module = types.ModuleType('langbot.pkg.core.app')
+core_app_module.Application = object
+sys.modules.setdefault('langbot.pkg.core.app', core_app_module)
+
+
+pytestmark = pytest.mark.asyncio
+
+
+async def _create_test_client(mcp_service: SimpleNamespace):
+    app = quart.Quart(__name__)
+    user_service = SimpleNamespace(
+        verify_jwt_token=AsyncMock(return_value='test@example.com'),
+        get_user_by_email=AsyncMock(return_value=SimpleNamespace(user='test@example.com')),
+    )
+    ap = SimpleNamespace(mcp_service=mcp_service, user_service=user_service)
+    MCPRouterGroup = import_module('langbot.pkg.api.http.controller.groups.resources.mcp').MCPRouterGroup
+    group = MCPRouterGroup(ap, app)
+    await group.initialize()
+    return app.test_client()
+
+
+async def test_mcp_server_route_accepts_encoded_slash_name():
+    mcp_service = SimpleNamespace(
+        get_mcp_server_by_name=AsyncMock(
+            return_value={
+                'uuid': 'test-uuid',
+                'name': 'pab1it0/prometheus',
+                'enable': True,
+                'mode': 'stdio',
+                'extra_args': {},
+            }
+        )
+    )
+    client = await _create_test_client(mcp_service)
+
+    response = await client.get(
+        '/api/v1/mcp/servers/pab1it0%2Fprometheus',
+        headers={'Authorization': 'Bearer test-token'},
+    )
+
+    assert response.status_code == 200
+    mcp_service.get_mcp_server_by_name.assert_awaited_once_with('pab1it0/prometheus')
+    payload = await response.get_json()
+    assert payload['data']['server']['name'] == 'pab1it0/prometheus'
+
+
+async def test_mcp_resource_route_accepts_encoded_slash_name():
+    mcp_service = SimpleNamespace(
+        get_mcp_server_by_name=AsyncMock(),
+        get_mcp_server_resources=AsyncMock(return_value=[]),
+        get_mcp_server_resource_templates=AsyncMock(return_value=[]),
+        get_runtime_info=AsyncMock(return_value={'resource_capabilities': {'subscribe': False}}),
+    )
+    client = await _create_test_client(mcp_service)
+
+    response = await client.get(
+        '/api/v1/mcp/servers/pab1it0%2Fprometheus/resources',
+        headers={'Authorization': 'Bearer test-token'},
+    )
+
+    assert response.status_code == 200
+    mcp_service.get_mcp_server_by_name.assert_not_awaited()
+    mcp_service.get_mcp_server_resources.assert_awaited_once_with('pab1it0/prometheus')
+    payload = await response.get_json()
+    assert payload['data']['resource_capabilities'] == {'subscribe': False}
@@ -162,3 +162,46 @@ async def test_runtime_pipeline_execute(mock_app, sample_query):

    # Verify stage was called
    mock_stage.process.assert_called_once()
+
+
+def test_runtime_pipeline_prefers_local_agent_mcp_resources(mock_app):
+    """Local Agent resource selection should override legacy extension prefs."""
+    pipelinemgr = get_pipelinemgr_module()
+    persistence_pipeline = get_persistence_pipeline_module()
+
+    pipeline_entity = Mock(spec=persistence_pipeline.LegacyPipeline)
+    pipeline_entity.config = {
+        'ai': {
+            'local-agent': {
+                'mcp-resources': [{'server_uuid': 'srv-new', 'uri': 'file:///new.md'}],
+                'mcp-resource-agent-read-enabled': False,
+            }
+        }
+    }
+    pipeline_entity.extensions_preferences = {
+        'mcp_resources': [{'server_uuid': 'srv-old', 'uri': 'file:///old.md'}],
+        'mcp_resource_agent_read_enabled': True,
+    }
+
+    runtime_pipeline = pipelinemgr.RuntimePipeline(mock_app, pipeline_entity, [])
+
+    assert runtime_pipeline.mcp_resource_attachments == [{'server_uuid': 'srv-new', 'uri': 'file:///new.md'}]
+    assert runtime_pipeline.mcp_resource_agent_read_enabled is False
+
+
+def test_runtime_pipeline_falls_back_to_extension_mcp_resources(mock_app):
+    """Existing extension prefs remain compatible until a Local Agent value exists."""
+    pipelinemgr = get_pipelinemgr_module()
+    persistence_pipeline = get_persistence_pipeline_module()
+
+    pipeline_entity = Mock(spec=persistence_pipeline.LegacyPipeline)
+    pipeline_entity.config = {'ai': {'local-agent': {}}}
+    pipeline_entity.extensions_preferences = {
+        'mcp_resources': [{'server_uuid': 'srv-old', 'uri': 'file:///old.md'}],
+        'mcp_resource_agent_read_enabled': False,
+    }
+
+    runtime_pipeline = pipelinemgr.RuntimePipeline(mock_app, pipeline_entity, [])
+
+    assert runtime_pipeline.mcp_resource_attachments == [{'server_uuid': 'srv-old', 'uri': 'file:///old.md'}]
+    assert runtime_pipeline.mcp_resource_agent_read_enabled is False
@@ -14,6 +14,7 @@ from __future__ import annotations
 import pytest
 from unittest.mock import AsyncMock, Mock
 from importlib import import_module
+from types import SimpleNamespace

 from tests.factories import (
    FakeApp,
@@ -431,3 +432,60 @@ class TestPreProcessorVariables:
        variables = result.new_query.variables
        assert 'group_name' in variables
        assert 'sender_name' in variables
+
+
+class TestPreProcessorToolSelection:
+    """Tests for Local Agent tool selection."""
+
+    @pytest.mark.asyncio
+    async def test_local_agent_filters_selected_tools(self):
+        """Only selected tools should be exposed when all-tools mode is off."""
+        preproc = get_preproc_module()
+
+        app = FakeApp()
+        mock_session = Mock()
+        mock_session.launcher_type = Mock(value='person')
+        mock_session.launcher_id = 12345
+        app.sess_mgr.get_session = AsyncMock(return_value=mock_session)
+
+        mock_conversation = Mock()
+        mock_conversation.prompt = Mock(messages=[])
+        mock_conversation.prompt.copy = Mock(return_value=Mock(messages=[]))
+        mock_conversation.messages = []
+        mock_conversation.uuid = None
+        app.sess_mgr.get_conversation = AsyncMock(return_value=mock_conversation)
+
+        mock_model = Mock()
+        mock_model.model_entity = Mock(uuid='primary-model-uuid', abilities=['func_call'])
+        app.model_mgr.get_model_by_uuid = AsyncMock(return_value=mock_model)
+        app.tool_mgr.get_all_tools = AsyncMock(
+            return_value=[
+                SimpleNamespace(name='exec'),
+                SimpleNamespace(name='plugin_tool'),
+                SimpleNamespace(name='mcp_tool'),
+            ]
+        )
+
+        mock_event_ctx = Mock()
+        mock_event_ctx.event = Mock(default_prompt=[], prompt=[])
+        app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
+
+        stage = preproc.PreProcessor(app)
+        query = text_query('hello')
+        query.pipeline_config = {
+            'ai': {
+                'runner': {'runner': 'local-agent'},
+                'local-agent': {
+                    'model': {'primary': 'primary-model-uuid', 'fallbacks': []},
+                    'prompt': 'default',
+                    'enable-all-tools': False,
+                    'tools': ['plugin_tool'],
+                },
+            },
+            'output': {'misc': {'at-sender': False}},
+            'trigger': {'misc': {}},
+        }
+
+        result = await stage.process(query, 'PreProcessor')
+
+        assert [tool.name for tool in result.new_query.use_funcs] == ['plugin_tool']
@@ -36,6 +36,11 @@ def get_entities_module():
    return import_module('langbot.pkg.pipeline.entities')


+def get_plugin_diagnostics_module():
+    """Lazy import for plugin diagnostic attribution helpers."""
+    return import_module('langbot.pkg.pipeline.plugin_diagnostics')
+
+
 def make_wrapper_config():
    """Create a pipeline config for wrapper tests."""
    return {
@@ -106,6 +111,45 @@ class TestResponseWrapperMessageChain:
        assert results[0].result_type == entities.ResultType.CONTINUE
        assert len(results[0].new_query.resp_message_chain) == 1

+    @pytest.mark.asyncio
+    async def test_message_chain_direct_append_consumes_pending_plugin_source(self):
+        """MessageChain replies from earlier plugin events keep attribution."""
+        wrapper = get_wrapper_module()
+
+        app = FakeApp()
+        stage = wrapper.ResponseWrapper(app)
+        await stage.initialize(make_wrapper_config())
+
+        reply_chain = platform_message.MessageChain([platform_message.Plain(text='response')])
+        query = text_query('hello')
+        query.pipeline_config = make_wrapper_config()
+        query.resp_messages = [reply_chain]
+        query.resp_message_chain = []
+        plugin_diagnostics = get_plugin_diagnostics_module()
+        plugin_diagnostics.record_pending_plugin_response_source(
+            query,
+            reply_chain,
+            [
+                {
+                    'kind': 'reply_message_chain',
+                    'plugin': {'author': 'tester', 'name': 'demo'},
+                }
+            ],
+            [{'manifest': {'metadata': {'author': 'observer', 'name': 'not-reply-source'}}}],
+            'PersonNormalMessageReceived',
+        )
+
+        results = []
+        async for result in stage.process(query, 'ResponseWrapper'):
+            results.append(result)
+
+        sources = plugin_diagnostics._get_response_sources(results[0].new_query, 0)
+        assert sources[0].plugin == {'author': 'tester', 'name': 'demo'}
+        assert sources[0].event_name == 'PersonNormalMessageReceived'
+        assert sources[0].is_approximate is False
+        assert '_plugin_response_sources' not in query.variables
+        assert '_plugin_pending_response_sources' not in query.variables
+

 class TestResponseWrapperCommand:
    """Tests for command response wrapping."""
@@ -421,6 +465,104 @@ class TestResponseWrapperCustomReply:
        chain = results[0].new_query.resp_message_chain[0]
        assert 'Custom reply' in str(chain)

+    @pytest.mark.asyncio
+    async def test_custom_reply_records_plugin_source(self):
+        """Plugin reply_message_chain should keep emitted plugin attribution."""
+        wrapper = get_wrapper_module()
+
+        app = FakeApp()
+        app.sess_mgr.get_session = AsyncMock(return_value=make_session())
+
+        custom_chain = platform_message.MessageChain([platform_message.Plain(text='Custom reply')])
+        mock_event_ctx = Mock()
+        mock_event_ctx.is_prevented_default = Mock(return_value=False)
+        mock_event_ctx.event = Mock()
+        mock_event_ctx.event.reply_message_chain = custom_chain
+        mock_event_ctx._emitted_plugins = [
+            {
+                'manifest': {'metadata': {'author': 'observer', 'name': 'not-reply-source'}},
+                'plugin_config': {'token': 'secret-token'},
+            },
+        ]
+        mock_event_ctx._response_sources = [
+            {
+                'kind': 'reply_message_chain',
+                'plugin': {'author': 'tester', 'name': 'demo'},
+            }
+        ]
+        app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
+
+        stage = wrapper.ResponseWrapper(app)
+        pipeline_config = make_wrapper_config()
+        await stage.initialize(pipeline_config)
+
+        query = text_query('hello')
+        query.pipeline_config = pipeline_config
+        query.resp_message_chain = []
+        assistant_resp = Mock()
+        assistant_resp.role = 'assistant'
+        assistant_resp.content = 'Default reply'
+        assistant_resp.tool_calls = None
+        assistant_resp.get_content_platform_message_chain = Mock(
+            return_value=platform_message.MessageChain([platform_message.Plain(text='Default reply')])
+        )
+        query.resp_messages = [assistant_resp]
+
+        results = []
+        async for result in stage.process(query, 'ResponseWrapper'):
+            results.append(result)
+
+        plugin_diagnostics = get_plugin_diagnostics_module()
+        sources = plugin_diagnostics._get_response_sources(results[0].new_query, 0)
+        assert sources[0].plugin == {'author': 'tester', 'name': 'demo'}
+        assert sources[0].event_name == 'NormalMessageResponded'
+        assert sources[0].is_approximate is False
+        assert 'secret-token' not in str(sources)
+        assert '_plugin_response_sources' not in query.variables
+
+    @pytest.mark.asyncio
+    async def test_custom_reply_falls_back_to_emitted_plugins_for_old_runtime(self):
+        """Older plugin runtimes without response_sources keep approximate attribution."""
+        wrapper = get_wrapper_module()
+
+        app = FakeApp()
+        app.sess_mgr.get_session = AsyncMock(return_value=make_session())
+
+        custom_chain = platform_message.MessageChain([platform_message.Plain(text='Custom reply')])
+        mock_event_ctx = Mock()
+        mock_event_ctx.is_prevented_default = Mock(return_value=False)
+        mock_event_ctx.event = Mock()
+        mock_event_ctx.event.reply_message_chain = custom_chain
+        mock_event_ctx._emitted_plugins = [
+            {'manifest': {'metadata': {'author': 'tester', 'name': 'demo'}}},
+        ]
+        app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
+
+        stage = wrapper.ResponseWrapper(app)
+        pipeline_config = make_wrapper_config()
+        await stage.initialize(pipeline_config)
+
+        query = text_query('hello')
+        query.pipeline_config = pipeline_config
+        query.resp_message_chain = []
+        assistant_resp = Mock()
+        assistant_resp.role = 'assistant'
+        assistant_resp.content = 'Default reply'
+        assistant_resp.tool_calls = None
+        assistant_resp.get_content_platform_message_chain = Mock(
+            return_value=platform_message.MessageChain([platform_message.Plain(text='Default reply')])
+        )
+        query.resp_messages = [assistant_resp]
+
+        results = []
+        async for result in stage.process(query, 'ResponseWrapper'):
+            results.append(result)
+
+        plugin_diagnostics = get_plugin_diagnostics_module()
+        sources = plugin_diagnostics._get_response_sources(results[0].new_query, 0)
+        assert sources[0].plugin == {'author': 'tester', 'name': 'demo'}
+        assert sources[0].is_approximate is True
+

 class TestResponseWrapperVariables:
    """Tests for bound plugins variable."""
@@ -0,0 +1,538 @@
+import pytest
+import aiocqhttp
+
+import langbot_plugin.api.entities.builtin.platform.message as platform_message
+from langbot.pkg.platform.sources.aiocqhttp import (
+    AiocqhttpAdapter,
+    AiocqhttpEventConverter,
+    AiocqhttpMessageConverter,
+)
+
+
+async def _convert_single(component: platform_message.MessageComponent):
+    chain = platform_message.MessageChain([component])
+    message, _, _ = await AiocqhttpMessageConverter.yiri2target(chain)
+    return message[0]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ('payload', 'expected'),
+    [
+        ('data:image/jpeg;base64,raw-image', 'base64://raw-image'),
+        ('raw-image', 'base64://raw-image'),
+        ('base64://raw-image', 'base64://raw-image'),
+    ],
+)
+async def test_image_base64_payload_is_normalized(payload, expected):
+    segment = await _convert_single(platform_message.Image(base64=payload))
+
+    assert segment.type == 'image'
+    assert segment.data['file'] == expected
+
+
+@pytest.mark.asyncio
+async def test_voice_data_uri_base64_payload_is_normalized():
+    segment = await _convert_single(platform_message.Voice(base64='data:audio/wav;base64,raw-voice'))
+
+    assert segment.type == 'record'
+    assert segment.data['file'] == 'base64://raw-voice'
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ('component', 'expected'),
+    [
+        (
+            platform_message.File(name='report.txt', base64='data:text/plain;base64,raw-file'),
+            {'file': 'base64://raw-file', 'name': 'report.txt'},
+        ),
+        (
+            platform_message.File(name='report.txt', base64='raw-file'),
+            {'file': 'base64://raw-file', 'name': 'report.txt'},
+        ),
+        (
+            platform_message.File(name='a.txt', url='http://example.com/a.txt'),
+            {'file': 'http://example.com/a.txt', 'name': 'a.txt'},
+        ),
+        (
+            platform_message.File(name='a.txt', path='/tmp/a.txt'),
+            {'file': '/tmp/a.txt', 'name': 'a.txt'},
+        ),
+    ],
+)
+async def test_file_message_uses_available_file_source(component, expected):
+    segment = await _convert_single(component)
+
+    assert segment.type == 'file'
+    assert segment.data == expected
+
+
+@pytest.mark.asyncio
+async def test_forward_image_base64_payload_is_normalized():
+    forward = platform_message.Forward(
+        node_list=[
+            platform_message.ForwardMessageNode(
+                sender_id='10001',
+                sender_name='Tester',
+                message_chain=platform_message.MessageChain(
+                    [platform_message.Image(base64='data:image/png;base64,raw-forward-image')]
+                ),
+            )
+        ]
+    )
+    messages = []
+
+    class Logger:
+        async def info(self, _message):
+            return None
+
+        async def error(self, _message):
+            return None
+
+    class Bot:
+        async def call_action(self, action, **kwargs):
+            assert action == 'send_forward_msg'
+            messages.append(kwargs)
+
+    platform = AiocqhttpAdapter.model_construct(
+        bot_account_id='10000',
+        config={},
+        logger=Logger(),
+        bot=Bot(),
+    )
+
+    await platform._send_forward_message(1000, forward)
+
+    assert messages[0]['messages'][0]['data']['content'][0] == {
+        'type': 'image',
+        'data': {'file': 'base64://raw-forward-image'},
+    }
+
+
+@pytest.mark.asyncio
+async def test_group_message_member_name_prefers_group_card():
+    event = aiocqhttp.Event(
+        {
+            'post_type': 'message',
+            'message_type': 'group',
+            'message_id': 1000,
+            'message': '',
+            'time': 1776491725,
+            'group_id': 2000,
+            'sender': {
+                'user_id': 3000,
+                'nickname': 'QQ Nickname',
+                'card': 'Group Card',
+                'role': 'member',
+                'title': 'Special Title',
+            },
+        }
+    )
+
+    class Bot:
+        async def get_group_info(self, group_id):
+            assert group_id == 2000
+            return {'group_id': group_id, 'group_name': 'Test Group'}
+
+    converted = await AiocqhttpEventConverter().target2yiri(event, Bot())
+
+    assert converted.sender.member_name == 'Group Card'
+    assert converted.sender.group.id == 2000
+    assert converted.sender.group.name == 'Test Group'
+    assert converted.sender.special_title == 'Special Title'
+
+
+@pytest.mark.asyncio
+async def test_group_message_member_name_falls_back_to_nickname():
+    event = aiocqhttp.Event(
+        {
+            'post_type': 'message',
+            'message_type': 'group',
+            'message_id': 1000,
+            'message': '',
+            'time': 1776491725,
+            'group_id': 2000,
+            'sender': {
+                'user_id': 3000,
+                'nickname': 'QQ Nickname',
+                'card': '',
+                'role': 'member',
+            },
+        }
+    )
+
+    converted = await AiocqhttpEventConverter().target2yiri(event)
+
+    assert converted.sender.member_name == 'QQ Nickname'
+
+
+@pytest.mark.asyncio
+async def test_group_message_special_title_uses_group_member_info_when_sender_title_is_empty():
+    event = aiocqhttp.Event(
+        {
+            'post_type': 'message',
+            'message_type': 'group',
+            'message_id': 1000,
+            'message': '',
+            'time': 1776491725,
+            'group_id': 2000,
+            'sender': {
+                'user_id': 3000,
+                'nickname': 'QQ Nickname',
+                'card': 'Group Card',
+                'role': 'member',
+                'title': '',
+            },
+        }
+    )
+
+    class Bot:
+        async def get_group_info(self, group_id):
+            return {'group_id': group_id, 'group_name': 'Test Group'}
+
+        async def get_group_member_info(self, group_id, user_id):
+            assert group_id == 2000
+            assert user_id == 3000
+            return {'group_id': group_id, 'user_id': user_id, 'title': 'Member Title'}
+
+    converted = await AiocqhttpEventConverter().target2yiri(event, Bot())
+
+    assert converted.sender.special_title == 'Member Title'
+
+
+@pytest.mark.asyncio
+async def test_group_message_special_title_does_not_lookup_when_sender_title_exists():
+    event = aiocqhttp.Event(
+        {
+            'post_type': 'message',
+            'message_type': 'group',
+            'message_id': 1000,
+            'message': '',
+            'time': 1776491725,
+            'group_id': 2000,
+            'sender': {
+                'user_id': 3000,
+                'nickname': 'QQ Nickname',
+                'card': 'Group Card',
+                'role': 'member',
+                'title': 'Event Title',
+            },
+        }
+    )
+
+    class Bot:
+        async def get_group_info(self, group_id):
+            return {'group_id': group_id, 'group_name': 'Test Group'}
+
+        async def get_group_member_info(self, group_id, user_id):
+            raise AssertionError('get_group_member_info should not be called')
+
+    converted = await AiocqhttpEventConverter().target2yiri(event, Bot())
+
+    assert converted.sender.special_title == 'Event Title'
+
+
+@pytest.mark.asyncio
+async def test_group_message_special_title_member_info_failure_is_cached(monkeypatch):
+    event = aiocqhttp.Event(
+        {
+            'post_type': 'message',
+            'message_type': 'group',
+            'message_id': 1000,
+            'message': '',
+            'time': 1776491725,
+            'group_id': 2000,
+            'sender': {
+                'user_id': 3000,
+                'nickname': 'QQ Nickname',
+                'card': 'Group Card',
+                'role': 'member',
+                'title': '',
+            },
+        }
+    )
+    now = 1000.0
+
+    class Bot:
+        member_info_calls = 0
+
+        async def get_group_info(self, group_id):
+            return {'group_id': group_id, 'group_name': 'Test Group'}
+
+        async def get_group_member_info(self, group_id, user_id):
+            self.member_info_calls += 1
+            raise RuntimeError('api unavailable')
+
+    monkeypatch.setattr('langbot.pkg.platform.sources.aiocqhttp.time.monotonic', lambda: now)
+
+    bot = Bot()
+    converter = AiocqhttpEventConverter()
+
+    first = await converter.target2yiri(event, bot)
+    second = await converter.target2yiri(event, bot)
+
+    assert first.sender.special_title == ''
+    assert second.sender.special_title == ''
+    assert bot.member_info_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_group_message_special_title_member_info_cache_expires(monkeypatch):
+    event = aiocqhttp.Event(
+        {
+            'post_type': 'message',
+            'message_type': 'group',
+            'message_id': 1000,
+            'message': '',
+            'time': 1776491725,
+            'group_id': 2000,
+            'sender': {
+                'user_id': 3000,
+                'nickname': 'QQ Nickname',
+                'card': 'Group Card',
+                'role': 'member',
+                'title': '',
+            },
+        }
+    )
+    now = 1000.0
+
+    class Bot:
+        member_info_calls = 0
+
+        async def get_group_info(self, group_id):
+            return {'group_id': group_id, 'group_name': 'Test Group'}
+
+        async def get_group_member_info(self, group_id, user_id):
+            self.member_info_calls += 1
+            return {
+                'group_id': group_id,
+                'user_id': user_id,
+                'title': f'Member Title {self.member_info_calls}',
+            }
+
+    monkeypatch.setattr('langbot.pkg.platform.sources.aiocqhttp.time.monotonic', lambda: now)
+
+    bot = Bot()
+    converter = AiocqhttpEventConverter()
+
+    first = await converter.target2yiri(event, bot)
+    now = 87401.0
+    second = await converter.target2yiri(event, bot)
+
+    assert first.sender.special_title == 'Member Title 1'
+    assert second.sender.special_title == 'Member Title 2'
+    assert bot.member_info_calls == 2
+
+
+@pytest.mark.asyncio
+async def test_group_message_special_title_retries_after_negative_cache_expires(monkeypatch):
+    event = aiocqhttp.Event(
+        {
+            'post_type': 'message',
+            'message_type': 'group',
+            'message_id': 1000,
+            'message': '',
+            'time': 1776491725,
+            'group_id': 2000,
+            'sender': {
+                'user_id': 3000,
+                'nickname': 'QQ Nickname',
+                'card': 'Group Card',
+                'role': 'member',
+                'title': '',
+            },
+        }
+    )
+    now = 1000.0
+
+    class Bot:
+        member_info_calls = 0
+
+        async def get_group_info(self, group_id):
+            return {'group_id': group_id, 'group_name': 'Test Group'}
+
+        async def get_group_member_info(self, group_id, user_id):
+            self.member_info_calls += 1
+            if self.member_info_calls == 1:
+                raise RuntimeError('api unavailable')
+            return {'group_id': group_id, 'user_id': user_id, 'title': 'Recovered Title'}
+
+    monkeypatch.setattr('langbot.pkg.platform.sources.aiocqhttp.time.monotonic', lambda: now)
+
+    bot = Bot()
+    converter = AiocqhttpEventConverter()
+
+    failed = await converter.target2yiri(event, bot)
+    now = 1601.0
+    recovered = await converter.target2yiri(event, bot)
+
+    assert failed.sender.special_title == ''
+    assert recovered.sender.special_title == 'Recovered Title'
+    assert bot.member_info_calls == 2
+
+
+@pytest.mark.asyncio
+async def test_group_message_group_name_is_cached(monkeypatch):
+    event = aiocqhttp.Event(
+        {
+            'post_type': 'message',
+            'message_type': 'group',
+            'message_id': 1000,
+            'message': '',
+            'time': 1776491725,
+            'group_id': 2000,
+            'sender': {
+                'user_id': 3000,
+                'nickname': 'QQ Nickname',
+                'card': 'Group Card',
+                'role': 'member',
+            },
+        }
+    )
+
+    class Bot:
+        calls = 0
+
+        async def get_group_info(self, group_id):
+            self.calls += 1
+            assert group_id == 2000
+            return {'group_id': group_id, 'group_name': 'Cached Group'}
+
+    monotonic = 1000.0
+    monkeypatch.setattr('langbot.pkg.platform.sources.aiocqhttp.time.monotonic', lambda: monotonic)
+
+    bot = Bot()
+    converter = AiocqhttpEventConverter()
+
+    first = await converter.target2yiri(event, bot)
+    second = await converter.target2yiri(event, bot)
+
+    assert first.sender.group.name == 'Cached Group'
+    assert second.sender.group.name == 'Cached Group'
+    assert bot.calls == 1
+
+
+@pytest.mark.asyncio
+async def test_group_message_group_name_cache_expires(monkeypatch):
+    event = aiocqhttp.Event(
+        {
+            'post_type': 'message',
+            'message_type': 'group',
+            'message_id': 1000,
+            'message': '',
+            'time': 1776491725,
+            'group_id': 2000,
+            'sender': {
+                'user_id': 3000,
+                'nickname': 'QQ Nickname',
+                'card': 'Group Card',
+                'role': 'member',
+            },
+        }
+    )
+    now = 1000.0
+
+    class Bot:
+        calls = 0
+
+        async def get_group_info(self, group_id):
+            self.calls += 1
+            return {'group_id': group_id, 'group_name': f'Group Name {self.calls}'}
+
+    monkeypatch.setattr('langbot.pkg.platform.sources.aiocqhttp.time.monotonic', lambda: now)
+
+    bot = Bot()
+    converter = AiocqhttpEventConverter()
+
+    first = await converter.target2yiri(event, bot)
+    now = 4601.0
+    second = await converter.target2yiri(event, bot)
+
+    assert first.sender.group.name == 'Group Name 1'
+    assert second.sender.group.name == 'Group Name 2'
+    assert bot.calls == 2
+
+
+@pytest.mark.asyncio
+async def test_group_message_group_name_uses_placeholder_when_lookup_fails(monkeypatch):
+    event = aiocqhttp.Event(
+        {
+            'post_type': 'message',
+            'message_type': 'group',
+            'message_id': 1000,
+            'message': '',
+            'time': 1776491725,
+            'group_id': 2000,
+            'sender': {
+                'user_id': 3000,
+                'nickname': 'QQ Nickname',
+                'card': 'Group Card',
+                'role': 'member',
+            },
+        }
+    )
+    now = 1000.0
+
+    class Bot:
+        calls = 0
+
+        async def get_group_info(self, group_id):
+            self.calls += 1
+            raise RuntimeError('api unavailable')
+
+    monkeypatch.setattr('langbot.pkg.platform.sources.aiocqhttp.time.monotonic', lambda: now)
+
+    bot = Bot()
+    converter = AiocqhttpEventConverter()
+
+    converted = await converter.target2yiri(event, bot)
+    cached_failure = await converter.target2yiri(event, bot)
+
+    assert converted.sender.group.name == 'Group 2000'
+    assert cached_failure.sender.group.name == 'Group 2000'
+    assert bot.calls == 1
+
+
+@pytest.mark.asyncio
+async def test_group_message_group_name_retries_after_negative_cache_expires(monkeypatch):
+    event = aiocqhttp.Event(
+        {
+            'post_type': 'message',
+            'message_type': 'group',
+            'message_id': 1000,
+            'message': '',
+            'time': 1776491725,
+            'group_id': 2000,
+            'sender': {
+                'user_id': 3000,
+                'nickname': 'QQ Nickname',
+                'card': 'Group Card',
+                'role': 'member',
+            },
+        }
+    )
+    now = 1000.0
+
+    class Bot:
+        calls = 0
+
+        async def get_group_info(self, group_id):
+            self.calls += 1
+            if self.calls == 1:
+                raise RuntimeError('api unavailable')
+            return {'group_id': group_id, 'group_name': 'Recovered Group'}
+
+    monkeypatch.setattr('langbot.pkg.platform.sources.aiocqhttp.time.monotonic', lambda: now)
+
+    bot = Bot()
+    converter = AiocqhttpEventConverter()
+
+    failed = await converter.target2yiri(event, bot)
+    now = 1061.0
+    recovered = await converter.target2yiri(event, bot)
+
+    assert failed.sender.group.name == 'Group 2000'
+    assert recovered.sender.group.name == 'Recovered Group'
+    assert bot.calls == 2
@@ -0,0 +1,91 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
+import langbot_plugin.api.entities.builtin.platform.message as platform_message
+from langbot.pkg.platform.sources.wecomcs import WecomCSAdapter
+
+
+class DummyLogger(abstract_platform_logger.AbstractEventLogger):
+    async def info(self, *args, **kwargs):
+        pass
+
+    async def debug(self, *args, **kwargs):
+        pass
+
+    async def warning(self, *args, **kwargs):
+        pass
+
+    async def error(self, *args, **kwargs):
+        pass
+
+
+def make_adapter():
+    return WecomCSAdapter(
+        config={
+            'corpid': 'corp-id',
+            'secret': 'secret',
+            'token': 'token',
+            'EncodingAESKey': 'encoding-key',
+        },
+        logger=DummyLogger(),
+    )
+
+
+@pytest.mark.asyncio
+async def test_send_message_sends_text_to_customer_service_user():
+    adapter = make_adapter()
+    adapter.bot_account_id = 'kf-test'
+    adapter.bot = SimpleNamespace(send_text_msg=AsyncMock())
+
+    message = platform_message.MessageChain([platform_message.Plain(text='hello')])
+
+    await adapter.send_message('person', 'uexternal-user', message)
+
+    adapter.bot.send_text_msg.assert_awaited_once()
+    kwargs = adapter.bot.send_text_msg.await_args.kwargs
+    assert kwargs['open_kfid'] == 'kf-test'
+    assert kwargs['external_userid'] == 'external-user'
+    assert kwargs['content'] == 'hello'
+    assert kwargs['msgid'].startswith('langbot_')
+
+
+@pytest.mark.asyncio
+async def test_send_message_allows_explicit_open_kfid_in_target_id():
+    adapter = make_adapter()
+    adapter.bot = SimpleNamespace(send_text_msg=AsyncMock())
+
+    message = platform_message.MessageChain([platform_message.Plain(text='hello')])
+
+    await adapter.send_message('person', 'kf-explicit|uexternal-user', message)
+
+    kwargs = adapter.bot.send_text_msg.await_args.kwargs
+    assert kwargs['open_kfid'] == 'kf-explicit'
+    assert kwargs['external_userid'] == 'external-user'
+
+
+@pytest.mark.asyncio
+async def test_send_message_requires_open_kfid():
+    adapter = make_adapter()
+    adapter.bot = SimpleNamespace(send_text_msg=AsyncMock())
+    message = platform_message.MessageChain([platform_message.Plain(text='hello')])
+
+    with pytest.raises(ValueError, match='open_kfid is required'):
+        await adapter.send_message('person', 'uexternal-user', message)
+
+    adapter.bot.send_text_msg.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_send_message_rejects_group_targets():
+    adapter = make_adapter()
+    adapter.bot_account_id = 'kf-test'
+    adapter.bot = SimpleNamespace(send_text_msg=AsyncMock())
+    message = platform_message.MessageChain([platform_message.Plain(text='hello')])
+
+    with pytest.raises(ValueError, match='only supports sending messages to person'):
+        await adapter.send_message('group', 'group-id', message)
+
+    adapter.bot.send_text_msg.assert_not_called()
@@ -13,6 +13,8 @@ import pytest
 from unittest.mock import Mock, AsyncMock
 from importlib import import_module

+from tests.factories import text_query
+

 def get_connector_module():
    """Lazy import to avoid circular import issues."""
@@ -132,6 +134,130 @@ class TestListPlugins:
        assert result[0]['debug'] is True


+class TestPluginDiagnostics:
+    @pytest.mark.asyncio
+    async def test_emit_event_preserves_response_sources(self):
+        connector = create_mock_connector()
+        query = text_query('hello')
+        event = query.message_event
+        object.__setattr__(event, 'query', query)
+        connector_module = get_connector_module()
+        original_from_event = connector_module.context.EventContext.from_event
+        original_model_validate = connector_module.context.EventContext.model_validate
+        response_sources = [
+            {
+                'kind': 'reply_message_chain',
+                'plugin': {'author': 'tester', 'name': 'demo'},
+            }
+        ]
+
+        async def emit_event_response(event_context, include_plugins=None):
+            return {
+                'event_context': event_context,
+                'emitted_plugins': [],
+                'response_sources': response_sources,
+            }
+
+        connector.handler = AsyncMock()
+        connector.handler.emit_event = AsyncMock(side_effect=emit_event_response)
+
+        fake_event_ctx = Mock()
+        event_dump = event.model_dump()
+        event_dump['event_name'] = 'FriendMessage'
+        fake_event_ctx.model_dump.return_value = {
+            'query_id': query.query_id,
+            'eid': 0,
+            'event_name': 'FriendMessage',
+            'event': event_dump,
+            'is_prevent_default': False,
+            'is_prevent_postorder': False,
+        }
+        connector_module.context.EventContext.from_event = Mock(return_value=fake_event_ctx)
+        parsed_event_ctx = Mock()
+        connector_module.context.EventContext.model_validate = Mock(return_value=parsed_event_ctx)
+        try:
+            event_ctx = await connector.emit_event(event)
+        finally:
+            connector_module.context.EventContext.from_event = original_from_event
+            connector_module.context.EventContext.model_validate = original_model_validate
+
+        assert event_ctx is parsed_event_ctx
+        assert event_ctx._response_sources == response_sources
+
+    @pytest.mark.asyncio
+    async def test_emit_event_leaves_response_sources_absent_for_old_runtime(self):
+        connector = create_mock_connector()
+        query = text_query('hello')
+        event = query.message_event
+        object.__setattr__(event, 'query', query)
+        connector_module = get_connector_module()
+        original_from_event = connector_module.context.EventContext.from_event
+        original_model_validate = connector_module.context.EventContext.model_validate
+
+        async def emit_event_response(event_context, include_plugins=None):
+            return {
+                'event_context': event_context,
+                'emitted_plugins': [
+                    {'manifest': {'metadata': {'author': 'tester', 'name': 'demo'}}},
+                ],
+            }
+
+        connector.handler = AsyncMock()
+        connector.handler.emit_event = AsyncMock(side_effect=emit_event_response)
+
+        fake_event_ctx = Mock()
+        event_dump = event.model_dump()
+        event_dump['event_name'] = 'FriendMessage'
+        fake_event_ctx.model_dump.return_value = {
+            'query_id': query.query_id,
+            'eid': 0,
+            'event_name': 'FriendMessage',
+            'event': event_dump,
+            'is_prevent_default': False,
+            'is_prevent_postorder': False,
+        }
+        connector_module.context.EventContext.from_event = Mock(return_value=fake_event_ctx)
+        parsed_event_ctx = Mock()
+        connector_module.context.EventContext.model_validate = Mock(return_value=parsed_event_ctx)
+        try:
+            event_ctx = await connector.emit_event(event)
+        finally:
+            connector_module.context.EventContext.from_event = original_from_event
+            connector_module.context.EventContext.model_validate = original_model_validate
+
+        assert '_response_sources' not in vars(event_ctx)
+        assert event_ctx._emitted_plugins == [
+            {'manifest': {'metadata': {'author': 'tester', 'name': 'demo'}}},
+        ]
+
+    @pytest.mark.asyncio
+    async def test_notify_plugin_diagnostic_skips_when_disabled(self):
+        connector_module = get_connector_module()
+
+        async def mock_disconnect(conn):
+            pass
+
+        mock_app = create_mock_app()
+        mock_app.instance_config.data = {'plugin': {'enable': False}}
+        connector = connector_module.PluginRuntimeConnector(mock_app, mock_disconnect)
+        connector.handler = AsyncMock()
+
+        await connector.notify_plugin_diagnostic({'code': 'response_delivery_failed'})
+
+        connector.handler.notify_plugin_diagnostic.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_notify_plugin_diagnostic_is_best_effort(self):
+        connector = create_mock_connector()
+        connector.handler = AsyncMock()
+        connector.handler.notify_plugin_diagnostic = AsyncMock(side_effect=RuntimeError('action not found'))
+
+        await connector.notify_plugin_diagnostic({'code': 'response_delivery_failed'})
+
+        connector.handler.notify_plugin_diagnostic.assert_awaited_once()
+        connector.ap.logger.debug.assert_called_once()
+
+
 class TestListKnowledgeEngines:
    """Tests for list_knowledge_engines method."""

@@ -159,6 +159,36 @@ class TestHandlerRagErrorResponse:
        assert 'KeyError' in response.message


+class TestHandlerPluginDiagnostic:
+    @pytest.mark.asyncio
+    async def test_notify_plugin_diagnostic_falls_back_to_raw_protocol_action(self):
+        """Diagnostic forwarding works before the SDK enum exists."""
+        app = SimpleNamespace()
+        app.logger = SimpleNamespace(debug=MagicMock())
+        runtime_handler = make_handler(app)
+        runtime_handler.call_action = AsyncMock(return_value={})
+
+        payload = {'code': 'response_delivery_failed'}
+        await runtime_handler.notify_plugin_diagnostic(payload)
+
+        action = runtime_handler.call_action.await_args.args[0]
+        assert action.value == 'plugin_diagnostic'
+        assert runtime_handler.call_action.await_args.args[1] is payload
+        assert runtime_handler.call_action.await_args.kwargs['timeout'] == 5
+
+    def test_langbot_to_runtime_action_uses_enum_when_available(self):
+        """The compatibility helper should prefer SDK enums once available."""
+        from langbot.pkg.plugin import handler as plugin_handler
+
+        sentinel = object()
+        original = plugin_handler.LangBotToRuntimeAction
+        plugin_handler.LangBotToRuntimeAction = SimpleNamespace(PLUGIN_DIAGNOSTIC=sentinel)
+        try:
+            assert plugin_handler._langbot_to_runtime_action('PLUGIN_DIAGNOSTIC', 'plugin_diagnostic') is sentinel
+        finally:
+            plugin_handler.LangBotToRuntimeAction = original
+
+
 class TestConstantsSemanticVersion:
    """Tests for version constant access."""

@@ -51,13 +51,15 @@ class TestRagRerankAction:
        app.model_mgr.get_rerank_model_by_uuid = AsyncMock(return_value=rerank_model)
        runtime_handler = make_handler(app)

-        response = await runtime_handler.actions[PluginToRuntimeAction.INVOKE_RERANK.value]({
-            'rerank_model_uuid': 'rerank-1',
-            'query': 'hello',
-            'documents': ['a', 'b'],
-            'top_k': 1,
-            'extra_args': {'return_documents': False},
-        })
+        response = await runtime_handler.actions[PluginToRuntimeAction.INVOKE_RERANK.value](
+            {
+                'rerank_model_uuid': 'rerank-1',
+                'query': 'hello',
+                'documents': ['a', 'b'],
+                'top_k': 1,
+                'extra_args': {'return_documents': False},
+            }
+        )

        assert response.code == 0
        assert response.data['results'] == [{'index': 1, 'relevance_score': 0.9}]
@@ -72,16 +74,16 @@ class TestRagRerankAction:
    @pytest.mark.asyncio
    async def test_returns_error_when_rerank_model_missing(self, app):
        """Missing rerank model returns an action error."""
-        app.model_mgr.get_rerank_model_by_uuid = AsyncMock(
-            side_effect=ValueError('not found')
-        )
+        app.model_mgr.get_rerank_model_by_uuid = AsyncMock(side_effect=ValueError('not found'))
        runtime_handler = make_handler(app)

-        response = await runtime_handler.actions[PluginToRuntimeAction.INVOKE_RERANK.value]({
-            'rerank_model_uuid': 'missing',
-            'query': 'hello',
-            'documents': ['a'],
-        })
+        response = await runtime_handler.actions[PluginToRuntimeAction.INVOKE_RERANK.value](
+            {
+                'rerank_model_uuid': 'missing',
+                'query': 'hello',
+                'documents': ['a'],
+            }
+        )

        assert response.code != 0
        assert 'Rerank model with rerank_model_uuid missing not found' in response.message
@@ -2,7 +2,7 @@

 import pytest

-from src.langbot.pkg.plugin.connector import PluginRuntimeConnector
+from langbot.pkg.plugin.connector import PluginRuntimeConnector


 def test_parse_plugin_id_accepts_author_name():
@@ -639,10 +639,13 @@ class TestGetRuntimeInfoDict:
        assert info['box_session_id'] == 'mcp-shared'
        assert info['box_enabled'] is True

-    def test_transient_test_session_is_isolated_from_shared(self, mcp_module):
-        """A transient test session (config-page "test", no persisted UUID)
-        must NOT share the live "mcp-shared" Box session. Regression: a failing
-        test churned the shared session and tore down healthy live servers."""
+    def test_transient_test_shares_session_but_isolated_by_process(self, mcp_module):
+        """A transient config-page "test" now shares the same 'mcp-shared' Box
+        session as live servers (so a test reuses the running container / live
+        process instead of a cold per-test session bootstrap). Isolation is at
+        the PROCESS level: the test runs under its own process_id and only ever
+        stops that process_id, so it cannot disturb another server's live
+        process or the shared session itself."""
        ap = _make_ap()
        ap.box_service.available = True
        transient = _make_session(
@@ -670,10 +673,12 @@ class TestGetRuntimeInfoDict:
        )
        assert transient.is_transient is True
        assert live.is_transient is False
-        # Isolated session id for the test, shared for the live server.
-        assert transient._build_box_session_id() == 'mcp-test-gen-uuid-123'
+        # Both share ONE Box session ...
+        assert transient._build_box_session_id() == 'mcp-shared'
        assert live._build_box_session_id() == 'mcp-shared'
-        assert transient._build_box_session_id() != live._build_box_session_id()
+        assert transient._build_box_session_id() == live._build_box_session_id()
+        # ... but are isolated by distinct process_ids within that session.
+        assert transient._box_stdio_runtime.process_id != live._box_stdio_runtime.process_id

    def test_stdio_session_refuses_when_box_unavailable(self, mcp_module):
        """Policy: when Box is configured but unavailable (disabled in config
@@ -824,3 +829,129 @@ async def test_init_box_stdio_server_stages_host_path_in_shared_workspace(mcp_mo
    assert process_payload['command'] == 'python'
    assert process_payload['args'] == ['/workspace/.mcp/u1/workspace/server.py']
    assert process_payload['cwd'] == '/workspace/.mcp/u1/workspace'
+
+
+@pytest.mark.asyncio
+async def test_stdio_handshake_raises_coldstart_retry_while_process_alive(mcp_module, tmp_path, monkeypatch):
+    """During a slow (npx) cold start the handshake fails while the managed
+    process is still alive. initialize() must raise _ColdStartRetry (so the
+    outer lifecycle loop reuses the live process and retries without stopping it
+    or consuming the fatal budget), NOT a fatal error."""
+    from contextlib import asynccontextmanager
+
+    mcp_stdio_module = sys.modules['langbot.pkg.provider.tools.loaders.mcp_stdio']
+
+    class ColdClientSession:
+        def __init__(self, *_args):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+        async def initialize(self):
+            # Process still cold-starting: handshake fails.
+            raise Exception('Connection closed')
+
+    @asynccontextmanager
+    async def fake_websocket_client(_url: str):
+        yield ('read-stream', 'write-stream')
+
+    monkeypatch.setattr(mcp_stdio_module, 'ClientSession', ColdClientSession)
+    monkeypatch.setattr(mcp_stdio_module, 'websocket_client', fake_websocket_client)
+    monkeypatch.setattr(mcp_stdio_module, '_HANDSHAKE_ATTEMPT_TIMEOUT_SEC', 1.0, raising=False)
+
+    ap = _make_ap()
+    ap.box_service.available = True
+    ap.box_service.create_session = AsyncMock(return_value={})
+    ap.box_service.start_managed_process = AsyncMock(return_value={})
+    ap.box_service.get_managed_process_websocket_url = Mock(return_value='ws://box/p')
+
+    session = _make_session(
+        mcp_module,
+        {
+            'name': 'slow',
+            'uuid': 'slow-uuid',
+            'mode': 'stdio',
+            'command': 'npx',
+            'args': ['-y', 'some-mcp'],
+        },
+        ap=ap,
+    )
+
+    # Process is NOT exited (still cold-starting) and not yet running for reuse.
+    async def _not_exited():
+        return False
+
+    session._box_stdio_runtime._managed_process_has_exited = _not_exited
+
+    async def _not_running():
+        return False
+
+    session._box_stdio_runtime._managed_process_is_running = _not_running
+
+    with pytest.raises(mcp_stdio_module._ColdStartRetry):
+        await session._init_box_stdio_server()
+
+    # Process was started exactly once (the retry will reuse it, not rebuild).
+    assert ap.box_service.start_managed_process.await_count == 1
+    await session.exit_stack.aclose()
+
+
+@pytest.mark.asyncio
+async def test_stdio_handshake_raises_fatal_when_process_exited(mcp_module, tmp_path, monkeypatch):
+    """If the handshake fails AND the process has definitively exited, that is a
+    real failure — initialize() must NOT swallow it as a cold-start retry."""
+    from contextlib import asynccontextmanager
+
+    mcp_stdio_module = sys.modules['langbot.pkg.provider.tools.loaders.mcp_stdio']
+
+    class DeadClientSession:
+        def __init__(self, *_args):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+        async def initialize(self):
+            raise Exception('Connection closed')
+
+    @asynccontextmanager
+    async def fake_websocket_client(_url: str):
+        yield ('read-stream', 'write-stream')
+
+    monkeypatch.setattr(mcp_stdio_module, 'ClientSession', DeadClientSession)
+    monkeypatch.setattr(mcp_stdio_module, 'websocket_client', fake_websocket_client)
+    monkeypatch.setattr(mcp_stdio_module, '_HANDSHAKE_ATTEMPT_TIMEOUT_SEC', 1.0, raising=False)
+
+    ap = _make_ap()
+    ap.box_service.available = True
+    ap.box_service.create_session = AsyncMock(return_value={})
+    ap.box_service.start_managed_process = AsyncMock(return_value={})
+    ap.box_service.get_managed_process_websocket_url = Mock(return_value='ws://box/p')
+
+    session = _make_session(
+        mcp_module,
+        {'name': 'dead', 'uuid': 'dead-uuid', 'mode': 'stdio', 'command': 'npx', 'args': ['-y', 'x']},
+        ap=ap,
+    )
+
+    async def _exited():
+        return True
+
+    session._box_stdio_runtime._managed_process_has_exited = _exited
+
+    async def _not_running():
+        return False
+
+    session._box_stdio_runtime._managed_process_is_running = _not_running
+
+    with pytest.raises(Exception) as ei:
+        await session._init_box_stdio_server()
+    assert not isinstance(ei.value, mcp_stdio_module._ColdStartRetry)
+    await session.exit_stack.aclose()
@@ -0,0 +1,278 @@
+from __future__ import annotations
+
+import base64
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from mcp import types as mcp_types
+
+from langbot.pkg.provider.tools.loaders.mcp import (
+    MCP_RESOURCE_CONTEXT_QUERY_KEY,
+    MCP_RESOURCE_TRACE_QUERY_KEY,
+    MCP_TOOL_LIST_RESOURCES,
+    MCP_TOOL_READ_RESOURCE,
+    MCPLoader,
+    MCPSessionStatus,
+    RuntimeMCPSession,
+)
+from langbot.pkg.telemetry import features as telemetry_features
+
+
+def _app() -> SimpleNamespace:
+    return SimpleNamespace(logger=Mock())
+
+
+def _connected_session(
+    *,
+    name: str = 'docs',
+    uuid: str = 'srv-1',
+    resources: list[dict] | None = None,
+    templates: list[dict] | None = None,
+) -> RuntimeMCPSession:
+    session = RuntimeMCPSession(name, {'uuid': uuid, 'mode': 'remote'}, True, _app())
+    session.status = MCPSessionStatus.CONNECTED
+    session.session = SimpleNamespace(read_resource=AsyncMock())
+    session.resources = resources or [
+        {
+            'uri': 'file:///README.md',
+            'name': 'README.md',
+            'title': '',
+            'description': '',
+            'mime_type': 'text/markdown',
+            'size': None,
+            'icons': [],
+            'annotations': {},
+            '_meta': {},
+        }
+    ]
+    session.resource_templates = templates or []
+    return session
+
+
+def _query() -> SimpleNamespace:
+    return SimpleNamespace(variables={})
+
+
+@pytest.mark.asyncio
+async def test_read_resource_envelope_truncates_caches_and_records_trace():
+    session = _connected_session()
+    session.session.read_resource.return_value = mcp_types.ReadResourceResult(
+        contents=[
+            mcp_types.TextResourceContents(
+                uri='file:///README.md',
+                mimeType='text/markdown',
+                text='abcdef',
+            )
+        ]
+    )
+    query = _query()
+
+    first = await session.read_resource_envelope(
+        'file:///README.md',
+        max_bytes=4,
+        source='ui_preview',
+        query=query,
+    )
+    second = await session.read_resource_envelope(
+        'file:///README.md',
+        max_bytes=4,
+        source='agent_tool',
+        query=query,
+    )
+
+    assert first['contents'][0]['text'] == 'abcd'
+    assert first['contents'][0]['bytes'] == 6
+    assert first['truncated'] is True
+    assert first['cache_hit'] is False
+    assert second['cache_hit'] is True
+    assert second['source'] == 'agent_tool'
+    assert session.session.read_resource.await_count == 1
+
+    traces = query.variables[MCP_RESOURCE_TRACE_QUERY_KEY]
+    assert [trace['source'] for trace in traces] == ['ui_preview', 'agent_tool']
+    assert traces[1]['cache_hit'] is True
+    assert query.variables[telemetry_features.FEATURES_KEY]['mcp_resource_reads'] == {
+        'ui_preview': 1,
+        'agent_tool': 1,
+    }
+
+
+@pytest.mark.asyncio
+async def test_read_resource_envelope_shares_byte_budget_across_text_contents():
+    session = _connected_session()
+    session.session.read_resource.return_value = mcp_types.ReadResourceResult(
+        contents=[
+            mcp_types.TextResourceContents(
+                uri='file:///README.md#first',
+                mimeType='text/plain',
+                text='abc',
+            ),
+            mcp_types.TextResourceContents(
+                uri='file:///README.md#second',
+                mimeType='text/plain',
+                text='def',
+            ),
+        ]
+    )
+
+    envelope = await session.read_resource_envelope('file:///README.md', max_bytes=4)
+
+    assert [item['text'] for item in envelope['contents']] == ['abc', 'd']
+    assert envelope['contents'][0]['truncated'] is False
+    assert envelope['contents'][1]['truncated'] is True
+    assert envelope['bytes'] == 6
+    assert envelope['truncated'] is True
+
+
+@pytest.mark.asyncio
+async def test_read_resource_envelope_omits_binary_by_default():
+    session = _connected_session(
+        resources=[
+            {
+                'uri': 'file:///image.png',
+                'name': 'image.png',
+                'title': '',
+                'description': '',
+                'mime_type': 'image/png',
+                'size': 4,
+                'icons': [],
+                'annotations': {},
+                '_meta': {},
+            }
+        ]
+    )
+    session.session.read_resource.return_value = mcp_types.ReadResourceResult(
+        contents=[
+            mcp_types.BlobResourceContents(
+                uri='file:///image.png',
+                mimeType='image/png',
+                blob=base64.b64encode(b'\x00\x01\x02\x03').decode(),
+            )
+        ]
+    )
+
+    envelope = await session.read_resource_envelope('file:///image.png')
+
+    content = envelope['contents'][0]
+    assert content['type'] == 'blob'
+    assert content['blob'] is None
+    assert content['bytes'] == 4
+    assert content['binary_omitted'] is True
+    assert envelope['truncated'] is True
+    assert envelope['warnings'] == ['Binary resource content omitted from response.']
+
+
+@pytest.mark.asyncio
+async def test_read_resource_envelope_rejects_unlisted_uri():
+    session = _connected_session()
+
+    with pytest.raises(ValueError, match='Resource URI is not available'):
+        await session.read_resource_envelope('file:///secret.txt')
+
+    session.session.read_resource.assert_not_called()
+
+
+def test_resource_uri_allowed_supports_listed_templates_conservatively():
+    session = _connected_session(
+        resources=[],
+        templates=[
+            {
+                'uri_template': 'repo://{owner}/{repo}/file/{path}',
+                'name': 'repository file',
+                'title': '',
+                'description': '',
+                'mime_type': 'text/plain',
+                'icons': [],
+                'annotations': {},
+                '_meta': {},
+            }
+        ],
+    )
+
+    assert session.resource_uri_allowed('repo://langbot-app/LangBot/file/src/main.py') is True
+    assert session.resource_uri_allowed('repo://langbot-app/LangBot/issues/1') is False
+    assert session.resource_uri_allowed('https://example.com/secret') is False
+
+
+@pytest.mark.asyncio
+async def test_mcp_loader_can_hide_synthetic_resource_tools():
+    loader = MCPLoader(_app())
+    session = _connected_session()
+    loader.sessions = {'docs': session}
+
+    with_resource_tools = await loader.get_tools(['srv-1'], include_resource_tools=True)
+    without_resource_tools = await loader.get_tools(['srv-1'], include_resource_tools=False)
+
+    assert {tool.name for tool in with_resource_tools} == {
+        MCP_TOOL_LIST_RESOURCES,
+        MCP_TOOL_READ_RESOURCE,
+    }
+    assert without_resource_tools == []
+
+
+@pytest.mark.asyncio
+async def test_mcp_loader_refuses_resource_tool_calls_when_agent_read_disabled():
+    loader = MCPLoader(_app())
+    session = _connected_session()
+    loader.sessions = {'docs': session}
+    query = SimpleNamespace(
+        variables={
+            '_pipeline_bound_mcp_servers': ['srv-1'],
+            '_pipeline_mcp_resource_agent_read_enabled': False,
+        }
+    )
+
+    result = await loader.invoke_tool(
+        MCP_TOOL_READ_RESOURCE,
+        {'server_name': 'docs', 'uri': 'file:///README.md'},
+        query,
+    )
+
+    assert result[0].text == 'Error: MCP resource agent reads are disabled.'
+    session.session.read_resource.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_build_resource_context_for_query_uses_only_bound_attached_text_resources():
+    loader = MCPLoader(_app())
+    docs = _connected_session(name='docs', uuid='srv-1')
+    docs.session.read_resource.return_value = mcp_types.ReadResourceResult(
+        contents=[
+            mcp_types.TextResourceContents(
+                uri='file:///README.md',
+                mimeType='text/markdown',
+                text='LangBot MCP resource context',
+            )
+        ]
+    )
+    other = _connected_session(name='other', uuid='srv-2')
+    other.session.read_resource.return_value = mcp_types.ReadResourceResult(
+        contents=[
+            mcp_types.TextResourceContents(
+                uri='file:///README.md',
+                mimeType='text/markdown',
+                text='must not be injected',
+            )
+        ]
+    )
+    loader.sessions = {'docs': docs, 'other': other}
+    query = SimpleNamespace(
+        variables={
+            '_pipeline_bound_mcp_servers': ['srv-1'],
+            '_pipeline_mcp_resource_attachments': [
+                {'server_uuid': 'srv-1', 'server_name': 'docs', 'uri': 'file:///README.md', 'mode': 'pinned'},
+                {'server_uuid': 'srv-2', 'server_name': 'other', 'uri': 'file:///README.md', 'mode': 'pinned'},
+            ],
+        }
+    )
+
+    context = await loader.build_resource_context_for_query(query)
+
+    assert '<mcp_resource ' in context
+    assert 'server="docs"' in context
+    assert 'LangBot MCP resource context' in context
+    assert 'must not be injected' not in context
+    assert query.variables[MCP_RESOURCE_CONTEXT_QUERY_KEY]['resource_count'] == 1
+    docs.session.read_resource.assert_awaited_once()
+    other.session.read_resource.assert_not_called()
@@ -15,13 +15,35 @@ from langbot.pkg.provider.tools.toolmgr import ToolManager


 class StubLoader:
-    def __init__(self, tools: list[resource_tool.LLMTool] | None = None, invoke_result=None):
+    def __init__(
+        self,
+        tools: list[resource_tool.LLMTool] | None = None,
+        invoke_result=None,
+        catalog_source: str = 'mcp',
+        catalog_source_name: str = 'fixture-server',
+    ):
        self._tools = tools or []
        self._invoke_result = invoke_result
+        self._catalog_source = catalog_source
+        self._catalog_source_name = catalog_source_name

    async def get_tools(self, *_args, **_kwargs):
        return self._tools

+    async def get_tool_catalog(self, *_args, **_kwargs):
+        return [
+            {
+                'name': tool.name,
+                'description': tool.description,
+                'human_desc': tool.human_desc,
+                'parameters': tool.parameters,
+                'source': self._catalog_source,
+                'source_name': self._catalog_source_name,
+                'source_id': self._catalog_source_name,
+            }
+            for tool in self._tools
+        ]
+
    async def has_tool(self, name: str) -> bool:
        return any(tool.name == name for tool in self._tools)

@@ -68,6 +90,28 @@ async def test_tool_manager_includes_skill_authoring_tools_when_requested():
    assert [tool.name for tool in tools] == ['exec', 'activate', 'plugin_tool', 'mcp_tool']


+@pytest.mark.asyncio
+async def test_tool_manager_catalog_labels_tool_sources():
+    manager = ToolManager(SimpleNamespace())
+    manager.native_tool_loader = StubLoader([make_tool('exec')])
+    manager.skill_tool_loader = StubLoader([make_tool('activate')])
+    manager.plugin_tool_loader = StubLoader(
+        [make_tool('plugin_tool')],
+        catalog_source='plugin',
+        catalog_source_name='fixture-plugin',
+    )
+    manager.mcp_tool_loader = StubLoader([make_tool('mcp_tool')])
+
+    catalog = await manager.get_tool_catalog(include_skill_authoring=True)
+
+    assert [(item['name'], item['source'], item['source_name']) for item in catalog] == [
+        ('exec', 'builtin', 'LangBot'),
+        ('activate', 'skill', 'LangBot'),
+        ('plugin_tool', 'plugin', 'fixture-plugin'),
+        ('mcp_tool', 'mcp', 'fixture-server'),
+    ]
+
+
@pytest.mark.asyncio
 async def test_tool_manager_routes_native_tool_calls():
    app = SimpleNamespace()
@@ -1,6 +1,6 @@
 from pathlib import Path

-from src.langbot.pkg.utils import paths
+from langbot.pkg.utils import paths


 def test_get_data_root_uses_source_root_in_repo_checkout():
@@ -118,7 +118,12 @@ async def test_preproc_enables_skill_authoring_tools_when_skill_service_availabl
    result = await stage.process(_make_query(), 'PreProcessor')

    assert result.result_type == entities_module.ResultType.CONTINUE
-    app.tool_mgr.get_all_tools.assert_awaited_once_with(None, None, include_skill_authoring=True)
+    app.tool_mgr.get_all_tools.assert_awaited_once_with(
+        None,
+        None,
+        include_skill_authoring=True,
+        include_mcp_resource_tools=True,
+    )


@pytest.mark.asyncio
@@ -131,7 +136,32 @@ async def test_preproc_disables_skill_authoring_tools_when_skill_service_missing
    result = await stage.process(_make_query(), 'PreProcessor')

    assert result.result_type == entities_module.ResultType.CONTINUE
-    app.tool_mgr.get_all_tools.assert_awaited_once_with(None, None, include_skill_authoring=False)
+    app.tool_mgr.get_all_tools.assert_awaited_once_with(
+        None,
+        None,
+        include_skill_authoring=False,
+        include_mcp_resource_tools=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_preproc_disables_mcp_resource_tools_when_agent_reading_is_disabled():
+    preproc_module, entities_module = _import_preproc_modules()
+
+    app = _make_app(skill_service=SimpleNamespace())
+    stage = preproc_module.PreProcessor(app)
+    query = _make_query()
+    query.variables['_pipeline_mcp_resource_agent_read_enabled'] = False
+
+    result = await stage.process(query, 'PreProcessor')
+
+    assert result.result_type == entities_module.ResultType.CONTINUE
+    app.tool_mgr.get_all_tools.assert_awaited_once_with(
+        None,
+        None,
+        include_skill_authoring=True,
+        include_mcp_resource_tools=False,
+    )


@pytest.mark.asyncio
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
dadachann	9fb4d27333	fix(mcp): use uniform session memory_mb=1024 to avoid BoxSessionConflictError All MCPs share one Box session (mcp-shared). When session memory_mb differed by command type (512 for python, 1024 for node), the second MCP to call create_session raised BoxSessionConflictError. Fix: always use 1024 MB for the shared session so python and node MCPs coexist without conflict.	2026-07-03 21:46:46 -04:00
Hyu	c5d7e3dcb1	fix(deps): pin langbot-plugin 0.4.12 (BoxManagedProcessSpec.memory_mb fix) (#2315 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-04 09:31:09 +08:00
Hyu	cd6d5d9c2f	fix(mcp): pin langbot-plugin 0.4.10 - per-process memory_mb for node OOM fix (#2314 ) * fix(mcp): bump default memory to 1024MB for node (npx) stdio MCP servers Node.js MCP servers (npx/bunx) were being OOM-killed (return_code=137) by the default 512MB nsjail cgroup_mem_max. Node V8 reserves large virtual address space and instantiates WebAssembly modules (undici llhttp) on startup, easily exceeding 512MB resident. This caused every node-based MCP (memory, sequential-thinking, filesystem, weather, docker, excel) to crash-loop. Fix: when the stdio command is npx/bunx/pnpm, default memory_mb to 1024 unless the operator explicitly set a value. Python/uvx servers keep the 512MB default. * chore(deps): pin langbot-plugin 0.4.10 (per-process memory_mb fix) * chore: update uv.lock for langbot-plugin 0.4.10 --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-04 09:03:43 +08:00
Hyu	1f7d9339fc	fix(mcp): bump default memory to 1024MB for node (npx) stdio MCP servers (#2313 ) Node.js MCP servers (npx/bunx) were being OOM-killed (return_code=137) by the default 512MB nsjail cgroup_mem_max. Node V8 reserves large virtual address space and instantiates WebAssembly modules (undici llhttp) on startup, easily exceeding 512MB resident. This caused every node-based MCP (memory, sequential-thinking, filesystem, weather, docker, excel) to crash-loop. Fix: when the stdio command is npx/bunx/pnpm, default memory_mb to 1024 unless the operator explicitly set a value. Python/uvx servers keep the 512MB default. Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-04 08:31:08 +08:00
RockChinQ	f3b5fcfb7c	fix mcp sidebar status sync (#2312 ) Co-authored-by: Hyu <chenhyu@proton.me>	2026-07-03 20:23:12 +08:00
Hyu	ffabb91bfe	fix(mcp): add missing MCPLogs import to MCPForm (#2311 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-03 17:59:02 +08:00
Hyu	96c84740db	feat(mcp): add logs tab to MCP server detail panel (frontend) (#2310 ) * feat(mcp): add logs tab to MCP server detail panel - Add MCPLogs component mirroring PluginLogs functionality - Add getMcpServerLogs API method to BackendClient - Add logs tab to MCPForm detail panel (edit mode only) - Add i18n keys (tabLogs, logsLevelAll, logsRefresh, logsAutoRefresh, logsEmpty) to all 8 locales - Auto-refresh logs every 3s with toggle - Filter logs by level (ALL/DEBUG/INFO/WARNING/ERROR) * style: fix prettier formatting in MCPForm.tsx logs tab * style: fix prettier in i18n locale files --------- Co-authored-by: dadachann <dadachann@users.noreply.github.com> Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-03 16:52:15 +08:00
Hyu	85b5b5b54b	feat(mcp): add MCP server log panel backend (#2308 ) - Add log buffer (_log_buffer: deque with maxlen=500) to RuntimeMCPSession - Capture stderr from Box managed process in monitor_process_health loop - Add get_mcp_server_logs service method with limit and level filtering - Add GET /servers/<server_name>/logs HTTP endpoint - Parse log level from stderr lines (error/warning/debug/info) - Tests passing: 211 passed, 12 skipped Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-03 16:05:23 +08:00
Hyu	28bffdef21	fix(mcp): set _preserve_managed_process before finally in cold-start path (#2309 ) _ColdStartRetry was caught in _lifecycle_loop_with_retry which set _preserve_managed_process = True — but by then the finally block inside _lifecycle_loop had already run and called _cleanup_box_stdio_session(), stopping the live managed process (return_code=143 SIGTERM). The cold-start retry then restarted a fresh process, eliminating the warm-up advantage. Fix: add an explicit except _ColdStartRetry in _lifecycle_loop that sets _preserve_managed_process = True before re-raising. The finally block then sees the flag and skips stop_managed_process, leaving the live process untouched for the next handshake attempt. Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-03 15:26:18 +08:00
Hyu	bf3c96026b	fix(mcp): fix cold-start connection drop by using lexically-intact owner exit stack (#2307 ) The previous _TransferredStack approach broke anyio lexical context: websocket_client/ClientSession use anyio task groups whose cancel scope is bound to the frame that entered them. Deferring their aclose via a transferred exit stack left the underlying memory streams closed once initialize() returned, so the very next request (refresh -> list_tools) failed with Connection closed. New design: - Attach on the owner exit stack (same task as the serve loop, lexically intact) - A cold-starting process makes initialize() fail; signal _ColdStartRetry up to the outer retry loop, which reuses the live process without consuming retry budget - _lifecycle_loop_with_retry handles _ColdStartRetry like _TransportReconnect: preserves process, no fatal budget, backs off 2s and retries - Two new unit tests: cold-start raises _ColdStartRetry (not fatal) when process is alive; raises fatal error when process has actually exited Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-03 14:58:40 +08:00
Hyu	ed9343c686	fix(mcp): retry the stdio handshake during slow (npx) cold starts (#2306 ) A node/npx stdio MCP server (e.g. firecrawl-mcp via npx -y) failed on first connect with Connection closed / Failed after 4 attempts, even though the process was fine. An npx cold start downloads+installs the package before the server can answer the MCP handshake (measured ~27s for a simple official server; longer for heavier ones). The old code attached the WS and called session.initialize() the instant the process was started, so the handshake ran before the process could answer and failed; the outer lifecycle retry then rebuilt the process, churning it in a loop. Verified decisively: attaching + initialize() against a mid-cold-start process times out on attempt 1 (process still installing) but SUCCEEDS at t+0.6s on attempt 2 once the process is ready. So the fix is to retry the handshake in place, not to rebuild the process. Changes (mcp_stdio.initialize): - Start the managed process ONCE, then loop attach WS -> ClientSession -> initialize() within the startup_timeout budget, tearing down each failed attempt cleanly, until the handshake succeeds or the budget elapses. A successful transport/session is transferred into the owner exit stack via a small _TransferredStack adapter. - Bound each attempt with asyncio.wait_for(initialize, _HANDSHAKE_ATTEMPT_TIMEOUT_SEC=10s) so a cold-starting process fails fast and retries instead of hanging until the transport drops. - Stop retrying ONLY when the process has DEFINITIVELY exited: new _managed_process_has_exited() (checks EXITED status) replaces the previous not-_managed_process_is_running() test, which false-negatived on a just-spawned process that had not yet reported RUNNING and made the loop bail to the outer rebuild path (relay then rejected the early re-attach with HTTP 400). Adds a unit test that fails the first two handshakes with the process alive and asserts the loop retries to success while starting the process exactly once. Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-03 14:16:57 +08:00
Hyu	9a8cdde86c	refactor(mcp): make MCP test reuse the shared Box session (near-instant tests) (#2304 ) * refactor(mcp): make MCP test reuse the shared Box session instead of a per-test session Testing an MCP server (config-page "test" button) previously spun up a fresh isolated mcp-test-<uuid> Box session every time: cold-start the container, run the dependency bootstrap, probe, then tear the whole session down. That is slow (tens of seconds) and, on an already-hosted server, wasteful — the server is already running in the shared session. Change the test to reuse the shared session / live process: - _build_box_session_id: transient tests now use mcp-shared, the same Box session as live servers, so a test reuses the running container (and, for an existing server, its live managed process) instead of a cold per-test session. - cleanup_session: a transient test no longer deletes the whole session (which under the shared model would kill every other MCP server in the container). It stops only its own process_id, exactly like a live server. Isolation is now at the process level (distinct process_id per server/test), not the session level. - test_mcp_server (persisted server): reuse the live connection with a real list_tools refresh/probe; only fall back to a full start() when there is no live connection to probe or the refresh fails, instead of an ERROR->start() rebuild. Trade-off: a failing test now shares the container with live servers rather than a throwaway session. Accepted deliberately in favour of near-instant tests; process-level isolation keeps a test from stopping another server's process. * chore(deps): pin langbot-plugin 0.4.9 for the nsjail RLIMIT_AS node/npx MCP fix --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-03 13:14:14 +08:00
Hyu	a2655e16f8	fix(mcp): survive transient WS transport drops for Box stdio MCP servers (#2303 ) * fix(mcp): survive transient WS transport drops for Box stdio MCP servers A Box-backed stdio MCP server (e.g. pab1it0/prometheus) would periodically error on the frontend with Box managed process exited unexpectedly / Failed after 4 attempts once the session had been alive for a while. Root cause: the managed MCP process lives in the Box runtime and SURVIVES a WebSocket transport drop, but _lifecycle_loop treated any monitor completion as a fatal process death. It then ran the finally-block cleanup — which STOPS the still-healthy managed process — and did a full 4-attempt exponential backoff rebuild. Under an occasionally-stalled single-worker event loop the mcp websocket client misses a ping/pong, the transport drops, and this self-inflicted teardown loop is what the user sees. Fixes: - _lifecycle_loop: when the health monitor completes, re-check the real managed-process state. If the process is still running, the transport merely dropped: raise an internal _TransportReconnect signal instead of Box managed process exited unexpectedly. - _lifecycle_loop_with_retry: handle _TransportReconnect as a free, uncounted reconnect (does not consume the fatal retry budget), so a long-lived session survives arbitrarily many transient drops. - finally-block: gate managed-process teardown on a _preserve_managed_process flag so a transport-only reconnect closes just the WS, not the process. - BoxStdioSessionRuntime.initialize: reuse an already-running managed process instead of stopping+rebuilding it (which also re-ran the slow dependency bootstrap); only (re)start when none is running. Adds _managed_process_is_running() helper. Pairs with langbot-plugin-sdk fix adding a server-driven WS heartbeat to the managed-process relay, which prevents most drops in the first place. * style(mcp): ruff format * chore(deps): pin langbot-plugin 0.4.8 for the managed-process WS heartbeat fix --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-03 12:24:24 +08:00
Hyu	f89611f39a	chore: release v4.10.5	2026-07-03 00:59:53 +08:00
RockChinQ	04a199a3b2	fix: preserve MCP server ids with slashes Fixes #2301	2026-07-03 00:17:29 +08:00
Hyu	f85278f98b	Fix monitoring CI regressions	2026-07-02 17:22:33 +08:00
Hyu	0ce382fc8b	Improve pipeline monitoring and AI tab resilience	2026-07-02 16:53:50 +08:00
Hyu	e32e515bc9	Tone down tool call bubbles	2026-07-02 16:53:50 +08:00
Hyu	c809e3d14f	Add tool call observability	2026-07-02 16:53:50 +08:00
Hyu	b1bc05d5d3	Improve monitoring conversation turns	2026-07-02 16:53:50 +08:00
Stevenqin	13dba887d5	fix(wecomcs): implement proactive text sends (#2300 )	2026-07-02 14:39:05 +08:00
彼方	cc7a13158e	fix(aiocqhttp): resolve group member metadata (#2298 ) * fix(aiocqhttp): resolve group member metadata * style: ruff format	2026-07-01 20:00:47 +08:00
RockChinQ	7de7c7f714	docs: update Product Hunt badge to featured	2026-06-30 08:23:55 -04:00
Junyan Qin	85923e3d7b	fix(migrations): shorten mcp resource revision id	2026-06-30 19:39:23 +08:00
Junyan Qin	c7c6c5dc51	fix(i18n): add skill tools tooltip translations	2026-06-30 19:37:57 +08:00
Junyan Qin	b056646696	fix(migrations): add mcp resource preferences	2026-06-30 19:36:04 +08:00
advancer-young	096ec1a8ce	feat(mcp): support mcp resources (#2215 ) * feat(mcp): support mcp resources * feat(web): split MCP resources into tab * docs: add MCP resources PR review * feat(mcp): productionize resource support * feat(mcp): scope local agent tools and resources * fix(web): gate space embedding models behind login * fix(web): prevent clipped space model CTA * test: update preproc resource tool expectations * fix(web): expose skill authoring tools in selector --------- Co-authored-by: yang.xiang <yang.xiang@advancegroup.com> Co-authored-by: Hyu <chenhyu@proton.me> Co-authored-by: Junyan Qin <rockchinq@gmail.com>	2026-06-30 19:16:30 +08:00
RockChinQ	2618e06492	docs: update 302.AI referral link to share.302ai.cn	2026-06-30 05:10:17 -04:00
彼方	5c8c0eb17b	test: use package import paths (#2297 )	2026-06-30 10:51:34 +08:00
彼方	ccc51522cf	fix(aiocqhttp): normalize base64 media payloads (#2296 ) * fix(aiocqhttp): normalize image and voice base64 payloads * fix(aiocqhttp): support file messages from base64 payloads * test(aiocqhttp): use package import path	2026-06-30 10:51:09 +08:00
彼方	6a99a83f2d	chore(scripts): mark scripts executable (#2295 )	2026-06-30 10:50:26 +08:00
Hyu	ebab5343cf	fix(i18n): add bots.admins keys to all locale files (#2292 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-06-27 17:38:30 +08:00
Hyu	b97cc800d3	feat(platform): migrate bot admins from config.yaml to database (#2289 ) * feat(platform): migrate bot admins from config.yaml to database - Add BotAdmin ORM model (bot_admins table) scoped per bot_uuid - Add Alembic migration 0007 to create table and migrate legacy config admins - Remove top-level admins key from config.yaml template - Add GET/POST/DELETE /api/v1/platform/bots/<uuid>/admins endpoints - Update cmdmgr privilege check to query bot_admins table (bot-scoped) - Add BotAdminsPanel frontend component in bot detail sessions tab - Add i18n keys (zh-Hans, en-US) * fix(ci): ruff/prettier format, fix test_importutil assertion * fix(ci): prettier format BackendClient.ts and i18n locales * fix(ci): eslint-prettier fix BackendClient.ts single-param formatting * 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 --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-06-27 17:18:19 +08:00
彼方	48905ea080	feat(plugin): report deferred response delivery failures (#2287 ) * feat(plugin): report deferred response delivery failures * style: fix ruff format issues in plugin_diagnostics and test_handler_actions --------- Co-authored-by: RockChinQ <45992437+RockChinQ@users.noreply.github.com>	2026-06-26 23:45:10 +08:00
Hyu	ddb77fc43c	fix(api): guard /set-password with allow_modify_login_info (#2288 ) The /change-password and /bind-space endpoints already refuse when system.allow_modify_login_info is false, but /set-password did not, leaving a path to alter login credentials on locked-down deployments (e.g. public demo instances). Apply the same guard. Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-06-26 16:35:50 +08:00