fix: align toolbox python env for run_command

modules/terminal_ops.py

@@ -7,6 +7,7 @@ import subprocess
 import shutil
 import time
 import signal
+import re
 from pathlib import Path
 from typing import Dict, Optional, Tuple, TYPE_CHECKING
 try:
@@ -39,8 +40,9 @@ class TerminalOperator:
     def __init__(self, project_path: str, container_session: Optional["ContainerHandle"] = None):
         self.project_path = Path(project_path).resolve()
         self.process = None
-        # 自动检测Python命令
+        # 自动检测Python命令，并记录虚拟环境变量（若存在）
-        self.python_cmd = self._detect_python_command()
+        self._python_env: Dict[str, str] = {}
+        self.python_cmd = self._detect_python_runtime()
         print(f"{OUTPUT_FORMATS['info']} 检测到Python命令: {self.python_cmd}")
         self._toolbox: Optional[ToolboxContainer] = None
         self.container_session: Optional["ContainerHandle"] = container_session
@@ -54,28 +56,64 @@ class TerminalOperator:
                 pass
             self._toolbox = None
     
-    def _detect_python_command(self) -> str:
+    def _detect_python_runtime(self) -> str:
         """
-        自动检测可用的Python命令
+        自动检测可用的 Python 命令，优先使用预装虚拟环境。
 
-        Returns:
+        1) 优先查找 AGENT_TOOLBOX_VENV /opt/agent-venv / 当前进程 VIRTUAL_ENV。
-            可用的Python命令（python、python3、py）
+        2) 找不到预装虚拟环境时，回退到系统可执行文件。
         """
-        # 按优先级尝试不同的Python命令
+        preferred = self._detect_preinstalled_python()
+        if preferred:
+            # 记录虚拟环境相关环境变量，供宿主机子进程复用
+            self._python_env = self._build_python_env(preferred)
+            return preferred
+
+        return self._detect_system_python()
+
+    def _detect_preinstalled_python(self) -> Optional[str]:
+        """尝试定位预装虚拟环境的 python 可执行文件。"""
+        candidates = []
+
+        # 环境变量优先（Dockerfile 已设置 AGENT_TOOLBOX_VENV=/opt/agent-venv）
+        env_venv = os.environ.get("AGENT_TOOLBOX_VENV")
+        if env_venv:
+            candidates.append(env_venv)
+
+        # 默认预装路径
+        candidates.append("/opt/agent-venv")
+
+        # 当前进程若已激活虚拟环境，也纳入候选
+        current_venv = os.environ.get("VIRTUAL_ENV")
+        if current_venv:
+            candidates.append(current_venv)
+
+        # 去重并依次检查
+        seen = set()
+        for raw in candidates:
+            if not raw or raw in seen:
+                continue
+            seen.add(raw)
+            root = Path(raw).expanduser()
+            bin_dir = root / ("Scripts" if sys.platform == "win32" else "bin")
+            for name in ("python3", "python"):
+                candidate = bin_dir / name
+                if candidate.exists() and os.access(candidate, os.X_OK):
+                    return str(candidate.resolve())
+        return None
+
+    def _detect_system_python(self) -> str:
+        """在系统 PATH 中探测可用的 Python3 可执行文件。"""
         commands_to_try = []
 
         if sys.platform == "win32":
-            # Windows优先顺序
             commands_to_try = ["python", "py", "python3"]
         else:
-            # Unix-like系统优先顺序
             commands_to_try = ["python3", "python"]
 
-        # 检测哪个命令可用
         for cmd in commands_to_try:
             if shutil.which(cmd):
                 try:
-                    # 验证是否真的可以运行
                     result = subprocess.run(
                         [cmd, "--version"],
                         capture_output=True,
@@ -83,16 +121,33 @@ class TerminalOperator:
                         timeout=5
                     )
                     if result.returncode == 0:
-                        # 检查版本是否为Python 3
                         output = result.stdout + result.stderr
                         if "Python 3" in output or "Python 2" not in output:
                             return cmd
-                except:
+                except Exception:
                     continue
 
-        # 如果都没找到，根据平台返回默认值
         return "python" if sys.platform == "win32" else "python3"
 
+    def _build_python_env(self, python_path: str) -> Dict[str, str]:
+        """构造与预装虚拟环境匹配的环境变量（仅作用于宿主机子进程）。"""
+        env: Dict[str, str] = {}
+        try:
+            py_path = Path(python_path).resolve()
+            bin_dir = py_path.parent
+            venv_dir = bin_dir.parent
+            env["VIRTUAL_ENV"] = str(venv_dir)
+            current_path = os.environ.get("PATH", "")
+            # 避免重复添加
+            prefix = str(bin_dir)
+            if current_path.startswith(prefix + os.pathsep) or current_path == prefix:
+                env["PATH"] = current_path
+            else:
+                env["PATH"] = f"{prefix}{os.pathsep}{current_path}" if current_path else prefix
+        except Exception:
+            pass
+        return env
+    
     def _get_toolbox(self) -> ToolboxContainer:
         if self._toolbox is None:
             self._toolbox = ToolboxContainer(
@@ -165,12 +220,9 @@ class TerminalOperator:
             }
         # 每次执行前重置工具容器（保持隔离），但下面改用一次性子进程执行，仍保留重置以兼容后续逻辑
         self._reset_toolbox()
