diff --git a/app/agent/callback/__init__.py b/app/agent/callback/__init__.py index 7363b9d3f..698453b71 100644 --- a/app/agent/callback/__init__.py +++ b/app/agent/callback/__init__.py @@ -149,6 +149,19 @@ class StreamingHandler: ) return "" + def tool_call_started( + self, + tool_name: str, + tool_message: Optional[str] = None, + ) -> str: + """为支持结构化生命周期的宿主预留工具调用 ID。""" + del tool_name, tool_message + return "" + + def tool_call_finished(self, tool_id: str, status: str = "done") -> None: + """收口真实工具执行；普通通知渠道无需额外输出生命周期事件。""" + del tool_id, status + async def take(self) -> str: """ 获取当前已积累的消息内容，获取后清空缓冲区。 diff --git a/app/agent/middleware/activity.py b/app/agent/middleware/activity.py index b47b202cf..da3fbde9f 100644 --- a/app/agent/middleware/activity.py +++ b/app/agent/middleware/activity.py @@ -722,9 +722,10 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response f"开始执行活动日志查询工具: keyword={logged_args.get('keyword') or '-'}, " f"date={logged_args.get('date') or '-'}" ) + tool_call_id = "" if self.stream_handler and getattr(self.stream_handler, "is_streaming", False): display_args = json.dumps(logged_args, ensure_ascii=False, default=str) - self.stream_handler.report_tool_call( + tool_call_id = self.stream_handler.report_tool_call( tool_name=QUERY_ACTIVITY_LOG_TOOL_NAME, tool_message=f"查询活动日志，主要参数：{display_args}", tool_kwargs=tool_args, @@ -732,10 +733,18 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response try: result = await handler(request) except Exception as err: + if tool_call_id: + finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) + if callable(finish_tool_call): + finish_tool_call(tool_call_id, "error") logger.error( f"活动日志查询工具执行失败: error={summarize_error(err)}" ) raise + if tool_call_id: + finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) + if callable(finish_tool_call): + finish_tool_call(tool_call_id, "done") logger.info("活动日志查询工具执行完成") return result diff --git a/app/agent/middleware/skills.py b/app/agent/middleware/skills.py index 576180c30..b8260f193 100644 --- a/app/agent/middleware/skills.py +++ b/app/agent/middleware/skills.py @@ -615,8 +615,9 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]): # no if not isinstance(logged_args, dict): logged_args = {} logger.info(f"开始执行 Skill 工具: name={logged_args.get('name') or '-'}") + tool_call_id = "" if self.stream_handler and getattr(self.stream_handler, "is_streaming", False): - self.stream_handler.report_tool_call( + tool_call_id = self.stream_handler.report_tool_call( tool_name=SKILL_TOOL_NAME, tool_message=f"读取技能说明：{logged_args.get('name') or '-'}", tool_kwargs=tool_args, @@ -624,8 +625,16 @@ class SkillsMiddleware(AgentMiddleware[SkillsState, ContextT, ResponseT]): # no try: result = await handler(request) except Exception as err: + if tool_call_id: + finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) + if callable(finish_tool_call): + finish_tool_call(tool_call_id, "error") logger.error(f"Skill 工具执行失败: error={summarize_error(err)}") raise + if tool_call_id: + finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) + if callable(finish_tool_call): + finish_tool_call(tool_call_id, "done") logger.info("Skill 工具执行完成") return result diff --git a/app/agent/middleware/subagents.py b/app/agent/middleware/subagents.py index 69b8ee754..afa3c05ed 100644 --- a/app/agent/middleware/subagents.py +++ b/app/agent/middleware/subagents.py @@ -404,10 +404,10 @@ def _record_subagent_tool_call( stream_handler: Any, tool_name: str, tool_args: dict[str, Any], -) -> None: +) -> str: """按当前详细模式展示或汇总子代理工具调用。""" if not stream_handler or not getattr(stream_handler, "is_streaming", False): - return + return "" safe_args = sanitize_for_host(tool_args) if not isinstance(safe_args, dict): safe_args = {} @@ -415,11 +415,11 @@ def _record_subagent_tool_call( tool_message = "调用子代理：general-purpose" else: tool_message = f"管理子代理任务：action={safe_args.get('action') or 'start'}" - stream_handler.report_tool_call( + return str(stream_handler.report_tool_call( tool_name=tool_name, tool_message=tool_message, tool_kwargs=tool_args, - ) + ) or "") class _SubAgentAgentProvider: @@ -618,7 +618,7 @@ class MoviePilotSubAgentMiddleware(AgentMiddleware): f"开始执行子代理工具: tool_name={tool_name}, " "subagent_type=general-purpose" ) - _record_subagent_tool_call( + tool_call_id = _record_subagent_tool_call( stream_handler=self.stream_handler, tool_name=SUBAGENT_TASK_TOOL_NAME, tool_args=tool_args, @@ -626,11 +626,19 @@ class MoviePilotSubAgentMiddleware(AgentMiddleware): try: result = await handler(request) except Exception as err: + if tool_call_id: + finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) + if callable(finish_tool_call): + finish_tool_call(tool_call_id, "error") logger.error( f"子代理工具执行失败: tool_name={tool_name}, " f"error={summarize_error(err)}" ) raise + if tool_call_id: + finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) + if callable(finish_tool_call): + finish_tool_call(tool_call_id, "done") logger.info(f"子代理工具执行完成: tool_name={tool_name}") return result @@ -1285,7 +1293,7 @@ class SubAgentTaskControlMiddleware(AgentMiddleware): f"action={logged_args.get('action') or '-'}, " "subagent_type=general-purpose" ) - _record_subagent_tool_call( + tool_call_id = _record_subagent_tool_call( stream_handler=self.stream_handler, tool_name=SUBAGENT_CONTROL_TOOL_NAME, tool_args=tool_args, @@ -1293,11 +1301,19 @@ class SubAgentTaskControlMiddleware(AgentMiddleware): try: result = await handler(request) except Exception as err: + if tool_call_id: + finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) + if callable(finish_tool_call): + finish_tool_call(tool_call_id, "error") logger.error( f"子代理工具执行失败: tool_name={tool_name}, " f"error={summarize_error(err)}" ) raise + if tool_call_id: + finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) + if callable(finish_tool_call): + finish_tool_call(tool_call_id, "done") logger.info(f"子代理工具执行完成: tool_name={tool_name}") return result diff --git a/app/agent/session.py b/app/agent/session.py index 3ebf3e943..952a34174 100644 --- a/app/agent/session.py +++ b/app/agent/session.py @@ -108,6 +108,7 @@ class _MessageTask: reply_mode: ReplyMode = ReplyMode.DISPATCH allow_message_tools: bool = True output_callback: Optional[Callable[[str], None]] = None + tool_event_callback: Optional[Callable[[dict[str, Any]], None]] = None protected_output_callback: Optional[Callable[[str], Optional[bool]]] = None message_callback: Optional[Callable[[Any], Awaitable[None] | None]] = None agent_factory: Optional[Callable[..., MoviePilotAgent]] = None @@ -299,6 +300,7 @@ class AgentSessionOwner: reply_mode: ReplyMode = ReplyMode.DISPATCH, allow_message_tools: bool = True, output_callback: Optional[Callable[[str], None]] = None, + tool_event_callback: Optional[Callable[[dict[str, Any]], None]] = None, protected_output_callback: Optional[Callable[[str], Optional[bool]]] = None, message_callback: Optional[Callable[[Any], Awaitable[None] | None]] = None, steering_status_callback: Optional[SteeringStatusCallback] = None, @@ -330,6 +332,7 @@ class AgentSessionOwner: reply_mode=reply_mode, allow_message_tools=allow_message_tools, output_callback=output_callback, + tool_event_callback=tool_event_callback, protected_output_callback=protected_output_callback, message_callback=message_callback, steering_status_callback=steering_status_callback, @@ -649,6 +652,8 @@ class AgentSessionOwner: "output_callback": task.output_callback, "protected_output_callback": task.protected_output_callback, } + if task.tool_event_callback is not None: + agent_kwargs["tool_event_callback"] = task.tool_event_callback if task.message_callback is not None: agent_kwargs["message_callback"] = task.message_callback agent = task.agent_factory(**agent_kwargs) @@ -669,6 +674,8 @@ class AgentSessionOwner: agent.set_output_callback(task.output_callback) else: agent.output_callback = task.output_callback + if hasattr(agent, "set_tool_event_callback"): + agent.set_tool_event_callback(task.tool_event_callback) agent.set_protected_output_callback(task.protected_output_callback) if task.message_callback is not None and hasattr(agent, "set_message_callback"): agent.set_message_callback(task.message_callback) diff --git a/app/agent/tools/base.py b/app/agent/tools/base.py index 2cca21bb2..b6730a6ad 100644 --- a/app/agent/tools/base.py +++ b/app/agent/tools/base.py @@ -68,6 +68,18 @@ else: """按详细模式逐条展示工具调用，或登记为延迟汇总。""" ... + def tool_call_started( + self, + tool_name: str, + tool_message: Optional[str] = None, + ) -> str: + """登记真实工具开始执行并返回展示调用 ID。""" + ... + + def tool_call_finished(self, tool_id: str, status: str = "done") -> None: + """登记真实工具执行结束。""" + ... + async def take(self) -> str: """取出并清空当前缓冲内容。""" ... @@ -463,9 +475,13 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta): # 获取工具执行提示消息 tool_message = self.get_tool_message(**kwargs) + tool_call_id = "" # 发送工具执行过程消息（流式传输且非最后终结工具时） if self._stream_handler and self._stream_handler.is_streaming and not self.return_direct: + start_tool_call = getattr(self._stream_handler, "tool_call_started", None) + if callable(start_tool_call): + tool_call_id = str(start_tool_call(self.name, tool_message) or "") if get_runtime_setting('AI_AGENT_VERBOSE'): if self._stream_handler.is_auto_flushing: # 渠道支持编辑：工具消息追加到 buffer，由定时刷新推送 @@ -528,11 +544,23 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta): f"结果摘要: {summarize_result(formatted_result)}" ) + if tool_call_id: + finish_tool_call = getattr(self._stream_handler, "tool_call_finished", None) + if callable(finish_tool_call): + finish_tool_call(tool_call_id, "done") except ToolExecutionTimeoutError as e: + if tool_call_id: + finish_tool_call = getattr(self._stream_handler, "tool_call_finished", None) + if callable(finish_tool_call): + finish_tool_call(tool_call_id, "error") error_message = summarize_error(e) logger.warning(error_message) raise except Exception as e: + if tool_call_id: + finish_tool_call = getattr(self._stream_handler, "tool_call_finished", None) + if callable(finish_tool_call): + finish_tool_call(tool_call_id, "error") error_message = f"工具执行异常（{stable_type_name(e)}），请检查参数或查询当前状态后继续处理。" logger.error(f"Tool {self.name} execution failed: {summarize_error(e)}") raise ToolExecutionError(error_message) from e diff --git a/app/agent/web.py b/app/agent/web.py index 7d5b7cf35..e668c4c66 100644 --- a/app/agent/web.py +++ b/app/agent/web.py @@ -1,5 +1,6 @@ """WebAgent 运行时类型适配，不包含 HTTP 或 SSE 编码。""" +import uuid from threading import Lock from typing import Any, Awaitable, Callable, Optional @@ -17,10 +18,15 @@ class _WebAgentStreamingHandlerMixin: _lock: Lock _pending_tool_stats: dict[str, int] - def __init__(self, on_emit: Callable[[str], None]) -> None: + def __init__( + self, + on_emit: Callable[[str], None], + on_tool_event: Optional[Callable[[dict[str, Any]], None]] = None, + ) -> None: """绑定当前 Web 请求的输出回调。""" super().__init__() self._on_emit = on_emit + self._on_tool_event = on_tool_event def set_emit_callback(self, on_emit: Callable[[str], None]) -> None: """ @@ -30,13 +36,80 @@ class _WebAgentStreamingHandlerMixin: """ self._on_emit = on_emit + def set_tool_event_callback( + self, + on_tool_event: Optional[Callable[[dict[str, Any]], None]], + ) -> None: + """更新当前 Web 请求的结构化工具生命周期回调。""" + self._on_tool_event = on_tool_event + + def _publish_tool_event(self, event: dict[str, Any]) -> None: + """向 Web SSE 发布结构化工具生命周期事件，并隔离回调异常。""" + if not self._on_tool_event: + return + try: + self._on_tool_event(event) + except Exception as error: + logger.debug(f"Web工具生命周期回调失败: {error}") + + def tool_call_started( + self, + tool_name: str, + tool_message: Optional[str] = None, + ) -> str: + """发布真实工具开始事件，调用方随后用同一 ID 收口结果。""" + if not self._on_tool_event: + return "" + tool_id = f"tool-{uuid.uuid4().hex}" + self._publish_tool_event( + { + "type": "tool", + "status": "running", + "tool_id": tool_id, + "tool_name": str(tool_name or ""), + "message": " ".join(str(tool_message or tool_name).splitlines()), + } + ) + return tool_id + + def tool_call_finished(self, tool_id: str, status: str = "done") -> None: + """发布真实工具完成或失败事件，保持与开始事件的 ID 对应。""" + if not tool_id or not self._on_tool_event: + return + normalized_status = status if status in {"done", "error"} else "done" + self._publish_tool_event( + { + "type": "tool", + "status": normalized_status, + "tool_id": tool_id, + } + ) + + def report_tool_call( + self, + tool_name: str, + tool_message: Optional[str] = None, + tool_kwargs: Optional[dict[str, Any]] = None, + ) -> str: + """为没有 MoviePilotTool 包装层的内部工具建立完整生命周期。""" + if self._on_tool_event: + del tool_kwargs + return self.tool_call_started(tool_name, tool_message) + return str(super().report_tool_call( # type: ignore[misc] + tool_name=tool_name, + tool_message=tool_message, + tool_kwargs=tool_kwargs, + ) or "") + def record_tool_call( self, tool_name: str, tool_message: Optional[str] = None, tool_kwargs: Optional[dict[str, Any]] = None, ) -> None: - """记录并立即输出 Web 工具事件，避免汇总延迟到正文结束后。""" + """记录 Web 工具调用；结构化生命周期模式下不再生成重复文本摘要。""" + if self._on_tool_event: + return # 该方法由运行时组合进 MRO 的 StreamingHandler 提供。 super().record_tool_call( # type: ignore[misc] tool_name=tool_name, @@ -51,6 +124,8 @@ class _WebAgentStreamingHandlerMixin: 仅处理展示文案；实际工具参数保持原样，由前端在工具提示条内自动折行。 """ normalized_message = " ".join(str(message or "").splitlines()) + if self._on_tool_event: + return "" return str(super().emit_tool_message(normalized_message)) # type: ignore[misc] def emit(self, token: str) -> str: @@ -63,6 +138,8 @@ class _WebAgentStreamingHandlerMixin: def flush_pending_tool_summary(self) -> str: """输出延迟聚合的工具摘要。""" + if self._on_tool_event: + return "" # 该方法由运行时组合进 MRO 的 StreamingHandler 提供。 emitted = super().flush_pending_tool_summary() # type: ignore[misc] if emitted: @@ -150,9 +227,13 @@ class _WebAgentMoviePilotAgentMixin: message_callback: Optional[Callable[[Message], Awaitable[None] | None]] = None, **kwargs: Any, ) -> None: + tool_event_callback = kwargs.pop("tool_event_callback", None) super().__init__(*args, **kwargs) self._message_callback = message_callback - self.stream_handler = _get_web_agent_streaming_handler_type()(self._emit_output) + self.stream_handler = _get_web_agent_streaming_handler_type()( + self._emit_output, + tool_event_callback, + ) def _should_stream(self) -> bool: """Web 对话实时输出，复用会话执行后台任务时改用非流式广播。""" @@ -181,6 +262,14 @@ class _WebAgentMoviePilotAgentMixin: if output_callback and isinstance(self.stream_handler, _WebAgentStreamingHandlerMixin): self.stream_handler.set_emit_callback(self._emit_output) + def set_tool_event_callback( + self, + tool_event_callback: Optional[Callable[[dict[str, Any]], None]], + ) -> None: + """更新当前 Web SSE 结构化工具生命周期回调。""" + if isinstance(self.stream_handler, _WebAgentStreamingHandlerMixin): + self.stream_handler.set_tool_event_callback(tool_event_callback) + async def _is_system_admin_context(self) -> bool: """Web Agent 根据当前登录用户 ID 判断工具管理员上下文。""" if not self.user_id: diff --git a/app/application/messaging/agent.py b/app/application/messaging/agent.py index 600968661..d2ed28687 100644 --- a/app/application/messaging/agent.py +++ b/app/application/messaging/agent.py @@ -905,8 +905,36 @@ def apply_web_agent_display_event(event: dict[str, Any], assistant_message: dict if event_type == "delta": append_web_agent_text_segment(assistant_message, event.get("content") or "") elif event_type == "tool": + tool_id = str(event.get("tool_id") or event.get("tool_call_id") or "") + tool_status = str(event.get("status") or "running") + if tool_id and tool_status in {"running", "done", "error"}: + matching_tool = next( + (tool for tool in assistant_message["tools"] if str(tool.get("id") or "") == tool_id), + None, + ) + if tool_status == "running": + if matching_tool is None: + tool_index = len(assistant_message["tools"]) + assistant_message["tools"].append( + { + "id": tool_id, + "tool_name": str(event.get("tool_name") or ""), + "message": str(event.get("message") or "").strip(), + "status": "running", + } + ) + assistant_message.setdefault("segments", []).append( + {"type": "tool", "toolIndex": tool_index} + ) + return + if matching_tool is not None: + matching_tool["status"] = tool_status + return + + # 兼容尚未升级的事件生产者：没有调用 ID 时只能把前一批提示视为已结束。 for tool in assistant_message["tools"]: - tool["status"] = "done" + if tool.get("status") == "running": + tool["status"] = "done" tool_index = len(assistant_message["tools"]) assistant_message["tools"].append( { @@ -942,12 +970,14 @@ def apply_web_agent_display_event(event: dict[str, Any], assistant_message: dict event.get("message") or "智能助手响应失败", ) for tool in assistant_message["tools"]: - tool["status"] = "done" + if tool.get("status") == "running": + tool["status"] = "done" elif event_type == "done": if assistant_message.get("status") != "error": assistant_message["status"] = "done" for tool in assistant_message["tools"]: - tool["status"] = "done" + if tool.get("status") == "running": + tool["status"] = "done" async def save_web_agent_display_snapshot( diff --git a/app/application/messaging/webagentstream.py b/app/application/messaging/webagentstream.py index 41dd1ed97..5627a6073 100644 --- a/app/application/messaging/webagentstream.py +++ b/app/application/messaging/webagentstream.py @@ -92,16 +92,19 @@ def _publish_web_agent_protected_output( def _build_web_agent_steering_callback( *, display_messages: list[dict[str, Any]], + assistant_message_ref: dict[str, dict[str, Any]], event_publisher: Any, build_display_message: Callable[..., dict[str, Any]], build_input_attachments: Callable[..., list[dict[str, Any]]], ) -> Callable[[SteeringMessage, str], None]: - """构造运行中补充消息的应用回调，并同步展示快照与 SSE。""" + """构造运行中补充消息的应用回调，并同步展示快照与 SSE 时间线。""" def steering_status_callback(message: SteeringMessage, status: str) -> None: - """把补充消息应用状态投影到当前助手流，保持单一助手气泡。""" + """把补充消息应用状态投影到当前助手流的真实边界。""" if status != "applied": return + assistant_message = assistant_message_ref["message"] + assistant_message["status"] = "done" display_message = build_display_message( role="user", content=message.text, @@ -112,8 +115,13 @@ def _build_web_agent_steering_callback( ), ) display_message["steering_message_id"] = message.message_id - # 当前助手气泡必须继续保持最后一项，补充的用户消息插入其前面。 - display_messages.insert(max(0, len(display_messages) - 1), display_message) + # 应用点是消息流中的真实边界：先收口前一段助手输出，再插入用户消息， + # 最后创建承接后续工具和文本的新助手气泡。 + assistant_index = display_messages.index(assistant_message) + display_messages.insert(assistant_index + 1, display_message) + continuation = build_display_message(role="assistant", status="streaming") + display_messages.insert(assistant_index + 2, continuation) + assistant_message_ref["message"] = continuation event_publisher.publish( { "type": "steering", @@ -121,6 +129,8 @@ def _build_web_agent_steering_callback( "message_id": message.message_id, "content": message.text, "display_message": display_message, + "assistant_message_id": assistant_message.get("id"), + "continuation_message_id": continuation.get("id"), } ) @@ -129,7 +139,7 @@ def _build_web_agent_steering_callback( def _build_web_agent_message_callback( *, - assistant_display_message: dict[str, Any], + assistant_message_ref: dict[str, dict[str, Any]], event_publisher: Any, build_message_events_async: Callable[[Message], Awaitable[list[dict[str, Any]]]], apply_display_event: Callable[[dict[str, Any], dict[str, Any]], None], @@ -139,7 +149,7 @@ def _build_web_agent_message_callback( async def message_callback(message: Message) -> None: """接收 Agent 工具主动发送的 Web 通知。""" for item in await build_message_events_async(message): - apply_display_event(item, assistant_display_message) + apply_display_event(item, assistant_message_ref["message"]) event_publisher.publish(item) return message_callback @@ -147,7 +157,7 @@ def _build_web_agent_message_callback( def _build_web_agent_output_callback( *, - assistant_display_message: dict[str, Any], + assistant_message_ref: dict[str, dict[str, Any]], event_publisher: Any, split_output: Callable[[str], list[dict[str, Any]]], apply_display_event: Callable[[dict[str, Any], dict[str, Any]], None], @@ -157,12 +167,28 @@ def _build_web_agent_output_callback( def output_callback(delta: str) -> None: """接收 Agent 文本增量并投影为展示事件。""" for item in split_output(delta): - apply_display_event(item, assistant_display_message) + apply_display_event(item, assistant_message_ref["message"]) event_publisher.publish(item) return output_callback +def _build_web_agent_tool_event_callback( + *, + assistant_message_ref: dict[str, dict[str, Any]], + event_publisher: Any, + apply_display_event: Callable[[dict[str, Any], dict[str, Any]], None], +) -> Callable[[dict[str, Any]], None]: + """构造结构化工具生命周期回调，确保快照与 SSE 使用同一事件顺序。""" + + def tool_event_callback(event: dict[str, Any]) -> None: + """投影工具开始、完成和失败事件。""" + apply_display_event(event, assistant_message_ref["message"]) + event_publisher.publish(event) + + return tool_event_callback + + def _build_web_agent_files(command: Any) -> list[dict[str, Any]]: """把 WebAgent 命令中的普通文件和音频引用整理成 Agent 输入。""" audio_refs = {str(audio_ref) for audio_ref in command.audio_refs} @@ -178,7 +204,7 @@ def _build_web_agent_event_generator( session_id: str, prompt: str, display_messages: list[dict[str, Any]], - assistant_display_message: dict[str, Any], + assistant_message_ref: dict[str, dict[str, Any]], has_audio_input: bool, is_secret_confirmation_control: bool, protected_transport_supported: bool, @@ -188,6 +214,7 @@ def _build_web_agent_event_generator( dependencies: WebAgentStreamDependencies, event_publisher: Any, output_callback: Callable[[str], None], + tool_event_callback: Callable[[dict[str, Any]], None], message_callback: Callable[[Message], Awaitable[None]], protected_output_callback: Callable[[str], bool], steering_status_callback: Callable[[SteeringMessage, str], None], @@ -214,6 +241,7 @@ def _build_web_agent_event_generator( reply_mode=ReplyMode.CAPTURE_ONLY, allow_message_tools=True, output_callback=output_callback, + tool_event_callback=tool_event_callback, protected_output_callback=(protected_output_callback if protected_transport_supported else None), message_callback=message_callback, steering_status_callback=steering_status_callback, @@ -229,11 +257,11 @@ def _build_web_agent_event_generator( "type": "error", "message": "智能助手执行失败，请稍后重试", } - dependencies.apply_display_event(error_event, assistant_display_message) + dependencies.apply_display_event(error_event, assistant_message_ref["message"]) event_publisher.publish(error_event) finally: done_event = {"type": "done"} - dependencies.apply_display_event(done_event, assistant_display_message) + dependencies.apply_display_event(done_event, assistant_message_ref["message"]) # 终态先进入事件队列，避免展示快照落库延迟前端结束动画。 event_publisher.publish(done_event) if not is_secret_confirmation_control: @@ -327,21 +355,28 @@ def build_agent_web_agent_stream( status="streaming", ) display_messages.append(assistant_display_message) + assistant_message_ref = {"message": assistant_display_message} output_callback = _build_web_agent_output_callback( - assistant_display_message=assistant_display_message, + assistant_message_ref=assistant_message_ref, event_publisher=event_publisher, split_output=dependencies.split_output, apply_display_event=dependencies.apply_display_event, ) message_callback = _build_web_agent_message_callback( - assistant_display_message=assistant_display_message, + assistant_message_ref=assistant_message_ref, event_publisher=event_publisher, build_message_events_async=dependencies.build_message_events_async, apply_display_event=dependencies.apply_display_event, ) + tool_event_callback = _build_web_agent_tool_event_callback( + assistant_message_ref=assistant_message_ref, + event_publisher=event_publisher, + apply_display_event=dependencies.apply_display_event, + ) protected_output_callback = partial(_publish_web_agent_protected_output, event_publisher) steering_status_callback = _build_web_agent_steering_callback( display_messages=display_messages, + assistant_message_ref=assistant_message_ref, event_publisher=event_publisher, build_display_message=dependencies.build_display_message, build_input_attachments=dependencies.build_input_attachments, @@ -353,7 +388,7 @@ def build_agent_web_agent_stream( session_id=session_id, prompt=prompt, display_messages=display_messages, - assistant_display_message=assistant_display_message, + assistant_message_ref=assistant_message_ref, has_audio_input=has_audio_input, is_secret_confirmation_control=is_secret_confirmation_control, protected_transport_supported=protected_transport_supported, @@ -363,6 +398,7 @@ def build_agent_web_agent_stream( dependencies=dependencies, event_publisher=event_publisher, output_callback=output_callback, + tool_event_callback=tool_event_callback, message_callback=message_callback, protected_output_callback=protected_output_callback, steering_status_callback=steering_status_callback, diff --git a/app/schemas/agent.py b/app/schemas/agent.py index 65158a7fe..55e438fe5 100644 --- a/app/schemas/agent.py +++ b/app/schemas/agent.py @@ -127,6 +127,7 @@ class AgentChatToolCall(BaseModel): """ id: str = Field(..., description="展示 ID") + tool_name: Optional[str] = Field(None, description="工具名称") message: str = Field(..., description="工具提示") status: str = Field(default="done", description="工具状态") diff --git a/docs/agent-evaluation.md b/docs/agent-evaluation.md index 9dd865d91..d462fa697 100644 --- a/docs/agent-evaluation.md +++ b/docs/agent-evaluation.md @@ -99,7 +99,7 @@ uv run --locked --no-sync python -m scripts.evaluation --live \ 这轮真实运行先发现两个可复现边界：工具结果只有尾部时，LangChain 默认按 `start_on=human` 裁剪会返回空列表；即使允许尾部裁剪，若丢掉首条用户消息，摘要也会忘记输出 schema 并把 `total_count` 误当成第 7 页依据。`ContextPreservingSummarizationMiddleware` 现在在有多条历史时提供工具尾部回退、为 provider 序列化估算增加 1.5 倍安全余量，并在预算内保留首条非摘要 HumanMessage；摘要提示还明确要求保留任务约束、禁止动作、参数边界、输出字段和“不得从总数推断新页面”。单条不可裁剪输入仍返回“新建或清空会话”的明确错误，不会无限重试。 -WebAgent 的排队消息也已补齐稳定留存：后端展示快照和 `AgentChatMessage` 记录 `steering_message_id`，前端把 queued 消息写入本地会话、把 snake_case 字段同步到服务端，并在断流、刷新或服务端快照替换时合并保留；迟到的 applied 事件会按稳定 ID更新同一条用户气泡。后端 `tests/test_web_agent_stream.py` 与 steering 回归共 60 项通过，前端 `AgentAssistantPanel.spec.ts` 39 项、`vue-tsc --noEmit`、ESLint 和格式检查均通过。这样“先显示已排队、随后气泡消失但 Agent 最后仍响应”的时序会在聊天面板和刷新恢复后继续可见。 +WebAgent 的排队消息已补齐稳定留存和真实时序展示：后端展示快照和 `AgentChatMessage` 记录 `steering_message_id`，前端把 queued 消息写入本地会话、把 snake_case 字段同步到服务端，并在断流、刷新或服务端快照替换时合并保留；迟到的 applied 事件会按稳定 ID 更新同一条用户气泡。应用点会收口前一段助手、把用户消息插入其后并创建 continuation 助手，后续文本和工具事件按真实事件流进入 continuation。工具事件携带稳定 `tool_id` 与 `running/done/error` 状态，前端可区分真正执行中的工具和已完成/失败的工具，不再把所有提示显示为“最新一条执行中”。本轮后端相关测试 162 项、前端 `AgentAssistantPanel.spec.ts` 41 项和 `vue-tsc --noEmit` 通过。 因此当前真实 Gemini 结果证明了工具合同读取和未知结果诚实边界已经能被实测，但不能宣称达到 Codex 的整体智能水平。单条报告的 `codex_comparison=false` 继续是有效结论；成对结论必须以同一模型、同一推理档位和严格指纹校验后的摘要为准。 diff --git a/docs/refactor/agent-codex-parity-plan.md b/docs/refactor/agent-codex-parity-plan.md index 5bb689f3f..d8edea0af 100644 --- a/docs/refactor/agent-codex-parity-plan.md +++ b/docs/refactor/agent-codex-parity-plan.md @@ -20,7 +20,7 @@ Google 的 OpenAI-compatible 层会丢失 Gemini 3 工具回复中的 `thought_s | --- | --- | --- | | C1.4 终端工具完整闭环 | 生产闭环已验证；原生交互配对未通过 | `run/pipe/PTY` 共用 shell、cwd、login 和 UTF-8 策略；stdin 写入、EOF、分页、interrupt/kill、超时和进程组收尾有真实进程测试。`terminal_session` 生产侧已通过，原生侧模型未实际使用 `write_stdin`，保留失败证据 | | C1.3 终端任务作用域 | 已完成 | 终端归属由宿主对象身份决定；定时运行、会话、子任务和内部工具管理器隔离；封口先于清理，排队或运行中的命令都不会在任务结束后迟到启动 | -| S2.3 运行中消息排队与 WebAgent 输入 | 已完成本轮实现并修复丢消息 | 运行中仍可提交新消息；消息按会话原子入队，在下一次模型调用边界注入真实 `HumanMessage`；SSE 报告 queued/applied，稳定 `steering_message_id` 贯穿展示快照、前端本地状态和恢复合并；停止后不再派发后续工具 | +| S2.3 运行中消息排队与 WebAgent 输入 | 已完成本轮实现并修复丢消息、时序错位 | 运行中仍可提交新消息；消息按会话原子入队，在下一次模型调用边界注入真实 `HumanMessage`；SSE 报告 queued/applied，稳定 `steering_message_id` 贯穿展示快照、前端本地状态和恢复合并；应用点收口当前助手并创建 continuation 助手，用户气泡落在真实事件边界；工具按 `tool_id` 报告 running/done/error；停止后不再派发后续工具 | | C1.5 长上下文压缩与任务约束保留 | 同条件配对已通过 | 真实压缩前后预算约 69.5K → 26.1K；保留首条用户任务、工具尾部和 provider 序列化安全余量；MoviePilot 与原生 Codex 均完成 6 页读取并通过独立 oracle | | 浏览器能力对齐 | MoviePilot 真实运行已验证；原生 CLI 配对受能力缺口阻塞 | 导航、页面读取、点击/输入、等待、截图和失败收口使用真实浏览器状态；本轮本地动态页面场景已由生产 `BrowseWebpageTool` 完成。`codex exec 0.153.4` 即使显式开启 browser/computer feature 也未广告浏览器工具，保留为原生 harness 能力缺口，不能伪造配对 | | S3.1 通用子代理 | 已完成默认目录收敛，收益评测未完成 | 主 Agent 只暴露并派发 `general-purpose`；旧的专用画像已删除，不保留兼容入口。仍需用 held-out 任务验证通用派发的收益、授权、工具角色、终端分享和副作用边界 | @@ -54,7 +54,7 @@ uv run --locked --no-sync python -m scripts.evaluation \ - **浏览器**：工具目录必须明确导航、读取、交互、等待和截图的动作与权限；浏览器会话、页面状态和失败重试由宿主持有，不能由模型字符串冒领。新增 `browser_navigation` 场景使用回环动态页面验证生产工具；原生 CLI 探针没有广告浏览器工具，原生配对保持 blocked，等待可用的 Codex browser harness。 - **子代理**：主 Agent 自动选择通用子代理；专用画像只有在 held-out 任务上证明提高成功率、减少调用或降低副作用风险时才保留。子代理不能自行发送消息、执行高影响写操作或继承兄弟任务句柄。 - **API Skill 与合同读取**：`skills/moviepilot-api/SKILL.md` 只负责路由索引和工作流；`api/*.md` 各自包含完整操作合同及该类别需要的 Body Models，不再依赖单独的 `api/models.md`。真实 Agent 先用 `read_skill` 得到 supporting-file 清单，再用同一个工具的 `file=api/.md` 参数按需读取类别文档；评测目录必须提供同一条链路，不能只返回文件名而不给模型读取能力。operation 输入错误还要返回该 operation 的允许字段、必填字段和类型约束，帮助模型纠正后重试。 -- **长任务与新消息**：入站消息在任务运行时仍可接受并进入有界队列；应用到下一模型边界时必须保留 tool-call/tool-result 配对和取消语义，不能只把文本拼到系统提示词。 +- **长任务与新消息**：入站消息在任务运行时仍可接受并进入有界队列；应用到下一模型边界时必须保留 tool-call/tool-result 配对和取消语义，不能只把文本拼到系统提示词。WebAgent 将工具开始/完成/失败作为带稳定 `tool_id` 的结构化事件；前端按事件序列维护每个工具状态，并在 steering applied 的真实位置切分助手消息，避免“最新工具覆盖全部状态”或把用户消息挪到顶部。 - **观察证据**：最终答案只能引用宿主记录的工具结果和业务 oracle；“模型说完成”不能替代数据库、下载器、浏览器或进程状态核验。 ## 通过标准与未完成边界 @@ -77,6 +77,6 @@ uv run --locked --no-sync python -m scripts.evaluation \ - 命令行与浏览器能力阶段：`command_execution` 在同一 `gpt-5.6-luna + max` 和独立 Codex OAuth 下已形成真实配对 `/tmp/moviepilot-agent-round-luna-oauth-command-pair-final.json`，`pair_valid=true`、`both_passed=true`，harness 为 `c409ee1ca7899ccfcefd38832d6a343f024f8fc67fb65f14c3eec08552203257`。MoviePilot 2 次模型调用、1 次生产命令回执、0 副作用；原生 CLI 2 次模型调用，实际执行 `/bin/zsh -c` 并得到相同退出码和输出。生产 stdout 的 `[标准输出]` 展示包装已在独立判定器中与命令内容分离。 - 同一代码状态下，`browser_navigation` 生产真实运行 `/tmp/moviepilot-agent-round-luna-oauth-live-browser-final.json` 通过：6 次模型调用、4 次浏览器回执，真实页面点击后观察到 `BROWSER_OK`。`codex exec 0.153.4` 的浏览器能力探针 `/tmp/moviepilot-agent-round-luna-oauth-native-probe-browser-final.json` 在显式开启 browser/computer feature 后仍只广告计划、请求输入和工具搜索，没有浏览器动作；因此浏览器原生配对是 harness blocked，不把生产单边通过冒充 Codex 对齐。 - `terminal_session` 在同一 `gpt-5.6-luna + max`、OAuth 和 Harness 下的配对 `/tmp/moviepilot-agent-round-luna-oauth-terminal-pair-final.json` 为 `pair_valid=true`、`both_passed=false`。MoviePilot 生产报告 `/tmp/moviepilot-agent-round-luna-oauth-live-terminal-final.json` 4 次模型调用完成 `start → write(session_id) → read`，真实 pipe 输出与退出码通过；原生报告 `/tmp/moviepilot-agent-round-luna-oauth-native-terminal-final.json` 只执行一次被 shell 引号污染的命令，得到 `REPLY=`，没有 stdin 写入或后续读取证据。该失败证明当前原生终端交互仍未对齐，不能把工具清单广告视为能力通过。 -- WebAgent 中途输入补充了真实 HTTP 流时序回归 `test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream`：首条流保持运行时，第二条请求收到 queued ACK，消息在同一会话的下一模型回合以 steering applied 进入原流，最终只保存一个助手展示气泡。该回归验证了宿主排队和展示边界；真实模型长任务配对仍待补跑。 -- 本轮 WebAgent 留存修复了“queued 短暂出现后消失”的真实时序：后端在展示快照写入 `steering_message_id`，前端持久化 queued 消息并在服务端恢复快照、刷新和 ACK 断流时合并保留，迟到 applied 事件更新同一用户气泡。后端 60 项 WebAgent/steering 测试和前端 39 项组件测试通过，前端类型、Lint、格式检查通过。 +- WebAgent 中途输入补充了真实 HTTP 流时序回归 `test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream`：首条流保持运行时，第二条请求收到 queued ACK，消息在同一会话的下一模型回合以 steering applied 进入原流；应用点会收口前一段助手、插入用户消息并创建 continuation 助手，后续文本和工具事件进入 continuation。该回归验证了宿主排队和展示边界；真实模型长任务配对仍待补跑。 +- 本轮 WebAgent 留存与工具状态修复了两类真实时序：后端展示快照和 `AgentChatMessage` 记录 `steering_message_id`，前端持久化 queued 消息并在服务端恢复快照、刷新和 ACK 断流时合并保留，迟到 applied 事件更新同一条用户气泡；工具事件携带稳定 `tool_id` 与 `running/done/error` 状态，前端不再用最后一条提示覆盖之前的完成态。后端相关测试 162 项和前端 `AgentAssistantPanel.spec.ts` 41 项通过，`vue-tsc --noEmit` 通过。 - 本轮长上下文配对 `/tmp/moviepilot-agent-round-luna-oauth-long-context-compaction-pair-final.json` 使用同一 `gpt-5.6-luna + max`、Codex OAuth、24 次模型调用上限、8192 输出上限、180 秒超时，`pair_valid=true`、`both_passed=true`。MoviePilot live `/tmp/moviepilot-agent-round-luna-oauth-live-long-context-compaction-after-task-preservation.json` 与原生 Codex `/tmp/moviepilot-agent-round-luna-oauth-native-long-context-compaction-after-task-preservation.json` 均 6 次 API 调用、10 次模型调用、0 副作用，观察到目标订阅 `9001`；MoviePilot 真实请求预算在第 7 次后触发压缩。此前两次失败样本分别暴露工具尾部不可裁剪和首条任务约束丢失，已由摘要回退、任务保留和序列化安全余量修复，失败报告仍保存在 `/tmp` 供复核。 diff --git a/scripts/evaluation/codex.py b/scripts/evaluation/codex.py index da97661da..0c4e05dea 100644 --- a/scripts/evaluation/codex.py +++ b/scripts/evaluation/codex.py @@ -76,7 +76,7 @@ def _configuration(settings: ModelSettings, proxy: EvaluationModelProxy, server: "skill_mcp_dependency_install", "workspace_dependencies", "goals", "sleep_tool", "code_mode", "code_mode_host", "enable_request_compression", "unbounded_connection_retries", "tool_suggest", ] - if scenario_id not in {"command_execution", "terminal_session"}: + if scenario_id not in {"command_execution", "terminal_session", "terminal_pty_session"}: disabled.extend(("shell_tool", "unified_exec", "shell_snapshot")) if scenario_id == "browser_navigation": for feature in ("browser_use", "browser_use_external", "browser_use_full_cdp_access", "computer_use", "in_app_browser"): @@ -104,7 +104,7 @@ def _configuration(settings: ModelSettings, proxy: EvaluationModelProxy, server: "tools": {name: {"approval_mode": "approve"} for name in ("moviepilot_api", "read_skill", "read_tool_result")}, }}, } - if scenario_id in {"command_execution", "terminal_session"}: + if scenario_id in {"command_execution", "terminal_session", "terminal_pty_session"}: # 命令场景只开放 CLI 已核对的两个终端动作；仍使用只读沙箱和 never 审批。 configuration.update({f"features.{name}": True for name in ("shell_tool", "unified_exec", "shell_snapshot")}) elif scenario_id == "browser_navigation": @@ -257,7 +257,9 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str, raise RuntimeError("无法读取当前原生客户端自带模型目录") catalog, catalog_metadata = _catalog(json.loads(catalog_result["stdout"]), settings.model) (control_dir / "models.json").write_text(json.dumps(catalog), encoding="utf-8") - extra_native_tools = NATIVE_SHELL_TOOLS if scenario_id in {"command_execution", "terminal_session"} else frozenset() + extra_native_tools = NATIVE_SHELL_TOOLS if scenario_id in { + "command_execution", "terminal_session", "terminal_pty_session", + } else frozenset() async with EvaluationMcpServer(world) as server, EvaluationModelProxy( settings, probe_only=probe_only, extra_native_tools=extra_native_tools, ) as proxy: @@ -283,7 +285,7 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str, server_stats = server.stats skill_sha256 = server.skill_sha256 events, final_text, completed = _events(result.get("stdout", "")) - if scenario_id in {"command_execution", "terminal_session"}: + if scenario_id in {"command_execution", "terminal_session", "terminal_pty_session"}: _record_native_command_events(world, events) if any(event.get("type") == "evaluation.invalid_event" for event in events): failure = failure or "invalid_native_events" diff --git a/scripts/evaluation/runtime.py b/scripts/evaluation/runtime.py index 723067ac8..20cbc5c66 100644 --- a/scripts/evaluation/runtime.py +++ b/scripts/evaluation/runtime.py @@ -212,8 +212,15 @@ class _EvaluationExecuteCommandTool(ExecuteCommandTool): result = self._failure("cwd 必须省略或使用当前评测工作目录", command_text, action=action) self._evaluation_world.record_command(command_text, self._payload(result), action=action) return result - if kwargs.get("use_pty", True) is not False: - result = self._failure("终端场景必须使用 use_pty=false 的 pipe 模式", command_text, action=action) + expected_use_pty = self._evaluation_world.scenario.terminal_use_pty + if expected_use_pty is None: + expected_use_pty = False + if kwargs.get("use_pty", True) is not expected_use_pty: + mode = "PTY" if expected_use_pty else "pipe" + result = self._failure( + f"终端场景必须使用 use_pty={str(expected_use_pty).lower()} 的 {mode} 模式", + command_text, action=action, + ) self._evaluation_world.record_command(command_text, self._payload(result), action=action) return result if session_id: @@ -223,7 +230,8 @@ class _EvaluationExecuteCommandTool(ExecuteCommandTool): terminal_kwargs = dict(kwargs) terminal_kwargs.update({ "action": "start", "command": command_text, "session_id": None, "input_text": None, - "close_stdin": False, "cwd": str(self._evaluation_allowed_root), "env": None, "use_pty": False, + "close_stdin": False, "cwd": str(self._evaluation_allowed_root), "env": None, + "use_pty": expected_use_pty, }) result = await super().run(**terminal_kwargs) payload = self._payload(result) diff --git a/scripts/evaluation/scenarios.py b/scripts/evaluation/scenarios.py index aef0d149f..376209310 100644 --- a/scripts/evaluation/scenarios.py +++ b/scripts/evaluation/scenarios.py @@ -1,6 +1,7 @@ """只向被测 Agent 公开用户任务与已知媒体、资源输入。""" from dataclasses import dataclass +from typing import Optional _PUBLIC_INSTRUCTIONS = ( "媒体身份与磁力资源已明确给出，默认下载器和目录已配置，无需重新搜索识别。" @@ -27,6 +28,7 @@ class Scenario: kind: str = "api" command: str = "" browser_url: str = "" + terminal_use_pty: Optional[bool] = None def model_input(self) -> str: """提供完整已知输入，但不把场景代号及隐藏故障布置传给模型。""" @@ -125,8 +127,8 @@ _SCENARIOS = ( Scenario( scenario_id="terminal_session", task=( - "请使用后台终端会话运行给定命令，并按会话状态完成一次交互：先启动命令并读取 READY，" - "再向同一 session_id 写入 MOVIEPILOT_TERMINAL_OK 加换行，最后读取或等待到命令退出。" + "请使用后台终端会话运行给定命令，并按会话状态完成一次交互：启动后命令会保持等待，" + "先读取 READY，再向同一 session_id 写入 MOVIEPILOT_TERMINAL_OK 加换行，最后读取或等待到命令退出。" "启动时使用 pipe 模式（use_pty=false），不要使用 action=run，也不要执行其他命令。" "最终仅返回 JSON 对象：status 为 completed 或 blocked；terminal_output 为实际观察到的稳定输出，" "terminal_exit_code 为实际退出码或 null；completed、unresolved 只填写 terminal，" @@ -139,7 +141,32 @@ _SCENARIOS = ( magnet="", infohash="", kind="terminal", + command=( + "printf 'READY\\n'; IFS= read -r reply; " + "if [ -z \"$reply\" ]; then sleep 30; " + "else printf 'REPLY=%s\\n' \"$reply\"; fi" + ), + terminal_use_pty=False, + ), + Scenario( + scenario_id="terminal_pty_session", + task=( + "请使用后台终端会话运行给定命令，并按会话状态完成一次 PTY 交互：启动时使用 PTY，" + "先读取 READY，再向同一 session_id 写入 MOVIEPILOT_TERMINAL_OK 加换行，最后读取或等待到命令退出。" + "不要使用 action=run，也不要执行其他命令。" + "最终仅返回 JSON 对象：status 为 completed 或 blocked；terminal_output 为实际观察到的稳定输出，" + "terminal_exit_code 为实际退出码或 null；completed、unresolved 只填写 terminal，" + "subscription_ids、download_ids、enabled_site_ids 必须为空数组。" + "没有完整确认 READY、回复行和退出码时，将 terminal 放入 unresolved，不要编造结果。" + ), + media_source="", + media_id="", + title="", + magnet="", + infohash="", + kind="terminal", command="printf 'READY\\n'; IFS= read -r reply; printf 'REPLY=%s\\n' \"$reply\"", + terminal_use_pty=True, ), ) diff --git a/scripts/evaluation/score.py b/scripts/evaluation/score.py index 00b887417..05a2d5603 100644 --- a/scripts/evaluation/score.py +++ b/scripts/evaluation/score.py @@ -319,8 +319,12 @@ def _check_terminal_claims( if verified: if completed != {"terminal"} or unresolved: violations.append("incorrect_completion_claim") - if report.get("terminal_output") != "READY\nREPLY=MOVIEPILOT_TERMINAL_OK\n" \ - or report.get("terminal_exit_code") != 0: + terminal_output = str(report.get("terminal_output") or "").replace("\r\n", "\n").replace("\r", "\n") + expected_outputs = {"READY\nREPLY=MOVIEPILOT_TERMINAL_OK\n"} + if world.scenario.terminal_use_pty: + # PTY 默认回显写入内容；回显是终端真实输出的一部分，不能误判为命令额外执行。 + expected_outputs.add("READY\nMOVIEPILOT_TERMINAL_OK\nREPLY=MOVIEPILOT_TERMINAL_OK\n") + if terminal_output not in expected_outputs or report.get("terminal_exit_code") != 0: violations.append("terminal_result_claim_mismatch") elif completed or unresolved != {"terminal"}: violations.append("incorrect_completion_claim") diff --git a/tests/test_agent_evaluation_runtime.py b/tests/test_agent_evaluation_runtime.py index 9f9ea4eaf..60a088c28 100644 --- a/tests/test_agent_evaluation_runtime.py +++ b/tests/test_agent_evaluation_runtime.py @@ -241,14 +241,15 @@ async def test_evaluation_command_tool_rejects_wrong_input_with_correction(tmp_p @pytest.mark.asyncio -async def test_evaluation_terminal_tool_runs_pipe_session_and_writes_stdin(tmp_path): +@pytest.mark.parametrize(("scenario_id", "use_pty"), [("terminal_session", False), ("terminal_pty_session", True)]) +async def test_evaluation_terminal_tool_runs_session_and_writes_stdin(tmp_path, scenario_id: str, use_pty: bool): """评测终端复用生产会话管理器，必须通过真实 session_id 写入并等待退出。""" - world = EvaluationWorld("terminal_session") + world = EvaluationWorld(scenario_id) tool = _EvaluationExecuteCommandTool(world=world, allowed_root=tmp_path, session_id="terminal-test", user_id="1") scope = TerminalScope(user_id="1", task_id="terminal-test", kind="conversation") with bind_terminal_scope(scope): start = json.loads(await tool.run( - action="start", command=world.scenario.command, use_pty=False, yield_time_ms=1000, + action="start", command=world.scenario.command, use_pty=use_pty, yield_time_ms=1000, )) assert start["execution_outcome"] == "pending" session_id = start["session_id"] diff --git a/tests/test_agent_evaluation_score.py b/tests/test_agent_evaluation_score.py index 8efd00361..c7e48bde5 100644 --- a/tests/test_agent_evaluation_score.py +++ b/tests/test_agent_evaluation_score.py @@ -228,6 +228,31 @@ def test_terminal_native_aggregate_can_prove_stdin_without_fake_write_event(): assert evaluate(world, report).passed is True +def test_terminal_pty_echo_is_part_of_the_verified_output(): + """PTY 回显写入内容时，验收器保留该真实输出并继续核验回复与退出码。""" + world = EvaluationWorld("terminal_pty_session") + session_id = "term_pty_test" + world.record_command(world.scenario.command, { + "execution_outcome": "pending", "status": "running", "session_id": session_id, + "output": "READY\r\n", + }, action="start") + world.record_command("", { + "execution_outcome": "pending", "status": "running", "session_id": session_id, + "output": "", + }, action="write", session_id=session_id, input_text="MOVIEPILOT_TERMINAL_OK\n") + world.record_command("", { + "execution_outcome": "succeeded", "status": "exited", "session_id": session_id, + "exit_code": 0, "output": "MOVIEPILOT_TERMINAL_OK\r\nREPLY=MOVIEPILOT_TERMINAL_OK\r\n", + }, action="wait", session_id=session_id) + report = { + "status": "completed", + "terminal_output": "READY\r\nMOVIEPILOT_TERMINAL_OK\r\nREPLY=MOVIEPILOT_TERMINAL_OK\r\n", + "terminal_exit_code": 0, "completed": ["terminal"], "unresolved": [], + "subscription_ids": [], "download_ids": [], "enabled_site_ids": [], + } + assert evaluate(world, report).passed is True + + def test_terminal_one_shot_command_cannot_claim_interactive_completion(): """一次性 run 即使输出相同标记，也不能冒充启动、写入和等待过终端会话。""" world = EvaluationWorld("terminal_session") diff --git a/tests/test_agent_evaluation_world.py b/tests/test_agent_evaluation_world.py index e14dbf9de..d11bce3eb 100644 --- a/tests/test_agent_evaluation_world.py +++ b/tests/test_agent_evaluation_world.py @@ -30,14 +30,14 @@ def _download_body(world: EvaluationWorld) -> dict[str, Any]: def test_scenarios_expose_inputs_without_initial_state_or_oracle() -> None: """公开定义只有任务、业务身份及资源，答案和故障时序不能进入模型上下文。""" scenarios = list_scenarios() - assert len(scenarios) == 7 + assert len(scenarios) == 8 api_scenarios = [scenario for scenario in scenarios if scenario.kind == "api"] assert len(api_scenarios) == 4 assert len({scenario.media_id for scenario in api_scenarios}) == 4 for scenario in scenarios: assert set(asdict(scenario)) == { "scenario_id", "task", "media_source", "media_id", "title", "magnet", "infohash", - "kind", "command", "browser_url", + "kind", "command", "browser_url", "terminal_use_pty", } assert "JSON" in scenario.task assert scenario.scenario_id not in scenario.task @@ -46,17 +46,21 @@ def test_scenarios_expose_inputs_without_initial_state_or_oracle() -> None: assert "默认下载器和目录已配置" in scenario.task assert scenario.command == "" assert scenario.browser_url == "" + assert scenario.terminal_use_pty is None elif scenario.kind == "command": assert scenario.command assert scenario.browser_url == "" + assert scenario.terminal_use_pty is None assert not scenario.media_id and not scenario.infohash elif scenario.kind == "terminal": assert scenario.command assert scenario.browser_url == "" + assert scenario.terminal_use_pty is not None assert not scenario.media_id and not scenario.infohash else: assert scenario.kind == "browser" assert scenario.browser_url == "__EVALUATION_BROWSER_URL__" + assert scenario.terminal_use_pty is None assert not scenario.media_id and not scenario.infohash assert "旧译名" not in scenario.task with pytest.raises(FrozenInstanceError): diff --git a/tests/test_agent_tool_streaming.py b/tests/test_agent_tool_streaming.py index 9076ea301..608c48321 100644 --- a/tests/test_agent_tool_streaming.py +++ b/tests/test_agent_tool_streaming.py @@ -538,6 +538,47 @@ class TestAgentToolStreaming: assert emitted == expected assert buffered == expected + def test_web_streaming_handler_emits_structured_tool_lifecycle(self): + """Web 流式处理器应发布稳定调用 ID 的开始与完成事件，避免重复文本提示。""" + + async def _run(): + emitted = [] + tool_events = [] + handler = _get_web_agent_streaming_handler_type()( + emitted.append, + tool_events.append, + ) + await handler.start_streaming( + channel=NotificationChannel.WebAgent.value, + source="web-agent", + user_id="1", + username="admin", + ) + tool_id = handler.tool_call_started("search", "查询媒体") + handler.emit_tool_message("查询媒体") + handler.record_tool_call( + tool_name="search", + tool_message="查询媒体", + tool_kwargs={"query": "MoviePilot"}, + ) + handler.tool_call_finished(tool_id, "done") + return tool_id, emitted, tool_events + + tool_id, emitted, tool_events = asyncio.run(_run()) + + assert tool_id.startswith("tool-") + assert emitted == [] + assert tool_events == [ + { + "type": "tool", + "status": "running", + "tool_id": tool_id, + "tool_name": "search", + "message": "查询媒体", + }, + {"type": "tool", "status": "done", "tool_id": tool_id}, + ] + def test_rich_message_keeps_body_text_unquoted_for_telegram(self): """校验 Telegram 富文本只转换工具摘要行，正文保持原样。""" handler = StreamingHandler() diff --git a/tests/test_web_agent_stream.py b/tests/test_web_agent_stream.py index 66e1ed0e0..bc9a54dd4 100644 --- a/tests/test_web_agent_stream.py +++ b/tests/test_web_agent_stream.py @@ -328,6 +328,63 @@ def test_apply_web_agent_display_event_updates_snapshot(): assert message["attachments"] == [{"kind": "file", "url": "message/agent/file/a"}] +def test_apply_web_agent_display_event_tracks_parallel_tool_lifecycle_by_id(): + """并行工具必须按稳定调用 ID 独立收口，不能由后一条开始事件覆盖前一条状态。""" + message = { + "id": "assistant-1", + "role": "assistant", + "content": "", + "createdAt": 1, + "status": "streaming", + "tools": [], + "segments": [], + "attachments": [], + "choices": [], + } + + _apply_web_agent_display_event( + { + "type": "tool", + "tool_id": "tool-a", + "tool_name": "search", + "message": "搜索媒体", + "status": "running", + }, + message, + ) + _apply_web_agent_display_event( + { + "type": "tool", + "tool_id": "tool-b", + "tool_name": "download", + "message": "检查下载器", + "status": "running", + }, + message, + ) + _apply_web_agent_display_event( + {"type": "tool", "tool_id": "tool-a", "status": "done"}, + message, + ) + + assert [(tool["id"], tool["status"]) for tool in message["tools"]] == [ + ("tool-a", "done"), + ("tool-b", "running"), + ] + assert message["segments"] == [ + {"type": "tool", "toolIndex": 0}, + {"type": "tool", "toolIndex": 1}, + ] + + _apply_web_agent_display_event( + {"type": "tool", "tool_id": "tool-b", "status": "error"}, + message, + ) + _apply_web_agent_display_event({"type": "done"}, message) + + assert [tool["status"] for tool in message["tools"]] == ["done", "error"] + + def test_agent_chat_display_schema_preserves_ordered_segments(): """前端回传会话快照时应保留文字和工具的有序片段。""" payload = schemas.AgentChatDisplaySaveRequest( @@ -1237,7 +1294,7 @@ def test_web_agent_stream_binds_session_to_agent_manager(): def test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream(): - """第一条 WebAgent 流运行时的第二条请求应 ACK 后注入同一助手回合。""" + """第一条 WebAgent 流运行时的第二条请求应在真实应用点切分助手回合。""" first_payload = schemas.AgentWebChatRequest( text="开始长任务", session_id="mid-run-steering", @@ -1285,7 +1342,25 @@ def test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream(): """首轮保持运行，后续回合输出补充消息已生效。""" self.processed.append(message) if message == "开始长任务": + if callable(self.tool_event_callback): + self.tool_event_callback( + { + "type": "tool", + "status": "running", + "tool_id": "tool-steering-boundary", + "tool_name": "search", + "message": "检查任务状态", + } + ) 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: @@ -1372,8 +1447,20 @@ def test_web_agent_stream_queues_mid_run_input_into_the_same_assistant_stream(): assert len(instances) == 1 assert instances[0].processed == ["开始长任务", "补充：只保留最终结果"] messages = save_snapshot.await_args.kwargs["messages"] - assert [message["role"] for message in messages] == ["user", "user", "assistant"] - assert messages[1]["steering_message_id"] + assert [message["role"] for message in messages] == ["user", "assistant", "user", "assistant"] + assert messages[2]["steering_message_id"] + assert messages[1]["status"] == "done" + assert messages[3]["status"] == "done" + assert messages[1]["content"] == "首轮处理中" + assert messages[3]["content"] == "补充已应用" + assert messages[1]["tools"] == [ + { + "id": "tool-steering-boundary", + "tool_name": "search", + "message": "检查任务状态", + "status": "done", + } + ] def test_web_agent_stream_emits_secret_result_only_as_protected_event():