Add OSS and commercial workspace boundaries

docs/multi-tenant/workspace-multi-user-architecture.md

@@ -0,0 +1,858 @@
+# LangBot 多租户与多用户改造方案
+
+## 目标
+
+本方案面向 LangBot 从“单实例单管理员”演进到 SaaS 友好的“多 workspace、多账户、多权限”架构。
+
+核心定义：
+
+- Account：登录主体。一个自然人或服务账号，可加入多个 workspace。
+- Workspace：租户边界。一个 workspace 内可拥有多个用户、机器人、流水线、模型、知识库、扩展、监控数据与 API Key。
+- Membership：账户与 workspace 的关系，承载角色与权限。
+- Role/Permission：workspace 内权限，不再用“是否是当前唯一用户”来决定访问能力。
+
+目标体验：
+
+- 新用户登录后可以创建 workspace、加入 workspace、切换 workspace。
+- 同一个账户可加入多个 workspace，每个 workspace 权限不同。
+- 一个 workspace 可邀请多个用户协作，并分别设置 owner/admin/editor/viewer 等权限。
+- 所有业务资源默认属于某个 workspace，所有 API 默认在当前 workspace 下工作。
+- Plugin SDK、MCP、知识库、模型调用、监控日志都能拿到稳定的 workspace 上下文，并且不跨租户泄露数据。
+
+## 调研结论
+
+### 当前 LangBot 的单用户假设
+
+LangBot 现在已经有 `users` 表和 JWT 登录，但仍是单用户/单租户模型：
+
+- `src/langbot/pkg/entity/persistence/user.py` 的 `User` 只保存 `user/password/account_type/space_*`，没有角色、状态、workspace 关系。
+- `src/langbot/pkg/api/http/service/user.py` 通过 `is_initialized()` 判断系统是否已有用户；`create_or_update_space_user()` 在系统已初始化且邮箱不匹配时拒绝新用户，这直接限制了多用户登录。
+- `src/langbot/pkg/api/http/controller/group.py` 的 `AuthType.USER_TOKEN` 验证后只向 handler 注入 `user_email`；JWT payload 也只有 `user`，没有 `account_id`、`workspace_id`、`role`、`permissions`。
+- `src/langbot/pkg/api/http/service/apikey.py` 的 API Key 只验证 key 是否存在，没有 owner、scope、workspace。
+- `web/src/app/infra/http/BaseHttpClient.ts` 从 `localStorage.token` 读取单个 token，并在所有请求上加 `Authorization`；前端没有 workspace selector，也没有当前 workspace 上下文。
+
+当前登录流程更像“初始化一个本地管理账号”，而不是 SaaS 账户体系。要支持多用户，必须把“初始化状态”和“首个 workspace 创建”拆开。
+
+### 业务资源当前都是全局资源
+
+主要持久化表没有租户字段：
+
+- Bot：`bots`
+- Pipeline：`legacy_pipelines`、`pipeline_run_records`
+- Model：`model_providers`、`llm_models`、`embedding_models`、`rerank_models`
+- Plugin：`plugin_settings`
+- MCP：`mcp_servers`
+- RAG：`knowledge_bases`、`knowledge_base_files`、`knowledge_base_chunks`
+- Monitoring：`monitoring_messages`、`monitoring_llm_calls`、`monitoring_sessions`、`monitoring_errors`、`monitoring_embedding_calls`、`monitoring_feedback`
+- API Key：`api_keys`
+- Webhook：`webhooks`
+- Metadata：`metadata`
+- Binary storage：`binary_storages`
+
+对应服务也直接 select 全表，例如：
+
+- `BotService.get_bots()` 返回所有 bot。
+- `PipelineService.get_pipelines()` 返回所有 pipeline。
+- `ModelProviderService.get_providers()` 返回所有 provider。
+- `MCPService.get_mcp_servers()` 返回所有 MCP server。
+- 插件和二进制存储没有 workspace 维度，插件 workspace storage 在 SDK 里还硬编码为 `default`。
+
+所以改造重点不是只给用户表加字段，而是给资源访问层统一加入 workspace scope。
+
+### 运行时也存在全局单例假设
+
+`src/langbot/pkg/core/stages/build_app.py` 初始化的是一个全局 `Application`，其中包含单例：
+
+- `platform_mgr`
+- `pipeline_mgr`
+- `model_mgr`
+- `tool_mgr`
+- `plugin_connector`
+- `sess_mgr`
+- `rag_mgr`
+- `vector_db_mgr`
+
+当前运行时把所有 bot、pipeline、model、plugin、MCP 都加载到同一套内存管理器。多租户改造需要决定：是共享运行时并在对象上带 workspace 过滤，还是每个 workspace 拆 runtime shard。第一阶段建议共享进程、强制 workspace-aware；等规模变大后再演进为按 workspace 分片。
+
+### Plugin SDK 对 workspace 的假设
+
+SDK 当前只认识 bot/pipeline/query/session，不认识租户：
+
+- `src/langbot_plugin/api/entities/builtin/pipeline/query.py` 的 `Query` 有 `query_id/launcher_type/launcher_id/sender_id/bot_uuid/pipeline_uuid`，没有 `workspace_id/account_id`。
+- `src/langbot_plugin/api/entities/builtin/provider/session.py` 的 `Session` 只按 `launcher_type + launcher_id` 表达会话。
+- `src/langbot_plugin/api/proxies/langbot_api.py` 暴露 `get_bots/get_llm_models/invoke_llm/list_tools/vector_*` 等 Host API，都是全局语义。
+- `src/langbot_plugin/runtime/io/handlers/plugin.py` 的 `set_workspace_storage/get_workspace_storage` 把 `owner_type` 设为 `workspace`，但 `owner` 固定为 `default`。
+- LangBot 侧 `src/langbot/pkg/plugin/handler.py` 处理插件请求时，会把 `GET_BOTS`、`GET_LLM_MODELS`、`VECTOR_*` 等转到全局服务。
+
+这意味着多租户落地时，不能只在 Web API 层过滤；插件可以通过 Host API 访问全局资源，所以 SDK/Runtime 通信也必须传递 workspace context。
+
+## 开源版与商业版产品边界
+
+LangBot 是开源软件，但多 workspace 能力本质上是 SaaS 控制面能力。如果把完整多 workspace、成员协作、订阅权益和配额代码都放进开源仓库，只靠本地 feature flag 或本地 license check，无法有效避免第三方 fork 后自建 SaaS。因此建议采用 open-core 架构：开源版保留单 workspace 执行能力，账户、订阅、权益和多 workspace 协作能力放到 LangBot Space/Cloud Control Plane 和商业模块中。
+
+### 版本边界
+
+推荐拆成三层：
+
+- `LangBot Core OSS`：开源，自托管，默认只有一个隐式 `default workspace`。它可以在数据结构上兼容 workspace，但产品能力上不提供创建多个 workspace、切换 workspace、成员邀请、成员权限管理、审计和多租户配额。
+- `LangBot Space / Cloud Control Plane`：托管控制面，负责 account、workspace、membership、subscription、billing、entitlement、license token、workspace quota、marketplace 权益等能力。
+- `LangBot Commercial Module`：商业闭源或私有包，承载多 workspace、团队协作、RBAC、自定义角色、审计日志、SAML/SSO、企业私有化授权等能力。
+
+企业私有化版本可以采用 `LangBot Core + Commercial Module + License Token` 的形式交付。开源 Core 仍然可独立运行，但只能作为单 workspace 自托管产品，不提供 SaaS 多租户控制面。
+
+### OSS 中如何保留兼容但不开放多 workspace
+
+为了让后续商业版复用同一套资源隔离模型，OSS 代码里可以保留 `workspace_uuid` 相关字段和默认 workspace 迁移，但应限制为单 workspace：
+
+- 首次初始化时创建一个 `Default Workspace`。
+- 所有资源自动归属这个 default workspace。
+- 不暴露 `POST /api/v1/workspaces`。
+- 不暴露 workspace switcher。
+- 不暴露成员邀请和成员角色管理。
+- 不支持一个 account 加入多个 workspace。
+- 不支持 workspace 数量大于 1。
+- 前端不展示 workspace selector。
+- API 层如果收到非 default workspace 的 `X-Workspace-Id`，直接拒绝。
+
+也就是说，OSS 可以是 workspace-aware，但不是 multi-workspace-enabled。这样做的价值是：代码结构提前适配租户隔离，未来商业版不用重写所有资源模型；同时开源版用户无法直接通过 UI/API 获得 SaaS 型多租户能力。
+
+### 账户、订阅和权益抽到 Space
+
+账户和订阅体系建议从 LangBot Core 中抽出，交给 Space 控制面：
+
+```text
+LangBot Space
+  -> Account
+  -> Workspace
+  -> Membership
+  -> Subscription
+  -> Entitlement
+  -> License Token
+
+LangBot Core
+  -> Validate entitlement / license
+  -> Run bots, pipelines, plugins, MCP, RAG
+  -> Enforce local resource scope
+  -> Report usage
+```
+
+这样做有几个原因：
+
+- 账号体系如果完全在本地，第三方容易直接改库创建 workspace/membership。
+- 订阅、配额和商业权益如果完全在本地，容易绕过。
+- Space 可以统一处理 OAuth、组织、邀请、付款、发票、套餐、权益、Marketplace 分发权限。
+- LangBot Core 只作为执行面消费 Space 下发的 entitlement，减少商业规则暴露。
+
+### Entitlement 设计
+
+Space 向 LangBot Core 下发签名权益，可以是在线校验，也可以为企业版提供短期/长期离线 license token。
+
+示例：
+
+```json
+{
+  "edition": "oss",
+  "workspace_limit": 1,
+  "member_limit": 1,
+  "multi_workspace": false,
+  "rbac": false,
+  "audit_log": false,
+  "custom_roles": false,
+  "sso": false,
+  "commercial_use": false,
+  "expires_at": 1893456000
+}
+```
+
+OSS 默认权益：
+
+- `workspace_limit = 1`
+- `member_limit = 1`
+- `multi_workspace = false`
+- `rbac = false`
+- `audit_log = false`
+- `sso = false`
+
+Cloud/Pro/Enterprise 权益：
+
+- `workspace_limit > 1`
+- `member_limit > 1`
+- `multi_workspace = true`
+- `rbac = true`
+- 可按套餐打开 audit、custom roles、SSO、usage reporting、enterprise support 等能力。
+
+Core 执行面需要在关键入口强制校验 entitlement：
+
+- 创建 workspace。
+- 邀请成员。
+- 修改成员角色。
+- 切换 workspace。
+- 创建超过 quota 的资源。
+- 开启商业模块功能。
+
+### 商业模块边界
+
+以下能力不建议进入 OSS 仓库的完整实现：
+
+- 多 workspace 创建和切换。
+- Workspace 成员邀请。
+- Workspace RBAC 和自定义角色。
+- Workspace 审计日志。
+- Workspace 级用量和配额管理。
+- 订阅、账单、发票。
+- 企业 SSO/SAML/OIDC。
+- 在线/离线 license 管理。
+- 多租户 SaaS 运营控制台。
+
+OSS 仓库可以保留接口占位、默认 workspace 兼容和必要的数据隔离字段，但完整交互、管理 UI、权益校验器和多 workspace policy engine 应由 Space 或商业模块提供。
+
+### 防自建 SaaS 的现实边界
+
+技术上无法 100% 阻止别人 fork 开源代码后自行改造。更可靠的策略是组合：
+
+- 不把完整商业多 workspace 实现放进 OSS。
+- Space 控制面提供账号、订阅、权益、Marketplace 和官方托管能力。
+- 商业模块闭源或私有分发。
+- 使用商标、云服务条款和商业 license 限制“自称官方 LangBot SaaS”或未经授权商用托管。
+- 如果当前开源 license 对托管商用限制不足，需要单独评估 license 策略，必要时引入 open-core license 或新增商业附加条款。具体 license 调整需要法律评审。
+
+结论：多 workspace 的底层 schema 可以在 OSS 中以 default workspace 兼容方式铺路，但多 workspace 产品能力、账户订阅权益、协作管理和 SaaS 控制面应放到 Space/商业模块，不作为开源版可直接使用的能力。
+
+## 推荐总体架构
+
+采用“单实例多 workspace，资源行级隔离，运行时上下文隔离”的架构：
+
+```mermaid
+flowchart TD
+    A["Account"] --> B["WorkspaceMembership"]
+    B --> C["Workspace"]
+    C --> D["Bots"]
+    C --> E["Pipelines"]
+    C --> F["Models & Providers"]
+    C --> G["Knowledge Bases"]
+    C --> H["Extensions: Plugins / MCP"]
+    C --> I["API Keys & Webhooks"]
+    C --> J["Monitoring"]
+    D --> K["Runtime Query"]
+    E --> K
+    K --> L["Plugin Runtime"]
+    K --> M["MCP Runtime"]
+    L --> N["Workspace-scoped Host APIs"]
+```
+
+原则：
+
+- 账户全局唯一，workspace 是所有业务资源的归属边界。
+- 所有 HTTP handler 在进入业务服务前解析出 `RequestContext(account, workspace, membership, permissions)`。
+- 所有 service 方法显式接收 `ctx` 或 `workspace_id`，禁止在业务服务里无条件 select 全表。
+- 运行时对象的 key 从 `uuid` 扩展为 `(workspace_id, uuid)` 或使用全局唯一 uuid 但必须记录 workspace_id 并校验。
+- 插件/MCP/知识库/模型调用都按 query 所属 workspace 过滤可用资源。
+
+## 数据模型设计
+
+### Account
+
+替代现有 `users` 的语义，建议保留表名但升级字段，避免过大迁移：
+
+字段建议：
+
+- `id`
+- `uuid`
+- `email`
+- `password_hash`
+- `display_name`
+- `avatar_url`
+- `account_type`: `local | space`
+- `status`: `active | disabled | deleted`
+- `space_account_uuid`
+- `space_access_token`
+- `space_refresh_token`
+- `space_access_token_expires_at`
+- `space_api_key`
+- `created_at`
+- `updated_at`
+
+兼容策略：
+
+- 旧字段 `user` 迁移为 `email`，可以短期保留 alias。
+- 旧 `password` 迁移为 `password_hash`，也可先保持列名不变，service 层改命名。
+- JWT 中不要继续只放 email，应放 `sub=account_uuid`。
+
+### Workspace
+
+新增 `workspaces`：
+
+- `uuid`
+- `name`
+- `slug`
+- `avatar_url`
+- `type`: `personal | team`
+- `status`: `active | suspended | deleted`
+- `default_language`
+- `created_by_account_uuid`
+- `created_at`
+- `updated_at`
+
+每个账户首次登录时自动创建一个 personal workspace。旧单用户实例迁移时创建一个 `Default Workspace`。
+
+### WorkspaceMembership
+
+新增 `workspace_memberships`：
+
+- `workspace_uuid`
+- `account_uuid`
+- `role`: `owner | admin | developer | operator | viewer`
+- `status`: `active | invited | disabled`
+- `invited_by_account_uuid`
+- `joined_at`
+- `created_at`
+- `updated_at`
+
+唯一索引：
+
+- `(workspace_uuid, account_uuid)`
+
+### WorkspaceInvitation
+
+新增 `workspace_invitations`：
+
+- `uuid`
+- `workspace_uuid`
+- `email`
+- `role`
+- `token_hash`
+- `expires_at`
+- `accepted_at`
+- `created_by_account_uuid`
+- `created_at`
+
+用于邀请外部用户加入 workspace。Space OAuth 登录时可以根据 email 自动匹配未接受邀请。
+
+### Role 与 Permission
+
+先用固定角色，后续再做自定义角色。
+
+建议权限：
+
+- `workspace.manage`
+- `member.view`
+- `member.invite`
+- `member.update_role`
+- `member.remove`
+- `bot.view`
+- `bot.manage`
+- `pipeline.view`
+- `pipeline.manage`
+- `model.view`
+- `model.manage`
+- `knowledge.view`
+- `knowledge.manage`
+- `extension.view`
+- `extension.manage`
+- `monitoring.view`
+- `apikey.manage`
+- `webhook.manage`
+- `billing.view`
+- `billing.manage`
+
+角色映射：
+
+| Role | 说明 | 权限 |
+| --- | --- | --- |
+| owner | workspace 拥有者 | 全部权限；可转让 owner；不可被其他角色移除 |
+| admin | 管理员 | 除 owner 转让和删除 workspace 外的全部权限 |
+| developer | 构建者 | 管理 bot、pipeline、model、knowledge、extension、webhook，可看监控 |
+| operator | 运营者 | 查看和启停 bot、查看监控、查看配置，不可改密钥和删除资源 |
+| viewer | 只读成员 | 只读资源和监控 |
+
+### 业务资源加 workspace_uuid
+
+以下表需要新增 `workspace_uuid`：
+
+- `bots`
+- `legacy_pipelines`
+- `pipeline_run_records`
+- `model_providers`
+- `llm_models`
+- `embedding_models`
+- `rerank_models`
+- `plugin_settings`
+- `mcp_servers`
+- `knowledge_bases`
+- `knowledge_base_files`
+- `knowledge_base_chunks`
+- `monitoring_*`
+- `api_keys`
+- `webhooks`
+- `binary_storages`
+- `metadata`
+
+索引建议：
+
+- 所有资源表加 `(workspace_uuid, created_at)` 或 `(workspace_uuid, updated_at)`。
+- 资源唯一键从单列改为 workspace 复合唯一：
+  - `bots.uuid` 可保持全局唯一，但查询仍必须带 workspace。
+  - `plugin_settings` 主键从 `(plugin_author, plugin_name)` 改为 `(workspace_uuid, plugin_author, plugin_name)`。
+  - `mcp_servers.name` 如果未来要求唯一，必须是 `(workspace_uuid, name)`。
+  - `metadata.key` 改为 `(workspace_uuid, key)`，系统级 metadata 单独放 `system_metadata` 或使用 `workspace_uuid=NULL`。
+  - `binary_storages.unique_key` 建议改为 `workspace_uuid + owner_type + owner + key` 的 hash。
+
+### API Key
+
+API Key 必须归属于 workspace：
+
+- `workspace_uuid`
+- `created_by_account_uuid`
+- `scopes`
+- `expires_at`
+- `last_used_at`
+- `status`
+
+验证 API Key 后生成 `RequestContext`：
+
+- `account_uuid=None` 或 service account uuid
+- `workspace_uuid=key.workspace_uuid`
+- `permissions=key.scopes`
+
+这样 `/api/v1/platform/bots/<uuid>/send_message` 之类接口不会跨 workspace 操作 bot。
+
+## 后端改造方案
+
+### RequestContext
+
+新增统一上下文对象，例如：
+
+```python
+class RequestContext:
+    account_uuid: str | None
+    workspace_uuid: str
+    role: str | None
+    permissions: set[str]
+    auth_type: Literal["user_token", "api_key"]
+```
+
+改造 `RouterGroup.route()`：
+
+- `USER_TOKEN`：验证 JWT，读取 `account_uuid`，再从 header/query/cookie 中解析 current workspace。
+- `API_KEY`：验证 API Key，直接得到 workspace。
+- `USER_TOKEN_OR_API_KEY`：两者都返回同一种 `RequestContext`。
+- handler 参数从可选 `user_email` 升级为可选 `ctx`；兼容期同时支持 `user_email`。
+
+当前 workspace 传递方式：
+
+- 推荐 header：`X-Workspace-Id: <workspace_uuid>`
+- Web 前端同时把当前 workspace 存在 localStorage。
+- 如果未传，后端用账户最近使用 workspace 或第一个 active membership。
+
+JWT payload：
+
+```json
+{
+  "sub": "account_uuid",
+  "email": "user@example.com",
+  "iss": "LangBot-...",
+  "exp": 1234567890
+}
+```
+
+不要把 workspace 写死在 JWT 里，否则切换 workspace 需要刷新 token。可以额外支持短 TTL workspace token，但第一阶段不必。
+
+### 服务层改造模式
+
+所有 service 方法增加 `ctx` 或 `workspace_uuid`：
+
+```python
+async def get_bots(self, ctx: RequestContext, include_secret: bool = True):
+    require(ctx, "bot.view")
+    query = sqlalchemy.select(Bot).where(Bot.workspace_uuid == ctx.workspace_uuid)
+```
+
+需要改造的服务：
+
+- `UserService`：拆成 AccountService + WorkspaceService 更清晰。
+- `ApiKeyService`：按 workspace 管理 key。
+- `BotService`：所有 bot 查询/创建/更新/删除按 workspace。
+- `PipelineService`：所有 pipeline 查询/默认 pipeline 按 workspace。
+- `ModelProviderService` 和 model services：按 workspace 隔离 provider 和 model。
+- `MCPService`：按 workspace 管理 MCP server，运行时按 workspace host。
+- `KnowledgeService/RAGRuntimeService`：按 workspace 管理 KB、文件、collection。
+- `MonitoringService`：记录和查询都带 workspace。
+- `WebhookService`：按 workspace 管理 webhook。
+- `PluginRuntimeConnector`：插件安装、设置、配置按 workspace。
+
+### HTTP API 形态
+
+保留现有路径，靠 `X-Workspace-Id` 表示当前 workspace，可减少前端和 SDK 破坏：
+
+- `GET /api/v1/workspaces`
+- `POST /api/v1/workspaces`
+- `GET /api/v1/workspaces/current`
+- `PUT /api/v1/workspaces/current`
+- `GET /api/v1/workspaces/<workspace_uuid>/members`
+- `POST /api/v1/workspaces/<workspace_uuid>/invitations`
+- `PUT /api/v1/workspaces/<workspace_uuid>/members/<account_uuid>`
+- `DELETE /api/v1/workspaces/<workspace_uuid>/members/<account_uuid>`
+
+现有资源 API：
+
+- `/api/v1/platform/bots`
+- `/api/v1/pipelines`
+- `/api/v1/provider/*`
+- `/api/v1/plugins`
+- `/api/v1/mcp`
+- `/api/v1/knowledge`
+
+继续保留，但必须从 `RequestContext.workspace_uuid` 过滤。
+
+对外 API Key 也使用相同路径，只是由 key 决定 workspace。
+
+### 初始化流程
+
+现有 `/api/v1/user/init` 含义改为“创建首个账号和首个 workspace”：
+
+1. 如果系统没有任何 account：
+   - 创建 account。
+   - 创建 personal/team workspace。
+   - 创建 owner membership。
+   - 创建默认 pipeline。
+   - 标记 wizard status 到该 workspace metadata。
+2. 如果系统已有 account：
+   - 禁止无邀请注册，除非配置允许公开注册。
+   - Space OAuth 登录后，如果没有 membership，引导创建 workspace 或接受邀请。
+
+`/api/v1/user/account-info` 不应再只返回 first user，应返回：
+
+- `initialized`
+- `registration_mode`
+- `space_enabled`
+- `default_login_methods`
+
+登录成功后前端调用 `/api/v1/workspaces` 选择 workspace。
+
+### 运行时隔离
+
+第一阶段采用共享进程 + workspace-aware runtime：
+
+- `RuntimeBot` 增加 `workspace_uuid`。
+- `RuntimePipeline` 增加 `workspace_uuid`。
+- `Query` 增加 `workspace_uuid`，从 bot/pipeline 派生。
+- `SessionManager.get_session()` key 从 `(launcher_type, launcher_id)` 改为 `(workspace_uuid, bot_uuid, launcher_type, launcher_id)`。
+- `PipelineManager.pipeline_dict` key 可保持 pipeline uuid，但所有 load/get 都校验 workspace；如果 uuid 不是全局唯一则改为 `(workspace_uuid, pipeline_uuid)`。
+- `ModelManager` provider/model 加 workspace 过滤；`get_model_by_uuid` 必须确保 query workspace 可访问。
+- `ToolManager` 中 MCP tools、plugin tools 按 query workspace 过滤。
+
+后续规模化时可演进：
+
+- workspace runtime shard：每个 workspace 一套 plugin runtime/MCP runtime。
+- 大客户独立进程或独立数据库。
+
+## Plugin SDK 与 Runtime 改造
+
+### Query/Event 增加 workspace context
+
+SDK `Query` 增加：
+
+- `workspace_uuid: str`
+- `workspace_slug: str | None`
+- `account_uuid: str | None`，仅 Web/API 触发时可能有，聊天平台消息通常为空。
+
+Event 模型通过 `event.query.workspace_uuid` 可拿到租户上下文；序列化时也应包含这些字段。
+
+向后兼容：
+
+- 字段可选，默认 `None`。
+- 老插件不感知这些字段也能跑。
+- 新插件可通过 `ctx.event.query.workspace_uuid` 或新增 `ctx.get_workspace()` 访问。
+
+### Host API 默认按当前 workspace 限制
+
+`LangBotAPIProxy` 的以下方法必须由 Host 端按 workspace 过滤：
+
+- `get_bots`
+- `get_bot_info`
+- `send_message`
+- `get_llm_models`
+- `invoke_llm`
+- `list_plugins_manifest`
+- `list_commands`
+- `list_tools`
+- `call_tool`
+- `invoke_embedding`
+- `vector_*`
+- `list_knowledge_bases`
+- `retrieve_knowledge`
+
+建议新增显式方法：
+
+- `get_workspace_info()`
+- `get_current_account()`
+- `get_workspace_storage(...)`
+
+但不要让插件传入任意 workspace id 来越权访问。插件请求的 workspace 应由 Runtime 根据当前 query/plugin connection 填充。
+
+### Workspace storage 修复
+
+当前 SDK runtime 中：
+
+```python
+data["owner_type"] = "workspace"
+data["owner"] = "default"
+```
+
+必须改为：
+
+- 如果请求来自 query/event：owner 为 `workspace_uuid`。
+- 如果请求来自后台插件任务：owner 为 plugin 安装所属 workspace。
+- Host 侧 `binary_storages` 加 `workspace_uuid`，并在 unique key 中包含 workspace。
+
+Plugin storage 建议也同时加 workspace：
+
+- 现在 plugin storage owner 是 `author/name`，这会导致同一插件在不同 workspace 的私有数据冲突。
+- 应改为 `(workspace_uuid, plugin_id, key)`。
+
+### 插件安装与配置
+
+`plugin_settings` 从全局变为 workspace-scoped：
+
+- 同一个插件可安装到多个 workspace。
+- 每个 workspace 有自己的 enabled、priority、config、install_source、install_info。
+- 插件 runtime 列表需要能按 workspace 过滤。
+
+实现路线有两种：
+
+1. 共享插件进程，插件代码只加载一份，设置和调用时附带 workspace。
+2. 每个 workspace 一个插件容器实例，隔离最彻底但资源占用更高。
+
+推荐第一阶段采用方案 1，但要求：
+
+- 所有 RuntimeToLangBot/PluginToRuntime action 都能携带 `workspace_uuid`。
+- 插件 config 获取时按 workspace 返回。
+- 插件 page API 请求必须校验当前用户在该 workspace 有访问权限。
+
+### MCP
+
+MCP server 是租户资源：
+
+- `mcp_servers.workspace_uuid`。
+- MCP session key 从 `server_name` 改为 `(workspace_uuid, server_name)` 或使用全局 uuid。
+- Pipeline extension preferences 中绑定 MCP server uuid 时，只能绑定同 workspace 的 server。
+- MCP tool 列表在 query 执行时按 query.workspace_uuid + pipeline 绑定关系过滤。
+
+## 前端改造
+
+### Workspace selector
+
+Home layout 顶部或 sidebar 增加 workspace selector：
+
+- 当前 workspace 名称和头像。
+- 切换 workspace 后写入 `localStorage.currentWorkspaceId`。
+- 所有请求自动带 `X-Workspace-Id`。
+- 切换后刷新 sidebar 数据和页面缓存。
+
+`BaseHttpClient` request interceptor 增加：
+
+```ts
+const workspaceId = localStorage.getItem("currentWorkspaceId");
+if (workspaceId) config.headers["X-Workspace-Id"] = workspaceId;
+```
+
+### 用户与成员管理页面
+
+新增页面：
+
+- `/home/workspace/settings`
+- `/home/workspace/members`
+- `/home/workspace/invitations`
+
+能力：
+
+- owner/admin 邀请成员。
+- owner/admin 修改成员角色。
+- owner 移除成员、转让 owner。
+- 所有人可切换 workspace。
+- viewer/operator 在 UI 上隐藏不可操作按钮，但后端仍做权限校验。
+
+### 登录与注册
+
+登录后流程：
+
+1. `authUser` 拿 token。
+2. `initializeUserInfo()` 获取 account info。
+3. `GET /api/v1/workspaces`。
+4. 如果没有 workspace：进入创建 workspace 向导。
+5. 如果有多个 workspace：默认进入最近使用 workspace，可切换。
+
+注册页不再表达“初始化管理员账号”，而是：
+
+- 首次系统启动：创建首个 owner + default workspace。
+- 后续：根据配置允许公开注册，或只能接受邀请。
+
+### 旧页面影响
+
+需要逐个检查这些页面的数据加载是否都依赖当前 workspace：
+
+- Bots
+- Pipelines
+- Plugins/Market/MCP
+- Knowledge
+- Monitoring
+- Models dialog
+- API integration dialog
+- Wizard
+
+## 迁移方案
+
+### 迁移阶段 0：准备
+
+- 引入 `workspaces`、`workspace_memberships`、`workspace_invitations`。
+- 给 `users` 增加 `uuid/status/display_name` 等字段。
+- 创建 `RequestContext`，但先不强制所有服务改完。
+
+### 迁移阶段 1：默认 workspace
+
+对现有实例执行迁移：
+
+1. 创建 `Default Workspace`。
+2. 找到现有第一个 user，设为 owner。
+3. 所有已有资源写入 `workspace_uuid=default_workspace_uuid`。
+4. `metadata` 迁入 default workspace；确实全局的配置放到 `system_metadata`。
+5. `binary_storages` 中 `owner_type=workspace, owner=default` 改为 owner 为 default workspace uuid。
+6. 插件 `plugin_settings` 归入 default workspace。
+
+### 迁移阶段 2：服务层强制 scope
+
+- 改所有 service 查询，必须要求 `workspace_uuid`。
+- API Key 迁移为 workspace key。
+- 所有写操作必须检查权限。
+- 监控和任务查询按 workspace 过滤。
+
+### 迁移阶段 3：运行时上下文
+
+- `Query`、`Session`、`RuntimeBot`、`RuntimePipeline` 增加 workspace。
+- Plugin/MCP/Model/RAG runtime 全部按 workspace 过滤。
+- 修复 SDK workspace storage。
+
+### 迁移阶段 4：前端多 workspace
+
+- 登录后 workspace 选择。
+- Header/sidebar workspace switcher。
+- 成员和邀请管理。
+- 所有 API 请求带 `X-Workspace-Id`。
+
+### 迁移阶段 5：安全收敛
+
+- 添加跨 workspace 越权测试。
+- 添加 API Key scope 测试。
+- 添加插件 Host API 过滤测试。
+- 添加 MCP 和 RAG 隔离测试。
+
+## 安全边界
+
+必须防的场景：
+
+- 用户 A 修改 URL 中 bot uuid，访问用户 B workspace 的 bot。
+- API Key 来自 workspace A，但调用 workspace B 的 bot。
+- 插件通过 `get_bots()` 枚举所有 workspace 的 bot。
+- 插件通过 `workspace_storage` 读取其它 workspace 的数据。
+- MCP server 名称相同导致 session 复用。
+- monitoring session_id 相同导致数据串租户。
+- Space OAuth 登录时，同 email 账户被错误绑定到已有本地 account。
+
+建议策略：
+
+- 所有资源访问都使用 `workspace_uuid + resource_id`。
+- 所有 service 方法入口做权限检查。
+- 插件 Host API 的 workspace 不信任插件入参，只信任 query/runtime connection 上下文。
+- API Key 只授予最小 scope，默认不允许成员管理。
+- owner 角色不能被普通 admin 移除或降权。
+
+## 实施优先级
+
+### P0：基础租户骨架
+
+- Account uuid/status。
+- Workspace / Membership / Invitation。
+- RequestContext。
+- JWT 改为 account uuid。
+- 前端 current workspace header。
+
+### P1：资源行级隔离
+
+- Bots、Pipelines、Models、MCP、Plugins、Knowledge、Monitoring、API Keys 全部加 workspace_uuid。
+- service 查询统一加 workspace filter。
+- 权限矩阵落地。
+
+### P2：运行时隔离
+
+- Query、Session、RuntimeBot、RuntimePipeline 加 workspace。
+- Plugin Host API 和 MCP tools 按 workspace 过滤。
+- SDK workspace storage 从 `default` 改为真实 workspace。
+
+### P3：协作体验
+
+- 邀请成员。
+- 成员列表和角色管理。
+- workspace switcher。
+- 最近使用 workspace。
+
+### P4：SaaS 运维增强
+
+- Workspace 级用量统计。
+- Workspace 级限额：max_bots/max_pipelines/max_extensions/tokens/storage。
+- 审计日志。
+- workspace suspend/delete。
+- 可选自定义角色。
+
+## 测试计划
+
+后端测试：
+
+- 账户可加入多个 workspace。
+- 同账户在不同 workspace 权限不同。
+- viewer 不能创建/修改资源。
+- API Key 只能访问所属 workspace。
+- 所有资源 list/get/update/delete 都不能跨 workspace。
+- 默认 workspace 迁移后旧数据可用。
+
+运行时测试：
+
+- 两个 workspace 使用相同 `launcher_id` 不共享 session。
+- 两个 workspace 使用相同 MCP server name 不共享 MCP session。
+- 插件 `get_bots()` 只能看到当前 workspace bot。
+- 插件 `workspace_storage` 在不同 workspace 读写隔离。
+- Pipeline 只调用当前 workspace 绑定的插件和 MCP tools。
+
+前端测试：
+
+- 登录后自动进入最近 workspace。
+- 切换 workspace 后列表数据变化。
+- 无权限按钮隐藏，直接调用 API 也被后端拒绝。
+- 邀请成员流程完整。
+
+迁移测试：
+
+- SQLite 老实例迁移。
+- PostgreSQL 老实例迁移。
+- 已有 local account 迁移为 default workspace owner。
+- 已有 Space account token 和 Space model provider API key 不丢失。
+
+## 关键实现注意事项
+
+- 不建议在第一版做数据库 schema-per-tenant。LangBot 当前 ORM 和运行时均以单库单表为主，先做 shared schema + workspace_uuid 成本更低。
+- 不建议每个 workspace 立即启动独立 plugin runtime。先共享 runtime，强制 action 带 workspace；大客户隔离可作为后续部署形态。
+- 不要只在前端过滤 workspace。插件、API Key、MCP、RAG 都能绕过前端，必须在后端和运行时层过滤。
+- `metadata` 要拆清楚：wizard status 属于 workspace，系统版本/迁移信息属于 system。
+- `users.user` 用 email 当主键语义不稳，应尽快引入 `account_uuid` 并让 JWT 以 uuid 为准。
+- `plugin_settings` 当前主键没有 workspace，改造时要先改主键/唯一约束，否则同插件无法在多个 workspace 配不同配置。
+
+## 建议落地顺序
+
+1. 新增 workspace/account/membership 表和 RequestContext。
+2. 迁移旧数据到 default workspace。
+3. 改 auth 和前端请求头，让每个请求都有 current workspace。
+4. 从最核心资源开始逐个加 scope：bot -> pipeline -> provider/model -> plugin/MCP -> knowledge -> monitoring。
+5. 改 SDK Query/Event 和 runtime storage。
+6. 上成员管理 UI 和邀请。
+7. 做越权测试和迁移测试。
+
+这个顺序的好处是可以较早让主 UI 在一个 workspace 下继续工作，同时把最危险的跨租户泄露面逐步收紧。

