fix: 修复 WebAgent 会话与选择状态 (#7266)

* fix(agent): preserve WebAgent session state

* test: separate subscription pressure coordination and owner waits

---------

Co-authored-by: jxxghp <jxxghp@gmail.com>

app/agent/session.py

@@ -177,6 +177,16 @@ class AgentManagerUnavailableError(RuntimeError):
 
     code = "agent_manager_unavailable"
 
+    def __init__(self, *args: object, task_accepted: bool = True) -> None:
+        """
+        标明异常发生时是否可能已接收任务，供交互入口决定能否安全重试。
+
+        :param args: 原有异常说明，保持 RuntimeError 调用兼容
+        :param task_accepted: 默认按可能已执行处理；明确准入拒绝时为 False
+        """
+        super().__init__(*args)
+        self.task_accepted = task_accepted
+
 
 class AgentManagerQueueFullError(RuntimeError):
     """
@@ -184,6 +194,8 @@ class AgentManagerQueueFullError(RuntimeError):
     """
 
     code = "agent_manager_queue_full"
+    # 排队已满发生在入队前，可安全重试原交互。
+    task_accepted = False
 
     def __init__(self, session_id: str, limit: int) -> None:
         """
@@ -275,10 +287,7 @@ class AgentSessionOwner:
         status["shutdown_pending"] = bool(
             pending_shutdown and not pending_shutdown.done()
         )
-        status["is_processing"] = (
-            session_id in self._session_workers
-            and not self._session_workers[session_id].done()
-        )
+        status["is_processing"] = self.is_session_busy(session_id)
         inbox = self._session_steering_inboxes.get(session_id)
         status["steering_pending"] = inbox.pending_count if inbox else 0
         status["steering_active"] = bool(inbox and inbox.running)
@@ -322,9 +331,10 @@ class AgentSessionOwner:
 
     def is_session_busy(self, session_id: str) -> bool:
         """
-        查询会话是否仍有正在执行或排队的任务。
+        查询实际执行或排队状态，空闲等待消息的 worker 不计入。
         """
-        return self._is_session_busy(session_id)
+        queue = self._session_queues.get(session_id)
+        return session_id in self._session_active_tasks or bool(queue and not queue.empty())
 
     def _expired_idle_sessions(self) -> list[tuple[str, str]]:
         """
@@ -406,7 +416,7 @@ class AgentSessionOwner:
             if not self._accepting_tasks:
                 if completion_future and not completion_future.done():
                     completion_future.cancel()
-                raise AgentManagerUnavailableError("AgentManager 未运行或已关闭")
+                raise AgentManagerUnavailableError("AgentManager 未运行或已关闭", task_accepted=False)
             pending_shutdown = self._session_shutdown_pending.get(session_id)
             if pending_shutdown:
                 if pending_shutdown.done():
@@ -415,7 +425,7 @@ class AgentSessionOwner:
                     if completion_future and not completion_future.done():
                         completion_future.cancel()
                     raise AgentManagerUnavailableError(
-                        f"Agent 会话 {session_id} 仍在停止，暂时不能接收新任务"
+                        f"Agent 会话 {session_id} 仍在停止，暂时不能接收新任务", task_accepted=False,
                     )
             self._record_session_activity(session_id, user_id)
             if await self._try_steer_channel_message(task):
@@ -610,8 +620,8 @@ class AgentSessionOwner:
                             queue_wait_ms,
                             3,
                         )
-                    await self._start_task_processing_status(task)
                     self._session_active_tasks[session_id] = task
+                    await self._start_task_processing_status(task)
                     record_metric(
                         "agent.active_tasks",
                         1,

app/api/endpoints/agent.py

@@ -350,9 +350,9 @@ async def web_agent_callback(
             ),
         )
 
