fix: 调整终端强约束范围并新增先读后改文件校验
This commit is contained in:
parent
2fd4782f5c
commit
c4e4e0f576
@ -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")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user