src/langbot/libs/dify_service_api/v1/client.py

@@ -109,61 +109,6 @@ class AsyncDifyServiceClient:
                     if chunk.startswith('data:'):
                         yield json.loads(chunk[5:])
 
-    async def workflow_submit(
-        self,
-        form_token: str,
-        workflow_run_id: str,
-        inputs: dict[str, typing.Any],
-        user: str,
-        action: str = '',
-        timeout: float = 120.0,
-    ) -> typing.AsyncGenerator[dict[str, typing.Any], None]:
-        """Submit human input to resume a paused workflow, then stream events.
-
-        1. POST /form/human_input/{form_token} to submit the form
-        2. GET /workflow/{task_id}/events to stream the resumed workflow events
-        """
-
-        headers = {
-            'Authorization': f'Bearer {self.api_key}',
-            'Content-Type': 'application/json',
-        }
-
-        async with httpx.AsyncClient(
-            base_url=self.base_url,
-            trust_env=True,
-            timeout=timeout,
-        ) as client:
-            # Step 1: Submit the form
-            payload: dict[str, typing.Any] = {
-                'inputs': inputs if isinstance(inputs, dict) else {},
-                'user': user,
-                'action': action,
-            }
-
-            submit_resp = await client.post(
-                f'/form/human_input/{form_token}',
-                headers=headers,
-                json=payload,
-            )
-            if submit_resp.status_code != 200:
-                raise DifyAPIError(f'{submit_resp.status_code} {submit_resp.text}')
-
-            # Step 2: Stream resumed workflow events
-            async with client.stream(
-                'GET',
-                f'/workflow/{workflow_run_id}/events',
-                headers={'Authorization': f'Bearer {self.api_key}'},
-                params={'user': user},
-            ) as r:
-                async for chunk in r.aiter_lines():
-                    if r.status_code != 200:
-                        raise DifyAPIError(f'{r.status_code} {chunk}')
-                    if chunk.strip() == '':
-                        continue
-                    if chunk.startswith('data:'):
-                        yield json.loads(chunk[5:])
-
     async def upload_file(
         self,
         file: httpx._types.FileTypes,

src/langbot/pkg/pipeline/pipelinemgr.py

@@ -157,7 +157,7 @@ class RuntimePipeline:
                     bot_message=query.resp_messages[-1],
                     message=result.user_notice,
                     quote_origin=query.pipeline_config['output']['misc']['quote-origin'],
-                    is_final=[msg.is_final for msg in query.resp_messages][-1],
+                    is_final=[msg.is_final for msg in query.resp_messages][0],
                 )
             else:
                 await query.adapter.reply_message(

src/langbot/pkg/pipeline/pool.py

@@ -42,13 +42,9 @@ class QueryPool:
         adapter: abstract_platform_adapter.AbstractMessagePlatformAdapter,
         pipeline_uuid: typing.Optional[str] = None,
         routed_by_rule: bool = False,
-        variables: typing.Optional[dict[str, typing.Any]] = None,
     ) -> pipeline_query.Query:
         async with self.condition:
             query_id = self.query_id_counter
