diff --git a/core/main_terminal_parts/tools_execution.py b/core/main_terminal_parts/tools_execution.py index 9b45299..0fc7718 100644 --- a/core/main_terminal_parts/tools_execution.py +++ b/core/main_terminal_parts/tools_execution.py @@ -90,9 +90,6 @@ class MainTerminalToolsExecutionMixin: "terminal_session", "terminal_input", "terminal_snapshot", - "sleep", - "run_command", - "run_python", } _SUB_AGENT_SERIES_TOOLS = { "create_sub_agent", @@ -126,6 +123,10 @@ class MainTerminalToolsExecutionMixin: def _skill_meta_key(skill_id: str) -> str: return f"skill_read::{skill_id}" + @staticmethod + def _file_read_meta_key() -> str: + return "read_file::visited_paths" + def _normalize_tool_path(self, path: Any) -> str: raw = str(path or "").strip().replace("\\", "/") if not raw: @@ -160,6 +161,85 @@ class MainTerminalToolsExecutionMixin: if normalized.endswith("skills/sub-agent-guide/skill.md"): self.context_manager._set_meta_flag(self._skill_meta_key("sub-agent-guide"), True) + def _get_visited_read_files(self) -> Set[str]: + raw = self.context_manager._get_meta_flag(self._file_read_meta_key(), []) + if isinstance(raw, set): + items = raw + elif isinstance(raw, list): + items = set(raw) + elif isinstance(raw, tuple): + items = set(raw) + else: + items = set() + + visited: Set[str] = set() + for item in items: + normalized = self._normalize_tool_path(item) + if normalized: + visited.add(normalized) + return visited + + def _mark_file_read_from_result(self, tool_name: str, arguments: Dict[str, Any], result: Dict[str, Any]) -> None: + if tool_name != "read_file": + return + if not isinstance(result, dict) or not result.get("success"): + return + normalized = self._normalize_tool_path(result.get("path") or arguments.get("path")) + if not normalized: + return + visited = self._get_visited_read_files() + if normalized in visited: + return + visited.add(normalized) + self.context_manager._set_meta_flag( + self._file_read_meta_key(), + sorted(visited), + ) + + def _has_read_file_in_conversation(self, path: Any) -> bool: + normalized = self._normalize_tool_path(path) + if not normalized: + return False + return normalized in self._get_visited_read_files() + + def _target_file_exists(self, path: Any) -> bool: + try: + valid, _, full_path = self.file_manager._validate_path(str(path or "")) + if not valid or full_path is None: + return False + return full_path.exists() and full_path.is_file() + except Exception: + return False + + def _check_read_before_edit_prerequisite(self, tool_name: str, arguments: Dict[str, Any]) -> Optional[Dict[str, Any]]: + target_path = arguments.get("file_path") + if not target_path: + return None + + if tool_name == "write_file": + append_mode = bool(arguments.get("append", False)) + if append_mode: + return None + # 覆盖模式仅对“已存在文件”执行先读后写约束; + # 新文件创建仍允许直接 write_file。 + if not self._target_file_exists(target_path): + return None + + if self._has_read_file_in_conversation(target_path): + return None + + return { + "success": False, + "error": f"调用 {tool_name} 前必须先使用 read_file 阅读目标文件", + "message": ( + f"文件 {target_path} 尚未在当前对话中通过 read_file 阅读。" + "请先使用 read_file(read/search/extract 任一模式)读取后再编辑。" + ), + "required_tool": "read_file", + "required_path": str(target_path), + "enforcement": "read_before_edit_required", + } + def _is_skill_enabled_for_enforcement(self, skill_id: str) -> bool: enabled_skill_ids = getattr(self, "enabled_skill_ids", None) if isinstance(enabled_skill_ids, set): @@ -273,6 +353,7 @@ class MainTerminalToolsExecutionMixin: elif tool_name == "read_file": result = self._handle_read_tool(arguments) self._mark_skill_read_from_result(tool_name, arguments, result) + self._mark_file_read_from_result(tool_name, arguments, result) elif tool_name in {"vlm_analyze", "ocr_image"}: path = arguments.get("path") prompt = arguments.get("prompt") @@ -479,6 +560,9 @@ class MainTerminalToolsExecutionMixin: print(f"📝 已更新文件备注: {old_path} -> {new_path}") elif tool_name == "write_file": + read_guard_error = self._check_read_before_edit_prerequisite(tool_name, arguments) + if read_guard_error: + return json.dumps(read_guard_error, ensure_ascii=False) path = arguments.get("file_path") content = arguments.get("content", "") append_flag = bool(arguments.get("append", False)) @@ -489,6 +573,9 @@ class MainTerminalToolsExecutionMixin: result = self.file_manager.write_file(path, content, mode=mode) elif tool_name == "edit_file": + read_guard_error = self._check_read_before_edit_prerequisite(tool_name, arguments) + if read_guard_error: + return json.dumps(read_guard_error, ensure_ascii=False) path = arguments.get("file_path") old_text = arguments.get("old_string") new_text = arguments.get("new_string")