-        # 替换命令中的python3为实际可用的命令
+        # 统一替换 python/python3，为保障虚拟环境命中只替换独立单词
-        if "python3" in command and self.python_cmd != "python3":
+        if re.search(r"\bpython3?\b", command):
-            command = command.replace("python3", self.python_cmd)
+            command = re.sub(r"\bpython3?\b", self.python_cmd, command)
-        elif "python" in command and "python3" not in command and self.python_cmd == "python3":
-            # 如果命令中有python（但不是python3），而系统使用python3
-            command = command.replace("python", self.python_cmd)
         
         # 验证命令
         valid, error = self._validate_command(command)
@@ -290,6 +342,8 @@ class TerminalOperator:
             # 统一环境，确保 Python 输出无缓冲
             env = os.environ.copy()
             env.setdefault("PYTHONUNBUFFERED", "1")
+            if self._python_env:
+                env.update(self._python_env)
 
             if use_shell:
                 process = await asyncio.create_subprocess_shell(

sub_agent/modules/terminal_ops.py

@@ -5,6 +5,7 @@ import sys
 import asyncio
 import subprocess
 import shutil
+import re
 from pathlib import Path
 from typing import Dict, Optional, Tuple
 try:
@@ -29,32 +30,49 @@ class TerminalOperator:
     def __init__(self, project_path: str):
         self.project_path = Path(project_path).resolve()
         self.process = None
-        # 自动检测Python命令
+        # 自动检测Python命令，并尝试复用预装虚拟环境
-        self.python_cmd = self._detect_python_command()
+        self._python_env: Dict[str, str] = {}
+        self.python_cmd = self._detect_python_runtime()
         print(f"{OUTPUT_FORMATS['info']} 检测到Python命令: {self.python_cmd}")
     
-    def _detect_python_command(self) -> str:
+    def _detect_python_runtime(self) -> str:
-        """
+        """优先选择预装虚拟环境的 Python，其次回退系统可执行文件。"""
-        自动检测可用的Python命令
+        preferred = self._detect_preinstalled_python()
+        if preferred:
+            self._python_env = self._build_python_env(preferred)
+            return preferred
+        return self._detect_system_python()
 
-        Returns:
+    def _detect_preinstalled_python(self) -> Optional[str]:
-            可用的Python命令（python、python3、py）
+        candidates = []
-        """
+        env_venv = os.environ.get("AGENT_TOOLBOX_VENV") or os.environ.get("VIRTUAL_ENV")
-        # 按优先级尝试不同的Python命令
+        if env_venv:
-        commands_to_try = []
+            candidates.append(env_venv)
 
+        candidates.append("/opt/agent-venv")
+
+        seen = set()
+        for raw in candidates:
+            if not raw or raw in seen:
+                continue
+            seen.add(raw)
+            root = Path(raw).expanduser()
+            bin_dir = root / ("Scripts" if sys.platform == "win32" else "bin")
+            for name in ("python3", "python"):
+                cand = bin_dir / name
+                if cand.exists() and os.access(cand, os.X_OK):
+                    return str(cand.resolve())
+        return None
+
+    def _detect_system_python(self) -> str:
         if sys.platform == "win32":
-            # Windows优先顺序
             commands_to_try = ["python", "py", "python3"]
         else:
-            # Unix-like系统优先顺序
             commands_to_try = ["python3", "python"]
 
-        # 检测哪个命令可用
         for cmd in commands_to_try:
             if shutil.which(cmd):
                 try:
-                    # 验证是否真的可以运行
                     result = subprocess.run(
                         [cmd, "--version"],
                         capture_output=True,
@@ -62,16 +80,30 @@ class TerminalOperator:
                         timeout=5
                     )
                     if result.returncode == 0:
-                        # 检查版本是否为Python 3
                         output = result.stdout + result.stderr
                         if "Python 3" in output or "Python 2" not in output:
                             return cmd
-                except:
+                except Exception:
                     continue
-        
-        # 如果都没找到，根据平台返回默认值
         return "python" if sys.platform == "win32" else "python3"
 
+    def _build_python_env(self, python_path: str) -> Dict[str, str]:
+        env: Dict[str, str] = {}
+        try:
+            py_path = Path(python_path).resolve()
+            bin_dir = py_path.parent
+            venv_dir = bin_dir.parent
+            env["VIRTUAL_ENV"] = str(venv_dir)
+            current_path = os.environ.get("PATH", "")
+            prefix = str(bin_dir)
+            if current_path.startswith(prefix + os.pathsep) or current_path == prefix:
+                env["PATH"] = current_path
+            else:
+                env["PATH"] = f"{prefix}{os.pathsep}{current_path}" if current_path else prefix
+        except Exception:
+            pass
+        return env
+        
     def _validate_command(self, command: str) -> Tuple[bool, str]:
         """验证命令安全性"""
         # 检查禁止的命令
@@ -111,12 +143,9 @@ class TerminalOperator:
         Returns:
             执行结果字典
         """
-        # 替换命令中的python3为实际可用的命令
+        # 替换 python/python3，确保命中预装虚拟环境
-        if "python3" in command and self.python_cmd != "python3":
+        if re.search(r"\bpython3?\b", command):
-            command = command.replace("python3", self.python_cmd)
+            command = re.sub(r"\bpython3?\b", self.python_cmd, command)
-        elif "python" in command and "python3" not in command and self.python_cmd == "python3":
-            # 如果命令中有python（但不是python3），而系统使用python3
-            command = command.replace("python", self.python_cmd)
         
         # 验证命令
         valid, error = self._validate_command(command)
@@ -151,12 +180,17 @@ class TerminalOperator:
         
         try:
             # 创建进程
+            env = os.environ.copy()
+            if self._python_env:
+                env.update(self._python_env)
+
             process = await asyncio.create_subprocess_shell(
                 command,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
                 cwd=str(work_path),
-                shell=True
+                shell=True,
+                env=env
             )
             
             # 等待执行完成