-            initial_variables: dict[str, typing.Any] = {'_routed_by_rule': routed_by_rule}
-            if variables:
-                initial_variables.update(variables)
             query = pipeline_query.Query(
                 bot_uuid=bot_uuid,
                 query_id=query_id,
@@ -57,7 +53,7 @@ class QueryPool:
                 sender_id=sender_id,
                 message_event=message_event,
                 message_chain=message_chain,
-                variables=initial_variables,
+                variables={'_routed_by_rule': routed_by_rule},
                 resp_messages=[],
                 resp_message_chain=[],
                 adapter=adapter,

src/langbot/pkg/pipeline/respback/respback.py

@@ -40,7 +40,7 @@ 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][-1]
+            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],

src/langbot/pkg/platform/botmgr.py

@@ -501,8 +501,6 @@ class PlatformManager:
             bot_entity.adapter_config,
             logger,
         )
-        if hasattr(adapter_inst, 'ap'):
-            adapter_inst.ap = self.ap
 
         # 如果 adapter 支持 set_bot_uuid 方法，设置 bot_uuid（用于统一 webhook）
         if hasattr(adapter_inst, 'set_bot_uuid'):

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

@@ -31,7 +31,6 @@ import langbot_plugin.api.entities.builtin.platform.message as platform_message
 import langbot_plugin.api.entities.builtin.platform.events as platform_events
 import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
 import langbot_plugin.api.definition.abstract.platform.event_logger as abstract_platform_logger
