From 99f4d3d1dd32b81f320b52319a91f9dd3a5ae139 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Wed, 12 Aug 2026 18:01:32 +0800 Subject: [PATCH] =?UTF-8?q?fix(server):=20=E5=AF=B9=E8=AF=9D=E7=BA=A7?= =?UTF-8?q?=E4=B8=BB=E4=BB=BB=E5=8A=A1=E9=97=A8=E9=97=B8=EF=BC=8C=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=E5=B9=B6=E5=8F=91=E4=B8=BB=E4=BB=BB=E5=8A=A1=E4=BA=A4?= =?UTF-8?q?=E5=8F=89=E5=86=99=E5=85=A5=E5=AF=B9=E8=AF=9D=E5=8E=86=E5=8F=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 事故：socketio 用户任务（不在 task_manager 注册）与完成通知轮询器 派发的通知任务并发写入共享 conversation_history，产生 assistant 乱序段， 导致 Kimi API 400（tool_call_id is not found）与通知永久丢失。 - 新增 server/main_task_gate.py：对话级主任务门闸（单写者不变量， 一个 WebTerminal 同一时刻只允许一个主聊天任务） - process_message_task 作为唯一入口获取/认领门闸，拿不到即拒绝； 通知链 token 经 session_data 移交，失败释放并回滚通知标记； _run_chat_task finally 兜底释放防泄漏 - execute_tool_calls 改守护包装（try/finally 复位 _tool_loop_active）， 杜绝并发交错导致标志卡死 - build_messages 增加孤儿 tool 消息剥离防御层（止血，不替代门闸） - AGENTS.md 新增 §12 记录门闸架构与改代码硬约束 --- AGENTS.md | 30 +++++++- core/main_terminal_parts/context/messages.py | 21 ++++++ server/chat_flow.py | 25 ++++++- server/chat_flow_task_main.py | 69 +++++++++++++++++- server/chat_flow_tool_loop.py | 19 ++++- server/main_task_gate.py | 73 ++++++++++++++++++++ server/tasks/models.py | 12 ++++ 7 files changed, 241 insertions(+), 8 deletions(-) create mode 100644 server/main_task_gate.py diff --git a/AGENTS.md b/AGENTS.md index 63a43154..cd0e57bb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -491,4 +491,32 @@ AI 执行以下流程时，每一步都要向用户说明在做什么： --- -注：本节按「现有架构 + 多智能体分支」方案描述，与项目记忆 `multi_agent_mode_design` 同步；如两者冲突以代码为准并同步修订本节。 \ No newline at end of file +注：本节按「现有架构 + 多智能体分支」方案描述，与项目记忆 `multi_agent_mode_design` 同步；如两者冲突以代码为准并同步修订本节。 + +--- + +## 12) 对话级主任务门闸与单写者不变量（2026-08-12） + +> 事故背景：一个对话并发运行了两个主聊天任务（socketio 用户任务 + 完成通知轮询器派发的通知任务），交叉写入共享 `conversation_history`，产生 `assistant→assistant→tool→tool` 乱序段，最终 API 400 `tool_call_id is not found`、通知永久丢失。完整根因与修复细节见项目记忆 `main_task_gate_and_parallel_universe`。 + +### 12.1 单写者不变量（核心约束） + +**一个 WebTerminal（≈ 一个打开的对话）同一时刻只允许存在一个主聊天任务。** 主任务包括：用户消息任务、后台完成通知派发任务、多智能体 idle 派发任务等一切会向 `conversation_history` 追加消息并请求模型的执行体。宁可拒绝/推迟新任务，也绝不并发写入。 + +### 12.2 门闸本体 + +- 实现：`server/main_task_gate.py`（进程内字典，key=terminal_id，value=token）。 +- 获取/认领：`acquire_or_claim_main_task_gate(terminal_id, owner_desc)`——门闸空闲或持有者是同一任务时返回 token，否则返回 None。 +- 释放：`release_main_task_gate(terminal_id, token)`，token 不匹配则拒绝（防止错误释放他人门闸）。 + +### 12.3 唯一入口与 token 移交 + +- **唯一入口**：`process_message_task`（`server/chat_flow.py`）是所有主任务的门闸入口——进入即获取/认领门闸，拿不到则向用户发 error 并返回，绝不强行执行；`finally` 中释放。 +- **通知链移交**：完成通知轮询器在派发前先预占门闸（`server/chat_flow_task_main.py`），token 经 `session_data["main_task_gate_token"]` 移交 `_run_chat_task` → `run_chat_task_sync` → `process_message_task`（持 token 认领，不重复获取）；派发失败时释放门闸并回滚已打的通知标记（`_rollback_completion_notice_marks`）。 +- **兜底释放**：`_run_chat_task`（`server/tasks/models.py`）外层 `finally` 兜底释放，防止异常路径门闸泄漏。 + +### 12.4 改代码注意事项（硬性） + +1. **新增任何主任务入口必须走门闸**：不要绕过 `process_message_task` 直接驱动一轮模型对话；多智能体 idle 派发（task_type="notice"）目前依赖 `_multi_agent_main_task_active` 标志，后续应统一纳管。 +2. **不要在 return 分支手写 `_tool_loop_active` 恢复**：`execute_tool_calls`（`server/chat_flow_tool_loop.py`）已改为守护包装（try/finally 复位，内层 `_execute_tool_calls_impl`），新增提前返回路径无需也不应手动操作该标志——并发交错「存旧值→置True→恢复旧值」正是此前标志卡死的原因。 +3. **不要依赖 build_messages 防御层掩盖并发问题**：`core/main_terminal_parts/context/messages.py` 的孤儿 tool 消息剥离只是「坏数据不再 400」的止血层，乱序段本身意味着历史已被污染；发现剥离 warning 日志应按事故排查，而不是视为正常。 \ No newline at end of file diff --git a/core/main_terminal_parts/context/messages.py b/core/main_terminal_parts/context/messages.py index 37002eac..3391d0c2 100644 --- a/core/main_terminal_parts/context/messages.py +++ b/core/main_terminal_parts/context/messages.py @@ -332,6 +332,12 @@ class MessagesMixin: conversation = context["conversation"] replaced_tool_count = 0 deep_compacted_skipped = 0 + orphan_tool_skipped = 0 + # 已随 assistant 消息发出的 tool_call id 集合：用于剥离孤儿 tool 消息。 + # 防御线：历史数据一旦被并发写入/截断损坏（assistant.tool_calls 与其 tool + # 响应错位），孤儿 tool 消息会让 API 直接 400（tool_call_id is not found）， + # 对话永久不可用。这里保证即使数据损坏，请求也始终合法（内容降级但不崩）。 + emitted_tool_call_ids = set() for idx, conv in enumerate(conversation): metadata = conv.get("metadata") or {} # 深压缩：被标记为 deep_compacted 的消息（整段已压缩前缀）原文保留在历史中用于展示， @@ -355,9 +361,19 @@ class MessagesMixin: tool_calls = conv.get("tool_calls") or [] if tool_calls and self._tool_calls_followed_by_tools(conversation, idx, tool_calls): message["tool_calls"] = tool_calls + for _tc in tool_calls: + _tcid = (_tc or {}).get("id") + if _tcid: + emitted_tool_call_ids.add(_tcid) messages.append(message) elif conv["role"] == "tool": + # 孤儿 tool 消息（父 assistant 被跳过/未携带 tool_calls/数据损坏） + # 直接剥离，避免 API 400；正常情况下不会发生。 + _tool_call_id = conv.get("tool_call_id") or "" + if not _tool_call_id or _tool_call_id not in emitted_tool_call_ids: + orphan_tool_skipped += 1 + continue if shallow_replace_enabled and metadata.get("auto_shallow_compacted"): messages.append({ "role": "tool", @@ -426,4 +442,9 @@ class MessagesMixin: print(f"[ContextCompression] build_messages 替换tool占位符: {replaced_tool_count} 条") if deep_compacted_skipped: print(f"[ContextCompression] build_messages 跳过已深压缩消息: {deep_compacted_skipped} 条") + if orphan_tool_skipped: + logger.warning( + f"[messages] build_messages 剥离孤儿 tool 消息 {orphan_tool_skipped} 条 " + f"（对话历史疑似损坏，建议检查 tool_call 配对）" + ) return messages diff --git a/server/chat_flow.py b/server/chat_flow.py index 6734048c..a6648f85 100644 --- a/server/chat_flow.py +++ b/server/chat_flow.py @@ -68,6 +68,7 @@ from .utils_common import ( STREAMING_DEBUG_LOG_FILE, ) from .security import rate_limited, format_tool_result_notice, compact_web_search_result, consume_socket_token, prune_socket_tokens, validate_csrf_request, requires_csrf_protection, get_csrf_token +from .main_task_gate import acquire_adopted_main_task_gate, release_main_task_gate from .monitor import cache_monitor_snapshot, get_cached_monitor_snapshot from .extensions import socketio from .state import ( @@ -126,11 +127,27 @@ def detect_malformed_tool_call(text): return _detect_malformed_tool_call(text) -def process_message_task(terminal: WebTerminal, message: str, images, sender, client_sid, workspace: UserWorkspace, username: str, videos=None, files=None): +def process_message_task(terminal: WebTerminal, message: str, images, sender, client_sid, workspace: UserWorkspace, username: str, videos=None, files=None, main_task_gate_token: Optional[str] = None): """在后台处理消息任务""" videos = videos or [] files = files or [] auto_user_message_event = bool(getattr(terminal, "_auto_user_message_event", False)) + + # 对话级主任务门闸（平行时空防护）：同一对话同一时刻只允许一个主任务写对话历史。 + # 通知链任务由轮询器预占门闸并移交 token（见 chat_flow_task_main 的 + # poll_completion_notifications）；其余入口竞争获取。所有路径统一 finally 释放。 + gate_token = acquire_adopted_main_task_gate(terminal, main_task_gate_token) + if gate_token is None: + conversation_id = getattr(getattr(terminal, "context_manager", None), "current_conversation_id", None) + debug_log(f"[MainTaskGate] 拒绝并发主任务: conv={conversation_id} client_sid={client_sid}") + sender('error', { + 'message': '当前对话已有任务在运行，请稍后再试。', + 'conversation_id': conversation_id, + 'task_id': getattr(terminal, "task_id", None) or client_sid, + 'client_sid': client_sid, + }) + return + try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) @@ -238,6 +255,8 @@ def process_message_task(terminal: WebTerminal, message: str, images, sender, cl finally: # 清理任务引用 clear_stop_flag(client_sid, username) + # 释放对话级主任务门闸（仅持有者可释放，重复调用为无操作） + release_main_task_gate(terminal, gate_token) # === 统一对外入口 === def start_chat_task(terminal, message: str, images: Any, sender, client_sid: str, workspace, username: str, videos: Any = None): @@ -255,6 +274,6 @@ def start_chat_task(terminal, message: str, images: Any, sender, client_sid: str ) -def run_chat_task_sync(terminal, message: str, images: Any, sender, client_sid: str, workspace, username: str, videos: Any = None, files: Any = None): +def run_chat_task_sync(terminal, message: str, images: Any, sender, client_sid: str, workspace, username: str, videos: Any = None, files: Any = None, main_task_gate_token: Optional[str] = None): """同步执行（测试/CLI 使用）。""" - return process_message_task(terminal, message, images, sender, client_sid, workspace, username, videos, files) + return process_message_task(terminal, message, images, sender, client_sid, workspace, username, videos, files, main_task_gate_token=main_task_gate_token) diff --git a/server/chat_flow_task_main.py b/server/chat_flow_task_main.py index 7ce7dcdd..151761cc 100644 --- a/server/chat_flow_task_main.py +++ b/server/chat_flow_task_main.py @@ -66,6 +66,7 @@ from .utils_common import ( STREAMING_DEBUG_LOG_FILE, ) from .security import rate_limited, compact_web_search_result, consume_socket_token, prune_socket_tokens, validate_csrf_request, requires_csrf_protection, get_csrf_token +from .main_task_gate import try_acquire_main_task_gate, release_main_task_gate from .monitor import cache_monitor_snapshot, get_cached_monitor_snapshot from .extensions import socketio from .state import ( @@ -518,6 +519,7 @@ async def _dispatch_completion_user_notice( user_message: str, extra_payload: Optional[Dict[str, Any]] = None, preceding_notices: Optional[List[Dict[str, Any]]] = None, + main_task_gate_token: Optional[str] = None, ): """复用子智能体完成后的 user 代发机制。 @@ -567,6 +569,9 @@ async def _dispatch_completion_user_notice( "model_key": getattr(web_terminal, "model_key", None), "message_source": message_source, } + # 派发方预占的主任务门闸 token 随任务移交，由任务线程认领（见 process_message_task） + if main_task_gate_token: + session_data["main_task_gate_token"] = main_task_gate_token ui_defaults = _user_message_ui_defaults( message_source, auto_user_message_event=True, @@ -806,6 +811,7 @@ def _collect_pending_completion_notices(*, web_terminal, conversation_id: str) - "sub_agent_notice": True, "message_source": "background_command", "background_command_notice": True, + "command_id": command_id, }, "sort_key": update.get("updated_at") or time.time(), }) @@ -814,6 +820,41 @@ def _collect_pending_completion_notices(*, web_terminal, conversation_id: str) - return notices +def _rollback_completion_notice_marks(*, web_terminal, notices: List[Dict[str, Any]]) -> None: + """派发失败时回滚 _collect_pending_completion_notices 的就地标记， + 避免通知被标记为已消费却从未派发（静默丢失）。 + """ + sub_manager = getattr(web_terminal, "sub_agent_manager", None) + announced = getattr(web_terminal, "_announced_sub_agent_tasks", set()) + bg_manager = getattr(web_terminal, "background_command_manager", None) + for item in notices or []: + payload = item.get("payload") or {} + if item.get("kind") == "sub_agent": + task_id = payload.get("task_id") + if not task_id: + continue + announced.discard(task_id) + task_info = sub_manager.tasks.get(task_id) if sub_manager else None + if isinstance(task_info, dict): + task_info["notified"] = False + elif item.get("kind") == "background_command": + command_id = payload.get("command_id") + if not command_id or not bg_manager: + continue + try: + with bg_manager._lock: + rec = bg_manager._records.get(str(command_id)) + if rec: + rec["notified"] = False + except Exception: + pass + if sub_manager: + try: + sub_manager._save_state() + except Exception: + pass + + def _has_pending_completion_work(*, web_terminal, conversation_id: str) -> bool: """是否还有运行中或待通知的传统后台任务（子智能体/后台 run_command）。 @@ -904,6 +945,16 @@ async def poll_completion_notifications(*, web_terminal, workspace, conversation await asyncio.sleep(1) continue + # 主任务门闸：仅当对话没有正在运行的主任务时才预占门闸。 + # 防止通知任务与主任务并发交叉写入对话历史（2026-08-12 平行时空事故）。 + # 预占成功后排发：门闸 token 随 session_data 移交给新任务线程认领释放； + # 排发失败则释放门闸并回滚通知标记，下轮重新收集。 + gate_token = try_acquire_main_task_gate(web_terminal) + if gate_token is None: + ma_debug("poll_completion_notifications_wait_main_task_gate", conversation_id=conversation_id) + await asyncio.sleep(1) + continue + ma_debug("poll_completion_notifications_collect", conversation_id=conversation_id, loop_count=loop_count) notices = _collect_pending_completion_notices( web_terminal=web_terminal, @@ -911,6 +962,10 @@ async def poll_completion_notifications(*, web_terminal, workspace, conversation ) ma_debug("poll_completion_notifications_collected", conversation_id=conversation_id, notice_count=len(notices)) + if not notices: + # 本轮没有待通知项：释放门闸后按原逻辑判断退出或继续 + release_main_task_gate(web_terminal, gate_token) + if notices: # 取出后剩余是否还有未完成/未通知的后台工作（决定前端是否保持等待态） has_remaining = _has_pending_completion_work( @@ -949,11 +1004,19 @@ async def poll_completion_notifications(*, web_terminal, workspace, conversation user_message=last_notice.get("message") or "", extra_payload=last_payload, preceding_notices=preceding_payload, + main_task_gate_token=gate_token, ) - except Exception: - pass + except Exception as exc: + # 派发失败：释放门闸 + 回滚通知标记，下轮轮询重新收集，杜绝静默丢失 + debug_log(f"[CompletionPoll] 派发完成通知失败，释放门闸并回滚标记: {exc}") + release_main_task_gate(web_terminal, gate_token) + _rollback_completion_notice_marks(web_terminal=web_terminal, notices=notices) + ma_debug("poll_completion_notifications_dispatch_failed", conversation_id=conversation_id, error=str(exc)) + await asyncio.sleep(5) + continue - # 本轮已派发一个后续任务，由该任务结束后重新 spawn 本轮询器继续消费剩余通知 + # 本轮已派发一个后续任务，门闸随任务移交（由任务线程 finally 释放）， + # 由该任务结束后重新 spawn 本轮询器继续消费剩余通知 return # 没有待通知项：若也没有运行中的后台工作，结束轮询 diff --git a/server/chat_flow_tool_loop.py b/server/chat_flow_tool_loop.py index c5857f58..336bbb23 100644 --- a/server/chat_flow_tool_loop.py +++ b/server/chat_flow_tool_loop.py @@ -261,7 +261,24 @@ async def _wait_for_user_questions(*, question_ids: List[str], username: str, ti return answered -async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, client_sid: str, username: str, iteration: int, conversation_id: Optional[str], last_tool_call_time: float, process_sub_agent_updates, process_background_command_updates, process_multi_agent_master_messages=process_multi_agent_master_messages, get_stop_flag, clear_stop_flag, workspace=None): +async def execute_tool_calls(**kwargs): + """_execute_tool_calls_impl 的守护包装：保证 _tool_loop_active 在任何退出路径都复位。 + + 历史教训（2026-08-12 平行时空事故）：标志的恢复逻辑散落在内层各 return 分支， + 并发交错或异常逃逸会把它永久卡在 True，导致完成通知轮询器死等、通知永远发不出。 + 内层各 return 点的恢复仍然保留（尽早复位缩短轮询器等待），本包装做最终兜底。 + """ + web_terminal = kwargs.get("web_terminal") + if web_terminal is None: + return await _execute_tool_calls_impl(**kwargs) + previous_tool_loop_active = getattr(web_terminal, "_tool_loop_active", False) + try: + return await _execute_tool_calls_impl(**kwargs) + finally: + web_terminal._tool_loop_active = previous_tool_loop_active + + +async def _execute_tool_calls_impl(*, web_terminal, tool_calls, sender, messages, client_sid: str, username: str, iteration: int, conversation_id: Optional[str], last_tool_call_time: float, process_sub_agent_updates, process_background_command_updates, process_multi_agent_master_messages=process_multi_agent_master_messages, get_stop_flag, clear_stop_flag, workspace=None): previous_tool_loop_active = getattr(web_terminal, "_tool_loop_active", False) web_terminal._tool_loop_active = True allowed_tool_names = set() diff --git a/server/main_task_gate.py b/server/main_task_gate.py new file mode 100644 index 00000000..27001f72 --- /dev/null +++ b/server/main_task_gate.py @@ -0,0 +1,73 @@ +"""对话级主任务门闸（单写者防护）。 + +背景（2026-08-12「平行时空」事故）：socketio 入口的主聊天任务不在 +task_manager 注册，`create_chat_task` 的单对话互斥对它们不可见；完成通知 +轮询器又只凭 `_tool_loop_active`（仅覆盖工具执行窗口）判断对话是否空闲， +于是在主任务两次工具循环的间隙里派发了通知任务——两个主任务并发交叉写入 +同一份 conversation_history，产生 assistant/assistant/tool/tool 乱序段， +下一轮请求重建消息时 tool 配对崩坏（API 400 tool_call_id is not found）； +并发 execute_tool_calls 还把 `_tool_loop_active` 永久卡成 True，通知全部死等。 + +不变量：**一个对话（= 一个 WebTerminal 实例）同一时刻只允许一个主聊天任务运行。** + +用法： +- 所有主任务入口统一收敛在 `process_message_task`（chat_flow.py），在此获取 + 门闸并在 finally 释放。 +- 通知派发链（完成通知轮询器）先 `try_acquire_main_task_gate` 预占，再通过 + session_data["main_task_gate_token"] 把 token 移交给新任务线程认领； + 派发失败时释放并回滚通知标记。 +""" +from __future__ import annotations + +import threading +import uuid +from typing import Optional + +_LOCK = threading.Lock() +_GATE_ATTR = "_main_task_gate_token" + + +def try_acquire_main_task_gate(terminal) -> Optional[str]: + """非阻塞获取门闸。成功返回 token；门闸已被占用返回 None。""" + if terminal is None: + return None + with _LOCK: + if getattr(terminal, _GATE_ATTR, None): + return None + token = uuid.uuid4().hex + setattr(terminal, _GATE_ATTR, token) + return token + + +def acquire_adopted_main_task_gate(terminal, token: Optional[str]) -> Optional[str]: + """认领派发方预占的门闸 token；认领失败则退化为竞争获取。 + + 返回 None 表示门闸被其他任务持有，调用方应放弃本次运行。 + """ + if terminal is None: + return None + with _LOCK: + current = getattr(terminal, _GATE_ATTR, None) + if token and current == token: + return token # 认领成功（门闸已由派发方持有） + if current: + return None # 被无关任务占用 + new_token = uuid.uuid4().hex + setattr(terminal, _GATE_ATTR, new_token) + return new_token + + +def release_main_task_gate(terminal, token: Optional[str]) -> None: + """释放门闸。只有持有者（token 匹配）才能释放，重复/过期调用为无操作。""" + if terminal is None or not token: + return + with _LOCK: + if getattr(terminal, _GATE_ATTR, None) == token: + setattr(terminal, _GATE_ATTR, None) + + +def is_main_task_gate_busy(terminal) -> bool: + """只读探测：当前是否有主任务持有门闸。""" + if terminal is None: + return False + return bool(getattr(terminal, _GATE_ATTR, None)) diff --git a/server/tasks/models.py b/server/tasks/models.py index 46ebe937..a8de5788 100644 --- a/server/tasks/models.py +++ b/server/tasks/models.py @@ -17,6 +17,7 @@ from flask import current_app, session from server.auth_helpers import api_login_required, get_current_username from server.context import get_user_resources, ensure_conversation_loaded from server.chat_flow import run_chat_task_sync +from server.main_task_gate import release_main_task_gate from server.work_timer import finalize_conversation_work_timer from server.state import stop_flags from server.utils_common import debug_log, log_conn_diag @@ -983,6 +984,8 @@ class TaskManager: username=username, videos=videos, files=files or [], + # 通知链任务认领轮询器预占的门闸（其余任务为 None，走竞争获取） + main_task_gate_token=(rec.session_data or {}).get("main_task_gate_token"), ) finally: try: @@ -1075,6 +1078,15 @@ class TaskManager: finally: # 清理 stop_flags stop_flags.pop(rec.task_id, None) + # 主任务门闸兜底释放：若任务线程在 process_message_task 认领前异常退出， + # 按 session_data 中的 token 释放，避免门闸泄漏导致对话永久被占用。 + # 正常路径下 process_message_task 已在 finally 释放，此处为无操作。 + try: + gate_token = (rec.session_data or {}).get("main_task_gate_token") + if gate_token and terminal: + release_main_task_gate(terminal, gate_token) + except Exception: + pass # 清理一次性配置 if terminal and hasattr(terminal, "max_iterations_override"): try: