from __future__ import annotations import asyncio import json import time import uuid from pathlib import Path from typing import Optional, Dict, Any, List from .utils_common import debug_log, brief_log from .state import MONITOR_FILE_TOOLS, MONITOR_MEMORY_TOOLS, MONITOR_SNAPSHOT_CHAR_LIMIT, MONITOR_MEMORY_ENTRY_LIMIT from .state import tool_approval_manager, user_question_manager from .monitor import cache_monitor_snapshot from .security import compact_web_search_result from .chat_flow_helpers import detect_tool_failure from .chat_flow_runner_helpers import resolve_monitor_path, resolve_monitor_memory, capture_monitor_snapshot from utils.tool_result_formatter import ( extract_mcp_content_for_context, format_tool_result_for_context, ) from utils.context_manager import AUTO_SHALLOW_PLACEHOLDER from config import TOOL_CALL_COOLDOWN from modules.personalization_manager import load_personalization_config, resolve_context_compression_settings from modules.auto_approval_service import run_auto_approval from modules.user_question_manager import format_user_question_answer from .deep_compression import run_deep_compression from .chat_flow_task_support import inject_runtime_user_message def _format_numbered_lines(lines: List[str], start_line_no: int) -> List[Dict[str, Any]]: return [ { "line_no": start_line_no + idx, "content": line.rstrip("\n"), } for idx, line in enumerate(lines) ] def _build_tool_approval_preview(web_terminal, function_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: args = arguments or {} preview: Dict[str, Any] = { "type": function_name, "tool_name": function_name, "arguments": args, } if function_name == "edit_file": file_path = args.get("file_path") old_string = args.get("old_string") new_string = args.get("new_string") replace_all_provided = "replace_all" in args replace_all = args.get("replace_all") if replace_all_provided else None preview["file_path"] = file_path preview["replace_all"] = replace_all if not file_path: preview["summary"] = "缺少 file_path" return preview if old_string is None or new_string is None: preview["summary"] = "缺少 old_string/new_string" return preview if not replace_all_provided: preview["summary"] = "缺少 replace_all（必须显式传入 true 或 false）" return preview if not isinstance(replace_all, bool): preview["summary"] = "replace_all 必须是 true 或 false" return preview old_text = str(old_string or "") short_old_text_notice = len(old_text.splitlines()) < 3 if short_old_text_notice: preview["notice"] = "提示：old_string 少于3行，允许继续执行；需要批量替换的场景可以单行或不足一行" try: valid, err, full_path = web_terminal.file_manager._validate_path(str(file_path)) if not valid or full_path is None: preview["summary"] = err or "路径校验失败" return preview resolved_path = str(Path(full_path)) preview["resolved_path"] = resolved_path if not full_path.exists() or not full_path.is_file(): preview["summary"] = "目标文件不存在，无法生成上下文预览" return preview content = full_path.read_text(encoding="utf-8", errors="ignore") new_text = str(new_string or "") old_lines = old_text.splitlines() new_lines = new_text.splitlines() idx = content.find(old_text) if old_text else -1 if idx >= 0: prefix = content[:idx] start_line_no = prefix.count("\n") + 1 end_line_no = start_line_no + max(1, len(old_lines)) - 1 all_lines = content.splitlines() before_start = max(1, start_line_no - 3) before = all_lines[before_start - 1:start_line_no - 1] after_end = min(len(all_lines), end_line_no + 3) after = all_lines[end_line_no:after_end] preview["edit_context"] = { "before": _format_numbered_lines(before, before_start), "old": _format_numbered_lines(old_lines or [""], start_line_no), "new": _format_numbered_lines(new_lines or [""], start_line_no), "after": _format_numbered_lines(after, end_line_no + 1), "old_start_line": start_line_no, "old_end_line": end_line_no, } mode_text = "全部匹配" if replace_all is True else "首个匹配" preview["summary"] = f"编辑 {file_path} 第 {start_line_no}-{end_line_no} 行（{mode_text}）" else: preview["edit_context"] = { "before": [], "old": _format_numbered_lines(old_lines or [""], 1), "new": _format_numbered_lines(new_lines or [""], 1), "after": [], "old_start_line": None, "old_end_line": None, } preview["summary"] = "未在文件中定位到 old_string，显示原始替换内容" if short_old_text_notice: preview["summary"] = f"{preview['summary']}（old_string 少于3行，已告知并继续）" except Exception as exc: preview["summary"] = f"生成编辑预览失败: {exc}" return preview if function_name in {"run_command", "terminal_input"}: preview["command"] = args.get("command") preview["summary"] = f"执行命令: {args.get('command') or ''}" return preview if function_name in {"run_python", "runpython"}: code = str(args.get("code") or "") timeout = args.get("timeout") preview["code"] = code preview["timeout"] = timeout # 兼容前端已有 command 展示分支 preview["command"] = code preview["summary"] = f"执行 Python 代码（timeout={timeout if timeout is not None else '未提供'}s）" return preview if function_name in {"create_file", "create_folder", "delete_file"}: preview["path"] = args.get("path") preview["summary"] = f"{function_name}: {args.get('path') or ''}" return preview if function_name == "rename_file": preview["old_path"] = args.get("old_path") preview["new_path"] = args.get("new_path") preview["summary"] = f"rename_file: {args.get('old_path') or ''} -> {args.get('new_path') or ''}" return preview if function_name == "write_file": content = str(args.get("content") or "") preview["file_path"] = args.get("file_path") preview["append"] = bool(args.get("append", False)) preview["content_preview"] = content preview["content_length"] = len(content) preview["summary"] = f"write_file: {args.get('file_path') or ''} ({'append' if preview['append'] else 'overwrite'})" return preview return preview def _is_permission_denied_result(result_data: Dict[str, Any]) -> bool: if not isinstance(result_data, dict): return False if result_data.get("success") is True: return False fragments: List[str] = [] for key in ("error", "message", "output"): value = result_data.get(key) if isinstance(value, str) and value.strip(): fragments.append(value.lower()) joined = "\n".join(fragments) if not joined: return False markers = ( "operation not permitted", "permission denied", "权限不足", "无权限", "不允许", "access denied", ) return any(marker in joined for marker in markers) def _inject_runtime_mode_notice( *, web_terminal, messages: List[Dict[str, Any]], content: str, sender=None, conversation_id: Optional[str] = None, ) -> None: inject_runtime_user_message( web_terminal=web_terminal, messages=messages, text=content, source="notify", sender=sender, conversation_id=conversation_id, inline=True, ) def _format_rejected_tool_text(reason: str) -> str: clean_reason = str(reason or "").strip() or "未提供" return f"工具调用被拒绝\n原因：{clean_reason}" async def _wait_for_tool_approval(*, approval_id: str, username: str, timeout_seconds: float = 3600.0) -> Dict[str, Any]: started = time.time() while True: row = tool_approval_manager.get(approval_id) if not row: return {"decision": "rejected", "reason": "审批请求不存在"} if row.get("username") != username: return {"decision": "rejected", "reason": "审批请求用户不匹配"} status = row.get("status") if status in {"approved", "rejected"}: return {"decision": status, "item": row} if (time.time() - started) >= timeout_seconds: return {"decision": "rejected", "reason": "审批超时"} await asyncio.sleep(0.2) def _safe_parse_tool_arguments_for_question(web_terminal, tool_call: Dict[str, Any]) -> Optional[Dict[str, Any]]: try: function = tool_call.get("function") or {} if function.get("name") != "ask_user": return None raw = function.get("arguments") or "{}" if hasattr(web_terminal, 'api_client') and hasattr(web_terminal.api_client, '_safe_tool_arguments_parse'): success, arguments, _error_msg = web_terminal.api_client._safe_tool_arguments_parse(raw, "ask_user") if success and isinstance(arguments, dict): return arguments return None parsed = json.loads(raw) if str(raw).strip() else {} return parsed if isinstance(parsed, dict) else None except Exception: return None async def _wait_for_user_questions(*, question_ids: List[str], username: str, timeout_seconds: float = 3600.0) -> Dict[str, Dict[str, Any]]: started = time.time() pending = {str(qid) for qid in question_ids if qid} answered: Dict[str, Dict[str, Any]] = {} while pending: for qid in list(pending): row = user_question_manager.get(qid) if not row: answered[qid] = {"status": "missing", "answer_text": "用户问题不存在。"} pending.remove(qid) continue if row.get("username") != username: answered[qid] = {"status": "forbidden", "answer_text": "用户问题所属用户不匹配。"} pending.remove(qid) continue if row.get("status") == "answered": answered[qid] = {**row, "answer_text": format_user_question_answer(row)} pending.remove(qid) if not pending: break if (time.time() - started) >= timeout_seconds: for qid in list(pending): answered[qid] = {"status": "timeout", "answer_text": "等待用户回答超时。"} pending.remove(qid) break await asyncio.sleep(0.2) return answered async def execute_tool_calls(*, web_terminal, tool_calls, sender, messages, client_sid: str, username: str, iteration: int, conversation_id: Optional[str], last_tool_call_time: float, process_sub_agent_updates, process_background_command_updates, maybe_mark_failure_from_message, mark_force_thinking, get_stop_flag, clear_stop_flag, workspace=None): previous_tool_loop_active = getattr(web_terminal, "_tool_loop_active", False) web_terminal._tool_loop_active = True allowed_tool_names = set() try: defined_tools = web_terminal.define_tools() or [] for tool in defined_tools: name = ((tool or {}).get("function") or {}).get("name") if isinstance(name, str) and name: allowed_tool_names.add(name) except Exception as exc: debug_log(f"构建工具白名单失败（降级继续）: {exc}") recent_tool_actions = list(getattr(web_terminal, "_recent_tool_actions", []) or []) user_question_results_by_tool_call_id: Dict[str, str] = {} user_question_id_by_tool_call_id: Dict[str, str] = {} ask_user_items: List[Dict[str, Any]] = [] for idx, pending_tool_call in enumerate(tool_calls or []): function = (pending_tool_call or {}).get("function") or {} if function.get("name") != "ask_user": continue arguments = _safe_parse_tool_arguments_for_question(web_terminal, pending_tool_call) if not isinstance(arguments, dict): continue question_text = str(arguments.get("question") or "").strip() if not question_text: continue ask_user_items.append({ "tool_call": pending_tool_call, "arguments": arguments, "order": idx, }) if ask_user_items: batch_id = f"question_batch_{uuid.uuid4().hex}" batch_total = len(ask_user_items) created_questions: List[Dict[str, Any]] = [] for batch_index, item in enumerate(ask_user_items, start=1): tool_call = item.get("tool_call") or {} arguments = item.get("arguments") or {} question = user_question_manager.create_question( username=username, conversation_id=conversation_id, task_id=getattr(web_terminal, "task_id", None), tool_call_id=tool_call.get("id"), question=arguments.get("question"), context=arguments.get("context"), options=arguments.get("options"), batch_id=batch_id, batch_index=batch_index, batch_total=batch_total, ) created_questions.append(question) if tool_call.get("id"): user_question_id_by_tool_call_id[str(tool_call.get("id"))] = question.get("question_id") sender('update_action', { 'preparing_id': tool_call.get("id"), 'status': 'awaiting_user_answer', 'result': { "success": False, "status": "awaiting_user_answer", "question_id": question.get("question_id"), "message": "等待用户回答" }, 'message': '等待用户回答', 'conversation_id': conversation_id }) sender('user_questions_required', { 'batch_id': batch_id, 'questions': created_questions, 'conversation_id': conversation_id, }) wait_answers = await _wait_for_user_questions( question_ids=[str(q.get("question_id") or "") for q in created_questions], username=username, ) for question in created_questions: qid = str(question.get("question_id") or "") answer_row = wait_answers.get(qid) or {} answer_text = str(answer_row.get("answer_text") or "用户未回答。").strip() or "用户未回答。" tool_call_id = str(question.get("tool_call_id") or "") if tool_call_id: user_question_results_by_tool_call_id[tool_call_id] = answer_text sender('user_questions_resolved', { 'batch_id': batch_id, 'question_ids': [q.get("question_id") for q in created_questions], 'conversation_id': conversation_id, }) # 执行每个工具 pending_runtime_mode_notices: List[str] = [] last_completed_tool_call_id: Optional[str] = None deep_compression_pending = False for tool_call in tool_calls: # 检查停止标志 client_stop_info = get_stop_flag(client_sid, username) if client_stop_info: stop_requested = client_stop_info.get('stop', False) if isinstance(client_stop_info, dict) else client_stop_info if stop_requested: debug_log("在工具调用过程中检测到停止状态") tool_call_id = tool_call.get("id") function_name = tool_call.get("function", {}).get("name") # 通知前端该工具已被取消，避免界面卡住 sender('update_action', { 'preparing_id': tool_call_id, 'status': 'cancelled', 'result': { "success": False, "status": "cancelled", "message": "命令执行被用户取消", "tool": function_name } }) # 在消息列表中记录取消结果，防止重新加载时仍显示运行中 if tool_call_id: messages.append({ "role": "tool", "tool_call_id": tool_call_id, "name": function_name, "content": "命令执行被用户取消", }) sender('task_stopped', { 'message': '命令执行被用户取消', 'reason': 'user_stop' }) clear_stop_flag(client_sid, username) web_terminal._tool_loop_active = previous_tool_loop_active return {"stopped": True, "last_tool_call_time": last_tool_call_time} # 工具调用间隔控制 current_time = time.time() if last_tool_call_time > 0: elapsed = current_time - last_tool_call_time if elapsed < TOOL_CALL_COOLDOWN: await asyncio.sleep(TOOL_CALL_COOLDOWN - elapsed) last_tool_call_time = time.time() function_name = tool_call["function"]["name"] arguments_str = tool_call["function"]["arguments"] tool_call_id = tool_call["id"] debug_log(f"准备解析JSON，工具: {function_name}, 参数长度: {len(arguments_str)}") debug_log(f"JSON参数前200字符: {arguments_str[:200]}") debug_log(f"JSON参数后200字符: {arguments_str[-200:]}") # 使用改进的参数解析方法 if hasattr(web_terminal, 'api_client') and hasattr(web_terminal.api_client, '_safe_tool_arguments_parse'): success, arguments, error_msg = web_terminal.api_client._safe_tool_arguments_parse(arguments_str, function_name) if not success: debug_log(f"安全解析失败: {error_msg}") error_text = f'工具参数解析失败: {error_msg}' error_payload = { "success": False, "error": error_text, "error_type": "parameter_format_error", "tool_name": function_name, "tool_call_id": tool_call_id, "message": error_text } sender('error', {'message': error_text}) sender('update_action', { 'preparing_id': tool_call_id, 'status': 'completed', 'result': error_payload, 'message': error_text }) error_content = json.dumps(error_payload, ensure_ascii=False) web_terminal.context_manager.add_conversation( "tool", error_content, tool_call_id=tool_call_id, name=function_name ) messages.append({ "role": "tool", "tool_call_id": tool_call_id, "name": function_name, "content": error_content }) continue debug_log(f"使用安全解析成功，参数键: {list(arguments.keys())}") else: # 回退到带有基本修复逻辑的解析 try: arguments = json.loads(arguments_str) if arguments_str.strip() else {} debug_log(f"直接JSON解析成功，参数键: {list(arguments.keys())}") except json.JSONDecodeError as e: debug_log(f"原始JSON解析失败: {e}") # 尝试基本的JSON修复 repaired_str = arguments_str.strip() repair_attempts = [] # 修复1: 未闭合字符串 if repaired_str.count('"') % 2 == 1: repaired_str += '"' repair_attempts.append("添加闭合引号") # 修复2: 未闭合JSON对象 if repaired_str.startswith('{') and not repaired_str.rstrip().endswith('}'): repaired_str = repaired_str.rstrip() + '}' repair_attempts.append("添加闭合括号") # 修复3: 截断的JSON（移除不完整的最后一个键值对） if not repair_attempts: # 如果前面的修复都没用上 last_comma = repaired_str.rfind(',') if last_comma > 0: repaired_str = repaired_str[:last_comma] + '}' repair_attempts.append("移除不完整的键值对") # 尝试解析修复后的JSON try: arguments = json.loads(repaired_str) debug_log(f"JSON修复成功: {', '.join(repair_attempts)}") debug_log(f"修复后参数键: {list(arguments.keys())}") except json.JSONDecodeError as repair_error: debug_log(f"JSON修复也失败: {repair_error}") debug_log(f"修复尝试: {repair_attempts}") debug_log(f"修复后内容前100字符: {repaired_str[:100]}") error_text = f'工具参数解析失败: {e}' error_payload = { "success": False, "error": error_text, "error_type": "parameter_format_error", "tool_name": function_name, "tool_call_id": tool_call_id, "message": error_text } sender('error', {'message': error_text}) sender('update_action', { 'preparing_id': tool_call_id, 'status': 'completed', 'result': error_payload, 'message': error_text }) error_content = json.dumps(error_payload, ensure_ascii=False) web_terminal.context_manager.add_conversation( "tool", error_content, tool_call_id=tool_call_id, name=function_name ) messages.append({ "role": "tool", "tool_call_id": tool_call_id, "name": function_name, "content": error_content }) continue # 严格校验：只允许执行当前回合明确下发给模型的工具。 # 某些模型会“幻觉”调用未下发工具（如 view_image），必须在执行层拦截。 if allowed_tool_names and function_name not in allowed_tool_names: if function_name == "view_image" and "vlm_analyze" in allowed_tool_names: if not arguments.get("prompt"): arguments["prompt"] = "请详细分析这张图片的内容，包括关键文字、主体对象与场景信息。" debug_log( "工具名自动纠正: view_image -> vlm_analyze (模型未获授权 view_image)" ) function_name = "vlm_analyze" else: denied_message = ( f"工具 {function_name} 不在当前模型可用工具列表中，已拒绝执行。" ) denied_payload = { "success": False, "status": "denied", "code": "tool_not_allowed", "tool": function_name, "message": denied_message, } sender('update_action', { 'preparing_id': tool_call_id, 'status': 'completed', 'result': denied_payload, 'message': denied_message, 'conversation_id': conversation_id }) denied_content = json.dumps(denied_payload, ensure_ascii=False) web_terminal.context_manager.add_conversation( "tool", denied_content, tool_call_id=tool_call_id, name=function_name ) messages.append({ "role": "tool", "tool_call_id": tool_call_id, "name": function_name, "content": denied_content }) continue debug_log(f"执行工具: {function_name} (ID: {tool_call_id})") previous_tool_actions = list(recent_tool_actions) recent_tool_actions.append({ "tool_name": function_name, "arguments": arguments, }) recent_tool_actions = recent_tool_actions[-3:] setattr(web_terminal, "_recent_tool_actions", list(recent_tool_actions)) permission_eval = web_terminal.evaluate_tool_permission(function_name, arguments) if not permission_eval.get("allowed", True): denied_message = permission_eval.get("message") or "当前权限模式不允许执行该工具。" denied_payload = { "success": False, "status": "denied", "code": permission_eval.get("code") or "permission_denied", "tool": function_name, "mode": permission_eval.get("mode"), "message": denied_message, } sender('update_action', { 'preparing_id': tool_call_id, 'status': 'completed', 'result': denied_payload, 'message': denied_message, 'conversation_id': conversation_id }) denied_content = json.dumps(denied_payload, ensure_ascii=False) web_terminal.context_manager.add_conversation( "tool", denied_content, tool_call_id=tool_call_id, name=function_name ) messages.append({ "role": "tool", "tool_call_id": tool_call_id, "name": function_name, "content": denied_content }) continue if permission_eval.get("requires_approval"): approval_preview = _build_tool_approval_preview(web_terminal, function_name, arguments) approval_item = tool_approval_manager.create_request( username=username, conversation_id=conversation_id, task_id=getattr(web_terminal, "task_id", None), tool_call_id=tool_call_id, tool_name=function_name, arguments=arguments, preview=approval_preview, ) sender('tool_approval_required', { 'approval': approval_item, 'conversation_id': conversation_id, }) sender('update_action', { 'preparing_id': tool_call_id, 'status': 'awaiting_approval', 'result': { "success": False, "status": "awaiting_approval", "approval_id": approval_item.get("approval_id"), "message": "等待用户审批" }, 'message': '等待用户审批', 'conversation_id': conversation_id }) wait_result = None permission_mode = str(permission_eval.get("mode") or "") if permission_mode == "auto_approval": approval_id = approval_item.get("approval_id") wait_result = await run_auto_approval( web_terminal=web_terminal, username=username, approval_id=approval_id, conversation_id=conversation_id, recent_tool_actions=previous_tool_actions, function_name=function_name, arguments=arguments, risk_markers=permission_eval.get("risk_markers") if isinstance(permission_eval, dict) else None, sender=sender, ) else: wait_result = await _wait_for_tool_approval( approval_id=approval_item.get("approval_id"), username=username, ) sender('tool_approval_resolved', { 'approval_id': approval_item.get("approval_id"), 'decision': wait_result.get("decision"), 'reason': ((wait_result.get("item") or {}).get("reason") or wait_result.get("reason")), 'conversation_id': conversation_id, }) if wait_result.get("decision") != "approved": reject_message = "操作被用户拒绝" if wait_result.get("reason") == "审批超时": reject_message = "审批超时，操作未执行" reason = ((wait_result.get("item") or {}).get("reason") or wait_result.get("reason") or "").strip() if reason: reject_message = f"{reject_message}：{reason}" reject_payload = { "success": False, "status": "rejected", "code": "approval_rejected", "tool": function_name, "message": reject_message, "approval_id": approval_item.get("approval_id"), } sender('update_action', { 'preparing_id': tool_call_id, 'status': 'completed', 'result': reject_payload, 'message': reject_message, 'conversation_id': conversation_id }) reject_content = _format_rejected_tool_text(reason or reject_message) web_terminal.context_manager.add_conversation( "tool", reject_content, tool_call_id=tool_call_id, name=function_name ) messages.append({ "role": "tool", "tool_call_id": tool_call_id, "name": function_name, "content": reject_content }) if permission_mode != "auto_approval": web_terminal._tool_loop_active = previous_tool_loop_active return { "stopped": False, "approval_rejected": True, "approval_message": reject_message, "last_tool_call_time": last_tool_call_time } continue # 发送工具开始事件 tool_display_id = f"tool_{iteration}_{function_name}_{time.time()}" monitor_snapshot = None snapshot_path = None memory_snapshot_type = None if function_name in MONITOR_FILE_TOOLS: snapshot_path = resolve_monitor_path(arguments) monitor_snapshot = capture_monitor_snapshot(web_terminal.file_manager, snapshot_path, MONITOR_SNAPSHOT_CHAR_LIMIT, debug_log) if monitor_snapshot: cache_monitor_snapshot(tool_display_id, 'before', monitor_snapshot) elif function_name in MONITOR_MEMORY_TOOLS: memory_snapshot_type = (arguments.get('memory_type') or 'main').lower() before_entries = None try: before_entries = resolve_monitor_memory(web_terminal.memory_manager._read_entries(memory_snapshot_type), MONITOR_MEMORY_ENTRY_LIMIT) except Exception as exc: debug_log(f"[MonitorSnapshot] 读取记忆失败: {memory_snapshot_type} ({exc})") if before_entries is not None: monitor_snapshot = { 'memory_type': memory_snapshot_type, 'entries': before_entries } cache_monitor_snapshot(tool_display_id, 'before', monitor_snapshot) sender('tool_start', { 'id': tool_display_id, 'name': function_name, 'arguments': arguments, 'preparing_id': tool_call_id, 'monitor_snapshot': monitor_snapshot, 'conversation_id': conversation_id }) brief_log(f"调用了工具: {function_name}") await asyncio.sleep(0.3) start_time = time.time() # 执行工具，同时监听停止标志 debug_log(f"[停止检测] 开始执行工具: {function_name}") tool_result = None tool_cancelled = False tool_task = None check_count = 0 if function_name == "ask_user" and str(tool_call_id) in user_question_results_by_tool_call_id: answer_text = user_question_results_by_tool_call_id.get(str(tool_call_id)) or "用户未回答。" qid = user_question_id_by_tool_call_id.get(str(tool_call_id)) tool_result = json.dumps({ "success": True, "status": "answered", "message": answer_text, "answer_text": answer_text, "question_id": qid, }, ensure_ascii=False) else: tool_task = asyncio.create_task(web_terminal.handle_tool_call(function_name, arguments)) # 在工具执行期间持续检查停止标志 while not tool_task.done(): await asyncio.sleep(0.1) # 每100ms检查一次 check_count += 1 client_stop_info = get_stop_flag(client_sid, username) if client_stop_info: stop_requested = client_stop_info.get('stop', False) if isinstance(client_stop_info, dict) else client_stop_info if stop_requested: debug_log(f"[停止检测] 工具执行过程中检测到停止请求(检查次数:{check_count})，立即取消工具") tool_task.cancel() tool_cancelled = True break debug_log(f"[停止检测] 工具执行完成，cancelled={tool_cancelled}, 检查次数={check_count}") # 获取工具结果或处理取消 if tool_cancelled: try: if tool_task is not None: await tool_task except asyncio.CancelledError: debug_log("[停止检测] 工具任务已被取消(CancelledError)") except Exception as e: debug_log(f"[停止检测] 工具任务取消时发生异常: {e}") # 返回取消消息 tool_result = json.dumps({ "success": False, "status": "cancelled", "message": "命令执行被用户取消" }, ensure_ascii=False) debug_log("[停止检测] 发送取消通知到前端") # 通知前端工具被取消 sender('update_action', { 'preparing_id': tool_call_id, 'status': 'cancelled', 'result': { "success": False, "status": "cancelled", "message": "命令执行被用户取消", "tool": function_name } }) # 记录取消结果到消息历史 messages.append({ "role": "tool", "tool_call_id": tool_call_id, "name": function_name, "content": "命令执行被用户取消", }) # 保存取消结果 web_terminal.context_manager.add_conversation( "tool", "命令执行被用户取消", tool_call_id=tool_call_id, name=function_name, metadata={"status": "cancelled"} ) debug_log("[停止检测] 取消结果已保存到对话历史") # 发送停止事件并清除标志 sender('task_stopped', { 'message': '命令执行被用户取消', 'reason': 'user_stop' }) clear_stop_flag(client_sid, username) debug_log("[停止检测] 返回stopped=True") web_terminal._tool_loop_active = previous_tool_loop_active return {"stopped": True, "last_tool_call_time": last_tool_call_time} else: if tool_result is None and tool_task is not None: tool_result = await tool_task if tool_result is None: tool_result = json.dumps({"success": False, "error": "工具未返回结果"}, ensure_ascii=False) debug_log(f"工具结果: {tool_result[:200]}...") execution_time = time.time() - start_time if execution_time < 1.5: await asyncio.sleep(1.5 - execution_time) # 更新工具状态 result_data = {} try: result_data = json.loads(tool_result) except: result_data = {'output': tool_result} # 批准模式下 run_command 采用“先只读执行，触发权限拒绝后再审批并单次可写重试”。 if ( function_name == "run_command" and permission_eval.get("mode") in {"approval", "auto_approval"} and not bool(arguments.get("_approval_write_granted", False)) and _is_permission_denied_result(result_data) ): approval_preview = _build_tool_approval_preview(web_terminal, function_name, arguments) approval_item = tool_approval_manager.create_request( username=username, conversation_id=conversation_id, task_id=getattr(web_terminal, "task_id", None), tool_call_id=tool_call_id, tool_name=function_name, arguments=arguments, preview=approval_preview, ) sender('tool_approval_required', { 'approval': approval_item, 'conversation_id': conversation_id, }) sender('update_action', { 'preparing_id': tool_call_id, 'status': 'awaiting_approval', 'result': { "success": False, "status": "awaiting_approval", "approval_id": approval_item.get("approval_id"), "message": "检测到写权限受限，等待用户审批后重试" }, 'message': '检测到写权限受限，等待用户审批', 'conversation_id': conversation_id }) wait_result = None permission_mode = str(permission_eval.get("mode") or "") if permission_mode == "auto_approval": approval_id = approval_item.get("approval_id") wait_result = await run_auto_approval( web_terminal=web_terminal, username=username, approval_id=approval_id, conversation_id=conversation_id, recent_tool_actions=previous_tool_actions, function_name=function_name, arguments=arguments, risk_markers=permission_eval.get("risk_markers") if isinstance(permission_eval, dict) else None, sender=sender, ) else: wait_result = await _wait_for_tool_approval( approval_id=approval_item.get("approval_id"), username=username, ) sender('tool_approval_resolved', { 'approval_id': approval_item.get("approval_id"), 'decision': wait_result.get("decision"), 'reason': ((wait_result.get("item") or {}).get("reason") or wait_result.get("reason")), 'conversation_id': conversation_id, }) if wait_result.get("decision") != "approved": reject_message = "操作被用户拒绝" if wait_result.get("reason") == "审批超时": reject_message = "审批超时，操作未执行" reason = ((wait_result.get("item") or {}).get("reason") or wait_result.get("reason") or "").strip() if reason: reject_message = f"{reject_message}：{reason}" reject_payload = { "success": False, "status": "rejected", "code": "approval_rejected", "tool": function_name, "message": reject_message, "approval_id": approval_item.get("approval_id"), } sender('update_action', { 'preparing_id': tool_call_id, 'status': 'completed', 'result': reject_payload, 'message': reject_message, 'conversation_id': conversation_id }) reject_content = _format_rejected_tool_text(reason or reject_message) web_terminal.context_manager.add_conversation( "tool", reject_content, tool_call_id=tool_call_id, name=function_name ) messages.append({ "role": "tool", "tool_call_id": tool_call_id, "name": function_name, "content": reject_content }) if permission_mode == "approval": web_terminal._tool_loop_active = previous_tool_loop_active return { "stopped": False, "approval_rejected": True, "approval_message": reject_message, "last_tool_call_time": last_tool_call_time } continue retry_arguments = dict(arguments or {}) retry_arguments["_approval_write_granted"] = True retry_tool_result = await web_terminal.handle_tool_call(function_name, retry_arguments) try: result_data = json.loads(retry_tool_result) tool_result = retry_tool_result except Exception: result_data = {"output": retry_tool_result} tool_result = retry_tool_result tool_failed = detect_tool_failure(result_data) action_status = 'completed' action_message = None awaiting_flag = False if function_name in {"write_file", "edit_file"}: diff_path = result_data.get("path") or arguments.get("file_path") summary = result_data.get("summary") or result_data.get("message") if summary: action_message = summary debug_log(f"{function_name} 执行完成: {summary or '无摘要'}") if function_name == "sleep": try: if isinstance(result_data, dict) and result_data.get("mode") == "wait_sub_agent_ids": waited_task_ids = result_data.get("waited_task_ids") or [] if not hasattr(web_terminal, "_announced_sub_agent_tasks"): web_terminal._announced_sub_agent_tasks = set() for tid in waited_task_ids: if tid: web_terminal._announced_sub_agent_tasks.add(str(tid)) except Exception: pass monitor_snapshot_after = None if function_name in MONITOR_FILE_TOOLS: result_path = None if isinstance(result_data, dict): result_path = resolve_monitor_path(result_data) if not result_path: candidate_path = result_data.get('path') if isinstance(candidate_path, str) and candidate_path.strip(): result_path = candidate_path.strip() if not result_path: result_path = resolve_monitor_path(arguments, snapshot_path) or snapshot_path monitor_snapshot_after = capture_monitor_snapshot(web_terminal.file_manager, result_path, MONITOR_SNAPSHOT_CHAR_LIMIT, debug_log) elif function_name in MONITOR_MEMORY_TOOLS: memory_after_type = str( arguments.get('memory_type') or (isinstance(result_data, dict) and result_data.get('memory_type')) or memory_snapshot_type or 'main' ).lower() after_entries = None try: after_entries = resolve_monitor_memory(web_terminal.memory_manager._read_entries(memory_after_type), MONITOR_MEMORY_ENTRY_LIMIT) except Exception as exc: debug_log(f"[MonitorSnapshot] 读取记忆失败(after): {memory_after_type} ({exc})") if after_entries is not None: monitor_snapshot_after = { 'memory_type': memory_after_type, 'entries': after_entries } mcp_parsed = None update_result_data = result_data if isinstance(function_name, str) and function_name.startswith("mcp__") and isinstance(result_data, dict): try: mcp_parsed = extract_mcp_content_for_context(result_data) update_result_data = mcp_parsed.get("sanitized_payload") or result_data except Exception: mcp_parsed = None update_result_data = result_data update_payload = { 'id': tool_display_id, 'status': action_status, 'result': update_result_data, 'preparing_id': tool_call_id, 'conversation_id': conversation_id } if action_message: update_payload['message'] = action_message if awaiting_flag: update_payload['awaiting_content'] = True if monitor_snapshot_after: update_payload['monitor_snapshot_after'] = monitor_snapshot_after cache_monitor_snapshot(tool_display_id, 'after', monitor_snapshot_after) sender('update_action', update_payload) if function_name in ['create_file', 'delete_file', 'rename_file', 'create_folder']: if not web_terminal.context_manager._is_host_mode_without_safety(): structure = web_terminal.context_manager.get_project_structure() sender('file_tree_update', structure) # ===== 增量保存：立即保存工具结果 ===== metadata_payload = None tool_images = None tool_videos = None tool_media_refs = None if isinstance(result_data, dict): if isinstance(function_name, str) and function_name.startswith("mcp__"): if not isinstance(mcp_parsed, dict): mcp_parsed = extract_mcp_content_for_context(result_data) tool_result_content = ( mcp_parsed.get("text") or format_tool_result_for_context(function_name, result_data, tool_result) ) mcp_media_items = mcp_parsed.get("media_items") or [] if mcp_media_items: tool_media_refs = [] for item in mcp_media_items: if not isinstance(item, dict): continue tool_media_refs.append( { "kind": item.get("kind"), "mime_type": item.get("mime_type"), "data_base64": item.get("data_base64"), "source": "mcp_content", "item_type": item.get("item_type"), "label": item.get("label"), "name": item.get("name"), "title": item.get("title"), "uri": item.get("uri"), "url": item.get("url"), "index": item.get("index"), } ) metadata_payload = { "tool_payload": mcp_parsed.get("sanitized_payload") or result_data } else: # 特殊处理 web_search：保留可供前端渲染的精简结构，以便历史记录复现搜索结果 if function_name == "web_search": try: tool_result_content = json.dumps(compact_web_search_result(result_data), ensure_ascii=False) except Exception: tool_result_content = tool_result else: tool_result_content = format_tool_result_for_context(function_name, result_data, tool_result) metadata_payload = {"tool_payload": result_data} else: tool_result_content = tool_result tool_message_content = tool_result_content # view_image: 将图片直接附加到 tool 结果中（不再插入 user 消息） if function_name == "view_image" and getattr(web_terminal, "pending_image_view", None): inj = web_terminal.pending_image_view web_terminal.pending_image_view = None if ( not tool_failed and isinstance(result_data, dict) and result_data.get("success") is not False ): img_path = inj.get("path") if isinstance(inj, dict) else None if img_path: tool_images = [img_path] if metadata_payload is None: metadata_payload = {} metadata_payload["tool_image_path"] = img_path sender('system_message', { 'content': f'系统已记录图片路径（不再附带二进制数据）: {img_path}' }) # view_video: 将视频直接附加到 tool 结果中（不再插入 user 消息） if function_name == "view_video" and getattr(web_terminal, "pending_video_view", None): inj = web_terminal.pending_video_view web_terminal.pending_video_view = None if ( not tool_failed and isinstance(result_data, dict) and result_data.get("success") is not False ): video_path = inj.get("path") if isinstance(inj, dict) else None if video_path: tool_videos = [video_path] if metadata_payload is None: metadata_payload = {} metadata_payload["tool_video_path"] = video_path sender('system_message', { 'content': f'系统已记录视频路径（不再附带二进制数据）: {video_path}' }) # 立即保存工具结果 saved_tool_message = web_terminal.context_manager.add_conversation( "tool", tool_result_content, tool_call_id=tool_call_id, name=function_name, metadata=metadata_payload, images=tool_images, videos=tool_videos, media_refs=tool_media_refs, ) saved_media_refs = [] if isinstance(saved_tool_message, dict): saved_media_refs = saved_tool_message.get("media_refs") or [] # 将工具结果即时组装为多模态消息，确保同一轮 tool loop 的下一次模型请求能真正看到媒体 if tool_images or tool_videos or saved_media_refs: try: tool_message_content = web_terminal.context_manager._build_content_with_images( str(tool_result_content or ""), tool_images or [], tool_videos or [], media_refs=saved_media_refs, ) except Exception: tool_message_content = tool_result_content # 添加到消息历史（用于 API 继续对话）。 # 必须在任何可能插入 user 消息/触发深层压缩总结之前完成，避免 assistant.tool_calls # 与对应 tool 结果之间夹入 user，尤其是同一轮并行 tool_calls 尚未全部返回时。 messages.append({ "role": "tool", "tool_call_id": tool_call_id, "name": function_name, "content": tool_message_content }) last_completed_tool_call_id = tool_call_id try: workspace_data_dir = getattr(workspace, "data_dir", None) if workspace else None personal_config = load_personalization_config(workspace_data_dir) if workspace_data_dir else {} except Exception: personal_config = {} compression_settings = resolve_context_compression_settings(personal_config) auto_shallow_enabled = bool(personal_config.get("auto_shallow_compress_enabled", False)) auto_deep_enabled = bool(personal_config.get("auto_deep_compress_enabled", False)) compressed_count = 0 try: compressed_count = int( web_terminal.context_manager.on_tool_call_finished( function_name, enable_shallow=auto_shallow_enabled, shallow_trigger_tokens=compression_settings["shallow_trigger_tokens"], deep_trigger_tokens=compression_settings["deep_trigger_tokens"], shallow_batch_size=compression_settings["shallow_max_replace_per_round"], shallow_keep_recent_tools=compression_settings["shallow_keep_recent_tools"], shallow_trigger_tool_calls_interval=compression_settings["shallow_trigger_tool_calls_interval"], shallow_keep_user_turn_tools=compression_settings.get("shallow_keep_user_turn_tools", 3), ) or 0 ) except Exception as exc: debug_log(f"[ContextCompression] 工具后浅压缩检测失败: {exc}") if compressed_count > 0: sender('shallow_compression', { "conversation_id": conversation_id, "compressed_count": compressed_count, "keep_recent_tools": compression_settings["shallow_keep_recent_tools"], "keep_user_turn_tools": compression_settings.get("shallow_keep_user_turn_tools", 3), }) # 关键修复：同一轮工具循环里，API 继续调用使用的是本地 messages（不是每次重建上下文）， # 因此需要把已打标的旧 tool 结果同步替换为占位符，避免日志看起来“弹窗触发但请求未替换”。 if auto_shallow_enabled and messages: try: marked_keys = set() for item in (web_terminal.context_manager.conversation_history or []): if not isinstance(item, dict) or item.get("role") != "tool": continue item_meta = item.get("metadata") or {} if not item_meta.get("auto_shallow_compacted"): continue marked_keys.add((str(item.get("tool_call_id") or ""), str(item.get("name") or ""))) if marked_keys: replaced_in_loop = 0 for msg in messages: if not isinstance(msg, dict) or msg.get("role") != "tool": continue key = (str(msg.get("tool_call_id") or ""), str(msg.get("name") or "")) if key in marked_keys and msg.get("content") != AUTO_SHALLOW_PLACEHOLDER: msg["content"] = AUTO_SHALLOW_PLACEHOLDER replaced_in_loop += 1 if replaced_in_loop > 0: debug_log(f"[ContextCompression] 同步替换本轮 messages 中已压缩 tool 结果: {replaced_in_loop}") except Exception as exc: debug_log(f"[ContextCompression] 同步替换本轮 messages 失败: {exc}") debug_log(f"💾 增量保存：工具结果 {function_name}") system_message = result_data.get("system_message") if isinstance(result_data, dict) else None if system_message: inject_runtime_user_message( web_terminal=web_terminal, messages=messages, text=system_message, source="sub_agent", sender=sender, conversation_id=conversation_id, after_tool_call_id=tool_call_id, inline=False, extra_metadata={"task_id": result_data.get("task_id")}, ) maybe_mark_failure_from_message(web_terminal, system_message) # 自动深层压缩（工具调用后触发） current_context_tokens = web_terminal.context_manager.get_current_context_tokens(conversation_id) if ( auto_deep_enabled and current_context_tokens > compression_settings["deep_trigger_tokens"] and not web_terminal.context_manager.is_compression_in_progress() ): # 只标记，不能在本工具刚结束时立即压缩。若本轮是并行 tool_calls， # 立即调用总结模型会在 assistant.tool_calls 与尚未补齐的 tool 结果之间插入 user prompt， # 触发 OpenAI 兼容接口的 tool message 顺序校验错误。 deep_compression_pending = True if not pending_runtime_mode_notices: try: if hasattr(web_terminal, "apply_pending_runtime_mode_changes"): pending_runtime_mode_notices = list(web_terminal.apply_pending_runtime_mode_changes() or []) except Exception as exc: debug_log(f"[RuntimeMode] 应用挂起模式变更失败: {exc}") await asyncio.sleep(0.2) if tool_failed: mark_force_thinking(web_terminal, reason=f"{function_name}_failed") # 自动深层压缩必须等待同一轮全部 tool_call 的 tool 消息都已写入 messages/历史后再触发。 if deep_compression_pending and not web_terminal.context_manager.is_compression_in_progress(): web_terminal.context_manager._set_meta_flag("is_ultra_long_conversation", True) sender('compression_state', { "conversation_id": conversation_id, "in_progress": True, "mode": "auto", "stage": "queued" }) deep_result = await run_deep_compression( web_terminal=web_terminal, workspace=workspace, conversation_id=conversation_id, mode="auto", sender=sender, ) if not deep_result.get("success"): sender('error', { "message": deep_result.get("error") or "自动深层压缩失败", "conversation_id": conversation_id, }) web_terminal._tool_loop_active = previous_tool_loop_active return { "stopped": False, "deep_compressed": True, "deep_result": deep_result, "last_tool_call_time": last_tool_call_time } # 子智能体/后台指令完成通知：必须等待同一轮全部 tool_call 都完成后再注入， # 避免在 assistant.tool_calls 与未完成的 tool 结果之间插入 system 消息导致 API 报错。 if last_completed_tool_call_id: debug_log( f"[SubAgent] after all tools finished -> poll updates inline after_tool_call_id={last_completed_tool_call_id}" ) await process_sub_agent_updates( messages=messages, inline=True, after_tool_call_id=last_completed_tool_call_id, web_terminal=web_terminal, sender=sender, debug_log=debug_log, maybe_mark_failure_from_message=maybe_mark_failure_from_message, ) debug_log( "[BgCmdDebug] after all tools finished -> poll background command updates " f"inline after_tool_call_id={last_completed_tool_call_id}" ) await process_background_command_updates( messages=messages, inline=True, after_tool_call_id=last_completed_tool_call_id, web_terminal=web_terminal, sender=sender, debug_log=debug_log, maybe_mark_failure_from_message=maybe_mark_failure_from_message, ) # 运行期模式通知：必须等待同一轮全部 tool_call 都完成后再注入， # 避免在 assistant.tool_calls 与对应 tool 消息之间插入 user 消息导致 API 报错。 if pending_runtime_mode_notices: for notice in pending_runtime_mode_notices: _inject_runtime_mode_notice( web_terminal=web_terminal, messages=messages, content=notice, sender=sender, conversation_id=conversation_id, ) # 运行期“引导对话”：需要等待同一轮全部工具执行结束后再注入， # 避免一轮内并行/多工具调用时过早插入。 try: from .tasks import task_manager runtime_guidance_items = task_manager.consume_runtime_guidance_for_injection( username=username, task_id=client_sid ) except Exception as exc: runtime_guidance_items = [] debug_log(f"[RuntimeGuidance] 读取引导队列失败: {exc}") if runtime_guidance_items: injected_count = 0 for raw_item in runtime_guidance_items: runtime_guidance_source = "guidance" if isinstance(raw_item, dict): runtime_guidance_text = str(raw_item.get("text") or "").strip() runtime_guidance_source = ( str(raw_item.get("source") or "guidance").strip().lower() or "guidance" ) else: runtime_guidance_text = str(raw_item or "").strip() if not runtime_guidance_text: continue inject_runtime_user_message( web_terminal=web_terminal, messages=messages, text=runtime_guidance_text, source=runtime_guidance_source, sender=sender, conversation_id=conversation_id, inline=True, ) injected_count += 1 if injected_count: debug_log( "[RuntimeGuidance] 已在工具结果批次结束后批量注入引导/通知 " f"task_id={client_sid} count={injected_count}" ) web_terminal._tool_loop_active = previous_tool_loop_active return {"stopped": False, "last_tool_call_time": last_tool_call_time}