-import langbot_plugin.api.entities.builtin.provider.session as provider_session
 
 
 class AESCipher(object):
@@ -771,7 +770,6 @@ CARD_ID_CACHE_MAX_LIFETIME = 20 * 60  # 20分钟
 class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
     bot: lark_oapi.ws.Client = pydantic.Field(exclude=True)
     api_client: lark_oapi.Client = pydantic.Field(exclude=True)
-    ap: typing.Any = pydantic.Field(exclude=True, default=None)
 
     bot_account_id: str  # 用于在流水线中识别at是否是本bot，直接以bot_name作为标识
     lark_tenant_key: str = pydantic.Field(exclude=True, default='')  # 飞书企业key
@@ -794,16 +792,6 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
     pending_monitoring_msg: dict[str, str]
     # Final: reply Lark message ID → (monitoring_message_id, timestamp) (used by feedback callbacks)
     reply_to_monitoring_msg: dict[str, tuple[str, float]]
-    reply_message_card_ids: dict[str, str]
-    card_sequence_dict: dict[str, int]
-    # card_id → set of source message ids registered against it (for cleanup)
-    card_id_to_source_ids: dict[str, set[str]]
-    # card_id → current streaming_txt content cache (needed for full aupdate during resume transition)
-    card_streaming_text: dict[str, str]
-    # card_id → pre-pause streaming_txt text (captured when resume first chunk arrives)
-    card_pre_pause_text: dict[str, str]
-    # set of card_ids that have already transitioned from "buttons visible" to "resume layout"
-    card_resume_transitioned: set[str]
     _MONITORING_MAPPING_TTL = 600  # 10 minutes
 
     seq: int  # 用于在发送卡片消息中识别消息顺序，直接以seq作为标识
@@ -824,134 +812,11 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         def sync_on_message(event: lark_oapi.im.v1.P2ImMessageReceiveV1):
             asyncio.create_task(on_message(event))
 
-        def schedule_on_app_loop(coro):
-            """Run a coroutine on the application event loop from sync callbacks."""
-            return asyncio.run_coroutine_threadsafe(coro, self.ap.event_loop)
-
         def sync_on_card_action(event):
             try:
-                action_value_raw = getattr(getattr(event.event, 'action', None), 'value', {})
-                # Parse JSON string values (from form action buttons)
-                if isinstance(action_value_raw, str):
-                    try:
-                        action_value_obj = json.loads(action_value_raw)
-                    except (json.JSONDecodeError, TypeError):
-                        action_value_obj = {}
-                else:
-                    action_value_obj = action_value_raw if isinstance(action_value_raw, dict) else {}
+                action_value_obj = getattr(getattr(event.event, 'action', None), 'value', {})
                 action_value = action_value_obj.get('feedback', '') if isinstance(action_value_obj, dict) else ''
 
-                # Handle Dify form action button clicks
-                if isinstance(action_value_obj, dict) and action_value_obj.get('form_action'):
-                    form_token = action_value_obj.get('form_token', '')
-                    workflow_run_id = action_value_obj.get('workflow_run_id', '')
-                    action_id = action_value_obj.get('action_id', '')
-                    session_key = action_value_obj.get('session_key', '')
-
-                    if session_key.startswith('group_') or session_key.startswith('g:'):
-                        launcher_type = provider_session.LauncherTypes.GROUP
-                        launcher_id = (
-                            session_key.split(':', 1)[1]
-                            if session_key.startswith('g:')
-                            else session_key[len('group_') :]
-                        )
-                    else:
-                        launcher_type = provider_session.LauncherTypes.PERSON
-                        launcher_id = (
-                            session_key.split(':', 1)[1]
-                            if session_key.startswith('p:')
-                            else session_key[len('person_') :]
-                        )
-
-                    # Find the bot entity to get bot_uuid and pipeline_uuid
-                    bot_uuid = ''
-                    pipeline_uuid = None
-                    for bot in self.ap.platform_mgr.bots:
-                        if bot.adapter is self:
-                            bot_uuid = bot.bot_entity.uuid
-                            pipeline_uuid = bot.bot_entity.use_pipeline_uuid
-                            break
-
-                    form_action_data = {
-                        'form_token': form_token,
-                        'workflow_run_id': workflow_run_id,
-                        'action_id': action_id,
-                        'user': f'{launcher_type.value}_{launcher_id}',
-                        'inputs': {},
-                    }
-
-                    context = getattr(event.event, 'context', None)
-                    open_message_id = getattr(context, 'open_message_id', None)
-                    source_time = datetime.datetime.now()
-                    event_time = source_time.timestamp()
-                    action_text = action_value_obj.get('action_id', 'confirm')
-                    message_chain = platform_message.MessageChain(
-                        [platform_message.Plain(text=f'[Form Action: {action_text}]')]
-                    )
-                    if open_message_id:
-                        message_chain.insert(
-                            0,
-                            platform_message.Source(
-                                id=open_message_id,
-                                time=source_time,
-                            ),
-                        )
-
-                    operator = getattr(event.event, 'operator', None)
-                    user_id = (
-                        getattr(operator, 'open_id', None) or getattr(operator, 'user_id', None) or str(launcher_id)
-                    )
-
-                    if launcher_type == provider_session.LauncherTypes.GROUP:
-                        synthetic_event = platform_events.GroupMessage(
-                            sender=platform_entities.GroupMember(
-                                id=user_id,
-                                member_name='',
-                                permission=platform_entities.Permission.Member,
-                                group=platform_entities.Group(
-                                    id=launcher_id,
-                                    name='',
-                                    permission=platform_entities.Permission.Member,
-                                ),
-                            ),
-                            message_chain=message_chain,
-                            time=event_time,
-                            source_platform_object=event,
-                        )
-                    else:
-                        synthetic_event = platform_events.FriendMessage(
-                            sender=platform_entities.Friend(
-                                id=user_id,
-                                nickname='',
-                                remark='',
-                            ),
-                            message_chain=message_chain,
-                            time=event_time,
-                            source_platform_object=event,
-                        )
-
-                    async def add_form_action_query():
-                        await self.ap.query_pool.add_query(
-                            bot_uuid=bot_uuid,
-                            launcher_type=launcher_type,
-                            launcher_id=launcher_id,
-                            sender_id=user_id,
-                            message_event=synthetic_event,
-                            message_chain=message_chain,
-                            adapter=self,
-                            pipeline_uuid=pipeline_uuid,
-                            variables={
-                                '_dify_form_action': form_action_data,
-                                '_routed_by_rule': True,
-                            },
-                        )
-
-                    schedule_on_app_loop(add_form_action_query())
-
-                    from lark_oapi.event.callback.model.p2_card_action_trigger import P2CardActionTriggerResponse
-
-                    return P2CardActionTriggerResponse({'toast': {'type': 'success', 'content': '操作成功'}})
-
                 if action_value == '有帮助':
                     feedback_type = 1
                 elif action_value == '无帮助':
@@ -992,14 +857,17 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                 )
 
                 if platform_events.FeedbackEvent in self.listeners:
-                    schedule_on_app_loop(self.listeners[platform_events.FeedbackEvent](feedback_event, self))
+                    loop = asyncio.get_event_loop()
+                    if loop.is_running():
+                        asyncio.create_task(self.listeners[platform_events.FeedbackEvent](feedback_event, self))
+                    else:
+                        loop.run_until_complete(self.listeners[platform_events.FeedbackEvent](feedback_event, self))
 
                 from lark_oapi.event.callback.model.p2_card_action_trigger import P2CardActionTriggerResponse
 
                 return P2CardActionTriggerResponse({'toast': {'type': 'success', 'content': '感谢您的反馈'}})
             except Exception:
-                traceback.print_exc()
-                schedule_on_app_loop(self.logger.error(f'Error in lark card action callback: {traceback.format_exc()}'))
+                asyncio.create_task(self.logger.error(f'Error in lark card action callback: {traceback.format_exc()}'))
                 from lark_oapi.event.callback.model.p2_card_action_trigger import P2CardActionTriggerResponse
 
                 return P2CardActionTriggerResponse({'toast': {'type': 'error', 'content': '反馈处理失败'}})
@@ -1025,12 +893,6 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             card_id_dict={},
             pending_monitoring_msg={},
             reply_to_monitoring_msg={},
-            reply_message_card_ids={},
-            card_sequence_dict={},
-            card_id_to_source_ids={},
-            card_streaming_text={},
-            card_pre_pause_text={},
-            card_resume_transitioned=set(),
             seq=1,
             listeners={},
             quart_app=quart_app,
@@ -1270,33 +1132,6 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         for k in expired:
             del self.reply_to_monitoring_msg[k]
 
-    def _next_card_sequence(self, card_id: str, suggested: int = 1) -> int:
-        """Return the next strictly increasing sequence for a card update."""
-        current = self.card_sequence_dict.get(card_id, 0)
-        next_seq = max(current + 1, suggested)
-        self.card_sequence_dict[card_id] = next_seq
-        return next_seq
-
-    def _register_card_for_source(self, card_id: str, *source_ids: str) -> None:
-        """Register a card_id under one or more source message ids."""
-        bucket = self.card_id_to_source_ids.setdefault(card_id, set())
-        for sid in source_ids:
-            if not sid:
-                continue
-            self.reply_message_card_ids[sid] = card_id
-            bucket.add(sid)
-
-    def _drop_card_state(self, card_id: str) -> None:
-        """Pop all per-card state for the given card_id."""
-        if not card_id:
-            return
-        for sid in self.card_id_to_source_ids.pop(card_id, set()):
-            self.reply_message_card_ids.pop(sid, None)
-        self.card_sequence_dict.pop(card_id, None)
-        self.card_streaming_text.pop(card_id, None)
-        self.card_pre_pause_text.pop(card_id, None)
-        self.card_resume_transitioned.discard(card_id)
-
     async def create_card_id(self, message_id):
         try:
             # self.logger.debug('飞书支持stream输出,创建卡片......')
@@ -1492,7 +1327,6 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             self.card_id_dict[message_id] = response.data.card_id
 
             card_id = response.data.card_id
-            self.card_sequence_dict[card_id] = 0
             return card_id
 
         except Exception as e:
@@ -1505,12 +1339,6 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         """
         # message_id = event.message_chain.message_id
 
-        source_message_id = str(event.message_chain.message_id)
-        existing_card_id = self.reply_message_card_ids.get(source_message_id)
-        if existing_card_id:
-            self.card_id_dict[message_id] = existing_card_id
-            return True
-
         card_id = await self.create_card_id(message_id)
         content = {
             'type': 'card',
@@ -1549,16 +1377,6 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             user_msg_id = event.message_chain.message_id
             reply_msg_id = getattr(response.data, 'message_id', None)
             monitoring_msg_id = self.pending_monitoring_msg.pop(user_msg_id, None)
-            # Register the card under both the user-incoming msg id (so a
-            # second reply_message_first_chunk for the same user message
-            # reuses this card) AND the bot-reply msg id (so a synthetic
-            # event from a form-button callback — whose Source.id equals
-            # the bot's card message id — hits the same card and renders
-            # the resume content into it).
-            if reply_msg_id:
-                self._register_card_for_source(card_id, str(user_msg_id), str(reply_msg_id))
-            else:
-                self._register_card_for_source(card_id, str(user_msg_id))
             if reply_msg_id and monitoring_msg_id:
                 self.reply_to_monitoring_msg[reply_msg_id] = (monitoring_msg_id, time.time())
                 self._cleanup_monitoring_mapping()
@@ -1567,93 +1385,6 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
 
         return True
 
-    async def _open_new_form_card(
-        self,
-        message_id: str,
-        message_source: platform_events.MessageEvent,
-        form_data: dict,
-    ) -> str | None:
-        """Spawn a fresh card to host a re-paused human-input prompt.
-
-        Creates a new card_id (rebinding ``self.card_id_dict[message_id]``),
-        replies it to the current incoming message so it appears as the next
-        step in the chat, registers the new reply_msg_id so subsequent button
-        callbacks resolve back to it, and renders the prompt + buttons on it.
-
-        Returns the new card_id, or ``None`` if creation failed (caller is
-        responsible for falling back to in-place update so the workflow
-        remains continuable).
-        """
-        source_message_id = getattr(message_source.message_chain, 'message_id', None)
-        if not source_message_id:
-            await self.logger.error('Cannot open new form card: source message_id missing')
-            return None
-
-        try:
-            new_card_id = await self.create_card_id(message_id)
-        except Exception:
-            await self.logger.error(f'Failed to create new form card: {traceback.format_exc()}')
-            return None
-
-        tenant_key = (
-            message_source.source_platform_object.header.tenant_key if message_source.source_platform_object else None
-        )
-        app_access_token = self.get_app_access_token()
-        tenant_access_token = self.get_tenant_access_token(tenant_key)
-        req_opt: RequestOption = (
-            RequestOption.builder()
-            .app_ticket(self.app_ticket)
-            .tenant_key(tenant_key)
-            .app_access_token(app_access_token)
-            .tenant_access_token(tenant_access_token)
-            .build()
-        )
-
-        content = {
-            'type': 'card',
-            'data': {'card_id': new_card_id, 'template_variable': {'content': ''}},
-        }
-        request: ReplyMessageRequest = (
-            ReplyMessageRequest.builder()
-            .message_id(str(source_message_id))
-            .request_body(
-                ReplyMessageRequestBody.builder()
-                .content(json.dumps(content))
-                .msg_type('interactive')
-                .uuid(str(uuid.uuid4()))
-                .build()
-            )
-            .build()
-        )
-
-        try:
-            response: ReplyMessageResponse = await self.api_client.im.v1.message.areply(request, req_opt)
-        except Exception:
-            await self.logger.error(f'Failed to send new form card: {traceback.format_exc()}')
-            return None
-
-        if not response.success():
-            await self.logger.error(
-                f'Failed to send new form card: code={response.code}, msg={response.msg}, '
-                f'log_id={response.get_log_id()}'
-            )
-            return None
-
-        reply_msg_id = getattr(response.data, 'message_id', None)
-        if reply_msg_id:
-            self._register_card_for_source(new_card_id, str(source_message_id), str(reply_msg_id))
-
-        sequence = self._next_card_sequence(new_card_id, 1)
-        await self._update_card_layout(
-            card_id=new_card_id,
-            message_source=message_source,
-            text_message='',
-            sequence=sequence,
-            form_data=form_data,
-            show_form_prompt=True,
-        )
-        return new_card_id
-
     async def reply_message(
         self,
         message_source: platform_events.MessageEvent,
@@ -1773,492 +1504,45 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
     ):
         """
         回复消息变成更新卡片消息
