fix(server): 对话级主任务门闸，修复并发主任务交叉写入对话历史

事故：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

@@ -492,3 +492,31 @@ AI 执行以下流程时，每一步都要向用户说明在做什么：
 ---
 
 注：本节按「现有架构 + 多智能体分支」方案描述，与项目记忆 `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 日志应按事故排查，而不是视为正常。

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

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)

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
 
         # 没有待通知项：若也没有运行中的后台工作，结束轮询

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()

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

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: