From a4eb688814b3ab6e65983203df766ebf6531de3e Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Mon, 20 Jul 2026 01:05:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(server,static):=20=E5=AF=B9=E8=AF=9D?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E7=8A=B6=E6=80=81=20REST=20=E5=AF=B9?= =?UTF-8?q?=E8=B4=A6=E4=B8=8E=20conversation=20=E7=BA=A7=20WebTerminal=20?= =?UTF-8?q?=E9=9A=94=E7=A6=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 议题1 运行状态对账： - 后端新增 GET /api/conversations//running-status 聚合主任务/子智能体/ 后台命令/多智能体四类运行状态 - 前端 3 个 1s probe 收敛为单个 2.5s 对账循环，事件为主、对账纠偏， 清理方向连续 2 次确认，恢复方向立即接管，修复轮询 404 卡死 议题4 conversation 级隔离： - terminal 缓存键改为 username::workspace::conversation 三段， 每个对话独立 WebTerminal/文件管理/终端/子智能体 - 同工作区多对话可并行运行，互斥维度从工作区改为同对话 - socket 终端面板按对话过滤广播，terminal 工具每对话最多 3 个 - 常驻内存 + 24h TTL 回收器（无活动且无运行任务才回收） - 子智能体 restore 按 owner_conversation_id 过滤，避免多 manager 重复恢复 --- core/main_terminal.py | 2 + core/web_terminal.py | 37 +- docs/conversation_level_terminal_design.md | 160 +++++ modules/sub_agent/manager.py | 11 + server/app_legacy.py | 7 +- server/context.py | 210 ++++++- server/multi_agent.py | 2 +- server/socket_handlers.py | 8 +- server/tasks/api.py | 61 +- server/tasks/models.py | 85 ++- static/src/App.vue | 1 + static/src/app/computed.ts | 4 +- static/src/app/methods/message/send.ts | 8 +- .../app/methods/taskPolling/compression.ts | 4 + static/src/app/methods/taskPolling/probe.ts | 555 +++++------------- static/src/app/methods/ui/terminal.ts | 7 +- static/src/app/state.ts | 6 +- .../src/components/panels/TerminalPanel.vue | 38 +- static/src/composables/useLegacySocket.ts | 4 +- 19 files changed, 742 insertions(+), 468 deletions(-) create mode 100644 docs/conversation_level_terminal_design.md diff --git a/core/main_terminal.py b/core/main_terminal.py index af095699..e7901f70 100644 --- a/core/main_terminal.py +++ b/core/main_terminal.py @@ -143,6 +143,8 @@ class MainTerminal(MainTerminalCommandMixin, MainTerminalContextMixin, MainTermi project_path=self.project_path, data_dir=str(self.data_dir), container_session=container_session, + # 对话级隔离（仅 WebTerminal 设置该属性；CLI MainTerminal 无此属性，getattr 回退 None） + owner_conversation_id=getattr(self, "_bound_conversation_id", None), ) self.sub_agent_manager.set_terminal(self) self.background_command_manager = BackgroundCommandManager( diff --git a/core/web_terminal.py b/core/web_terminal.py index 7cbd9f89..d7d2c9cb 100644 --- a/core/web_terminal.py +++ b/core/web_terminal.py @@ -30,10 +30,27 @@ class WebTerminal(MainTerminal): """Web版本的终端，继承自MainTerminal，包含对话持久化功能""" def _ensure_conversation(self): - """确保Web端在首次进入时自动加载或创建对话""" + """确保Web端在首次进入时自动加载或创建对话。 + + 对话级 terminal（_bound_conversation_id 非空）：加载绑定的对话； + 加载失败不 fallback 到“最近对话”（避免串对话），保持无对话状态， + 由任务执行链路的 ensure_conversation_loaded 显式处理。 + 工作区级 terminal（未绑定）：保持原有“最近对话”行为。 + """ if self.context_manager.current_conversation_id: return + bound_id = getattr(self, "_bound_conversation_id", None) + if bound_id: + if not str(bound_id).startswith("conv_"): + bound_id = f"conv_{bound_id}" + result = self.load_conversation(bound_id, restore_model=False) + if result.get("success"): + print(f"[WebTerminal] 已加载绑定对话: {bound_id}") + else: + print(f"[WebTerminal] 警告：绑定对话 {bound_id} 加载失败: {result.get('message') or result.get('error')}") + return + latest_list = self.context_manager.get_conversation_list(limit=1, offset=0) conversations = latest_list.get("conversations", []) if latest_list else [] @@ -53,15 +70,29 @@ class WebTerminal(MainTerminal): print(f"[WebTerminal] 自动创建新对话: {conversation_id}") def __init__( - self, - project_path: str, + self, + project_path: str, thinking_mode: bool = False, run_mode: Optional[str] = None, message_callback: Optional[Callable] = None, data_dir: Optional[str] = None, container_session: Optional["ContainerHandle"] = None, usage_tracker: Optional[object] = None, + conversation_id: Optional[str] = None, ): + # 对话级隔离：绑定指定对话（必须在 super().__init__ 之前设置， + # 因为父类初始化会调用 self._ensure_conversation()） + self._bound_conversation_id = conversation_id + # 24h TTL 回收器判定用：最近活动时间 + self.last_activity_at = time.time() + # 对话级 terminal：广播回调注入 conversation_id，前端按对话过滤终端事件 + if message_callback is not None and conversation_id: + _raw_callback = message_callback + def message_callback(event_type, data, _cb=_raw_callback, _cid=conversation_id): + if isinstance(data, dict): + data = dict(data) + data.setdefault("conversation_id", _cid) + return _cb(event_type, data) # 调用父类初始化（包含对话持久化功能） super().__init__( project_path, diff --git a/docs/conversation_level_terminal_design.md b/docs/conversation_level_terminal_design.md new file mode 100644 index 00000000..5f9baf89 --- /dev/null +++ b/docs/conversation_level_terminal_design.md @@ -0,0 +1,160 @@ +# Conversation 级 WebTerminal 隔离改造——交接文档（压缩后必读） + +> 写于 2026-07-19，供上下文压缩后的对话继续工作。 +> **读完本文档即可直接开工，无需重新调研代码**。行号基于 2026-07-19 的代码，若有漂移以搜索符号名为准。 +> 相关项目记忆：`runtime_state_loading_concurrency_redesign`（4 议题讨论总索引）。 + +--- + +## 0. 当前状态一句话 + +议题 1（运行状态 REST 对账）✅ 已完成并端到端验证；**议题 4（conversation 级 WebTerminal 隔离）✅ 已于 2026-07-20 完成并端到端验证**（见 §7）。议题 2（子智能体按需加载）、议题 3（worktree）用户明确暂缓，是未来可做项。 + +## 1. 用户已拍板的设计决策（不要重新讨论） + +1. **单对话单 WebTerminal**：每个对话有自己的 context_manager、file_manager、TerminalManager（shell）、SubAgentManager、BackgroundCommandManager。 +2. **生命周期：常驻 + 24 小时 TTL**。实例随对话首次运行创建并常驻；仅当「超过 24 小时无活动 且 无运行中任务」才回收。不追求精确释放（原因见 §4 不可序列化清单）。 +3. **worktree 后置**：先做 terminal 隔离。用户观点：worktree 只能硬隔离文件编辑工具（路径锚定），run_command 只能软隔离（cwd 锚定+agent 自觉），Claude Code/Codex 也一样，工作量不大，后面再做。 +4. 对话互斥维度：**同工作区互斥 → 同对话互斥**（同一对话仍禁止两个主 chat 任务并行，防止串写对话历史）。 + +## 2. 议题 1 已完成内容（文件 + 行号，勿重复调研） + +### 后端 + +- `server/tasks/models.py` + - `TaskManager.get_conversation_running_status(terminal, conversation_id)`（约 604-665 行）：新增。聚合 3 类后台状态，返回 `{has_running_sub_agents, has_running_background_commands, has_running_multi_agent}`。sub_agents 排除 multi_agent_mode 任务；multi_agent 判定 = 非终态多智能体任务 或 `MultiAgentState` 有 running 实例 或 `has_pending_master_messages()`。 + - `_has_running_background(rec, terminal)`（约 667-682 行）：改为委托上面的方法，保持旧语义（多智能体计入 sub_agents）。 + - `create_chat_task` 互斥逻辑在 **147-155 行**（⚠️ 议题 4 要改这里）。 + - `list_tasks(username, workspace_id)` 在约 206 行。 +- `server/tasks/api.py` + - `GET /api/conversations//running-status?workspace_id=`（约 38-90 行，插在 `create_task_api` 之前）。返回 `{is_main_running, main_task_id, main_task_type, has_running_sub_agents, has_running_background_commands, has_running_multi_agent, is_truly_active}`。 + +### 前端 + +- `static/src/app/methods/taskPolling/probe.ts`（已整体重写，@ts-nocheck，现约 180 行） + - `startRunningStateReconcile()`：幂等启动 2.5s 对账循环 + 立即对账一次。 + - `reconcileRunningStateOnce()`：核心逻辑。恢复方向立即执行（server active 本地空闲 → `taskStore.resumeTask(main_task_id)` + 设 waiting 标志；多智能体只设 taskInProgress 不设 waitingForSubAgent）；清理方向保守（`taskStore.isPolling` 活跃则不干预；否则需连续 2 次 server idle 确认才清 streamingMessage/taskInProgress/waiting 标志 + `clearTaskState()`）。网络失败一律不变更状态。 + - `startWaitingTaskProbe/stopWaitingTaskProbe/startMultiAgentTaskProbe/stopMultiAgentTaskProbe`：保留为委托/no-op（调用点未动：lifecycle.ts 400/407/418/419/502/531、messaging.ts 163/165、scroll.ts 31-32）。 + - `restoreSubAgentWaitingState(retry)`：简化为启动对账循环（被 compression.ts 调用）。 +- `static/src/app/state.ts` ~29 行：`runningStateReconcileTimer: null` + `_runningStateIdleStreak: 0`。 +- `static/src/app/methods/taskPolling/compression.ts` ~96-103 行：`restoreTaskState()` 入口启动对账。 +- `static/src/app/methods/message/send.ts` ~312-316 行：`taskInProgress = true` 后启动对账。 + +### 验证资产 + +- 验证脚本：`_experiments/verify_running_status_api.py`（test_client 4 项断言：401/空闲/活动/终态，全过）。 +- 已验证：冒烟 6/6、前端构建、真实服务 playwright 端到端（2.5s 精确间隔、外部建任务前端自动接管轮询、完成后正常收尾）。 +- 测试副作用：`conv_20260718_005231_752`（张雪峰对话）多了一条"对账测试通过"消息，可删。 +- **既有 bug（与议题 1 无关，未修）**：assistant 消息 `reasoning_content` 前端渲染两遍（对话文件数据正常，刷新仍复现，历史加载渲染路径问题）。 + +## 3. 议题 4 改动地图（已普查，直接照此施工） + +### 3.1 核心缓存与资源链路 + +- `server/state.py:28`：`user_terminals: Dict[str, WebTerminal] = {}` —— 全局缓存，当前 key = `username::workspace_id`。 +- `server/context.py` + - `_make_terminal_key(username, workspace_id)` **49 行**：返回 `f"{username}::{workspace_id}"`。→ 议题 4 改为加 conversation_id 段。 + - `get_user_resources(username, workspace_id, update_session)` **131 行**：所有资源获取的统一入口。 + - host 模式分支 ~143-280 行：term_key 在 **205 行**生成、**280 行**写入缓存；含「project_path 变化时重建 terminal」逻辑（~210-260 行）。 + - web 模式分支：**377 行**生成 term_key、**413 行**写入缓存。 + - ⚠️ 调用点普查：**34 处调用 / 14 个文件**。分布：`server/api_v1.py` 14 处、`server/multi_agent.py` 6 处、`server/tasks/api.py` 4 处、`server/tasks/models.py` 3 处、`server/context.py` 2 处、`server/app_legacy.py` 2 处、`server/tasks/skills.py` 1 处、`server/socket_handlers.py` 1 处。 + - 改造策略建议：不改 `get_user_resources` 签名，新增 `get_conversation_resources(username, workspace_id, conversation_id)`，逐调用点迁移；chat 任务链路（tasks/models.py、tasks/api.py、chat_flow*）优先。 + +### 3.2 WebTerminal 本体 + +- `core/web_terminal.py` + - `class WebTerminal(MainTerminal)` ~30 行。 + - `_ensure_conversation()` **33-52 行**：当前自动加载「最近对话」→ 议题 4 改为加载**指定 conversation_id**（或新对话则 start_new）。 + - `__init__` **55-100 行**：调用父类 init → 创建 TerminalManager（83 行）→ attach。 + - `create_new_conversation()` ~110 行起。 + - `close()` ~809 行：`terminal_manager.close_all()`。 +- `core/main_terminal.py:142`：SubAgentManager 每 WebTerminal 一个（非全局单例）——对话级化天然兼容。 +- `modules/terminal_manager.py:65`：TerminalManager（max_terminals 限制是 per-manager 的，对话级后每对话一个 manager，互不干扰）。 +- `modules/persistent_terminal/start.py`：shell 是**惰性启动**（Popen 在 167/208/269/346 行，首次用才起进程）——这是内存可控的关键，保持不变。 + +### 3.3 互斥与前端 + +- `server/tasks/models.py:147-155`：chat 任务互斥（`task_type="chat"` 时同用户同工作区禁止并发，409 "当前工作区已有运行中的任务"）。→ 改为同 conversation_id 互斥。 +- `static/src/app/methods/message/send.ts:25`：前端发送拦截。 +- `static/src/app/computed.ts:220`：`currentWorkspaceHasRunningTask`（仅 host/docker 模式生效）→ 改为按对话判定，**可直接复用议题 1 的 running-status 接口**。 +- 限制 B（运行中禁切传统/多智能体模式，纯前端）：`static/src/app/methods/ui/mode.ts:73` + QuickMenu.vue disabled。用户尚未拍板何时放开，默认暂不动。 + +### 3.4 回收器（新建） + +- 定期扫描 `state.user_terminals`，回收条件：当前时间 - 最后活动时间 > 24h 且该对话无运行任务（可用 `TaskManager.get_conversation_running_status` 判定）。 +- 回收动作：`terminal.close()`（关 shell）→ 从 user_terminals 移除。对话历史本来就有磁盘持久化，无损。 +- 需要给 WebTerminal 加 `last_activity_at` 字段（任务创建/工具调用时刷新）。 + +### 3.5 需要注意的关联点 + +- `restore_running_tasks()`（`modules/sub_agent/manager.py:279`）：启动时恢复多智能体子智能体，当前按 terminal 恢复 → 对话级后恢复逻辑要按对话 terminal 走。 +- `stop_flags`（`server/state.py`）：按 task_id 全局 dict，task 本就属于具体对话，理论上不用大改，但要检查清理时机。 +- 多智能体模式：`multi_agent_mode` 是对话级 metadata，对话级 terminal 后 `web_terminal.multi_agent_mode` 属性语义要重新对齐（当前是 terminal 级属性，chat_flow_task_main.py 多处 `getattr(web_terminal, "multi_agent_mode", False)`）。 +- 后台通知轮询（`poll_completion_notifications` / `poll_multi_agent_notifications`，server/chat_flow_task_main.py）：持有 terminal 引用，对话级后注意引用来源。 + +## 4. 为什么选「常驻+TTL」而不是「空闲释放」（勿重新争论） + +不可序列化的状态（释放即丢，无法 100% 恢复）： +1. **shell 进程状态**：cwd、`export` 变量、激活的 venv、后台 job（OS 进程态）。快照重放只能覆盖 90%，且错误是静默的（AI 在错误目录执行命令）。 +2. **子智能体 asyncio 状态**：运行中/idle 的 Task、等待唤醒的 Event、ask_master 的 Future。现有 restore 也只能恢复到「强制 idle」。 +3. **后台命令进程句柄**。 + +内存估算（已给用户确认）：单对话常驻 ~10-25MB（shell 惰性），30 活跃对话 ~300-750MB，本机单用户可控。 + +## 5. 服务端当前状态 + +- 服务当前**未在运行**（用户关过，我起的 8091 后台进程也已结束）。重启命令：`python3 -m server.app --port 8091`（host 模式，`/host-login` 免密登录；POST 需先 GET `/api/csrf-token` 拿 `token` 字段放 `X-CSRF-Token` 头）。 +- 前端构建命令：`npm run build --silent 2>&1 | tail -n 5`。 +- 验证命令：`python3 -m py_compile ` + `python3 -m unittest test.test_server_refactor_smoke`。 + +## 6. 下一步（压缩后从这里继续） + +议题 4 已全部完成。未来可做：议题 2（子智能体按需加载）、议题 3（worktree）、`currentWorkspaceHasRunningTask` 更名为 `currentConversationHasRunningTask`（语义已变、名字未改）。 + +--- + +## 7. 议题 4 实现记录（2026-07-20 完成） + +### 实际改动（与原计划偏差：未新增 `get_conversation_resources`，改为 `get_user_resources` 加 `conversation_id=None` 可选参数——向后兼容、更 DRY，迁移即传参） + +**后端** +- `server/context.py` + - `_make_terminal_key(username, workspace_id, conversation_id=None)`：三段键 `u::w::cid` + - `_wrap_callback_with_conversation_id(callback, cid)`：广播 data 注入 cid（setdefault） + - `_touch_terminal_activity(terminal, cid)`：刷新 `last_activity_at` + - `attach_user_broadcast`：读 `terminal._bound_conversation_id` 自动包装 + - `get_user_resources(..., conversation_id=None)`：host/web 两分支 term_key 带对话段；**container_key 保持工作区级**（docker 一工作区一容器，用户确认）；WebTerminal 创建传 cid；return 前 touch + - `with_terminal`：从 query/JSON body 读 conversation_id（/api/terminals 等自动支持） + - `get_terminal_for_sid(sid, conversation_id=None)` + - 回收器（文件末尾）：`CONVERSATION_TERMINAL_TTL_SECONDS`（默认 24h，env 可调）/ `reap_idle_conversation_terminals(now)`（可测试）/ `_conversation_terminal_reaper_loop` / `start_conversation_terminal_reaper`（幂等）；回收=保存对话+close_all shell+close MCP+pop；仅三段键、过期、无运行工作（主任务+3类后台，判定失败保守不回收） +- `core/web_terminal.py`：`__init__` 加 `conversation_id=None`（super 前设 `_bound_conversation_id`、`last_activity_at`；message_callback 包装注入 cid）；`_ensure_conversation` 绑定对话优先加载（conv_ 前缀规范化，失败不 fallback） +- `core/main_terminal.py`：创建 SubAgentManager 透传 `owner_conversation_id=getattr(self,'_bound_conversation_id',None)` +- `modules/sub_agent/manager.py`：`owner_conversation_id` 属性；`restore_running_tasks` 过滤——仅恢复本对话任务，**工作区级 manager 不再恢复（行为变化：重启后多智能体任务延迟到对话激活时恢复）** +- `server/tasks/models.py`：chat 互斥改同对话（`_norm_cid` 规范化比较，文案“当前对话已有运行中的任务”）；235/264/739 三处传 `conversation_id=rec.conversation_id` +- `server/tasks/api.py`：running-status、create_task(×2)、cancel 四处传 cid +- `server/multi_agent.py`：`list_active_sub_agents_api` 传 cid（其余 5 处保持工作区级——无对话上下文/仅 workspace） +- `server/socket_handlers.py`：`terminal_subscribe`/`get_terminal_output` 从 data 读 cid +- `server/app_legacy.py`：`start_background_jobs` 接入回收器启动 +- 保持工作区级不动：`server/conversation.py`（对话列表/标题/创建）、`server/api_v1.py` 14 处、`server/tasks/skills.py`、socket connect（轻量服务实例，shell 惰性） + +**前端** +- `static/src/app/computed.ts` `currentWorkspaceHasRunningTask`：`!==` → `===`（当前对话判定；displayLockEngaged 随之按对话锁） +- `static/src/app/methods/message/send.ts`：拦截文案“当前对话正在运行” +- `static/src/components/panels/TerminalPanel.vue`：+`conversationId` prop；`acceptTerminalBroadcast` 过滤 7 个广播事件（started/list_update/output/input/closed/reset/switched；响应类 subscribed/history 不过滤）；watch conversationId 清空重订阅；emit 带 cid +- `static/src/App.vue`：TerminalPanel 传 `:conversation-id="currentConversationId"` +- `static/src/app/methods/ui/terminal.ts`：fetchTerminalCount/subscribeTerminalEvents/switchTerminalSession 带 cid +- `static/src/composables/useLegacySocket.ts`：connect 时 terminal_subscribe 带 cid + +### 验证结果（全部通过） +- py_compile + 冒烟 6/6 + 前端构建 +- `_experiments/verify_conversation_level_terminal.py`（可复用回归）9 项断言：键隔离/容器共享/绑定/Manager 独立/同对话互斥/异对话并行/广播注入/回收器 3 态 +- 真实服务 playwright 原子脚本：**PARALLEL_PROOF {a_running:true, b_running:true}**；A 运行中 B 页面零 is-disabled；B 发送不被拦；同对话 409 +- 终端隔离实测：A 创建 `test-shell-a` 后 A 列表 1 个 / B 列表 0 个；`/api/terminals` 返回 `max_allowed:3`（每对话 3 终端） +- 议题 1 回归：双任务完成后双方 is_truly_active=false，对账循环正常收尾 + +### 遗留与注意 +- **行为变化**：重启后多智能体任务不再立即恢复，延迟到对话激活（restore 过滤的必然结果） +- 测试副作用：语音工作区的 conv_20260720_001402_595/857 两个测试对话（A 内有运行中的 test-shell-a）；张雪峰对话（conv_20260718_005231_752）有议题 1 时的测试消息 +- 24h 回收器真实时长未实测（单元覆盖三态；env `CONVERSATION_TERMINAL_TTL_SECONDS` 可调短测） +- TerminalPanel 广播过滤前端 UI 未单独截图（逻辑随构建通过；后端 API 层隔离已实测） +- 既有 bug 仍未修：assistant reasoning_content 前端渲染两遍（与议题 1/4 无关） diff --git a/modules/sub_agent/manager.py b/modules/sub_agent/manager.py index 76ecc3a9..46a36ae8 100644 --- a/modules/sub_agent/manager.py +++ b/modules/sub_agent/manager.py @@ -47,9 +47,15 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi project_path: str, data_dir: str, container_session: Optional["ContainerHandle"] = None, + owner_conversation_id: Optional[str] = None, ): self.project_path = Path(project_path).resolve() self.data_dir = Path(data_dir).resolve() + # 对话级隔离：本 manager 所属 terminal 绑定的对话。 + # 非空时 restore_running_tasks 只恢复该对话的多智能体任务； + # 为空（工作区级服务 terminal）时不恢复，避免同工作区多个 + # 对话级 manager 重复恢复同一批任务（任务延迟到对话激活时恢复）。 + self.owner_conversation_id = owner_conversation_id # 子智能体任务和状态按 data_dir 隔离（web 模式下按用户/工作区自动隔离） self.base_dir = self.data_dir / "sub_agent_tasks" self.state_file = self.data_dir / "sub_agents.json" @@ -718,12 +724,17 @@ class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMi restored = 0 terminal_statuses = TERMINAL_STATUSES.union({"terminated"}) + # 对话级隔离：仅恢复本 manager 绑定对话的任务；未绑定（工作区级 + # 服务 terminal）不恢复，任务延迟到对应对话激活（创建对话级 terminal）时恢复。 + owner_cid = getattr(self, "owner_conversation_id", None) for task_id, task in list(self.tasks.items()): if not isinstance(task, dict): continue # 仅恢复多智能体模式任务；传统子智能体保持原有清理逻辑 if not task.get("multi_agent_mode"): continue + if not owner_cid or task.get("conversation_id") != owner_cid: + continue status = task.get("status", "running") if status in terminal_statuses: continue diff --git a/server/app_legacy.py b/server/app_legacy.py index 5d258b37..ba23f6d8 100644 --- a/server/app_legacy.py +++ b/server/app_legacy.py @@ -440,13 +440,18 @@ def idle_reaper_loop(): def start_background_jobs(): - """启动一次性的后台任务（容器空闲回收）。""" + """启动一次性的后台任务（容器空闲回收 + 对话级 terminal TTL 回收）。""" global _idle_reaper_started if _idle_reaper_started: return _idle_reaper_started = True _load_last_active_cache() socketio.start_background_task(idle_reaper_loop) + try: + from .context import start_conversation_terminal_reaper + start_conversation_terminal_reaper() + except Exception as exc: + debug_log(f"[ConvTerminalReaper] 启动失败: {exc}") TITLE_DEBUG_DIR = Path(LOGS_DIR).expanduser().resolve() / "title_debug" TITLE_DEBUG_FILE = TITLE_DEBUG_DIR / "title_generation.log" diff --git a/server/context.py b/server/context.py index 02813ddb..322aa054 100644 --- a/server/context.py +++ b/server/context.py @@ -1,8 +1,10 @@ """用户终端与工作区相关的共享辅助函数。""" from __future__ import annotations +import os +import time from functools import wraps from typing import Optional, Tuple, Dict, Any -from flask import session, jsonify, has_request_context +from flask import session, jsonify, has_request_context, request from core.web_terminal import WebTerminal from modules.gui_file_manager import GuiFileManager @@ -39,15 +41,62 @@ def make_terminal_callback(username: str): def attach_user_broadcast(terminal: WebTerminal, username: str): - """确保终端的广播函数指向当前用户的房间""" + """确保终端的广播函数指向当前用户的房间。 + + 对话级 terminal 的回调会额外包装注入 conversation_id（见 _wrap_callback_with_conversation_id）。 + """ callback = make_terminal_callback(username) + callback = _wrap_callback_with_conversation_id( + callback, getattr(terminal, "_bound_conversation_id", None) + ) terminal.message_callback = callback if terminal.terminal_manager: terminal.terminal_manager.broadcast = callback -def _make_terminal_key(username: str, workspace_id: Optional[str] = None) -> str: - return f"{username}::{workspace_id}" if workspace_id else username +def _make_terminal_key( + username: str, + workspace_id: Optional[str] = None, + conversation_id: Optional[str] = None, +) -> str: + """终端缓存键。 + + 对话级隔离：传入 conversation_id 时键为 ``username::workspace_id::conversation_id``， + 每个对话拥有独立的 WebTerminal（context/file/terminal manager、子智能体等）。 + 不传 conversation_id 时保持旧的两段键（工作区级服务实例，供对话列表等无对话上下文的 API 使用）。 + """ + base = f"{username}::{workspace_id}" if workspace_id else username + if conversation_id: + base = f"{base}::{conversation_id}" + return base + + +def _wrap_callback_with_conversation_id(callback, conversation_id: Optional[str]): + """包装广播回调，为 dict 类型的事件数据注入 conversation_id（setdefault，不覆盖已有值）。 + + 对话级 terminal 的广播（shell 输出、terminal 列表等）仍发到用户房间， + 前端按 conversation_id 过滤，避免同工作区多个对话的终端事件互相串扰。 + """ + if not callback or not conversation_id: + return callback + + def _wrapped(event_type, data): + if isinstance(data, dict): + data = dict(data) + data.setdefault("conversation_id", conversation_id) + return callback(event_type, data) + + return _wrapped + + +def _touch_terminal_activity(terminal: Optional[WebTerminal], conversation_id: Optional[str]) -> None: + """对话级 terminal：刷新最近活动时间（供 24h TTL 回收器判定）。""" + if not terminal or not conversation_id: + return + try: + terminal.last_activity_at = time.time() + except Exception: + pass def _set_terminal_workspace_label(terminal: WebTerminal, label: Optional[str]) -> None: @@ -132,7 +181,14 @@ def get_user_resources( username: Optional[str] = None, workspace_id: Optional[str] = None, update_session: bool = True, + conversation_id: Optional[str] = None, ) -> Tuple[Optional[WebTerminal], Optional['modules.user_manager.UserWorkspace']]: + """获取用户终端与工作区资源。 + + conversation_id 非空时返回对话级 terminal（每对话独立的 shell/文件/子智能体状态， + 常驻内存 + 24h 无活动回收）；为空时返回工作区级服务 terminal（对话列表等无对话 + 上下文的 API 使用）。容器句柄始终按工作区级共享。 + """ from modules.user_manager import UserWorkspace username = (username or get_current_username()) if not username: @@ -202,8 +258,10 @@ def get_user_resources( workspace.workspace_id = host_workspace.get("workspace_id") or "default" workspace_id_value = getattr(workspace, "workspace_id", None) or host_workspace.get("workspace_id") or "default" - term_key = _make_terminal_key("host", workspace_id_value) - container_handle = state.container_manager.ensure_container("host", str(project_path), container_key=term_key, preferred_mode="host") + term_key = _make_terminal_key("host", workspace_id_value, conversation_id) + # 容器句柄始终按工作区级共享：对话级 terminal 在同一容器内起独立 shell 进程 + container_key = _make_terminal_key("host", workspace_id_value) + container_handle = state.container_manager.ensure_container("host", str(project_path), container_key=container_key, preferred_mode="host") usage_tracker = None # 宿主机模式不计配额 terminal = state.user_terminals.get(term_key) target_project_path = str(project_path) @@ -273,7 +331,8 @@ def get_user_resources( message_callback=make_terminal_callback("host"), data_dir=str(data_dir), container_session=container_handle, - usage_tracker=usage_tracker + usage_tracker=usage_tracker, + conversation_id=conversation_id, ) if terminal.terminal_manager: terminal.terminal_manager.broadcast = terminal.message_callback @@ -341,6 +400,7 @@ def get_user_resources( _apply_workspace_personalization_preferences(terminal, workspace) _ensure_workspace_skills_synced(terminal, workspace) + _touch_terminal_activity(terminal, conversation_id) write_host_workspace_debug( "context.get_user_resources.host.return", terminal_id=id(terminal), @@ -374,8 +434,10 @@ def get_user_resources( except Exception: pass workspace_id_value = getattr(workspace, "workspace_id", None) or "default" - term_key = _make_terminal_key(username, workspace_id_value) - container_handle = state.container_manager.ensure_container(username, str(workspace.project_path), container_key=term_key, preferred_mode="docker") + term_key = _make_terminal_key(username, workspace_id_value, conversation_id) + # 容器句柄始终按工作区级共享（docker 模式：一个工作区/项目一个容器） + container_key = _make_terminal_key(username, workspace_id_value) + container_handle = state.container_manager.ensure_container(username, str(workspace.project_path), container_key=container_key, preferred_mode="docker") usage_tracker = None if is_api_user else get_or_create_usage_tracker(username, workspace) terminal = state.user_terminals.get(term_key) if not terminal: @@ -406,7 +468,8 @@ def get_user_resources( message_callback=make_terminal_callback(username), data_dir=str(workspace.data_dir), container_session=container_handle, - usage_tracker=usage_tracker + usage_tracker=usage_tracker, + conversation_id=conversation_id, ) if terminal.terminal_manager: terminal.terminal_manager.broadcast = terminal.message_callback @@ -480,6 +543,7 @@ def get_user_resources( _apply_workspace_personalization_preferences(terminal, workspace, update_session=update_session) _ensure_workspace_skills_synced(terminal, workspace) + _touch_terminal_activity(terminal, conversation_id) return terminal, workspace @@ -557,12 +621,25 @@ def apply_conversation_overrides(terminal: WebTerminal, workspace, conversation_ def with_terminal(func): - """注入用户专属终端和工作区""" + """注入用户专属终端和工作区。 + + 请求携带 conversation_id（query 参数或 JSON body）时返回该对话的对话级 terminal， + 否则返回工作区级服务 terminal。 + """ @wraps(func) def wrapper(*args, **kwargs): username = get_current_username() + conversation_id = None try: - terminal, workspace = get_user_resources(username) + conversation_id = (request.args.get("conversation_id") or "").strip() or None + if not conversation_id and request.is_json: + body = request.get_json(silent=True) or {} + if isinstance(body, dict): + conversation_id = (body.get("conversation_id") or "").strip() or None + except Exception: + conversation_id = None + try: + terminal, workspace = get_user_resources(username, conversation_id=conversation_id) except RuntimeError as exc: return jsonify({"error": str(exc), "code": "resource_busy"}), 503 if not terminal or not workspace: @@ -576,12 +653,12 @@ def with_terminal(func): return wrapper -def get_terminal_for_sid(sid: str): +def get_terminal_for_sid(sid: str, conversation_id: Optional[str] = None): username = state.connection_users.get(sid) if not username: return None, None, None try: - terminal, workspace = get_user_resources(username) + terminal, workspace = get_user_resources(username, conversation_id=conversation_id) except RuntimeError: return username, None, None return username, terminal, workspace @@ -715,6 +792,109 @@ def reset_system_state(terminal: Optional[WebTerminal]): debug_log(f"错误详情: {traceback.format_exc()}") +# ====== 对话级 terminal 24h TTL 回收器 ====== +# 对话级 terminal（key 为 username::workspace_id::conversation_id 三段）常驻内存， +# 仅当「超过 TTL 无活动 且 该对话无运行中工作」时回收；工作区级服务 terminal 不回收。 +CONVERSATION_TERMINAL_TTL_SECONDS = float(os.environ.get("CONVERSATION_TERMINAL_TTL_SECONDS", str(24 * 3600))) +CONVERSATION_TERMINAL_REAP_INTERVAL_SECONDS = float(os.environ.get("CONVERSATION_TERMINAL_REAP_INTERVAL_SECONDS", "600")) +_conversation_terminal_reaper_started = False + + +def _conversation_terminal_has_running_work( + username: str, + workspace_id: str, + conversation_id: str, + terminal: WebTerminal, +) -> bool: + """判定对话是否仍有运行中的工作（主任务/子智能体/后台命令/多智能体）。 + + 判定失败时保守返回 True（不回收）。 + """ + try: + from .tasks import task_manager + active_statuses = {"pending", "running", "cancel_requested"} + for rec in task_manager.list_tasks(username, workspace_id): + if rec.conversation_id == conversation_id and rec.status in active_statuses: + return True + status = task_manager.get_conversation_running_status(terminal, conversation_id) + if any(bool(v) for v in (status or {}).values()): + return True + except Exception as exc: + debug_log(f"[ConvTerminalReaper] 运行状态判定失败 {conversation_id}: {exc}") + return True + return False + + +def reap_idle_conversation_terminals(now: Optional[float] = None) -> int: + """回收超过 TTL 且无运行任务的对话级 terminal，返回回收数量（可测试）。""" + now = now or time.time() + reaped = 0 + for term_key, terminal in list(state.user_terminals.items()): + parts = term_key.split("::") + if len(parts) < 3: + continue # 工作区级服务 terminal 不回收 + username, workspace_id = parts[0], parts[1] + conversation_id = "::".join(parts[2:]) + try: + last_active = float(getattr(terminal, "last_activity_at", None)) + except (TypeError, ValueError): + last_active = None + if last_active is None: + # 无时间戳实例（旧版本创建）：补上当前时间，下轮再判定 + try: + terminal.last_activity_at = now + except Exception: + pass + continue + if now - last_active < CONVERSATION_TERMINAL_TTL_SECONDS: + continue + if _conversation_terminal_has_running_work(username, workspace_id, conversation_id, terminal): + continue + try: + cm = getattr(terminal, "context_manager", None) + if cm and getattr(cm, "current_conversation_id", None): + cm.save_current_conversation() + except Exception as exc: + debug_log(f"[ConvTerminalReaper] 保存对话失败 {conversation_id}: {exc}") + try: + tm = getattr(terminal, "terminal_manager", None) + if tm: + tm.close_all() + except Exception as exc: + debug_log(f"[ConvTerminalReaper] 关闭 shell 失败 {conversation_id}: {exc}") + try: + mcp = getattr(terminal, "mcp_client_manager", None) + if mcp: + mcp.close_all_clients() + except Exception as exc: + debug_log(f"[ConvTerminalReaper] 关闭 MCP 失败 {conversation_id}: {exc}") + state.user_terminals.pop(term_key, None) + reaped += 1 + debug_log(f"[ConvTerminalReaper] 已回收对话级 terminal: {term_key} (idle {int(now - last_active)}s)") + return reaped + + +def _conversation_terminal_reaper_loop(): + """后台循环：定期扫描回收空闲对话级 terminal。""" + while True: + try: + reap_idle_conversation_terminals() + time.sleep(CONVERSATION_TERMINAL_REAP_INTERVAL_SECONDS) + except Exception as exc: + debug_log(f"[ConvTerminalReaper] 后台循环异常: {exc}") + time.sleep(CONVERSATION_TERMINAL_REAP_INTERVAL_SECONDS) + + +def start_conversation_terminal_reaper(): + """幂等启动对话级 terminal TTL 回收后台线程。""" + global _conversation_terminal_reaper_started + if _conversation_terminal_reaper_started: + return + _conversation_terminal_reaper_started = True + from .extensions import socketio + socketio.start_background_task(_conversation_terminal_reaper_loop) + + __all__ = [ "get_user_resources", "with_terminal", @@ -727,4 +907,6 @@ __all__ = [ "get_or_create_usage_tracker", "emit_user_quota_update", "attach_user_broadcast", + "reap_idle_conversation_terminals", + "start_conversation_terminal_reaper", ] diff --git a/server/multi_agent.py b/server/multi_agent.py index 0d963b1e..9e3eac2b 100644 --- a/server/multi_agent.py +++ b/server/multi_agent.py @@ -399,7 +399,7 @@ def list_active_sub_agents_api(): conversation_id = (request.args.get("conversation_id") or "").strip() if not conversation_id: return jsonify({"success": False, "error": "缺少 conversation_id 参数"}), 400 - terminal, _ = get_user_resources(username) + terminal, _ = get_user_resources(username, conversation_id=conversation_id) if not terminal: return jsonify({"success": False, "error": "工作区未就绪"}), 503 sub_agent_manager = getattr(terminal, "sub_agent_manager", None) diff --git a/server/socket_handlers.py b/server/socket_handlers.py index 98111902..d746f743 100644 --- a/server/socket_handlers.py +++ b/server/socket_handlers.py @@ -154,7 +154,8 @@ def handle_terminal_subscribe(data): session_name = data.get('session') subscribe_all = data.get('all', False) - username, terminal, _ = get_terminal_for_sid(request.sid) + conv_id = (data.get('conversation_id') or '').strip() or None + username, terminal, _ = get_terminal_for_sid(request.sid, conversation_id=conv_id) if not username or not terminal or not terminal.terminal_manager: emit('error', {'message': 'Terminal system not initialized'}) return @@ -211,8 +212,9 @@ def handle_get_terminal_output(data): """获取终端输出历史""" session_name = data.get('session') lines = data.get('lines', 50) - - username, terminal, _ = get_terminal_for_sid(request.sid) + + conv_id = (data.get('conversation_id') or '').strip() or None + username, terminal, _ = get_terminal_for_sid(request.sid, conversation_id=conv_id) if not terminal or not terminal.terminal_manager: emit('error', {'message': 'Terminal system not initialized'}) return diff --git a/server/tasks/api.py b/server/tasks/api.py index fc2d24c4..3780a1be 100644 --- a/server/tasks/api.py +++ b/server/tasks/api.py @@ -51,6 +51,61 @@ def list_tasks_api(): ] }) +@tasks_bp.route("/api/conversations//running-status", methods=["GET"]) +@api_login_required +def get_conversation_running_status_api(conversation_id: str): + """REST 对账接口：聚合某对话的完整运行状态。 + + 前端架构：事件流（250ms 任务轮询 + socket）负责即时性，本接口周期对账负责 + 正确性，冲突以对账为准。聚合四类状态：主 task / 传统后台子智能体 / 后台命令 / + 多智能体实例与待消费消息。 + """ + username = get_current_username() + conversation_id = (conversation_id or "").strip() + if not conversation_id: + return jsonify({"success": False, "error": "缺少 conversation_id"}), 400 + + # ① 主 task：内存中该对话是否有活动任务（取最新一条） + active_statuses = {"pending", "running", "cancel_requested"} + main_rec = None + for rec in task_manager.list_tasks(username): + if rec.conversation_id != conversation_id or rec.status not in active_statuses: + continue + if main_rec is None or rec.created_at > main_rec.created_at: + main_rec = rec + + # ②③④ 需要 terminal（sub_agent_manager / background_command_manager 挂在 terminal 上） + workspace_id = ( + (request.args.get("workspace_id") or "").strip() + or (main_rec.workspace_id if main_rec else "") + or (session.get("workspace_id") or "") + ) or None + bg_status = { + "has_running_sub_agents": False, + "has_running_background_commands": False, + "has_running_multi_agent": False, + } + try: + terminal, _workspace = get_user_resources(username, workspace_id=workspace_id, conversation_id=conversation_id) + except Exception as exc: + debug_log(f"[TaskAPI] running-status 获取终端失败: {exc}") + terminal = None + if terminal: + bg_status = task_manager.get_conversation_running_status(terminal, conversation_id) + + is_main_running = main_rec is not None + return jsonify({ + "success": True, + "data": { + "conversation_id": conversation_id, + "is_main_running": is_main_running, + "main_task_id": main_rec.task_id if main_rec else None, + "main_task_type": getattr(main_rec, "task_type", "chat") if main_rec else None, + **bg_status, + "is_truly_active": is_main_running or any(bg_status.values()), + } + }) + @tasks_bp.route("/api/tasks", methods=["POST"]) @api_login_required def create_task_api(): @@ -72,7 +127,7 @@ def create_task_api(): # 避免新对话错误继承旧目标状态。 if not goal_mode: try: - _terminal, workspace = get_user_resources(username, workspace_id) + _terminal, workspace = get_user_resources(username, workspace_id, conversation_id=conversation_id) if workspace: gsm = GoalStateManager(workspace.data_dir) if gsm.is_active(): @@ -89,7 +144,7 @@ def create_task_api(): f"count={len(raw_skill_refs) if isinstance(raw_skill_refs, list) else 'N/A'} " f"refs={raw_skill_refs!r}" ) - _terminal, workspace = get_user_resources(username, workspace_id) + _terminal, workspace = get_user_resources(username, workspace_id, conversation_id=conversation_id) if not workspace: debug_log(f"[SkillsAPI] create_task workspace unavailable user={username} ws={workspace_id}") return jsonify({"success": False, "error": "工作区不可用"}), 400 @@ -209,7 +264,7 @@ def cancel_task_api(task_id: str): # 用户取消任务时，一并停止该工作区的目标模式，避免后续新对话继承旧目标。 try: if ok and rec.workspace_id: - _, workspace = get_user_resources(username, rec.workspace_id) + _, workspace = get_user_resources(username, rec.workspace_id, conversation_id=rec.conversation_id) if workspace: gsm = GoalStateManager(workspace.data_dir) if gsm.is_active(): diff --git a/server/tasks/models.py b/server/tasks/models.py index 102b8ccf..5fd50010 100644 --- a/server/tasks/models.py +++ b/server/tasks/models.py @@ -144,12 +144,22 @@ class TaskManager: raise ValueError("run_mode 只支持 fast/thinking/deep") run_mode = normalized normalized_task_type = str(task_type or "chat").strip().lower() or "chat" - # 单工作区互斥：普通 chat 任务禁止同一用户同一工作区并发； + # 单对话互斥：普通 chat 任务禁止同一对话并发（防串写对话历史）； + # 同工作区不同对话允许并行（对话级 terminal 隔离）。 # notice（通知触发）任务允许与已完成的 chat 任务共存，用于后台通知重入。 if normalized_task_type == "chat": - existing = [t for t in self.list_tasks(username, workspace_id) if t.status in {"pending", "running"} and getattr(t, "task_type", "chat") == "chat"] + def _norm_cid(cid): + cid = str(cid or "").strip() + return cid[5:] if cid.startswith("conv_") else cid + target_cid = _norm_cid(conversation_id) + existing = [ + t for t in self.list_tasks(username, workspace_id) + if t.status in {"pending", "running"} + and getattr(t, "task_type", "chat") == "chat" + and _norm_cid(getattr(t, "conversation_id", None)) == target_cid + ] if existing: - raise RuntimeError("当前工作区已有运行中的任务，请稍后再试。") + raise RuntimeError("当前对话已有运行中的任务，请稍后再试。") task_id = str(uuid.uuid4()) record = TaskRecord(task_id, username, workspace_id, message, conversation_id, model_key, thinking_mode, run_mode, max_iterations, task_type=normalized_task_type) # 记录当前 session 快照，便于后台线程内使用 @@ -232,7 +242,7 @@ class TaskManager: terminal = entry.get('terminal') if not terminal and rec.workspace_id: try: - terminal, _ = get_user_resources(username, rec.workspace_id) + terminal, _ = get_user_resources(username, rec.workspace_id, conversation_id=rec.conversation_id) except Exception: pass @@ -261,7 +271,7 @@ class TaskManager: # 2. 停止目标模式 if rec.workspace_id: try: - _, workspace = get_user_resources(username, rec.workspace_id) + _, workspace = get_user_resources(username, rec.workspace_id, conversation_id=rec.conversation_id) if workspace: gsm = GoalStateManager(workspace.data_dir) if gsm.is_active(): @@ -617,38 +627,77 @@ class TaskManager: return has_running_background @staticmethod - def _has_running_background(rec: TaskRecord, terminal: Optional[Any]) -> Dict[str, bool]: - """检测指定任务/对话是否还有运行中的后台子智能体或后台命令。""" - result = {"has_running_sub_agents": False, "has_running_background_commands": False} - if not terminal or not rec.conversation_id: + def get_conversation_running_status(terminal: Optional[Any], conversation_id: Optional[str]) -> Dict[str, bool]: + """按对话聚合后台运行状态（REST 对账接口用）。 + + 返回三类后台工作，语义互不重叠： + - has_running_sub_agents: 传统后台子智能体（排除多智能体实例任务） + - has_running_background_commands: 后台命令（含终态未通知） + - has_running_multi_agent: 多智能体实例在跑、非终态多智能体任务、 + 或有待消费的 pending master 消息 + """ + result = { + "has_running_sub_agents": False, + "has_running_background_commands": False, + "has_running_multi_agent": False, + } + if not terminal or not conversation_id: return result sub_agent_manager = getattr(terminal, 'sub_agent_manager', None) if sub_agent_manager: try: - sub_agent_manager.reconcile_task_states(conversation_id=rec.conversation_id) + sub_agent_manager.reconcile_task_states(conversation_id=conversation_id) + terminal_statuses = SUB_AGENT_TERMINAL_STATUSES.union({"terminated"}) for task_info in sub_agent_manager.tasks.values(): - if task_info.get('conversation_id') != rec.conversation_id: + if task_info.get('conversation_id') != conversation_id: continue status = task_info.get('status') - if status not in SUB_AGENT_TERMINAL_STATUSES.union({"terminated"}): + if status in terminal_statuses: + continue + if task_info.get('multi_agent_mode'): + result["has_running_multi_agent"] = True + else: result["has_running_sub_agents"] = True - break except Exception as exc: - debug_log(f"[Task] 检查后台子智能体失败: {exc}") + debug_log(f"[Task] 对账检查子智能体失败: {exc}") + try: + state = sub_agent_manager.get_multi_agent_state(conversation_id) + if state: + has_running_instance = any(a.status == "running" for a in state.list_all()) + if has_running_instance or state.has_pending_master_messages(): + result["has_running_multi_agent"] = True + except Exception as exc: + debug_log(f"[Task] 对账检查多智能体状态失败: {exc}") bg_manager = getattr(terminal, 'background_command_manager', None) if bg_manager: try: - bg_manager.reconcile_stale_records(conversation_id=rec.conversation_id) - waiting_items = bg_manager.list_waiting_items(rec.conversation_id) + bg_manager.reconcile_stale_records(conversation_id=conversation_id) + waiting_items = bg_manager.list_waiting_items(conversation_id) if waiting_items: result["has_running_background_commands"] = True except Exception as exc: - debug_log(f"[Task] 检查后台命令失败: {exc}") + debug_log(f"[Task] 对账检查后台命令失败: {exc}") return result + @staticmethod + def _has_running_background(rec: TaskRecord, terminal: Optional[Any]) -> Dict[str, bool]: + """检测指定任务/对话是否还有运行中的后台子智能体或后台命令。 + + 兼容旧语义：多智能体实例任务也计入 sub_agents。 + """ + result = {"has_running_sub_agents": False, "has_running_background_commands": False} + if not rec: + return result + status = TaskManager.get_conversation_running_status(terminal, rec.conversation_id) + result["has_running_sub_agents"] = ( + status["has_running_sub_agents"] or status["has_running_multi_agent"] + ) + result["has_running_background_commands"] = status["has_running_background_commands"] + return result + def _append_event(self, rec: TaskRecord, event_type: str, data: Dict[str, Any]): if isinstance(data, dict): data = dict(data) @@ -697,7 +746,7 @@ class TaskManager: ) except Exception: pass - terminal, workspace = get_user_resources(username, workspace_id=workspace_id) + terminal, workspace = get_user_resources(username, workspace_id=workspace_id, conversation_id=rec.conversation_id) if not terminal or not workspace: raise RuntimeError("系统未初始化") stop_hint = bool(stop_flags.get(rec.task_id, {}).get("stop")) diff --git a/static/src/App.vue b/static/src/App.vue index 9f949fa4..fece20e4 100644 --- a/static/src/App.vue +++ b/static/src/App.vue @@ -394,6 +394,7 @@ diff --git a/static/src/app/computed.ts b/static/src/app/computed.ts index 4da9d484..bef9e19b 100644 --- a/static/src/app/computed.ts +++ b/static/src/app/computed.ts @@ -218,6 +218,8 @@ export const computed = { return !!this.compressionInProgress || !!this.compressing || !!this.currentWorkspaceHasRunningTask; }, currentWorkspaceHasRunningTask() { + // 对话级隔离后：仅当「当前对话」有运行中任务时才拦截发送； + // 同工作区其他对话的任务不再阻塞本对话（多对话并行）。 if (!(this.versioningHostMode || this.dockerProjectMode) || !this.currentHostWorkspaceId) { return false; } @@ -226,7 +228,7 @@ export const computed = { const status = String(task?.status || ''); return ( task?.workspace_id === this.currentHostWorkspaceId && - task?.conversation_id !== this.currentConversationId && + task?.conversation_id === this.currentConversationId && ['pending', 'running', 'cancel_requested'].includes(status) ); }); diff --git a/static/src/app/methods/message/send.ts b/static/src/app/methods/message/send.ts index 61dcffb3..914e3418 100644 --- a/static/src/app/methods/message/send.ts +++ b/static/src/app/methods/message/send.ts @@ -24,8 +24,8 @@ export const sendMethods = { (Array.isArray(this.selectedVideos) && this.selectedVideos.length > 0); if (this.currentWorkspaceHasRunningTask) { this.uiPushToast({ - title: '当前工作区正在运行', - message: '请先在“运行中的任务”中查看或等待任务完成后再发送新消息。', + title: '当前对话正在运行', + message: '请等待当前对话任务完成后再发送新消息；同工作区的其他对话可正常并行。', type: 'warning' }); return; @@ -309,6 +309,10 @@ export const sendMethods = { // 标记任务进行中，直到任务完成或用户手动停止 this.taskInProgress = true; + // 启动运行状态对账循环（幂等）：发送消息后确保对账兜底在运行 + if (typeof this.startRunningStateReconcile === 'function') { + this.startRunningStateReconcile(); + } const localMessageSource = usePresetText ? (options?.source === 'runtime_queue_manual_guide' ? 'guidance' : 'presend') : 'user'; this.chatAddUserMessage(message, images, videos, [], localMessageSource); // 关键体验修复：用户发送后立刻显示 assistant 头部 + 工作中计时 + 等待提示， diff --git a/static/src/app/methods/taskPolling/compression.ts b/static/src/app/methods/taskPolling/compression.ts index 0fd089df..1159b8e6 100644 --- a/static/src/app/methods/taskPolling/compression.ts +++ b/static/src/app/methods/taskPolling/compression.ts @@ -96,6 +96,10 @@ export const compressionMethods = { async restoreTaskState() { // 清理已处理的事件索引 this.clearProcessedEvents(); + // 启动运行状态对账循环（幂等）：事件负责即时性，对账负责正确性 + if (typeof this.startRunningStateReconcile === 'function') { + this.startRunningStateReconcile(); + } restoreDebugLog('restore:start', { currentConversationId: this.currentConversationId, streamingMessage: this.streamingMessage, diff --git a/static/src/app/methods/taskPolling/probe.ts b/static/src/app/methods/taskPolling/probe.ts index ca26b6b2..342a4c97 100644 --- a/static/src/app/methods/taskPolling/probe.ts +++ b/static/src/app/methods/taskPolling/probe.ts @@ -1,22 +1,7 @@ // @ts-nocheck -import { debugLog, goalModeDebugLog } from '../common'; +import { debugLog } from '../common'; import { useTaskStore } from '../../../stores/task'; -import { getMessageVisibility, messageStartsWork } from '../../../utils/messageVisibility'; -import { - debugNotifyLog, - keyNotifyLog, - jsonDebug, - userMDebug, - isRestoreDebugEnabled, - restoreDebugLog, - isSystemAutoUserMessagePayload, - isRuntimeModeNoticePayload, - resolveUserMessageSource, - resolveUserMessageMetadata, - isEmptyAssistantPlaceholderMessage, - getOptimisticUserEchoTarget, - findRecentMatchingUserMessage, -} from './shared'; +import { debugNotifyLog } from './shared'; export const probeMethods = { scheduleTodoListRefresh(delayMs = 120) { @@ -32,414 +17,158 @@ export const probeMethods = { } }, delay); }, - stopWaitingTaskProbe() { - if (this.waitingTaskProbeTimer) { - debugNotifyLog('[DEBUG_NOTIFY][ui] stopWaitingTaskProbe'); - clearInterval(this.waitingTaskProbeTimer); - this.waitingTaskProbeTimer = null; + // ---------- 运行状态对账（事件为主，对账纠偏） ---------- + // 架构：事件流（250ms 任务轮询 + socket）负责即时性；本对账循环每 2.5s 调用 + // GET /api/conversations//running-status 获取服务端权威聚合状态，负责正确性。 + // 冲突以对账为准：恢复方向立即执行，清理方向需连续 2 次服务端空闲确认 + // （防止 notice 任务创建间隙、多智能体 idle dispatch 间隙造成误清抖动）。 + startRunningStateReconcile() { + if (this.runningStateReconcileTimer) { + return; } - this._waitingProbeStableEmptyCount = 0; + this._runningStateIdleStreak = 0; + this.runningStateReconcileTimer = setInterval(() => { + void this.reconcileRunningStateOnce(); + }, 2500); + void this.reconcileRunningStateOnce(); + }, + stopRunningStateReconcile() { + if (this.runningStateReconcileTimer) { + clearInterval(this.runningStateReconcileTimer); + this.runningStateReconcileTimer = null; + } + this._runningStateIdleStreak = 0; + }, + async reconcileRunningStateOnce() { + const conversationId = this.currentConversationId; + if (!conversationId) { + return; + } + let status: any = null; + try { + const wsParam = + (this.versioningHostMode || this.dockerProjectMode) && this.currentHostWorkspaceId + ? `?workspace_id=${encodeURIComponent(this.currentHostWorkspaceId)}` + : ''; + const resp = await fetch( + `/api/conversations/${encodeURIComponent(conversationId)}/running-status${wsParam}` + ); + if (!resp.ok) { + return; // 网络/鉴权异常不做任何状态变更，避免误清 + } + const result = await resp.json().catch(() => ({})); + if (!result?.success || !result?.data) { + return; + } + status = result.data; + } catch (error) { + return; // 请求失败不做任何状态变更 + } + if (this.currentConversationId !== conversationId) { + return; // 请求期间切换了对话，丢弃过期结果 + } + + const taskStore = useTaskStore(); + + if (status.is_truly_active) { + this._runningStateIdleStreak = 0; + // 恢复方向（立即执行，宁多勿漏）：服务端活跃但本地缺轮询/缺标志位 + if (status.is_main_running && status.main_task_id && !taskStore.hasActiveTask) { + debugNotifyLog('[DEBUG_NOTIFY][ui] reconcile:resume-main-task', { + taskId: status.main_task_id, + conversationId + }); + if (typeof this.clearProcessedEvents === 'function') { + this.clearProcessedEvents(); + } + taskStore.resumeTask(status.main_task_id, { + status: 'running', + resetOffset: true, + eventHandler: (event: any) => this.handleTaskEvent(event) + }); + this.taskInProgress = true; + } + if (status.has_running_sub_agents) { + this.waitingForSubAgent = true; + this.taskInProgress = true; + } + if (status.has_running_background_commands) { + this.waitingForBackgroundCommand = true; + this.taskInProgress = true; + } + if (status.has_running_multi_agent) { + // 多智能体活跃不阻塞输入区（不设置 waitingForSubAgent） + this.taskInProgress = true; + } + return; + } + + // 清理方向（保守，需连续确认）：服务端确认空闲 + const localUiActive = + this.streamingMessage || + this.taskInProgress || + this.waitingForSubAgent || + this.waitingForBackgroundCommand; + if (!localUiActive && !taskStore.hasActiveTask) { + this._runningStateIdleStreak = 0; + return; // 双端一致空闲 + } + if (taskStore.isPolling) { + // 事件轮询仍活跃：终态事件会正常收尾，对账不干预（避免截断尾部事件） + return; + } + this._runningStateIdleStreak = (this._runningStateIdleStreak || 0) + 1; + if (this._runningStateIdleStreak < 2) { + return; + } + this._runningStateIdleStreak = 0; + debugLog('[TaskPolling] 对账清理：服务端确认对话已空闲，清除本地卡住的运行状态', { + conversationId, + streamingMessage: this.streamingMessage, + taskInProgress: this.taskInProgress, + waitingForSubAgent: this.waitingForSubAgent, + waitingForBackgroundCommand: this.waitingForBackgroundCommand, + taskStatus: taskStore.taskStatus + }); + this.streamingMessage = false; + this.taskInProgress = false; + this.stopRequested = false; + this.waitingForSubAgent = false; + this.waitingForBackgroundCommand = false; + if (typeof this.clearPendingTools === 'function') { + this.clearPendingTools('reconcile_auto_clear'); + } + this.clearTaskState(); + this.$forceUpdate(); + }, + // 兼容旧调用点：原"等待子智能体探测"与"多智能体任务探测"已收敛为单一对账循环， + // 对账循环按会话常驻，不再随单个探测启动/停止。 + stopWaitingTaskProbe() { + // no-op：保留方法签名避免改动所有调用点 }, startWaitingTaskProbe() { - if (this.waitingTaskProbeTimer) { - debugNotifyLog('[DEBUG_NOTIFY][ui] startWaitingTaskProbe:already-started'); - return; - } - debugNotifyLog('[DEBUG_NOTIFY][ui] startWaitingTaskProbe'); - keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] startWaitingTaskProbe', { - conversationId: this.currentConversationId - }); - this.waitingTaskProbeTimer = setInterval(async () => { - if (!this.waitingForSubAgent) { - this.stopWaitingTaskProbe(); - return; - } - const resumed = await this.tryResumeWaitingNoticeTask(); - if (resumed) { - debugNotifyLog('[DEBUG_NOTIFY][ui] waitingTaskProbe:resumed-and-stop'); - keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] waitingTaskProbe:resumed-and-stop'); - this.stopWaitingTaskProbe(); - return; - } - - const state = await this.inspectWaitingCompletionState(); - if (state.hasPending) { - this._waitingProbeStableEmptyCount = 0; - this.waitingForBackgroundCommand = !!state.hasBackgroundPending; - return; - } - - this._waitingProbeStableEmptyCount = (this._waitingProbeStableEmptyCount || 0) + 1; - if (this._waitingProbeStableEmptyCount >= 2) { - debugLog('[TaskPolling] 等待态兜底清理：未发现运行中的后台子智能体/后台指令'); - this.streamingMessage = false; - this.taskInProgress = false; - this.stopRequested = false; - this.waitingForSubAgent = false; - this.waitingForBackgroundCommand = false; - if (typeof this.clearPendingTools === 'function') { - this.clearPendingTools('waiting_probe_auto_clear'); - } - this.clearTaskState(); - this.stopWaitingTaskProbe(); - this.$forceUpdate(); - } - }, 1000); + this.startRunningStateReconcile(); }, - // ---------- 多智能体任务探测 ---------- + // ---------- 多智能体任务探测（已收敛为对账循环的兼容入口） ---------- stopMultiAgentTaskProbe() { - if (this.multiAgentTaskProbeTimer) { - debugNotifyLog('[DEBUG_NOTIFY][ui] stopMultiAgentTaskProbe'); - goalModeDebugLog('probe.stopMultiAgentTaskProbe', { - conversationId: this.currentConversationId, - taskInProgress: this.taskInProgress - }); - clearInterval(this.multiAgentTaskProbeTimer); - this.multiAgentTaskProbeTimer = null; - } + // no-op：对账循环常驻，保留方法签名避免改动所有调用点 }, startMultiAgentTaskProbe() { - if (this.multiAgentTaskProbeTimer) { - debugNotifyLog('[DEBUG_NOTIFY][ui] startMultiAgentTaskProbe:already-started'); - return; - } - debugNotifyLog('[DEBUG_NOTIFY][ui] startMultiAgentTaskProbe'); - keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] startMultiAgentTaskProbe', { - conversationId: this.currentConversationId - }); - goalModeDebugLog('probe.startMultiAgentTaskProbe', { - conversationId: this.currentConversationId, - taskInProgress: this.taskInProgress - }); - this.multiAgentTaskProbeTimer = setInterval(async () => { - if (!this.taskInProgress || !this.currentConversationId) { - debugNotifyLog('[DEBUG_NOTIFY][ui] multiAgentTaskProbe:stop-no-task-in-progress'); - goalModeDebugLog('probe.multiAgentTaskProbe_stop', { - reason: 'no_task_in_progress', - taskInProgress: this.taskInProgress, - currentConversationId: this.currentConversationId - }); - this.stopMultiAgentTaskProbe(); - return; - } - goalModeDebugLog('probe.multiAgentTaskProbe_tick', { - currentConversationId: this.currentConversationId - }); - const resumed = await this.tryResumeRunningTask(); - if (resumed) { - debugNotifyLog('[DEBUG_NOTIFY][ui] multiAgentTaskProbe:resumed-and-stop'); - keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] multiAgentTaskProbe:resumed-and-stop'); - this.stopMultiAgentTaskProbe(); - } - }, 1000); - }, - async tryResumeRunningTask() { - if (!this.currentConversationId) { - return false; - } - try { - const { useTaskStore } = await import('../../../stores/task'); - const taskStore = useTaskStore(); - if (taskStore.hasActiveTask) { - debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeRunningTask:already-active', { - taskId: taskStore.currentTaskId, - status: taskStore.taskStatus - }); - return true; - } - - const resp = await fetch('/api/tasks'); - if (!resp.ok) { - debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeRunningTask:tasks-api-not-ok', { - status: resp.status - }); - return false; - } - const result = await resp.json().catch(() => ({})); - const tasks = Array.isArray(result?.data) ? result.data : []; - debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeRunningTask:tasks', { - currentConversationId: this.currentConversationId, - count: tasks.length, - running: tasks - .filter((t: any) => t?.status === 'running') - .map((t: any) => ({ - task_id: t?.task_id, - conversation_id: t?.conversation_id, - status: t?.status - })) - }); - const runningTask = tasks.find( - (task: any) => - (task?.status === 'running' || task?.status === 'pending') && - task?.conversation_id === this.currentConversationId && - (!(this.versioningHostMode || this.dockerProjectMode) || - !this.currentHostWorkspaceId || - task?.workspace_id === this.currentHostWorkspaceId) - ); - if (!runningTask?.task_id) { - debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeRunningTask:no-matched-task'); - goalModeDebugLog('probe.tryResumeRunningTask_no_match', { - currentConversationId: this.currentConversationId, - tasksCount: tasks.length - }); - return false; - } - - debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeRunningTask:resume', { - task_id: runningTask.task_id - }); - goalModeDebugLog('probe.tryResumeRunningTask_resume', { - taskId: runningTask.task_id, - conversationId: this.currentConversationId - }); - keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] tryResumeRunningTask:resume', { - task_id: runningTask.task_id, - conversationId: this.currentConversationId - }); - if (typeof this.clearProcessedEvents === 'function') { - this.clearProcessedEvents(); - } - taskStore.resumeTask(runningTask.task_id, { - status: 'running', - resetOffset: true, - eventHandler: (event: any) => this.handleTaskEvent(event) - }); - return true; - } catch (error) { - console.warn('[TaskPolling] 接管运行中任务失败:', error); - return false; - } - }, - - async inspectWaitingCompletionState() { - const terminalStatuses = new Set(['completed', 'failed', 'timeout', 'terminated', 'cancelled']); - let hasSubAgentPending = false; - let hasBackgroundPending = false; - let resolved = true; - - try { - const [subResp, bgResp] = await Promise.all([ - fetch('/api/sub_agents'), - fetch('/api/background_commands?limit=200') - ]); - - if (subResp.ok) { - const subResult = await subResp.json().catch(() => ({})); - const tasks = Array.isArray(subResult?.data) ? subResult.data : []; - const relevantTasks = tasks.filter((task: any) => { - if (!task) return false; - if (task.conversation_id && task.conversation_id !== this.currentConversationId) - return false; - return true; - }); - hasSubAgentPending = relevantTasks.some((task: any) => { - if (!task?.run_in_background) return false; - const status = String(task?.status || '').toLowerCase(); - return !terminalStatuses.has(status) || !!task?.notice_pending; - }); - } else { - resolved = false; - } - - if (bgResp.ok) { - const bgResult = await bgResp.json().catch(() => ({})); - const commands = Array.isArray(bgResult?.data) ? bgResult.data : []; - hasBackgroundPending = commands.some((command: any) => { - if (!command) return false; - const status = String(command?.status || '').toLowerCase(); - return !terminalStatuses.has(status) || !!command?.notice_pending; - }); - } else { - resolved = false; - } - } catch (error) { - debugNotifyLog('[DEBUG_NOTIFY][ui] inspectWaitingCompletionState:error', { - error: error?.message || String(error) - }); - return { - resolved: false, - hasPending: true, - hasBackgroundPending: this.waitingForBackgroundCommand - }; - } - - return { - resolved, - hasPending: !resolved || hasSubAgentPending || hasBackgroundPending, - hasBackgroundPending - }; - }, - async tryResumeWaitingNoticeTask() { - if (!this.waitingForSubAgent || !this.currentConversationId) { - debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:skip', { - waitingForSubAgent: this.waitingForSubAgent, - currentConversationId: this.currentConversationId - }); - return false; - } - try { - const { useTaskStore } = await import('../../../stores/task'); - const taskStore = useTaskStore(); - if (taskStore.hasActiveTask) { - debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:already-active', { - taskId: taskStore.currentTaskId, - status: taskStore.taskStatus - }); - return true; - } - - const resp = await fetch('/api/tasks'); - if (!resp.ok) { - debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:tasks-api-not-ok', { - status: resp.status - }); - return false; - } - const result = await resp.json().catch(() => ({})); - const tasks = Array.isArray(result?.data) ? result.data : []; - debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:tasks', { - currentConversationId: this.currentConversationId, - count: tasks.length, - running: tasks - .filter((t: any) => t?.status === 'running') - .map((t: any) => ({ - task_id: t?.task_id, - conversation_id: t?.conversation_id, - status: t?.status - })) - }); - const runningTask = tasks.find( - (task: any) => - task?.status === 'running' && - task?.conversation_id === this.currentConversationId && - (!(this.versioningHostMode || this.dockerProjectMode) || - !this.currentHostWorkspaceId || - task?.workspace_id === this.currentHostWorkspaceId) - ); - if (!runningTask?.task_id) { - debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:no-matched-task'); - return false; - } - - debugNotifyLog('[DEBUG_NOTIFY][ui] tryResumeWaitingNoticeTask:resume', { - task_id: runningTask.task_id - }); - keyNotifyLog('[DEBUG_NOTIFY_KEY][ui] tryResumeWaitingNoticeTask:resume', { - task_id: runningTask.task_id, - conversationId: this.currentConversationId - }); - // 关键修复：切到新的通知任务前，清空旧任务事件去重索引。 - // 否则新任务 event idx 与旧任务重叠时会被误判为“重复事件”而丢失 user_message。 - if (typeof this.clearProcessedEvents === 'function') { - this.clearProcessedEvents(); - } - taskStore.resumeTask(runningTask.task_id, { - status: 'running', - resetOffset: true, - eventHandler: (event: any) => this.handleTaskEvent(event) - }); - return true; - } catch (error) { - console.warn('[TaskPolling] 接管等待任务失败:', error); - return false; - } + this.startRunningStateReconcile(); }, async restoreSubAgentWaitingState(retry = 0) { - try { - if (!this.currentConversationId) { - if (retry < 5) { - setTimeout(() => { - this.restoreSubAgentWaitingState(retry + 1); - }, 300); - } - return; + // 初始化恢复与周期对账共用同一接口：启动对账循环即会立即对账一次， + // 服务端活跃则恢复等待态/事件轮询，服务端空闲则不设置任何运行标志。 + if (!this.currentConversationId) { + if (retry < 5) { + setTimeout(() => { + this.restoreSubAgentWaitingState(retry + 1); + }, 300); } - - const response = await fetch('/api/sub_agents'); - if (!response.ok) { - debugLog('[TaskPolling] 获取子智能体状态失败'); - return; - } - const result = await response.json(); - if (!result.success) { - debugLog('[TaskPolling] 子智能体状态响应无效'); - return; - } - - const tasks = Array.isArray(result.data) ? result.data : []; - debugLog('[TaskPolling] 子智能体状态响应', { - total: tasks.length, - currentConversationId: this.currentConversationId, - tasks: tasks.map((task: any) => ({ - task_id: task?.task_id, - status: task?.status, - run_in_background: task?.run_in_background, - notice_pending: task?.notice_pending, - conversation_id: task?.conversation_id - })) - }); - const terminalStatuses = new Set(['completed', 'failed', 'timeout', 'terminated']); - const relevant = tasks.filter((task: any) => { - if (task && task.conversation_id && task.conversation_id !== this.currentConversationId) { - return false; - } - return true; - }); - - const running = relevant.filter( - (task: any) => task?.run_in_background && !terminalStatuses.has(task?.status) - ); - const pendingNotice = relevant.filter((task: any) => task?.notice_pending); - let backgroundPending = false; - try { - const bgResp = await fetch('/api/background_commands?limit=200'); - if (bgResp.ok) { - const bgResult = await bgResp.json().catch(() => ({})); - const bgCommands = Array.isArray(bgResult?.data) ? bgResult.data : []; - backgroundPending = bgCommands.some((command: any) => { - if (!command) return false; - const status = String(command?.status || '').toLowerCase(); - return !terminalStatuses.has(status) || !!command?.notice_pending; - }); - } - } catch (bgErr) { - console.warn('[TaskPolling] 获取后台指令状态失败:', bgErr); - } - - if (running.length || pendingNotice.length || backgroundPending) { - debugLog('[TaskPolling] 恢复子智能体等待状态', { - running: running.length, - pendingNotice: pendingNotice.length, - backgroundPending, - runningTasks: running.map((task: any) => ({ - task_id: task?.task_id, - status: task?.status - })), - pendingTasks: pendingNotice.map((task: any) => ({ - task_id: task?.task_id, - status: task?.status - })) - }); - this.waitingForSubAgent = true; - this.waitingForBackgroundCommand = backgroundPending; - this.taskInProgress = true; - this.streamingMessage = false; - this.stopRequested = false; - this.startWaitingTaskProbe(); - this.$forceUpdate(); - } else { - this.waitingForSubAgent = false; - this.waitingForBackgroundCommand = false; - this.stopWaitingTaskProbe(); - if (!this.streamingMessage) { - this.taskInProgress = false; - } - this.$forceUpdate(); - } - } catch (error) { - console.error('[TaskPolling] 恢复子智能体等待状态失败:', error); + return; } + this.startRunningStateReconcile(); }, clearProcessedEvents() { if (this._processedEventIndices) { diff --git a/static/src/app/methods/ui/terminal.ts b/static/src/app/methods/ui/terminal.ts index db06f82a..fefd6b15 100644 --- a/static/src/app/methods/ui/terminal.ts +++ b/static/src/app/methods/ui/terminal.ts @@ -36,7 +36,7 @@ export const terminalMethods = { subscribeTerminalEvents() { const socket = this.socket; if (!socket) return; - socket.emit('terminal_subscribe', { all: true }); + socket.emit('terminal_subscribe', { all: true, conversation_id: this.currentConversationId || undefined }); }, setTerminalSessions(sessions: Record) { this.terminalSessions = sessions; @@ -47,12 +47,13 @@ export const terminalMethods = { switchTerminalSession(name: string) { this.terminalActiveSession = name; if (this.socket) { - this.socket.emit('get_terminal_output', { session: name, lines: 0 }); + this.socket.emit('get_terminal_output', { session: name, lines: 0, conversation_id: this.currentConversationId || undefined }); } }, async fetchTerminalCount() { try { - const res = await fetch('/api/terminals'); + const cid = encodeURIComponent(this.currentConversationId || ''); + const res = await fetch(`/api/terminals?conversation_id=${cid}`); if (!res.ok) return; const data = await res.json(); if (data?.sessions) { diff --git a/static/src/app/state.ts b/static/src/app/state.ts index 0fbdb125..ef772aca 100644 --- a/static/src/app/state.ts +++ b/static/src/app/state.ts @@ -25,8 +25,10 @@ export function dataState() { toolStacks: new Map(), // 当前任务是否仍在进行中（用于保持输入区的"停止"状态） taskInProgress: false, - // 等待后台通知期间，用于发现并接管新建任务的轮询定时器 - waitingTaskProbeTimer: null, + // 对话运行状态对账定时器（事件为主、2.5s 对账纠偏，冲突以对账为准） + runningStateReconcileTimer: null, + // 对账清理方向的连续空闲确认计数（防 notice/idle dispatch 间隙误清） + _runningStateIdleStreak: 0, // 宿主机多工作区任务列表后台刷新定时器（用于当前未查看运行对话时同步完成态） runningWorkspaceTasksRefreshTimer: null, // 运行期消息堆积（提前发送 / 引导对话） diff --git a/static/src/components/panels/TerminalPanel.vue b/static/src/components/panels/TerminalPanel.vue index 5dd86ff7..445f4f0d 100644 --- a/static/src/components/panels/TerminalPanel.vue +++ b/static/src/components/panels/TerminalPanel.vue @@ -50,6 +50,7 @@ defineOptions({ name: 'TerminalPanel' }); const props = defineProps<{ width: number; workspaceId?: string; + conversationId?: string; }>(); const emit = defineEmits<{ @@ -74,6 +75,15 @@ const _historyLastEventTime = ref>({}); const sessionKeys = computed(() => Object.keys(sessions.value)); +// 对话级隔离：广播类终端事件（started/list_update/output/input/closed/reset/switched） +// 由后端注入 conversation_id，仅处理当前对话的事件； +// 响应类事件（terminal_subscribed / *_history）是请求的直接回应，无此字段，不过滤。 +function acceptTerminalBroadcast(data: any): boolean { + const cid = data?.conversation_id; + if (!cid || !props.conversationId) return false; + return cid === props.conversationId; +} + // ---- 主题适配 ---- function getCurrentTheme(): 'light' | 'dark' { const theme = document.documentElement.getAttribute('data-theme'); @@ -218,7 +228,7 @@ function switchToSession(name: string) { _historyLastEventTime.value = updatedTimes; if (socket?.connected) { console.log('[TerminalPanel] requesting history for:', name); - socket.emit('get_terminal_output', { session: name, lines: 0 }); + socket.emit('get_terminal_output', { session: name, lines: 0, conversation_id: props.conversationId }); } } // renderSessionLog 由 watch(activeSession) 统一触发，此处不重复调用 @@ -257,6 +267,23 @@ watch(() => props.workspaceId, (newId, oldId) => { } }); +// 对话切换时重置并重新订阅（对话级 terminal：各对话 shell 互相独立） +watch(() => props.conversationId, (newId, oldId) => { + if (newId !== oldId) { + console.log('[TerminalPanel] conversation changed:', oldId, '->', newId); + sessions.value = {}; + activeSession.value = ''; + sessionLogs.value = {}; + sessionHydrated.value = {}; + _historyReady.value = {}; + _historyLastEventTime.value = {}; + if (term) term.clear(); + if (socket?.connected) { + socket.emit('terminal_subscribe', { all: true, conversation_id: newId }); + } + } +}); + // ---- socket 初始化 ---- async function initSocket() { console.log('[TerminalPanel] initSocket start'); @@ -293,7 +320,7 @@ async function initSocket() { socket.on('connect', () => { console.log('[TerminalPanel] socket connected'); - socket!.emit('terminal_subscribe', { all: true }); + socket!.emit('terminal_subscribe', { all: true, conversation_id: props.conversationId }); }); socket.on('disconnect', () => { @@ -315,6 +342,7 @@ async function initSocket() { }); socket.on('terminal_started', (data: any) => { + if (!acceptTerminalBroadcast(data)) return; console.log('[TerminalPanel] terminal_started:', data); sessions.value = { ...sessions.value, @@ -324,6 +352,7 @@ async function initSocket() { }); socket.on('terminal_list_update', (data: any) => { + if (!acceptTerminalBroadcast(data)) return; console.log('[TerminalPanel] terminal_list_update:', data); if (data?.terminals) { const map: Record = {}; @@ -336,6 +365,7 @@ async function initSocket() { }); socket.on('terminal_output', (data: any) => { + if (!acceptTerminalBroadcast(data)) return; const s = data.session; if (!s) return; const delta = (data.data || '') as string; @@ -363,6 +393,7 @@ async function initSocket() { }); socket.on('terminal_input', (data: any) => { + if (!acceptTerminalBroadcast(data)) return; console.log('[TerminalPanel] terminal_input:', data.session); // 原样显示后端输出，此处不重复写入 // 仅用于新终端自动切换 @@ -370,6 +401,7 @@ async function initSocket() { }); socket.on('terminal_closed', (data: any) => { + if (!acceptTerminalBroadcast(data)) return; console.log('[TerminalPanel] terminal_closed:', data); const updated = { ...sessions.value }; delete updated[data.session]; @@ -381,6 +413,7 @@ async function initSocket() { }); socket.on('terminal_reset', (data: any) => { + if (!acceptTerminalBroadcast(data)) return; console.log('[TerminalPanel] terminal_reset:', data); const target = data.session || activeSession.value; if (target) { @@ -394,6 +427,7 @@ async function initSocket() { }); socket.on('terminal_switched', (data: any) => { + if (!acceptTerminalBroadcast(data)) return; console.log('[TerminalPanel] terminal_switched:', data); if (data.current) switchToSession(data.current); }); diff --git a/static/src/composables/useLegacySocket.ts b/static/src/composables/useLegacySocket.ts index 0e1951b8..36406e2f 100644 --- a/static/src/composables/useLegacySocket.ts +++ b/static/src/composables/useLegacySocket.ts @@ -625,8 +625,8 @@ export async function initializeLegacySocket(ctx: any) { ctx.isConnected = true; socketLog('WebSocket已连接'); - // 自动订阅终端事件，确保终端面板数据始终就绪 - ctx.socket.emit('terminal_subscribe', { all: true }); + // 自动订阅终端事件，确保终端面板数据始终就绪（对话级：带当前 conversation_id） + ctx.socket.emit('terminal_subscribe', { all: true, conversation_id: ctx.currentConversationId || undefined }); // 首次连接且历史已在加载/已就绪时，避免重复清空与重复动画 if (!isReconnect && (historyLoadingSame || historyReady)) {