-
-        Supports Dify form-action resume:  when the runner yields a chunk with
-        ``_resume_from_form=True``, the card transitions from buttons to a
-        grey "已选择" notice and a new ``streaming_txt_resume`` element is added
-        for subsequent resume chunks to stream into.
-
-        When ``_open_new_card=True`` on the final chunk, the existing card is
-        left as-is and the pipeline will create a new card (with fresh form
-        buttons) for the re-pause.
         """
+        # self.seq += 1
         message_id = bot_message.resp_message_id
         msg_seq = bot_message.msg_sequence
+        if msg_seq % 8 == 0 or is_final:
+            text_elements, media_items = await self.message_converter.yiri2target(message, self.api_client)
 
-        form_data = getattr(bot_message, '_form_data', None)
-        resume_from = getattr(bot_message, '_resume_from_form', False)
-        action_title = getattr(bot_message, '_resume_action_title', '')
-        resume_node_title = getattr(bot_message, '_resume_node_title', '')
-        open_new_card = getattr(bot_message, '_open_new_card', False)
-        if action_title:
-            if resume_node_title:
-                selected_notice = f'**{resume_node_title}**\n已选择：{action_title}'
-            else:
-                selected_notice = f'**已选择**：{action_title}'
-        else:
-            selected_notice = ''
+            text_message = ''
+            if text_elements:
+                parts = []
+                for paragraph in text_elements:
+                    para_text = ''.join(ele['text'] for ele in paragraph if ele['tag'] in ('text', 'md'))
+                    if para_text:
+                        parts.append(para_text)
+                text_message = '\n\n'.join(parts)
 
-        # ── decide whether this chunk needs a card update ────────────────────
-        card_id = self.card_id_dict.get(message_id)
-        if not card_id:
-            return
+            # content = {
+            #     'type': 'card_json',
+            #     'data': {'card_id': self.card_id_dict[message_id], 'elements': {'content': text_message}},
+            # }
 
-        # ── convert message chain → text ─────────────────────────────────────
-        text_elements, media_items = await self.message_converter.yiri2target(message, self.api_client)
-
-        text_message = ''
-        if text_elements:
-            parts = []
-            for paragraph in text_elements:
-                para_text = ''.join(ele['text'] for ele in paragraph if ele['tag'] in ('text', 'md'))
-                if para_text:
-                    parts.append(para_text)
-            text_message = '\n\n'.join(parts)
-
-        tenant_key = (
-            message_source.source_platform_object.header.tenant_key if message_source.source_platform_object else None
-        )
-        app_access_token = self.get_app_access_token()
-        tenant_access_token = self.get_tenant_access_token(tenant_key)
-        req_opt: RequestOption = (
-            RequestOption.builder()
-            .app_ticket(self.app_ticket)
-            .tenant_key(tenant_key)
-            .app_access_token(app_access_token)
-            .tenant_access_token(tenant_access_token)
-            .build()
-        )
-
-        card_sequence = self._next_card_sequence(card_id, msg_seq)
-
-        # ── RESUME: first chunk after button click ───────────────────────────
-        if resume_from and card_id not in self.card_resume_transitioned:
-            # Transition the card from the form state into resume mode.
-            # Preserve the text that was shown before the pause, and seed the
-            # resume placeholder with the current resume content if we already
-            # have any on the first yielded chunk.
-            pre_pause_text = self.card_pre_pause_text.get(card_id) or self.card_streaming_text.get(card_id, '')
-            initial_resume_text = text_message or '\u200b'
-            await self._update_card_layout(
-                card_id=card_id,
-                message_source=message_source,
-                text_message=pre_pause_text,
-                sequence=card_sequence,
-                form_data=None,
-                notice_text=selected_notice,
-                resume_placeholder_text=initial_resume_text,
-            )
-            self.card_resume_transitioned.add(card_id)
-            self.card_pre_pause_text[card_id] = pre_pause_text
-            self.card_streaming_text[card_id] = text_message
-            if not is_final:
-                return
-
-        # ── RESUME: subsequent chunks → full card update ─────────────────────
-        if resume_from and card_id in self.card_resume_transitioned:
-            cached = self.card_streaming_text.get(card_id, '')
-            if text_message != cached:
-                self.card_streaming_text[card_id] = text_message
-                pre_pause_text = self.card_pre_pause_text.get(card_id, '')
-                await self._update_card_layout(
-                    card_id=card_id,
-                    message_source=message_source,
-                    text_message=pre_pause_text,
-                    sequence=card_sequence,
-                    form_data=None,
-                    notice_text=selected_notice,
-                    resume_placeholder_text=text_message,
-                )
-            if not is_final:
-                return
-
-        # ── NORMAL streaming (non-resume): update streaming_txt in-place ──────
-        if not resume_from and (msg_seq % 8 == 0 or is_final):
-            cached = self.card_streaming_text.get(card_id)
-            if text_message != cached:
-                self.card_streaming_text[card_id] = text_message
-                request: ContentCardElementRequest = (
-                    ContentCardElementRequest.builder()
-                    .card_id(card_id)
-                    .element_id('streaming_txt')
-                    .request_body(
-                        ContentCardElementRequestBody.builder().content(text_message).sequence(card_sequence).build()
-                    )
+            request: ContentCardElementRequest = (
+                ContentCardElementRequest.builder()
+                .card_id(self.card_id_dict[message_id])
+                .element_id('streaming_txt')
+                .request_body(
+                    ContentCardElementRequestBody.builder()
+                    # .uuid("a0d69e20-1dd1-458b-k525-dfeca4015204")
+                    .content(text_message)
+                    .sequence(msg_seq)
                     .build()
                 )
-                response: ContentCardElementResponse = await self.api_client.cardkit.v1.card_element.acontent(
-                    request, req_opt
-                )
-                if not response.success():
-                    raise Exception(
-                        f'client.cardkit.v1.card_element.acontent failed, code: {response.code}, '
-                        f'msg: {response.msg}, log_id: {response.get_log_id()}, '
-                        f'resp: \n{json.dumps(json.loads(response.raw.content), indent=4, ensure_ascii=False)}'
-                    )
-
-        # ── FINAL chunk: full card layout update ─────────────────────────────
-        if is_final:
-            final_seq = self._next_card_sequence(card_id, card_sequence + 1)
-            pre_pause = self.card_pre_pause_text.get(card_id, text_message)
-            resume_cached = self.card_streaming_text.get(card_id, '')
-            if form_data:
-                if open_new_card:
-                    # The old card has already been laid out into resume mode
-                    # by the resume-transition block above (notice + resume
-                    # placeholder). Finalise it as a frozen step snapshot and
-                    # spawn a brand-new card to host the next human-input
-                    # prompt — each step stays visible as its own card in the
-                    # chat history.
-                    new_card_id = await self._open_new_form_card(message_id, message_source, form_data)
-                    if new_card_id is None:
-                        # Fallback: keep the existing in-place behaviour so the
-                        # workflow remains continuable even if creating the
-                        # new card failed.
-                        await self._update_card_layout(
-                            card_id=card_id,
-                            message_source=message_source,
-                            text_message=pre_pause,
-                            sequence=final_seq,
-                            form_data=form_data,
-                            resume_placeholder_text=resume_cached,
-                            show_form_prompt=True,
-                        )
-                        self.card_streaming_text.pop(card_id, None)
-                        self.card_pre_pause_text.pop(card_id, None)
-                    else:
-                        # The old card is now a frozen snapshot; let go of its
-                        # streaming-side state but keep its source registrations
-                        # intact (no _drop_card_state) so historical button
-                        # callbacks aimed at it can still be matched if needed.
-                        self.card_streaming_text.pop(card_id, None)
-                        self.card_pre_pause_text.pop(card_id, None)
-                        self.card_resume_transitioned.discard(card_id)
-                else:
-                    # Initial pause path: render prompt + buttons in place on
-                    # the current card.
-                    await self._update_card_layout(
-                        card_id=card_id,
-                        message_source=message_source,
-                        text_message=text_message,
-                        sequence=final_seq,
-                        form_data=form_data,
-                        show_form_prompt=True,
-                    )
-                    # The human-input prompt itself is rendered as buttons only
-                    # on Lark, so do not keep the hidden fallback text around;
-                    # otherwise it will resurface after the button click.
-                    self.card_streaming_text[card_id] = ''
-                    self.card_pre_pause_text[card_id] = ''
-            else:
-                # Normal finish: keep pre-pause + resume content visible,
-                # remove buttons/notice, drop the resume placeholder.
-                await self._update_card_layout(
-                    card_id=card_id,
-                    message_source=message_source,
-                    text_message=pre_pause,
-                    sequence=final_seq,
-                    form_data=None,
-                    notice_text=selected_notice if resume_from else '',
-                    resume_placeholder_text=resume_cached,
-                )
-                self._drop_card_state(card_id)
-            self.card_id_dict.pop(message_id, None)
-
-        # ── media (images / files) appended at the end ───────────────────────
-        if is_final and media_items:
-            for media in media_items:
-                media_request: ReplyMessageRequest = (
-                    ReplyMessageRequest.builder()
-                    .message_id(message_source.message_chain.message_id)
-                    .request_body(
-                        ReplyMessageRequestBody.builder()
-                        .content(json.dumps(media['content']))
-                        .msg_type(media['msg_type'])
-                        .reply_in_thread(False)
-                        .uuid(str(uuid.uuid4()))
-                        .build()
-                    )
-                    .build()
-                )
-                media_response: ReplyMessageResponse = await self.api_client.im.v1.message.areply(
-                    media_request, req_opt
-                )
-                if not media_response.success():
-                    raise Exception(
-                        f'client.im.v1.message.reply ({media["msg_type"]}) failed, code: {media_response.code}, msg: {media_response.msg}, log_id: {media_response.get_log_id()}'
-                    )
-
-    async def _add_form_buttons_to_card(
-        self,
-        card_id: str,
-        message_source: platform_events.MessageEvent,
-        form_data: dict,
-        text_message: str = '',
-        sequence: int = 1,
-    ):
-        """Update the entire card to include form action buttons.
-
-        Uses card.aupdate to replace the card JSON with a template that
-        includes the streaming text content plus interactive buttons.
-        """
-        await self._update_card_layout(
-            card_id=card_id,
-            message_source=message_source,
-            text_message=text_message,
-            sequence=sequence,
-            form_data=form_data,
-        )
-
-    async def _remove_form_buttons_from_card(
-        self,
-        card_id: str,
-        message_source: platform_events.MessageEvent,
-        text_message: str = '',
-        sequence: int = 1,
-    ):
-        """Replace the human-input card layout with the plain final layout."""
-        await self._update_card_layout(
-            card_id=card_id,
-            message_source=message_source,
-            text_message=text_message,
-            sequence=sequence,
-            form_data=None,
-        )
-
-    async def _update_card_layout(
-        self,
-        card_id: str,
-        message_source: platform_events.MessageEvent,
-        text_message: str = '',
-        sequence: int = 1,
-        form_data: dict | None = None,
-        notice_text: str = '',
-        resume_placeholder_text: str = '',
-        show_form_prompt: bool = True,
-    ):
-        """Update the entire card layout.
-
-        • form_data → show interactive buttons (initial Dify pause)
-        • notice_text → replace buttons with a grey "已选择" notice (resume transition)
-        • resume_placeholder_text → add a streaming_txt_resume markdown element
-        """
-        form_data = form_data or {}
-        actions = form_data.get('actions', [])
-        form_token = form_data.get('form_token', '')
-        workflow_run_id = form_data.get('workflow_run_id', '')
-        node_title = form_data.get('node_title', '') or 'Human Input Required'
-        form_content = form_data.get('form_content', '')
-
-        # When form_data is set, the visible content is rendered inside the
-        # interactive container, so the top streaming text should stay empty
-        # to avoid duplicate text above the action area.
-        #
-        # For resume notice state, keep the existing text visible in the card
-        # and only add the grey "selected" notice below it.
-        if form_data:
-            render_text_message = ''
-        else:
-            render_text_message = text_message
-
-        # Determine session key from message source
-        if isinstance(message_source, platform_events.GroupMessage):
-            session_key = f'group_{message_source.group.id}'
-        else:
-            session_key = f'person_{message_source.sender.id}'
-
-        # Build button elements matching the existing card template's thumbsup/down format
-        action_buttons = []
-        for action in actions:
-            action_id = action.get('id', '')
-            action_title = action.get('title', action_id)
-            button_style = action.get('button_style', 'default')
-
-            if button_style == 'primary':
-                lark_button_type = 'primary'
-            elif button_style == 'danger':
-                lark_button_type = 'danger'
-            else:
-                lark_button_type = 'default'
-
-            action_buttons.append(
-                {
-                    'tag': 'button',
-                    'text': {'tag': 'plain_text', 'content': action_title},
-                    'type': lark_button_type,
-                    'width': 'fill',
-                    'size': 'medium',
-                    'hover_tips': {'tag': 'plain_text', 'content': action_title},
-                    'behaviors': [
-                        {
-                            'type': 'callback',
-                            'value': {
-                                'form_action': True,
-                                'form_token': form_token,
-                                'workflow_run_id': workflow_run_id,
-                                'action_id': action_id,
-                                'session_key': session_key,
-                            },
-                        }
-                    ],
-                    'margin': '0px 0px 0px 0px',
-                }
+                .build()
             )
 
-        interactive_elements = []
-        if form_data:
-            if show_form_prompt:
-                interactive_elements = [
-                    {
-                        'tag': 'markdown',
-                        'content': f'**[Human Input Required] {node_title}**',
-                        'text_align': 'left',
-                        'text_size': 'normal',
-                        'margin': '0px 0px 4px 0px',
-                    }
-                ]
-                if form_content:
-                    interactive_elements.append(
-                        {
-                            'tag': 'markdown',
-                            'content': form_content,
-                            'text_align': 'left',
-                            'text_size': 'normal',
-                            'margin': '0px 0px 8px 0px',
-                        }
-                    )
-            interactive_elements.append(
-                {
-                    'tag': 'column_set',
-                    'horizontal_spacing': '8px',
-                    'horizontal_align': 'left',
-                    'margin': '0px 0px 0px 0px',
-                    'columns': [
-                        {
-                            'tag': 'column',
-                            'width': 'weighted',
-                            'elements': [btn],
-                            'padding': '0px 0px 0px 0px',
-                        }
-                        for btn in action_buttons
-                    ],
-                }
-            )
+            if is_final and bot_message.tool_calls is None:
+                # self.seq = 1  # 消息回复结束之后重置seq
+                self.card_id_dict.pop(message_id)  # 清理已经使用过的卡片
 
-        # Build the full card JSON with buttons, same structure as create_card_id
-        # ── mid_section: either form buttons, resume notice, or empty ──
-        mid_section_elements = []
-        if form_data:
-            mid_section_elements = [
-                {
-                    'tag': 'interactive_container',
-                    'margin': '12px 0px 8px 0px',
-                    'padding': '12px 12px 12px 12px',
-                    'has_border': True,
-                    'elements': interactive_elements,
-                },
-                {'tag': 'hr', 'margin': '0px 0px 0px 0px'},
-            ]
-        elif notice_text:
-            mid_section_elements = [
-                {
-                    'tag': 'markdown',
-                    'content': notice_text,
-                    'text_align': 'left',
-                    'text_size': 'normal',
-                    'margin': '8px 0px 4px 0px',
-                    'text_color': 'grey',
-                },
-                {'tag': 'hr', 'margin': '0px 0px 0px 0px'},
-            ]
-
-        # ── resume placeholder element (empty, filled via acontent on each chunk) ──
-        resume_elements = []
-        if resume_placeholder_text:
-            resume_elements = [
-                {
-                    'tag': 'markdown',
-                    'content': resume_placeholder_text,
-                    'text_align': 'left',
-                    'text_size': 'normal',
-                    'margin': '0px 0px 0px 0px',
-                    'element_id': 'streaming_txt_resume',
-                },
-            ]
-
-        card_data = {
-            'schema': '2.0',
-            'config': {
-                'update_multi': True,
-                'streaming_mode': False,
-            },
-            'body': {
-                'direction': 'vertical',
-                'padding': '12px 12px 12px 12px',
-                'elements': [
-                    {
-                        'tag': 'div',
-                        'text': {
-                            'tag': 'plain_text',
-                            'content': 'LangBot',
-                            'text_size': 'normal',
-                            'text_align': 'left',
-                            'text_color': 'default',
-                        },
-                        'icon': {
-                            'tag': 'custom_icon',
-                            'img_key': 'img_v3_02p3_05c65d5d-9bad-440a-a2fb-c89571bfd5bg',
-                        },
-                    },
-                    {
-                        'tag': 'markdown',
-                        'content': render_text_message,
-                        'text_align': 'left',
-                        'text_size': 'normal',
-                        'margin': '0px 0px 0px 0px',
-                        'element_id': 'streaming_txt',
-                    },
-                    *mid_section_elements,
-                    *resume_elements,
-                    {
-                        'tag': 'column_set',
-                        'horizontal_spacing': '12px',
-                        'horizontal_align': 'right',
-                        'columns': [
-                            {
-                                'tag': 'column',
-                                'width': 'weighted',
-                                'elements': [
-                                    {
-                                        'tag': 'markdown',
-                                        'content': '<font color="grey-600">以上内容由 AI 生成，仅供参考。更多详细、准确信息可点击引用链接查看</font>',
-                                        'text_align': 'left',
-                                        'text_size': 'notation',
-                                        'margin': '4px 0px 0px 0px',
-                                        'icon': {
-                                            'tag': 'standard_icon',
-                                            'token': 'robot_outlined',
-                                            'color': 'grey',
-                                        },
-                                    }
-                                ],
-                                'padding': '0px 0px 0px 0px',
-                                'direction': 'vertical',
-                                'horizontal_spacing': '8px',
-                                'vertical_spacing': '8px',
-                                'horizontal_align': 'left',
-                                'vertical_align': 'top',
-                                'margin': '0px 0px 0px 0px',
-                                'weight': 1,
-                            }
-                        ],
-                        'margin': '0px 0px 4px 0px',
-                    },
-                ],
-            },
-        }
-
-        try:
             tenant_key = (
                 message_source.source_platform_object.header.tenant_key
                 if message_source.source_platform_object
@@ -2274,27 +1558,39 @@ class LarkAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                 .tenant_access_token(tenant_access_token)
                 .build()
             )
+            # 发起请求
+            response: ContentCardElementResponse = self.api_client.cardkit.v1.card_element.content(request, req_opt)
 
-            request: UpdateCardRequest = (
-                UpdateCardRequest.builder()
-                .card_id(card_id)
-                .request_body(
-                    UpdateCardRequestBody.builder()
-                    .sequence(sequence)
-                    .uuid(str(uuid.uuid4()))
-                    .card(Card.builder().type('card_json').data(json.dumps(card_data)).build())
-                    .build()
-                )
-                .build()
-            )
-            response: UpdateCardResponse = await self.api_client.cardkit.v1.card.aupdate(request, req_opt)
+            # 处理失败返回
             if not response.success():
-                await self.logger.error(
-                    f'Failed to update lark card with form buttons: code={response.code}, msg={response.msg}, '
-                    f'log_id={response.get_log_id()}, resp={getattr(getattr(response, "raw", None), "content", None)}'
+                raise Exception(
+                    f'client.im.v1.message.patch failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}, resp: \n{json.dumps(json.loads(response.raw.content), indent=4, ensure_ascii=False)}'
                 )
-        except Exception:
-            await self.logger.error(f'Error updating lark card with form buttons: {traceback.format_exc()}')
+                return
+
+            # Send media messages when streaming is done
+            if is_final and media_items:
+                for media in media_items:
+                    media_request: ReplyMessageRequest = (
+                        ReplyMessageRequest.builder()
+                        .message_id(message_source.message_chain.message_id)
+                        .request_body(
+                            ReplyMessageRequestBody.builder()
+                            .content(json.dumps(media['content']))
+                            .msg_type(media['msg_type'])
+                            .reply_in_thread(False)
+                            .uuid(str(uuid.uuid4()))
+                            .build()
+                        )
+                        .build()
+                    )
+                    media_response: ReplyMessageResponse = await self.api_client.im.v1.message.areply(
+                        media_request, req_opt
+                    )
+                    if not media_response.success():
+                        raise Exception(
+                            f'client.im.v1.message.reply ({media["msg_type"]}) failed, code: {media_response.code}, msg: {media_response.msg}, log_id: {media_response.get_log_id()}'
+                        )
 
     async def is_muted(self, group_id: int) -> bool:
         return False

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

@@ -1,14 +1,14 @@
 from __future__ import annotations
+import time
 
 
 import telegram
 import telegram.ext
-from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
-from telegram.ext import ApplicationBuilder, ContextTypes, MessageHandler, CallbackQueryHandler, filters
+from telegram import Update
+from telegram.ext import ApplicationBuilder, ContextTypes, MessageHandler, filters
 import telegramify_markdown
 import typing
 import traceback
-import json
 import base64
 import pydantic
 
@@ -189,7 +189,6 @@ class TelegramEventConverter(abstract_platform_adapter.AbstractEventConverter):
 class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
     bot: telegram.Bot = pydantic.Field(exclude=True)
     application: telegram.ext.Application = pydantic.Field(exclude=True)
-    ap: typing.Any = pydantic.Field(exclude=True, default=None)
 
     message_converter: TelegramMessageConverter = TelegramMessageConverter()
     event_converter: TelegramEventConverter = TelegramEventConverter()
@@ -225,102 +224,6 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                 telegram_callback,
             )
         )
-
-        async def callback_query_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
-            query = update.callback_query
-            await query.answer()
-            try:
-                data = json.loads(query.data)
-                if data.get('form_action') or data.get('f'):
-                    import langbot_plugin.api.entities.builtin.provider.session as provider_session
-
-                    workflow_run_id = data.get('workflow_run_id', '')
-                    w_suffix = data.get('w', '')
-                    action_id = data.get('action_id') or data.get('a', '')
-                    session_key = data.get('session_key') or data.get('s', '')
-
-                    if session_key.startswith('group_') or session_key.startswith('g:'):
-                        launcher_type = provider_session.LauncherTypes.GROUP
-                        launcher_id = (
-                            session_key.split(':', 1)[1]
-                            if session_key.startswith('g:')
-                            else session_key[len('group_') :]
-                        )
-                    else:
-                        launcher_type = provider_session.LauncherTypes.PERSON
-                        launcher_id = (
-                            session_key.split(':', 1)[1]
-                            if session_key.startswith('p:')
-                            else session_key[len('person_') :]
-                        )
-
-                    user_id = str(query.from_user.id)
-
-                    # Find bot_uuid and pipeline_uuid
-                    bot_uuid = ''
-                    pipeline_uuid = None
-                    for b in self.ap.platform_mgr.bots:
-                        if b.adapter is self:
-                            bot_uuid = b.bot_entity.uuid
-                            pipeline_uuid = b.bot_entity.use_pipeline_uuid
-                            break
-
-                    form_action_data = {
-                        'workflow_run_id': workflow_run_id,
-                        'w_suffix': w_suffix,
-                        'action_id': action_id,
-                        'user': f'{launcher_type.value}_{launcher_id}',
-                        'inputs': {},
-                    }
-
-                    message_chain = platform_message.MessageChain(
-                        [platform_message.Plain(text=f'[Form Action: {action_id}]')]
-                    )
-
-                    if launcher_type == provider_session.LauncherTypes.GROUP:
-                        synthetic_event = platform_events.GroupMessage(
-                            sender=platform_entities.GroupMember(
-                                id=user_id,
-                                member_name='',
-                                permission=platform_entities.Permission.Member,
-                                group=platform_entities.Group(
-                                    id=launcher_id,
-                                    name='',
-                                    permission=platform_entities.Permission.Member,
-                                ),
-                            ),
-                            message_chain=message_chain,
-                            source_platform_object=update,
-                        )
-                    else:
-                        synthetic_event = platform_events.FriendMessage(
-                            sender=platform_entities.Friend(
-                                id=user_id,
-                                nickname='',
-                                remark='',
-                            ),
-                            message_chain=message_chain,
-                            source_platform_object=update,
-                        )
-
-                    await self.ap.query_pool.add_query(
-                        bot_uuid=bot_uuid,
-                        launcher_type=launcher_type,
-                        launcher_id=launcher_id,
-                        sender_id=user_id,
-                        message_event=synthetic_event,
-                        message_chain=message_chain,
-                        adapter=self,
-                        pipeline_uuid=pipeline_uuid,
-                        variables={
-                            '_dify_form_action': form_action_data,
-                            '_routed_by_rule': True,
-                        },
-                    )
-            except Exception:
-                await self.logger.error(f'Error in telegram callback query: {traceback.format_exc()}')
-
-        application.add_handler(CallbackQueryHandler(callback_query_handler))
         super().__init__(
             config=config,
             logger=logger,
@@ -416,19 +319,14 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         update = event.source_platform_object
         chat_id = update.effective_chat.id
         chat_type = update.effective_chat.type
-        effective_message = update.effective_message
-        message_thread_id = getattr(effective_message, 'message_thread_id', None) if effective_message else None
+        message_thread_id = update.message.message_thread_id
 
         if chat_type == 'private':
-            import time as _time
-
-            draft_id = int(_time.time() * 1000)
+            draft_id = int(time.time() * 1000)
             self.msg_stream_id[message_id] = ('private', draft_id)
+
             args = self._build_message_args(chat_id, 'Thinking...', message_thread_id, draft_id=draft_id)
-            try:
-                await self.bot.send_message_draft(**args)
-            except (telegram.error.RetryAfter, telegram.error.BadRequest):
-                pass
+            await self.bot.send_message_draft(**args)
         else:
             args = self._build_message_args(chat_id, 'Thinking...', message_thread_id)
             send_msg = await self.bot.send_message(**args)
@@ -449,13 +347,12 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         assert isinstance(message_source.source_platform_object, Update)
         update = message_source.source_platform_object
         chat_id = update.effective_chat.id
-        effective_message = update.effective_message
-        message_thread_id = getattr(effective_message, 'message_thread_id', None) if effective_message else None
+        message_thread_id = update.message.message_thread_id
 
         if message_id not in self.msg_stream_id:
             return
 
-        chat_mode, stream_id = self.msg_stream_id[message_id]
+        chat_mode, draft_id = self.msg_stream_id[message_id]
         components = await TelegramMessageConverter.yiri2target(message, self.bot)
 
         if not components or components[0]['type'] != 'text':
@@ -464,42 +361,16 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             return
 
         content = components[0]['text']
-        form_data = getattr(bot_message, '_form_data', None)
-
-        if form_data and is_final:
-            self.msg_stream_id.pop(message_id, None)
-            await self._send_form_action_buttons(message_source, form_data)
-            return
 
         if chat_mode == 'private':
-            # Streaming via draft (ephemeral preview in the chat input area)
-            if (msg_seq - 1) % 8 == 0 or is_final:
-                args = self._build_message_args(chat_id, content, message_thread_id, draft_id=stream_id)
-                try:
-                    await self.bot.send_message_draft(**args)
-                except telegram.error.BadRequest as exc:
-                    if 'Message_too_long' in str(exc):
-                        args['text'] = content[:4000] + '\n\n… (truncated)'
-                        try:
-                            await self.bot.send_message_draft(**args)
-                        except telegram.error.RetryAfter:
-                            pass
-                    else:
-                        pass  # Ignore other draft errors (cosmetic)
+            args = self._build_message_args(chat_id, content, message_thread_id, draft_id=draft_id)
+            await self.bot.send_message_draft(**args)
             if is_final and bot_message.tool_calls is None:
-                # Finalise: send the real message, discard the draft
-                args = self._build_message_args(chat_id, content, message_thread_id)
-                try:
-                    await self.bot.send_message(**args)
-                except telegram.error.BadRequest as exc:
-                    if 'Message_too_long' in str(exc):
-                        args['text'] = content[:4000] + '\n\n… (truncated)'
-                        await self.bot.send_message(**args)
-                    else:
-                        raise
+                del args['draft_id']
+                await self.bot.send_message(**args)
                 self.msg_stream_id.pop(message_id)
         else:
-            # Streaming via edit_message_text (persistent message)
+            stream_id = draft_id
             if (msg_seq - 1) % 8 == 0 or is_final:
                 args = {
                     'message_id': stream_id,
@@ -508,68 +379,11 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                 }
                 if self.config.get('markdown_card', False):
                     args['parse_mode'] = 'MarkdownV2'
-                try:
-                    await self.bot.edit_message_text(**args)
-                except telegram.error.BadRequest as exc:
-                    if 'Message_too_long' in str(exc):
-                        args['text'] = self._process_markdown(content[:4000] + '\n\n… (truncated)')
-                        await self.bot.edit_message_text(**args)
-                    else:
-                        raise
+                await self.bot.edit_message_text(**args)
 
             if is_final and bot_message.tool_calls is None:
                 self.msg_stream_id.pop(message_id)
 
-    async def _send_form_action_buttons(
-        self,
-        message_source: platform_events.MessageEvent,
-        form_data: dict,
-    ):
-        """Send inline keyboard buttons for Dify human_input_required form actions."""
-        actions = form_data.get('actions', [])
-        node_title = form_data.get('node_title', '')
-        form_content = form_data.get('form_content', '')
-        workflow_run_id = form_data.get('workflow_run_id', '')
-        # Telegram callback_data is capped at 64 bytes, so we identify the
-        # paused workflow by the last 8 chars of workflow_run_id (unique
-        # within a session with overwhelming probability).
-        w_suffix = workflow_run_id[-8:] if workflow_run_id else ''
-
-        if isinstance(message_source, platform_events.GroupMessage):
-            session_key = f'g:{message_source.group.id}'
-        else:
-            session_key = f'p:{message_source.sender.id}'
-
-        keyboard = []
-        for action in actions:
-            action_id = action.get('id', '')
-            action_title = action.get('title', action_id)
-            callback_payload = {'f': 1, 'a': action_id, 's': session_key}
-            if w_suffix:
-                callback_payload['w'] = w_suffix
-            callback_data = json.dumps(callback_payload, separators=(',', ':'))
-            keyboard.append([InlineKeyboardButton(action_title, callback_data=callback_data)])
-
-        reply_markup = InlineKeyboardMarkup(keyboard)
-
-        update = message_source.source_platform_object
-        chat_id = update.effective_chat.id
-        effective_message = update.effective_message
-        message_thread_id = getattr(effective_message, 'message_thread_id', None) if effective_message else None
-
-        text_lines = [f'[{node_title}] Please select an action:']
-        if form_content:
-            text_lines.insert(0, form_content)
-        args = {
-            'chat_id': chat_id,
-            'text': '\n\n'.join(text_lines),
-            'reply_markup': reply_markup,
-        }
-        if message_thread_id:
-            args['message_thread_id'] = message_thread_id
-
-        await self.bot.send_message(**args)
-
     def get_launcher_id(self, event: platform_events.MessageEvent) -> str | None:
         if not isinstance(event.source_platform_object, Update):
             return None

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

@@ -2,11 +2,9 @@ from __future__ import annotations
 
 import typing
 import json
-import time
 import uuid
 import base64
 import mimetypes
-from collections import OrderedDict
 
 
 from langbot.pkg.provider import runner
@@ -18,102 +16,6 @@ from langbot.libs.dify_service_api.v1 import client, errors
 import httpx
 
 
-# Module-level store for paused-workflow form state, keyed by session key
-# (launcher_type_value + "_" + launcher_id). Each session holds an
-# insertion-ordered dict of form_token -> form_data, allowing multiple
-# Dify workflows to be paused simultaneously for the same session.
-_PENDING_FORMS: dict[str, 'OrderedDict[str, dict[str, typing.Any]]'] = {}
-_PENDING_FORM_DEFAULT_TTL = 30 * 60  # 30 minutes safety cap
-
-
-def _session_key_from_query(query: pipeline_query.Query) -> str:
-    return f'{query.session.launcher_type.value}_{query.session.launcher_id}'
-
-
-def _prune_pending_forms(now: float | None = None) -> None:
-    if now is None:
-        now = time.time()
-    for session_key in list(_PENDING_FORMS.keys()):
-        forms = _PENDING_FORMS[session_key]
-        expired_tokens = [token for token, data in forms.items() if data.get('_expires_at', 0) <= now]
-        for token in expired_tokens:
-            forms.pop(token, None)
-        if not forms:
-            _PENDING_FORMS.pop(session_key, None)
-
-
-def _set_pending_form(session_key: str, form_data: dict[str, typing.Any]) -> None:
-    _prune_pending_forms()
-    stored = dict(form_data)
-    expiration_time = stored.get('expiration_time')
-    try:
-        expiration_ts = float(expiration_time) if expiration_time is not None else 0.0
-    except (TypeError, ValueError):
-        expiration_ts = 0.0
-    stored['_expires_at'] = expiration_ts or (time.time() + _PENDING_FORM_DEFAULT_TTL)
-    form_token = str(stored.get('form_token') or '')
-    forms = _PENDING_FORMS.setdefault(session_key, OrderedDict())
-    # Re-insert at the end so this becomes the "latest" entry
-    forms.pop(form_token, None)
-    forms[form_token] = stored
-
-
-def _get_pending_form_by_token(session_key: str, form_token: str) -> dict[str, typing.Any] | None:
-    _prune_pending_forms()
-    forms = _PENDING_FORMS.get(session_key)
-    if not forms or not form_token:
-        return None
-    return forms.get(form_token)
-
-
-def _get_pending_form_by_w_suffix(session_key: str, w_suffix: str) -> dict[str, typing.Any] | None:
-    """Look up a pending form whose workflow_run_id ends with the given suffix.
-
-    Used by adapters (e.g. Telegram) whose callback payload is too small to
-    carry the full form_token / workflow_run_id.
-    """
-    _prune_pending_forms()
-    forms = _PENDING_FORMS.get(session_key)
-    if not forms or not w_suffix:
-        return None
-    for token in reversed(forms):
-        form = forms[token]
-        if str(form.get('workflow_run_id', '')).endswith(w_suffix):
-            return form
-    return None
-
-
-def _get_latest_pending_form(session_key: str) -> dict[str, typing.Any] | None:
-    _prune_pending_forms()
-    forms = _PENDING_FORMS.get(session_key)
-    if not forms:
-        return None
-    return forms[next(reversed(forms))]
-
-
-def _iter_pending_forms(session_key: str) -> typing.Iterator[dict[str, typing.Any]]:
-    """Iterate pending forms for a session, newest-first."""
-    _prune_pending_forms()
-    forms = _PENDING_FORMS.get(session_key)
-    if not forms:
-        return
-    for token in reversed(list(forms.keys())):
-        yield forms[token]
-
-
-def _clear_pending_form(session_key: str, form_token: str | None = None) -> None:
-    """Clear one specific pending form (by token) or all forms for the session."""
-    forms = _PENDING_FORMS.get(session_key)
-    if not forms:
-        return
-    if form_token is None:
-        _PENDING_FORMS.pop(session_key, None)
-        return
-    forms.pop(form_token, None)
-    if not forms:
-        _PENDING_FORMS.pop(session_key, None)
-
-
 @runner.runner_class('dify-service-api')
 class DifyServiceAPIRunner(runner.RequestRunner):
     """Dify Service API 对话请求器"""
@@ -433,140 +335,11 @@ class DifyServiceAPIRunner(runner.RequestRunner):
 
         query.session.using_conversation.uuid = chunk['conversation_id']
 
-    async def _submit_workflow_form_blocking(
-        self, form_action: dict
-    ) -> typing.AsyncGenerator[provider_message.Message, None]:
-        """Submit human input to resume a paused Dify workflow (non-streaming)."""
-
-        form_token = form_action['form_token']
-        workflow_run_id = form_action['workflow_run_id']
-        user = form_action['user']
-        action_id = form_action.get('action_id', '')
-        inputs = form_action.get('inputs', {})
-
-        async for chunk in self.dify_client.workflow_submit(
-            form_token=form_token,
-            workflow_run_id=workflow_run_id,
-            inputs=inputs,
-            user=user,
-            action=action_id,
-            timeout=120,
-        ):
-            self.ap.logger.debug('dify-workflow-submit-chunk: ' + str(chunk))
-
-            if chunk['event'] == 'workflow_finished':
-                if chunk['data'].get('error'):
-                    raise errors.DifyAPIError(chunk['data']['error'])
-                content, _ = self._process_thinking_content(chunk['data']['outputs']['summary'])
-                yield provider_message.Message(
-                    role='assistant',
-                    content=content,
-                )
-
-    def _resolve_pending_form(self, session_key: str, form_action: dict) -> dict | None:
-        """Locate the pending form this action targets.
-
-        Tries identifiers in order of specificity: form_token, full
-        workflow_run_id, workflow_run_id suffix (Telegram-style compact id),
-        then falls back to the newest pending form for the session.
-        """
-        form_token = form_action.get('form_token')
-        if form_token:
-            form = _get_pending_form_by_token(session_key, form_token)
-            if form:
-                return form
-
-        workflow_run_id = form_action.get('workflow_run_id')
-        if workflow_run_id:
-            for form in _iter_pending_forms(session_key):
-                if form.get('workflow_run_id') == workflow_run_id:
-                    return form
-
-        w_suffix = form_action.get('w_suffix')
-        if w_suffix:
-            form = _get_pending_form_by_w_suffix(session_key, w_suffix)
-            if form:
-                return form
-
-        return _get_latest_pending_form(session_key)
-
-    def _merge_pending_form_action(self, session_key: str, form_action: dict | None) -> dict | None:
-        """Backfill resume fields from the matching pending form."""
-        if not form_action:
-            return None
-
-        merged_action = dict(form_action)
-        merged_action.pop('w_suffix', None)
-        pending_form = self._resolve_pending_form(session_key, form_action)
-        if pending_form:
-            merged_action['form_token'] = merged_action.get('form_token') or pending_form.get('form_token', '')
-            merged_action['workflow_run_id'] = merged_action.get('workflow_run_id') or pending_form.get(
-                'workflow_run_id', ''
-            )
-            merged_action.setdefault('inputs', pending_form.get('inputs', {}))
-            merged_action.setdefault('user', pending_form.get('user', ''))
-            merged_action.setdefault('node_title', pending_form.get('node_title', ''))
-
-            # Resolve clicked action's display title from the stored actions list
-            if 'action_title' not in merged_action:
-                clicked_id = merged_action.get('action_id', '')
-                for action in pending_form.get('actions', []):
-                    if str(action.get('id', '')) == str(clicked_id):
-                        merged_action['action_title'] = action.get('title', clicked_id)
-                        break
-
-        return merged_action
-
-    def _match_pending_form_action(self, session_key: str, user_text: str) -> dict | None:
-        """Match plain text replies against pending Dify form actions.
-
-        Iterates all pending forms newest-first; the first action whose
-        title/id matches the text wins. This means when multiple forms are
-        pending with the same button label, the most recent one resolves.
-        """
-        normalized_text = user_text.strip().lower()
-        if not normalized_text:
-            return None
-
-        for pending_form in _iter_pending_forms(session_key):
-            for action in pending_form.get('actions', []):
-                titles = {
-                    str(action.get('title', '')).strip().lower(),
-                    str(action.get('id', '')).strip().lower(),
-                }
-                if normalized_text in titles:
-                    return {
-                        'form_token': pending_form.get('form_token', ''),
-                        'workflow_run_id': pending_form.get('workflow_run_id', ''),
-                        'action_id': action.get('id', ''),
-                        'action_title': action.get('title', action.get('id', '')),
-                        'node_title': pending_form.get('node_title', ''),
-                        'inputs': pending_form.get('inputs', {}),
-                        'user': pending_form.get('user', ''),
-                    }
-
-        return None
-
     async def _workflow_messages(
         self, query: pipeline_query.Query
     ) -> typing.AsyncGenerator[provider_message.Message, None]:
         """调用工作流"""
 
-        # Check if this is a form action resume (button click or text match)
-        form_action_raw = query.variables.get('_dify_form_action')
-        session_key = _session_key_from_query(query)
-
-        if form_action_raw:
-            form_action = self._merge_pending_form_action(session_key, form_action_raw)
-        else:
-            form_action = self._match_pending_form_action(session_key, str(query.message_chain))
-
-        if form_action:
-            _clear_pending_form(session_key, form_action.get('form_token') or None)
-            async for msg in self._submit_workflow_form_blocking(form_action):
-                yield msg
-            return
-
         if not query.session.using_conversation.uuid:
             query.session.using_conversation.uuid = str(uuid.uuid4())
 
@@ -593,7 +366,6 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         }
 
         inputs.update(query.variables)
-        human_input_yielded = False
 
         async for chunk in self.dify_client.workflow_run(
             inputs=inputs,
@@ -605,46 +377,6 @@ class DifyServiceAPIRunner(runner.RequestRunner):
             if chunk['event'] in ignored_events:
                 continue
 
-            if chunk['event'] == 'workflow_paused':
-                reasons = chunk['data'].get('reasons', [])
-                workflow_run_id = chunk['data'].get('workflow_run_id', '')
-                for reason in reasons:
-                    if reason.get('TYPE') == 'human_input_required':
-                        form_content = reason.get('form_content', '')
-                        actions = reason.get('actions', [])
-                        node_title = reason.get('node_title', '')
-
-                        _set_pending_form(
-                            _session_key_from_query(query),
-                            {
-                                'workflow_run_id': workflow_run_id,
-                                'form_id': reason.get('form_id'),
-                                'form_token': reason.get('form_token'),
-                                'node_id': reason.get('node_id'),
-                                'node_title': node_title,
-                                'form_content': form_content,
-                                'inputs': reason.get('inputs', {}),
-                                'actions': actions,
-                                'expiration_time': reason.get('expiration_time'),
-                                'user': f'{query.session.launcher_type.value}_{query.session.launcher_id}',
-                            },
-                        )
-
-                        query.variables['_dify_form_render'] = {
-                            'form_content': form_content,
-                            'actions': actions,
-                            'node_title': node_title,
-                        }
-
-                        action_lines = '\n'.join(f'- [{a.get("title", a.get("id", ""))}]' for a in actions)
-                        display_text = f'[Human Input Required] {node_title}\n{form_content}\n{action_lines}'
-
-                        human_input_yielded = True
-                        yield provider_message.Message(
-                            role='assistant',
-                            content=display_text,
-                        )
-
             if chunk['event'] == 'node_started':
                 if chunk['data']['node_type'] == 'start' or chunk['data']['node_type'] == 'end':
                     continue
@@ -667,8 +399,6 @@ class DifyServiceAPIRunner(runner.RequestRunner):
                 yield msg
 
             elif chunk['event'] == 'workflow_finished':
-                if human_input_yielded:
-                    break
                 if chunk['data']['error']:
                     raise errors.DifyAPIError(chunk['data']['error'])
                 content, _ = self._process_thinking_content(chunk['data']['outputs']['summary'])
@@ -906,153 +636,11 @@ class DifyServiceAPIRunner(runner.RequestRunner):
 
         query.session.using_conversation.uuid = chunk['conversation_id']
 
-    async def _submit_workflow_form(
-        self, form_action: dict
-    ) -> typing.AsyncGenerator[provider_message.MessageChunk, None]:
-        """Submit human input to resume a paused Dify workflow."""
-
-        form_token = form_action['form_token']
-        workflow_run_id = form_action['workflow_run_id']
-        user = form_action['user']
-        action_id = form_action.get('action_id', '')
-        action_title = form_action.get('action_title', '') or action_id
-        node_title = form_action.get('node_title', '')
-        inputs = form_action.get('inputs', {})
-
-        messsage_idx = 0
-        is_final = False
-        think_start = False
-        think_end = False
-        workflow_contents = ''
-        repause_form_data: dict | None = None
-
-        remove_think = self.pipeline_config['output'].get('misc', {}).get('remove-think')
-        async for chunk in self.dify_client.workflow_submit(
-            form_token=form_token,
-            workflow_run_id=workflow_run_id,
-            inputs=inputs,
-            user=user,
-            action=action_id,
-            timeout=120,
-        ):
-            self.ap.logger.debug('dify-workflow-submit-chunk: ' + str(chunk))
-
-            yield_this_iteration = False
-
-            if chunk['event'] == 'workflow_finished':
-                is_final = True
-                yield_this_iteration = True
-                if chunk['data'].get('error'):
-                    raise errors.DifyAPIError(chunk['data']['error'])
-
-            if chunk['event'] == 'workflow_paused':
-                reasons = chunk['data'].get('reasons', [])
-                new_run_id = chunk['data'].get('workflow_run_id', workflow_run_id)
-                for reason in reasons:
-                    if reason.get('TYPE') != 'human_input_required':
-                        continue
-                    form_content = reason.get('form_content', '')
-                    actions = reason.get('actions', [])
-                    # Use a distinct name — `node_title` (the just-resolved step)
-                    # must keep its value so the resume notice on the previous
-                    # card still shows which step the user acted on.
-                    paused_node_title = reason.get('node_title', '')
-                    raw_inputs = reason.get('inputs', {})
-
-                    _set_pending_form(
-                        user,
-                        {
-                            'workflow_run_id': new_run_id,
-                            'form_id': reason.get('form_id'),
-                            'form_token': reason.get('form_token'),
-                            'node_id': reason.get('node_id'),
-                            'node_title': paused_node_title,
-                            'form_content': form_content,
-                            'inputs': raw_inputs if isinstance(raw_inputs, dict) else {},
-                            'actions': actions,
-                            'expiration_time': reason.get('expiration_time'),
-                            'user': user,
-                        },
-                    )
-
-                    repause_form_data = {
-                        'form_content': form_content,
-                        'actions': actions,
-                        'node_title': paused_node_title,
-                        'workflow_run_id': new_run_id,
-                        'form_token': reason.get('form_token', ''),
-                    }
-                    # Ensure the final chunk has non-empty content so
-                    # ResponseWrapper (which skips empty-content chunks) lets it
-                    # propagate to SendResponseBackStage. Use a zero-width space
-                    # so neither Lark nor Telegram renders visible noise — the
-                    # adapter substitutes its own card text from _form_data.
-                    if not workflow_contents:
-                        workflow_contents = '​'
-                    is_final = True
-                    yield_this_iteration = True
-                    break
-
-            if chunk['event'] == 'text_chunk':
-                messsage_idx += 1
-                if remove_think:
-                    if '<think>' in chunk['data']['text'] and not think_start:
-                        think_start = True
-                        continue
-                    if '</think>' in chunk['data']['text'] and not think_end:
-                        import re
-
-                        content = re.sub(r'^\n</think>', '', chunk['data']['text'])
-                        workflow_contents += content
-                        think_end = True
-                    elif think_end:
-                        workflow_contents += chunk['data']['text']
-                    if think_start:
-                        continue
-                else:
-                    workflow_contents += chunk['data']['text']
-                if messsage_idx % 8 == 0:
-                    yield_this_iteration = True
-
-            if yield_this_iteration:
-                msg = provider_message.MessageChunk(
-                    role='assistant',
-                    content=workflow_contents,
-                    is_final=is_final,
-                )
-                msg._resume_from_form = True
-                if action_title:
-                    msg._resume_action_title = action_title
-                if node_title:
-                    msg._resume_node_title = node_title
-                if is_final and repause_form_data:
-                    msg._form_data = repause_form_data
-                    msg._open_new_card = True
-                yield msg
-                if is_final:
-                    return
-
     async def _workflow_messages_chunk(
         self, query: pipeline_query.Query
     ) -> typing.AsyncGenerator[provider_message.MessageChunk, None]:
         """调用工作流"""
 
-        # Check if this is a form action resume (button click or text match)
-        form_action_raw = query.variables.get('_dify_form_action')
-        session_key = _session_key_from_query(query)
-
-        if form_action_raw:
-            form_action = self._merge_pending_form_action(session_key, form_action_raw)
-        else:
-            form_action = self._match_pending_form_action(session_key, str(query.message_chain))
-
-        if form_action:
-            _clear_pending_form(session_key, form_action.get('form_token') or None)
-            # Resume paused workflow via submit endpoint
-            async for msg in self._submit_workflow_form(form_action):
-                yield msg
-            return
-
         if not query.session.using_conversation.uuid:
             query.session.using_conversation.uuid = str(uuid.uuid4())
 
@@ -1084,13 +672,6 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         think_start = False
         think_end = False
         workflow_contents = ''
-        workflow_run_id = ''
-        human_input_yielded = False
-
-        # Saved form data to attach to the final MessageChunk so the adapter
-        # can detect it when is_final=True and render buttons.
-        pending_form_data = None
-        display_text = ''
 
         remove_think = self.pipeline_config['output'].get('misc', '').get('remove-think')
         async for chunk in self.dify_client.workflow_run(
@@ -1101,62 +682,7 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         ):
             self.ap.logger.debug('dify-workflow-chunk: ' + str(chunk))
             if chunk['event'] in ignored_events:
-                if chunk['event'] == 'workflow_started':
-                    workflow_run_id = chunk['data'].get('workflow_run_id', '')
                 continue
-
-            if chunk['event'] == 'workflow_paused':
-                reasons = chunk['data'].get('reasons', [])
-                workflow_run_id = chunk['data'].get('workflow_run_id', workflow_run_id)
-                for reason in reasons:
-                    if reason.get('TYPE') == 'human_input_required':
-                        form_content = reason.get('form_content', '')
-                        actions = reason.get('actions', [])
-                        node_title = reason.get('node_title', '')
-
-                        # Persist form state in module-level store keyed by session
-                        raw_inputs = reason.get('inputs', {})
-                        _set_pending_form(
-                            _session_key_from_query(query),
-                            {
-                                'workflow_run_id': workflow_run_id,
-                                'form_id': reason.get('form_id'),
-                                'form_token': reason.get('form_token'),
-                                'node_id': reason.get('node_id'),
-                                'node_title': node_title,
-                                'form_content': form_content,
-                                'inputs': raw_inputs if isinstance(raw_inputs, dict) else {},
-                                'actions': actions,
-                                'expiration_time': reason.get('expiration_time'),
-                                'user': f'{query.session.launcher_type.value}_{query.session.launcher_id}',
-                            },
-                        )
-
-                        # Pass form render metadata to downstream stages
-                        query.variables['_dify_form_render'] = {
-                            'form_content': form_content,
-                            'actions': actions,
-                            'node_title': node_title,
-                        }
-
-                        action_lines = '\n'.join(f'- [{a.get("title", a.get("id", ""))}]' for a in actions)
-                        display_text = f'[Human Input Required] {node_title}\n{form_content}\n{action_lines}'
-                        workflow_contents += display_text + '\n'
-
-                        # Save form data to attach to the final chunk later.
-                        # We do NOT yield here — the form content will be sent
-                        # as the final MessageChunk (with is_final=True and
-                        # _form_data) so the adapter can update the card and
-                        # add buttons in one pass.
-                        pending_form_data = {
-                            'form_content': form_content,
-                            'actions': actions,
-                            'node_title': node_title,
-                            'workflow_run_id': workflow_run_id,
-                            'form_token': reason.get('form_token', ''),
-                        }
-                        human_input_yielded = True
-
             if chunk['event'] == 'workflow_finished':
                 is_final = True
                 if chunk['data']['error']:
@@ -1204,29 +730,11 @@ class DifyServiceAPIRunner(runner.RequestRunner):
                 yield msg
 
             if messsage_idx % 8 == 0 or is_final:
-                final_content = workflow_contents if workflow_contents.strip() else ''
-                msg = provider_message.MessageChunk(
+                yield provider_message.MessageChunk(
                     role='assistant',
-                    content=final_content,
+                    content=workflow_contents,
                     is_final=is_final,
                 )
-                # Attach form data to the final chunk for the adapter
-                if is_final and pending_form_data:
-                    msg._form_data = pending_form_data
-                    pending_form_data = None
-                yield msg
-
-        # If the stream ended after workflow_paused without a
-        # workflow_finished event, yield a final chunk so the adapter
-        # can update the card and add buttons.
-        if human_input_yielded and not is_final:
-            msg = provider_message.MessageChunk(
-                role='assistant',
-                content=workflow_contents or display_text,
-                is_final=True,
-            )
-            msg._form_data = pending_form_data
-            yield msg
 
     async def run(self, query: pipeline_query.Query) -> typing.AsyncGenerator[provider_message.Message, None]:
         """运行请求"""