-    result = web_agent_application.resolve_web_agent_choice_payload(
+    result = await web_agent_application.resolve_and_save_web_agent_choice(
         callback_data=payload.callback_data,
-        user_id=str(current_user.id),
+        current_user=current_user,
     )
     if not result:
         return _SchemaResponse(success=False, message="该选择已失效，请重新发起选择")

app/application/messaging/agent.py

@@ -10,7 +10,7 @@ import shutil
 import time
 import uuid
 from collections import deque
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from datetime import datetime
 from functools import partial
 from pathlib import Path
@@ -498,9 +498,9 @@ def _apply_web_agent_tool_event(event: dict[str, Any], assistant_message: dict[s
             matching_tool["status"] = tool_status
         return
 
-    # 兼容尚未升级的事件生产者：没有调用 ID 时只能把前一批提示视为已结束。
+    # 旧提示没有调用 ID，只收口上一条旧提示，保留结构化并行工具的实际状态。
     for tool in assistant_message["tools"]:
-        if tool.get("status") == "running":
+        if tool.get("status") == "running" and tool.get("tool_name") is None:
             tool["status"] = "done"
     tool_index = len(assistant_message["tools"])
     assistant_message["tools"].append(
@@ -566,14 +566,14 @@ def apply_web_agent_display_event(event: dict[str, Any], assistant_message: dict
             )
         for tool in assistant_message["tools"]:
             if tool.get("status") == "running":
-                tool["status"] = "done"
+                tool["status"] = "error"
     elif event_type == "done":
         assistant_message["thinking"] = False
         if assistant_message.get("status") != "error":
             assistant_message["status"] = "done"
         for tool in assistant_message["tools"]:
             if tool.get("status") == "running":
-                tool["status"] = "done"
+                tool["status"] = "done" if tool.get("tool_name") is None else "error"
 
 
 async def save_web_agent_display_snapshot(
@@ -1143,7 +1143,7 @@ def resolve_web_agent_choice_payload(callback_data: str, user_id: str) -> Option
         return None
 
     request_id, option_index = parsed
-    resolved = agent_interaction.agent_interaction_manager.resolve(
+    resolved = agent_interaction.agent_interaction_manager.resolve_web_choice(
         request_id=request_id,
         option_index=option_index,
         user_id=str(user_id),
@@ -1151,10 +1151,11 @@ def resolve_web_agent_choice_payload(callback_data: str, user_id: str) -> Option
     if not resolved:
         return None
 
-    request, option = resolved
+    request, option, already_selected = resolved
     buttons, button_rows = agent_interaction.build_agent_choice_button_rows(request)
     selected_description = option.description or option.label
     return {
+        "already_selected": already_selected,
         "message": option.value,
         "session_id": request.session_id,
         "display_message": selected_description,
@@ -1181,6 +1182,42 @@ def resolve_web_agent_choice_payload(callback_data: str, user_id: str) -> Option
     }
 
 
+async def resolve_and_save_web_agent_choice(
+    callback_data: str, current_user: AgentChatPrincipal,
+    service: Optional[AgentChatService] = None,
+    persistence: Optional[AgentChatPersistenceService] = None,
+) -> Optional[dict[str, Any]]:
+    """
+    校验会话归属并在返回选择反馈前持久化，失败不会派发 Agent。
+    """
+    parsed = agent_interaction.parse_agent_choice_callback(callback_data)
+    if parsed is None:
+        return None
+    request = agent_interaction.agent_interaction_manager.get_request(parsed[0], str(current_user.id))
+    if request is None:
+        return None
+    service = service or get_configured_agent_chat_service()
+    existing = await service.get(request.session_id)
+    if existing is not None and not can_access_agent_chat(existing, current_user):
+        return None
+    result = resolve_web_agent_choice_payload(callback_data, str(current_user.id))
+    if result is None:
+        return None
+    receipt = build_web_agent_display_message(role="user")
+    receipt["id"] = f"choice-selection-{request.request_id}"
+    receipt["choice_selection"] = result["choice_selection"]
+    try:
+        await save_web_agent_display_snapshot(
+            session_id=request.session_id, current_user=current_user, messages=[receipt],
+            service=service, persistence=persistence, append=True,
+        )
+    except Exception:
+        agent_interaction.agent_interaction_manager.finish_web_choice(request.request_id, str(current_user.id), persisted=False)
+        raise
+    agent_interaction.agent_interaction_manager.finish_web_choice(request.request_id, str(current_user.id), persisted=True)
+    return result
+
+
 def build_web_agent_message_events(
     message: Message,
     *,
@@ -1654,6 +1691,22 @@ async def build_web_agent_stream(
         return WebAgentStreamResult(
             events=_single_web_agent_event({"type": "error", "message": "会话不存在或无权访问"})
         )
+    if command.choice_selection:
+        choice_id = str(command.choice_selection.get("choice_id") or "")
+        claimed = agent_interaction.agent_interaction_manager.get_web_choice(choice_id, str(current_user.id), session_id)
+        if claimed is None:
+            return WebAgentStreamResult(events=_single_web_agent_event({
+                "type": "error", "message": "该选择已处理或不属于当前会话",
+            }))
+        choice_request, option = claimed
+        canonical = resolve_web_agent_choice_payload(
+            agent_interaction.build_agent_choice_callback(choice_id, choice_request.selected_index or 1), str(current_user.id),
+        )
+        if canonical is None:
+            return WebAgentStreamResult(events=_single_web_agent_event({"type": "error", "message": "该选择已失效"}))
+        prompt = option.value
+        display_prompt = option.description or option.label
+        command = replace(command, text=prompt, display_text=display_prompt, choice_selection=canonical["choice_selection"])
     is_secret_confirmation_candidate = (
         prompt in {"确认", "取消"} and not command.images and not command.audio_refs and not command.files
     )
@@ -1678,7 +1731,7 @@ async def build_web_agent_stream(
             )
         )
 
-    is_traditional_message = not agent_interaction.is_agent_learning_command(prompt) and (
+    is_traditional_message = not command.choice_selection and not agent_interaction.is_agent_learning_command(prompt) and (
         is_web_agent_traditional_message(prompt) or has_web_agent_traditional_interaction(str(current_user.id))
     )
     if is_traditional_message:
@@ -1743,7 +1796,7 @@ async def build_web_agent_stream(
             )
         )
     # 活动 Agent 运行内的补充输入走同一张图；无活动运行时才创建新一轮。
-    steering_message = await submit_web_agent_steering(
+    steering_message = None if command.choice_selection else await submit_web_agent_steering(
         manager=manager,
         session_id=session_id,
         user_id=str(current_user.id),

app/application/messaging/interaction/agent.py

@@ -82,6 +82,9 @@ class PendingAgentInteraction:
     :param prompt: 向用户说明选择目的的提示正文
     :param options: 按展示顺序排列的可选项
     :param created_at: 会话创建的本地时间，用于计算过期时间
+    :ivar selected_index: 已确认的选项序号；未选择时为 None
+    :ivar selection_persisted: Web 选择结果是否已完成持久化
+    :ivar execution_claimed: Web 选择是否已交给一次 Agent 执行
     """
 
     request_id: str
@@ -94,6 +97,9 @@ class PendingAgentInteraction:
     prompt: str
     options: list[AgentInteractionOption]
     created_at: datetime = field(default_factory=datetime.now)
+    selected_index: Optional[int] = field(default=None, init=False)
+    selection_persisted: bool = field(default=False, init=False)
+    execution_claimed: bool = field(default=False, init=False)
 
 
 def build_agent_choice_button_rows(
@@ -203,5 +209,79 @@ class AgentInteractionManager:
         with self._lock:
             self._pending_interactions.clear()
 
+    def get_request(self, request_id: str, user_id: str) -> Optional[PendingAgentInteraction]:
+        """
+        读取当前用户的交互归属，供异步会话授权使用，不消费请求。
+        """
+        with self._lock:
+            self._cleanup_locked()
+            request = self._pending_interactions.get(request_id)
+            return request if request and request.user_id == str(user_id) else None
+
+    def resolve_web_choice(
+        self, request_id: str, option_index: int, user_id: str,
+    ) -> Optional[tuple[PendingAgentInteraction, AgentInteractionOption, bool]]:
+        """
+        保留 Web 选择收据；重复按钮始终返回首次选项。
+        """
+        with self._lock:
+            self._cleanup_locked()
+            request = self._pending_interactions.get(request_id)
+            if not request or request.user_id != str(user_id) or not 1 <= option_index <= len(request.options):
+                return None
+            repeated = request.execution_claimed
+            if request.selected_index is None:
+                request.selected_index = option_index
+            return request, request.options[request.selected_index - 1], repeated
+
+    def finish_web_choice(self, request_id: str, user_id: str, *, persisted: bool) -> None:
+        """
+        确认选择落库；失败保留首次选项，重试不会换成另一按钮。
+        """
+        with self._lock:
+            request = self._pending_interactions.get(request_id)
+            if request is None or request.user_id != str(user_id):
+                return
+            if persisted:
+                request.selection_persisted = True
+
+    def get_web_choice(
+        self, request_id: str, user_id: str, session_id: str,
+    ) -> Optional[tuple[PendingAgentInteraction, AgentInteractionOption]]:
+        """
+        读取已落库的同一用户、同一会话选择，不领取执行。
+        """
+        with self._lock:
+            self._cleanup_locked()
+            request = self._pending_interactions.get(request_id)
+            if (request is None or request.user_id != str(user_id) or request.session_id != session_id
+                    or request.selected_index is None or not request.selection_persisted):
+                return None
+            return request, request.options[request.selected_index - 1]
+
+    def release_web_choice(self, request_id: str, user_id: str, session_id: str) -> None:
+        """
+        仅在执行入口明确拒绝接收任务时归还领取，允许原选项重试。
+        """
+        with self._lock:
+            request = self._pending_interactions.get(request_id)
+            if request and request.user_id == str(user_id) and request.session_id == session_id:
+                request.execution_claimed = False
+
+    def claim_web_choice(
+        self, request_id: str, user_id: str, session_id: str,
+    ) -> Optional[tuple[PendingAgentInteraction, AgentInteractionOption]]:
+        """
+        仅为同一用户和会话领取一次已落库选择，重复流请求不再执行业务。
+        """
+        with self._lock:
+            self._cleanup_locked()
+            request = self._pending_interactions.get(request_id)
+            if (request is None or request.user_id != str(user_id) or request.session_id != session_id
+                    or request.selected_index is None or not request.selection_persisted or request.execution_claimed):
+                return None
+            request.execution_claimed = True
+            return request, request.options[request.selected_index - 1]
+
 
 agent_interaction_manager = AgentInteractionManager()

app/application/messaging/webagent/stream.py

@@ -11,6 +11,7 @@ from typing import Any, AsyncIterator, Awaitable, Callable, Coroutine, Optional,
 
 from app.agent.steering import SteeringMessage
 from app.application import agent as agent_application
+from app.application.messaging.interaction.agent import agent_interaction_manager
 from app.runtime.log import logger
 from app.runtime.stop import runtime_stop_state
 from app.schemas.message import Message
@@ -244,13 +245,18 @@ def _build_web_agent_tool_event_callback(
     """
     构造结构化工具生命周期回调，确保快照与 SSE 使用同一事件顺序。
     """
+    tool_messages: dict[str, dict[str, Any]] = {}
 
     def tool_event_callback(event: dict[str, Any]) -> None:
         """
         投影工具开始、完成和失败事件。
         """
-        display_event = _attach_assistant_message_id(event, assistant_message_ref)
-        apply_display_event(display_event, assistant_message_ref["message"])
+        tool_id = str(event.get("tool_id") or event.get("tool_call_id") or "")
+        target = tool_messages.setdefault(tool_id, assistant_message_ref["message"]) if tool_id else assistant_message_ref["message"]
+        display_event = _attach_assistant_message_id(event, {"message": target})
+        if tool_id and target.get("id"):
+            display_event = {**display_event, "assistant_message_id": target["id"]}
+        apply_display_event(display_event, target)
         event_publisher.publish(display_event)
 
     return tool_event_callback
@@ -266,6 +272,32 @@ def _build_web_agent_files(command: Any) -> list[dict[str, Any]]:
     return files
 
 
+def _finish_web_agent_display_messages(
+    messages: list[dict[str, Any]],
+    apply_display_event: Callable[[dict[str, Any], dict[str, Any]], None],
+) -> None:
+    """
+    主运行终止时收口本轮所有助手段，保留工具已报告的终态。
+    """
+    for message in messages:
+        if message.get("role") == "assistant":
+            apply_display_event({"type": "done"}, message)
+
+
+def _prepare_web_agent_run(command: Any, current_user: Any, session_id: str) -> tuple[Any, Any, bool]:
+    """
+    确认执行服务和 Agent 工厂就绪后，才领取本次选择。
+    """
+    manager = agent_application.get_running_agent_manager()
+    if manager is None:
+        raise RuntimeError("智能助手服务尚未就绪，请稍后重试。")
+    agent_factory = agent_application.get_web_agent_type()
+    choice_id = str((command.choice_selection or {}).get("choice_id") or "")
+    if choice_id and agent_interaction_manager.claim_web_choice(choice_id, str(current_user.id), session_id) is None:
+        raise RuntimeError("该选择已领取执行")
+    return manager, agent_factory, bool(choice_id)
+
+
 def _build_web_agent_event_generator(
     *,
     command: Any,
@@ -297,10 +329,10 @@ def _build_web_agent_event_generator(
         """
         后台执行 Agent，并在完成后持久化展示快照。
         """
+        choice_id = str((command.choice_selection or {}).get("choice_id") or "")
+        choice_claimed = False
         try:
-            runtime_manager = agent_application.get_running_agent_manager()
-            if runtime_manager is None:
-                raise RuntimeError("智能助手服务尚未就绪，请稍后重试。")
+            runtime_manager, agent_factory, choice_claimed = _prepare_web_agent_run(command, current_user, session_id)
             await runtime_manager.process_message(
                 session_id=session_id,
                 user_id=str(current_user.id),
@@ -318,13 +350,16 @@ def _build_web_agent_event_generator(
                 protected_output_callback=(protected_output_callback if protected_transport_supported else None),
                 message_callback=message_callback,
                 steering_status_callback=steering_status_callback,
-                agent_factory=agent_application.get_web_agent_type(),
+                agent_factory=agent_factory,
                 wait_for_completion=True,
             )
         except asyncio.CancelledError:
             # 显式停止会话沿用正常终止语义；服务关闭由 manager 的稳定异常分支处理。
             pass
         except Exception as err:
+            if choice_claimed and getattr(err, "task_accepted", True) is False:
+                agent_interaction_manager.release_web_choice(choice_id, str(current_user.id), session_id)
+                choice_claimed = False
             logger.error(f"Web智能助手执行失败: {str(err)}")
             error_event = _attach_assistant_message_id({
                 "type": "error",
@@ -334,10 +369,10 @@ def _build_web_agent_event_generator(
             event_publisher.publish(error_event)
         finally:
             done_event = _attach_assistant_message_id({"type": "done"}, assistant_message_ref)
-            dependencies.apply_display_event(done_event, assistant_message_ref["message"])
+            _finish_web_agent_display_messages(display_messages, dependencies.apply_display_event)
             # 终态先进入事件队列，避免展示快照落库延迟前端结束动画。
             event_publisher.publish(done_event)
-            if not is_secret_confirmation_control:
+            if not is_secret_confirmation_control and (not choice_id or choice_claimed):
                 try:
                     await dependencies.save_display_snapshot(
                         session_id=session_id,
@@ -425,14 +460,15 @@ def build_agent_web_agent_stream(
         audio_refs=command.audio_refs,
     )
     display_messages = []
-    if command.echo_user and not is_secret_confirmation_control:
+    if (command.echo_user or command.choice_selection) and not is_secret_confirmation_control:
         user_display_message = dependencies.build_display_message(
             role="user",
-            content=display_prompt or prompt,
+            content=(display_prompt or prompt) if command.echo_user else "",
             attachments=user_attachments,
         )
         if command.choice_selection:
             user_display_message["choice_selection"] = command.choice_selection
+            user_display_message["id"] = f"choice-selection-{command.choice_selection['choice_id']}"
         display_messages.append(user_display_message)
     assistant_display_message = dependencies.build_display_message(
         role="assistant",

app/db/oper/agentchat.py

@@ -252,6 +252,9 @@ class AgentChatOper(DbOper):
         )
         if not chat:
             return None
+        from app.domain.chat import merge_choice_state
+
+        normalized_messages = merge_choice_state(chat.display_messages or [], normalized_messages)
         normalized_title = (
             chat.title
             if self.has_custom_title(chat.title)

app/domain/chat.py

@@ -0,0 +1,76 @@
+"""
+会话展示快照中选择结果的合并规则，不负责交互执行或持久化。
+"""
+
+from copy import deepcopy
+from typing import Any
+
+
+def _choice_selections(messages: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
+    """
+    优先采用完整选择回执，旧记录只有已选卡片时从卡片恢复。
+    """
+    selections: dict[str, dict[str, Any]] = {}
+    for message in messages:
+        selected = message.get("choice_selection")
+        if isinstance(selected, dict) and selected.get("choice_id"):
+            selections.setdefault(str(selected["choice_id"]), selected)
+    for message in messages:
+        for choice in message.get("choices") or []:
+            if isinstance(choice, dict) and choice.get("status") == "selected" and choice.get("id"):
+                selections.setdefault(str(choice["id"]), {
+                    "choice_id": str(choice["id"]),
+                    "prompt": choice.get("prompt") or "",
+                    "buttons": choice.get("buttons") or [],
+                    "button_rows": choice.get("button_rows") or [],
+                    **{key: choice.get(key) for key in (
+                        "title", "selected_label", "selected_value", "selected_description",
+                    )},
+                })
+    return selections
+
+
+def merge_choice_state(
+    previous: list[dict[str, Any]], messages: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """
+    保留首次已选结果，旧客户端的 pending 快照不能降级已选卡片。
+
+    :param previous: 当前持久化的权威快照
+    :param messages: 本次准备保存的完整快照，不原地修改
+    :return: 应用已选结果并按稳定 ID 合并选择回显后的快照
+    """
+    selections = _choice_selections(previous)
+    for choice_id, selection in _choice_selections(messages).items():
+        selections.setdefault(choice_id, selection)
+    result: list[dict[str, Any]] = []
+    receipts: dict[str, int] = {}
+    received_choices: set[str] = set()
+    for original in messages:
+        message = deepcopy(original)
+        selected = message.get("choice_selection")
+        if isinstance(selected, dict) and selected.get("choice_id"):
+            choice_id = str(selected["choice_id"])
+            received_choices.add(choice_id)
+            message["choice_selection"] = deepcopy(selections[choice_id])
+        for choice in message.get("choices") or []:
+            if not isinstance(choice, dict) or str(choice.get("id") or "") not in selections:
+                continue
+            selection = selections[str(choice["id"])]
+            choice.update({"status": "selected", **{
+                key: selection.get(key) for key in ("selected_label", "selected_value", "selected_description")
+            }})
+        key = str(message.get("id") or "") if selected else ""
+        if key and key in receipts:
+            result[receipts[key]] = message
+        else:
+            if key:
+                receipts[key] = len(result)
+            result.append(message)
+    for choice_id, selection in selections.items():
+        if choice_id not in received_choices:
+            result.append({
+                "id": f"choice-selection-{choice_id}", "role": "user", "content": "",
+                "choice_selection": deepcopy(selection),
+            })
+    return result

app/schemas/agent.py

@@ -279,6 +279,7 @@ class AgentWebCallbackData(BaseModel):
     """
 
     message: str = Field(description="下一条用户消息")
+    already_selected: bool = Field(default=False, description="选择是否已领取执行；为 True 时仅恢复首次反馈，不再派发消息")
     display_message: str = Field(default="", description="前端展示消息")
     session_id: Optional[str] = Field(default=None, description="Agent 会话 ID")
     traditional: bool = Field(default=False, description="是否为传统消息链回调")

docs/architecture-overview.md

@@ -820,8 +820,8 @@ flowchart LR
 
 | 指标 | 当前值 |
 |---|---:|
-| Python 模块 | 1144 |
-| 内部导入边 | 9,899 |
+| Python 模块 | 1145 |
+| 内部导入边 | 9,904 |
 | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） |

docs/mcp-api.md

@@ -34,6 +34,18 @@
 升级前历史没有来源标记，界面显示“未记录”，不会按当前订阅设置追溯推断。
 普通订阅的首次提交与洗版订阅补缺在活动中都标记为首次订阅。
 
+### WebAgent 会话状态
+
+`POST /api/v1/message/agent/callback` 在返回前保存首次选择。`already_selected=true` 表示
+该选择已领取执行，客户端仅恢复原反馈；已选但尚未执行时返回 `false`，可重试原选项。
+后端按用户、会话和选择 ID 领取一次执行；入队前明确拒绝可重试，执行结果未知时不重放。
+`echo_user=false` 仅隐藏用户回显，选择结果仍保存；完整快照替换保留隐藏选择收据，
+即使旧快照不含卡片，也不能让后续快照把已选卡片改回 `pending`。普通消息仍按完整快照替换。
+
+会话 `is_processing` 只表示实际执行或排队，空闲等待下一条消息的 worker 不计入。
+工具事件沿用 `tool_id` 与 `assistant_message_id`，跨 steering 后仍更新原助手段。
+全流结束收口所有助手段；缺少终态的结构化工具标为 `error`，旧文本提示保留完成兼容。
+
 ### 资源搜索诊断
 
 Web 影视精确与标题 SSE 的 `manual_paging=true` 模式在最终 `done` 事件增加可选

docs/rules/05-architecture.md

@@ -1319,6 +1319,7 @@ driven workflow registration.
 | `app/application/server/report.py` | Server reporting use cases over injected local readers and transport callbacks |
 | `app/application/server/share.py` | Server sharing use cases over injected repositories and transport callbacks |
 | `app/domain/plugin.py` | Pure local-source, generation-compatibility and Release-install-plan rules |
+| `app/domain/chat.py` | 会话展示快照的纯选择状态合并规则；保留首次选择，不读取交互管理器或持久化服务 |
 | `app/adapters/external/plugin/client.py` | `PluginMarketTransport`, the unique market transport, index/Release and local-candidate owner |
 | `app/adapters/system/plugin/package.py` | `PluginPackageManager`, the unique install, checkpoint, backup, restore and physical-delete owner |
 | `app/adapters/system/plugin/dependency.py` | `PluginDependencyInstaller`, enabled-source dependency inspection and constrained per-plugin recovery owner; `async_install` has two completion-owned worker boundaries for request preparation and fallback planning, while package subprocess cancellation remains with `PluginRuntimeHealth` |

scripts/validation/subscription_governance_scale.py

@@ -345,6 +345,8 @@ class _ScaleTorrentHelper:
 
 # 线程同步最多等待 5 秒，避免验收异常时无限挂起。
 _SITE_PRESSURE_SYNC_TIMEOUT = 5.0
+# 执行方退出单独计时，故障注入可缩短收口等待而不影响请求竞争场景的建立。
+_SITE_PRESSURE_JOIN_TIMEOUT = 5.0
 
 
 class _ScaleSiteRequestBoundary:
@@ -574,7 +576,7 @@ def _run_site_pressure_case(
         contention_observed = pressure_event.wait(timeout=_SITE_PRESSURE_SYNC_TIMEOUT)
         boundary.release_first.set()
         for thread in threads:
-            thread.join(timeout=_SITE_PRESSURE_SYNC_TIMEOUT)
+            thread.join(timeout=_SITE_PRESSURE_JOIN_TIMEOUT)
         owners_finished = all(not thread.is_alive() for thread in threads)
 
         initial_deferrals = [

skills/moviepilot-api/SKILL.md

@@ -377,6 +377,10 @@ token exchange downloads retain their existing credential rules.
 - `agent_task` pins a real execution account at creation. Channel senders must uniquely bind to an active MoviePilot user. Background API calls retain that account's permissions, and invalid identity fails the run. Confirm legacy tasks with `update` in their original conversation or recreate them; never substitute a same-name account or fall back to the superuser.
 - Traditional message sessions are scoped by channel, configured source, and user ID. A stop, clear, status query, or confirmation targets the current source's session.
 - Background WebAgent completion appends the current turn to display history. Frontend complete snapshots keep their existing replacement contract.
+- Complete snapshot replacement retains hidden choice receipts so delayed snapshots cannot restore a selected card to pending; ordinary messages retain replacement behavior.
+- WebAgent choice callbacks save the first selection before replying. `already_selected=true` means
+  execution was claimed; restore its feedback without sending another message. Unclaimed selections
+  can retry the same value. Idle waiting workers do not count toward `is_processing`.
 - WebAgent temporary attachments are scoped to their creating account. Download, voice transcription, and attachment events accept only references registered to the current account; cross-account access is treated as missing.
 - File creation with `overwrite=false` rejects a concurrent creator. Versioned edits reject a peer's intervening write; reread the file and use its current SHA before retrying.
 - Market skill installation preserves an explicitly selected GitHub branch. If that branch is unavailable, treat installation as failed instead of silently switching branches.

tests/fixtures/architecture/dependency-baseline.json

@@ -959,8 +959,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 9899,
-  "edge_sha256": "fa4b6dd62edd439a931b77de55e0debf482fcc09c2caa223027a214959df9124",
+  "edge_count": 9904,
+  "edge_sha256": "4fbaf07e93f98d96204ced4acdd8be21af9ac11e2a7c3aa9bbce1b4c0785a5b6",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -3953,6 +3953,9 @@
     "app.application.messaging.webagent.stream -> app.agent.steering",
     "app.application.messaging.webagent.stream -> app.application",
     "app.application.messaging.webagent.stream -> app.application.agent",
+    "app.application.messaging.webagent.stream -> app.application.messaging",
+    "app.application.messaging.webagent.stream -> app.application.messaging.interaction",
+    "app.application.messaging.webagent.stream -> app.application.messaging.interaction.agent",
     "app.application.messaging.webagent.stream -> app.runtime",
     "app.application.messaging.webagent.stream -> app.runtime.log",
     "app.application.messaging.webagent.stream -> app.runtime.stop",
@@ -6785,6 +6788,8 @@
     "app.db.oper.agentchat -> app.db.models.agentchat",
     "app.db.oper.agentchat -> app.db.oper",
     "app.db.oper.agentchat -> app.db.oper.agentinvocation",
+    "app.db.oper.agentchat -> app.domain",
+    "app.db.oper.agentchat -> app.domain.chat",
     "app.db.oper.agentchat -> app.schemas",
     "app.db.oper.agentchat -> app.schemas.types",
     "app.db.oper.agentinvocation -> app.db",
@@ -10862,7 +10867,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 1144,
+  "module_count": 1145,
   "modules": [
     "app",
     "app.adapters",
@@ -11539,6 +11544,7 @@
     "app.doctor.runner",
     "app.domain",
     "app.domain.artwork",
+    "app.domain.chat",
     "app.domain.classification",
     "app.domain.classification.conditions",
     "app.domain.classification.evaluator",

tests/test_agent_session_consistency.py

@@ -0,0 +1,306 @@
+"""
+选择回执、执行状态和展示快照的一致性回归。
+"""
+
+import asyncio
+from datetime import datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, Mock
+from uuid import uuid4
+
+import pytest
+
+from app.agent.session import AgentSessionOwner, _MessageTask
+from app.application.messaging import agent as web_agent
+from app.application.messaging.interaction.agent import (
+    AgentInteractionOption,
+    agent_interaction_manager,
+    build_agent_choice_callback,
+)
+from app.db.oper.agentchat import AgentChatOper
+from app.db.session import SessionFactory
+from app.domain.chat import merge_choice_state
+from app.schemas.types import NotificationChannel
+
+
+@pytest.fixture(autouse=True)
+def clear_choices():
+    """
+    隔离进程内选择收据，不影响其它会话测试。
+    """
+    agent_interaction_manager.clear()
+    yield
+    agent_interaction_manager.clear()
+
+
+@pytest.fixture
+def choice():
+    """
+    在临时数据库中建立带待选卡片的用户会话。
+    """
+    session_id = f"web-agent:consistency-{uuid4().hex}"
+    user = SimpleNamespace(id=501, name="choice-user", is_superuser=False)
+    request = agent_interaction_manager.create_request(
+        session_id=session_id, user_id=str(user.id), channel=NotificationChannel.WebAgent.value,
+        source="web-agent", username=user.name, title="请选择", prompt="选择分类",
+        options=[AgentInteractionOption("电影", "选择电影"), AgentInteractionOption("电视剧", "选择电视剧")],
+    )
+    messages = [{"id": "assistant-choice", "role": "assistant", "content": "", "choices": [{
+        "id": request.request_id, "status": "pending", "prompt": request.prompt,
+    }]}]
+    with SessionFactory() as db:
+        AgentChatOper(db).save_display_messages(session_id=session_id, user_id=str(user.id), messages=messages)
+        db.commit()
+    return request, user, messages
+
+
+@pytest.mark.asyncio
+async def test_choice_callback_persists_first_selection_and_replays_without_changing_option(choice):
+    """
+    回调返回前已落库；重复点击其它按钮仍返回首次选择且不新增回显。
+    """
+    request, user, _ = choice
+    first = await web_agent.resolve_and_save_web_agent_choice(build_agent_choice_callback(request.request_id, 1), user)
+    repeated = await web_agent.resolve_and_save_web_agent_choice(build_agent_choice_callback(request.request_id, 2), user)
+    assert first["already_selected"] is False
+    assert repeated["already_selected"] is False
+    assert repeated["message"] == first["message"] == "选择电影"
+    assert repeated["feedback"] == first["feedback"]
+    assert agent_interaction_manager.claim_web_choice(request.request_id, str(user.id), request.session_id) is not None
+    executed = await web_agent.resolve_and_save_web_agent_choice(build_agent_choice_callback(request.request_id, 2), user)
+    assert executed["already_selected"] is True and executed["message"] == first["message"]
+    with SessionFactory() as db:
+        row = AgentChatOper(db).get(request.session_id)
+        assert row.display_messages[0]["choices"][0]["status"] == "selected"
+        assert row.display_messages[0]["choices"][0]["selected_value"] == "选择电影"
+        receipts = [message for message in row.display_messages if message.get("choice_selection")]
+        assert len(receipts) == 1 and receipts[0]["content"] == ""
+
+
+@pytest.mark.asyncio
+async def test_choice_rejects_other_user_and_reassigned_session_without_consuming(choice):
+    """
+    选择收据与持久会话授权同时校验，越权不能读出或消费原选项。
+    """
+    request, user, _ = choice
+    callback = build_agent_choice_callback(request.request_id, 1)
+    assert await web_agent.resolve_and_save_web_agent_choice(callback, SimpleNamespace(id=502)) is None
+    service = SimpleNamespace(get=AsyncMock(return_value=SimpleNamespace(user_id="502", username="other")))
+    assert await web_agent.resolve_and_save_web_agent_choice(callback, user, service=service) is None
+    assert request.selected_index is None
+
+
+@pytest.mark.asyncio
+async def test_choice_write_failure_does_not_claim_execution_and_can_retry(choice, monkeypatch):
+    """
+    持久化失败原样报错，Agent 不会领取未落库的选择。
+    """
+    request, user, _ = choice
+    callback = build_agent_choice_callback(request.request_id, 1)
+    with monkeypatch.context() as patcher:
+        patcher.setattr(web_agent, "save_web_agent_display_snapshot", AsyncMock(side_effect=RuntimeError("write failed")))
+        with pytest.raises(RuntimeError, match="write failed"):
+            await web_agent.resolve_and_save_web_agent_choice(callback, user)
+    assert agent_interaction_manager.claim_web_choice(request.request_id, str(user.id), request.session_id) is None
+    result = await web_agent.resolve_and_save_web_agent_choice(callback, user)
+    assert result["already_selected"] is False
+
+
+@pytest.mark.asyncio
+async def test_choice_stream_claim_is_once_only_and_uses_server_selection(choice, monkeypatch):
+    """
+    即使重复直接提交流请求，选择业务也只准入一次，正文采用服务端原选项。
+    """
+    request, user, _ = choice
+    resolved = await web_agent.resolve_and_save_web_agent_choice(build_agent_choice_callback(request.request_id, 1), user)
+    manager = SimpleNamespace(submit_steering_message=AsyncMock(return_value=None), process_message=AsyncMock(return_value="完成"))
+    monkeypatch.setattr(web_agent.agent_application, "get_running_agent_manager", lambda: manager)
+    monkeypatch.setattr(web_agent.agent_application, "get_web_agent_type", lambda: object)
+    monkeypatch.setattr(web_agent, "get_api_runtime_config_snapshot", lambda: SimpleNamespace(ai_agent_enable=True))
+    monkeypatch.setattr(web_agent, "bind_web_agent_user_session", Mock())
+    save = AsyncMock()
+    monkeypatch.setattr(web_agent, "save_web_agent_display_snapshot", save)
+    command = web_agent.WebAgentStreamCommand(
+        text="客户端改写正文", session_id=request.session_id, echo_user=False,
+        choice_selection=resolved["choice_selection"],
+    )
+    first = await web_agent.build_web_agent_stream(command, current_user=user,
+        is_disconnected=AsyncMock(return_value=False), protected_transport_supported=False)
+    assert [event["type"] async for event in first.events] == ["start", "done"]
+    second = await web_agent.build_web_agent_stream(command, current_user=user,
+        is_disconnected=AsyncMock(return_value=False), protected_transport_supported=False)
+    assert "error" in [event["type"] async for event in second.events]
+    await web_agent.wait_web_agent_background_tasks()
+    manager.process_message.assert_awaited_once()
+    assert manager.process_message.await_args.kwargs["message"] == "选择电影"
+    receipt = save.await_args.kwargs["messages"][0]
+    assert receipt["content"] == "" and receipt["choice_selection"]["selected_value"] == "选择电影"
+    assert agent_interaction_manager.claim_web_choice(request.request_id, str(user.id), "another-session") is None
+
+
+def test_selected_card_survives_old_pending_snapshot_without_mutating_inputs(choice):
+    """
+    旧客户端替换快照时保留首次已选结果，输入对象保持不变。
+    """
+    request, _, original = choice
+    selected = {"choice_id": request.request_id, "selected_label": "电影", "selected_value": "选择电影"}
+    previous = merge_choice_state([], [*original, {"id": "receipt", "choice_selection": selected}])
+    before_card = merge_choice_state(previous, [])
+    assert before_card == [{
+        "id": f"choice-selection-{request.request_id}", "role": "user", "content": "", "choice_selection": selected,
+    }]
+    result = merge_choice_state(before_card, original)
+    assert result[0]["choices"][0]["status"] == "selected"
+    assert original[0]["choices"][0]["status"] == "pending"
+    assert merge_choice_state(result, [{"choice_selection": selected}])[0]["choice_selection"]["choice_id"] == request.request_id
+
+
+@pytest.mark.asyncio
+async def test_idle_worker_is_not_publicly_processing_but_still_protects_lifecycle():
+    """
+    空闲 worker 不显示执行中，仍阻止生命周期提前释放会话。
+    """
+    owner = AgentSessionOwner(memory=Mock())
+    session_id = "idle-status"
+    worker = asyncio.create_task(asyncio.Event().wait())
+    owner._session_workers[session_id] = worker
+    owner._session_last_used[session_id] = ("1", datetime.now() - timedelta(days=2))
+    queue = asyncio.Queue()
+    owner._session_queues[session_id] = queue
+    task = _MessageTask(session_id=session_id, user_id="1", message="工作")
+    try:
+        assert owner.get_session_status(session_id)["is_processing"] is False
+        assert owner.is_session_busy(session_id) is False
+        assert owner._is_session_busy(session_id) is True
+        assert owner._expired_idle_sessions() == []
+        queue.put_nowait(task)
+        assert owner.is_session_busy(session_id) is True
+        queue.get_nowait()
+        owner._session_active_tasks[session_id] = task
+        assert owner.get_session_status(session_id)["is_processing"] is True
+        owner._session_active_tasks.clear()
+        assert owner.is_session_busy(session_id) is False
+    finally:
+        worker.cancel()
+        await asyncio.gather(worker, return_exceptions=True)
+
+
+def test_legacy_tool_hint_does_not_complete_structured_parallel_tool():
+    """
+    旧文本提示不会提前收口仍在运行的结构化工具。
+    """
+    message = web_agent.build_web_agent_display_message(role="assistant", status="streaming")
+    web_agent.apply_web_agent_display_event({"type": "tool", "tool_id": "active", "status": "running"}, message)
+    web_agent.apply_web_agent_display_event({"type": "tool", "message": "旧提示"}, message)
+    assert message["tools"][0]["status"] == "running"
+    web_agent.apply_web_agent_display_event({"type": "done"}, message)
+    assert [tool["status"] for tool in message["tools"]] == ["error", "done"]
+
+
+def test_database_replacement_cannot_restore_selected_card_to_pending(choice):
+    """
+    持久化旧完整快照后重新读取，已选卡片仍保留首次选择。
+    """
+    request, user, original = choice
+    with SessionFactory() as db:
+        oper = AgentChatOper(db)
+        oper.append_display_messages(session_id=request.session_id, user_id=str(user.id), messages=[{
+            "id": "receipt", "choice_selection": {
+                "choice_id": request.request_id, "selected_label": "电影", "selected_value": "选择电影",
+            },
+        }])
+        db.commit()
+    with SessionFactory() as db:
+        oper = AgentChatOper(db)
+        oper.save_display_messages(session_id=request.session_id, user_id=str(user.id), messages=[])
+        db.commit()
+    with SessionFactory() as db:
+        oper = AgentChatOper(db)
+        oper.save_display_messages(session_id=request.session_id, user_id=str(user.id), messages=original)
+        db.commit()
+    with SessionFactory() as db:
+        row = AgentChatOper(db).get(request.session_id)
+        assert row.display_messages[0]["choices"][0]["status"] == "selected"
+        assert row.display_messages[0]["choices"][0]["selected_value"] == "选择电影"
+
+
+@pytest.mark.asyncio
+async def test_concurrent_choice_write_failure_keeps_first_option_and_successful_receipt(choice, monkeypatch):
+    """
+    并发回调中一笔写失败不重置另一笔在途选择，首次选项保持不变。
+    """
+    request, user, _ = choice
+    entered, release = asyncio.Event(), asyncio.Event()
+    count = 0
+
+    async def save(**_kwargs):
+        """
+        第一笔写入等待后失败，第二笔确认成功。
+        """
+        nonlocal count
+        count += 1
+        if count == 1:
+            entered.set()
+            await release.wait()
+            raise RuntimeError("first write failed")
+
+    monkeypatch.setattr(web_agent, "save_web_agent_display_snapshot", save)
+    first = asyncio.create_task(web_agent.resolve_and_save_web_agent_choice(build_agent_choice_callback(request.request_id, 1), user))
+    await entered.wait()
+    second = await web_agent.resolve_and_save_web_agent_choice(build_agent_choice_callback(request.request_id, 2), user)
+    release.set()
+    with pytest.raises(RuntimeError, match="first write failed"):
+        await first
+    assert second["message"] == "选择电影" and second["already_selected"] is False
+    assert request.selected_index == 1 and request.selection_persisted is True
+    assert agent_interaction_manager.claim_web_choice(request.request_id, str(user.id), request.session_id) is not None
+
+
+@pytest.mark.asyncio
+async def test_stream_construction_failure_does_not_consume_choice_execution(choice, monkeypatch):
+    """
+    回调后断网或流构建失败，原选择仍可提交一次 canonical 执行。
+    """
+    from app.application.messaging.webagent import stream as stream_module
+
+    request, user, _ = choice
+    callback = build_agent_choice_callback(request.request_id, 1)
+    result = await web_agent.resolve_and_save_web_agent_choice(callback, user)
+    repeated = await web_agent.resolve_and_save_web_agent_choice(callback, user)
+    assert repeated["already_selected"] is False
+    monkeypatch.setattr(web_agent.agent_application, "get_running_agent_manager", lambda: SimpleNamespace())
+    monkeypatch.setattr(web_agent, "get_api_runtime_config_snapshot", lambda: SimpleNamespace(ai_agent_enable=True))
+    monkeypatch.setattr(stream_module, "build_agent_web_agent_stream", Mock(side_effect=RuntimeError("stream setup failed")))
+    command = web_agent.WebAgentStreamCommand(text=result["message"], session_id=request.session_id,
+                                             choice_selection=result["choice_selection"], echo_user=False)
+    with pytest.raises(RuntimeError, match="stream setup failed"):
+        await web_agent.build_web_agent_stream(command, current_user=user,
+            is_disconnected=AsyncMock(return_value=False), protected_transport_supported=False)
+    assert request.execution_claimed is False
+    assert agent_interaction_manager.claim_web_choice(request.request_id, str(user.id), request.session_id) is not None
+
+
+@pytest.mark.asyncio
+async def test_explicit_agent_admission_rejection_releases_choice_for_retry(choice, monkeypatch):
+    """
+    Agent 入队前明确拒绝时归还选择，未知执行失败保留领取防止重复业务。
+    """
+    from app.agent.session import AgentManagerQueueFullError
+
+    request, user, _ = choice
+    result = await web_agent.resolve_and_save_web_agent_choice(build_agent_choice_callback(request.request_id, 1), user)
+    manager = SimpleNamespace(submit_steering_message=AsyncMock(return_value=None),
+                              process_message=AsyncMock(side_effect=AgentManagerQueueFullError(request.session_id, 8)))
+    monkeypatch.setattr(web_agent.agent_application, "get_running_agent_manager", lambda: manager)
+    monkeypatch.setattr(web_agent.agent_application, "get_web_agent_type", lambda: object)
+    monkeypatch.setattr(web_agent, "get_api_runtime_config_snapshot", lambda: SimpleNamespace(ai_agent_enable=True))
+    monkeypatch.setattr(web_agent, "bind_web_agent_user_session", Mock())
+    command = web_agent.WebAgentStreamCommand(text=result["message"], session_id=request.session_id,
+                                             choice_selection=result["choice_selection"], echo_user=False)
+    stream = await web_agent.build_web_agent_stream(command, current_user=user,
+        is_disconnected=AsyncMock(return_value=False), protected_transport_supported=False)
+    assert "error" in [event["type"] async for event in stream.events]
+    await web_agent.wait_web_agent_background_tasks()
+    assert request.execution_claimed is False
+    retry = await web_agent.resolve_and_save_web_agent_choice(build_agent_choice_callback(request.request_id, 2), user)
+    assert retry["already_selected"] is False and retry["message"] == "选择电影"

tests/test_subscription_governance_scale.py

@@ -252,14 +252,31 @@ def test_scale_validator_rejects_fake_site_pressure(monkeypatch, tmp_path, mutat
     assert result["site_request_boundary_peak"] >= 2
 
 
-def test_scale_validator_rejects_unfinished_site_wrapper(monkeypatch, tmp_path):
+@pytest.mark.parametrize("request_start_delay", [0.0, 0.1])
+def test_scale_validator_rejects_unfinished_site_wrapper(
+    monkeypatch, tmp_path, request_start_delay,
+):
     """
     延后已登记但 wrapper 尚未返回时，压力门禁必须拒绝并暴露未收口 owner。
     """
     original_wrapper = SearchChain._search_site_torrents_with_budget
     release_stalled = threading.Event()
+    rejection_stalled = threading.Event()
     threads_before = set(threading.enumerate())
 
+    class DelayedRequestBoundary(scale._ScaleSiteRequestBoundary):
+        """
+        模拟慢速 CI 在首个请求进入前的调度延迟，不能提前耗尽协调预算。
+        """
+
+        def request(self, *, owner):
+            """
+            只延迟首个请求，延迟超过故障收口的 50 毫秒等待。
+            """
+            if request_start_delay and not self.first_started.is_set():
+                threading.Event().wait(request_start_delay)
+            return super().request(owner=owner)
+
     def stall_after_rejection(self, *, site, keyword, mtype, page):
         """
         模拟预算拒绝已记录、调用方却未取得返回值的挂起路径。
@@ -270,6 +287,7 @@ def test_scale_validator_rejects_unfinished_site_wrapper(monkeypatch, tmp_path):
             """
             在拒绝已登记后阻塞原始 wrapper 的返回。
             """
+            rejection_stalled.set()
             original_record_deferred(deferral)
             release_stalled.wait()
 
@@ -290,7 +308,8 @@ def test_scale_validator_rejects_unfinished_site_wrapper(monkeypatch, tmp_path):
         "_search_site_torrents_with_budget",
         stall_after_rejection,
     )
-    monkeypatch.setattr(scale, "_SITE_PRESSURE_SYNC_TIMEOUT", 0.05)
+    monkeypatch.setattr(scale, "_ScaleSiteRequestBoundary", DelayedRequestBoundary)
+    monkeypatch.setattr(scale, "_SITE_PRESSURE_JOIN_TIMEOUT", 0.05)
 
     try:
         result = _run_durable_governance(
@@ -298,6 +317,8 @@ def test_scale_validator_rejects_unfinished_site_wrapper(monkeypatch, tmp_path):
             tmp_path,
         )
 
+        assert rejection_stalled.is_set(), "必须实际进入拒绝后的挂起路径"
+        assert result["site_observations"][0]["request_active_at_rejection"] == 1
         assert result["site_pressure_valid"] is False
         assert result["site_pressure_concurrency_verified"] is False
         assert result["site_observations"][0]["owners_finished"] is False

tests/test_web_agent_stream.py

@@ -3,6 +3,7 @@
 """
 
 import asyncio
+import json
 import time
 from queue import Queue
 from threading import Event as ThreadEvent
@@ -1563,7 +1564,8 @@ def test_web_agent_stream_binds_session_to_agent_manager():
             worker.cancel()
 
 
-def test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream():
+@pytest.mark.parametrize("finish_old_tool", [False, True])
+def test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream(finish_old_tool):
     """
     第一条 WebAgent 流运行时的第二条请求应在真实应用点切分助手回合。
     """
@@ -1639,17 +1641,13 @@ def test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream():
                             }
                         )
                     self.output_callback("首轮处理中")
-                    if callable(self.tool_event_callback):
-                        self.tool_event_callback(
-                            {
-                                "type": "tool",
-                                "status": "done",
-                                "tool_id": "tool-steering-boundary",
-                            }
-                        )
                     started.set()
                     await release.wait()
                 else:
+                    if finish_old_tool and callable(self.tool_event_callback):
+                        self.tool_event_callback({
+                            "type": "tool", "status": "done", "tool_id": "tool-steering-boundary",
+                        })
                     self.output_callback("补充已应用")
                 return message
 
@@ -1749,9 +1747,13 @@ def test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream():
             "id": "tool-steering-boundary",
             "tool_name": "search",
             "message": "检查任务状态",
-            "status": "done",
+            "status": "done" if finish_old_tool else "error",
         }
     ]
+    assert messages[3]["tools"] == []
+    tool_events = [json.loads(line[6:]) for line in first_body.splitlines() if line.startswith("data: ")]
+    tool_events = [event for event in tool_events if event.get("tool_id") == "tool-steering-boundary"]
+    assert all(event["assistant_message_id"] == messages[1]["id"] for event in tool_events)
 
 
 def test_web_agent_stream_emits_secret_result_only_as_protected_event():