fix: reuse terminal container for command exec

core/main_terminal.py

@@ -127,6 +127,8 @@ class MainTerminal:
             broadcast_callback=None,  # CLI模式不需要广播
             container_session=container_session,
         )
+        # 让 run_command/run_python 复用终端容器，保持环境一致
+        self.terminal_ops.attach_terminal_manager(self.terminal_manager)
         self._apply_container_session(container_session)
 
         self.todo_manager = TodoManager(self.context_manager)

core/web_terminal.py

@@ -82,6 +82,8 @@ class WebTerminal(MainTerminal):
             broadcast_callback=message_callback,
             container_session=self.container_session
         )
+        # 让 run_command/run_python 与实时终端共享同一容器环境
+        self.terminal_ops.attach_terminal_manager(self.terminal_manager)
         
         print(f"[WebTerminal] 初始化完成，项目路径: {project_path}")
         print(f"[WebTerminal] 初始模式: {self.run_mode}")

modules/terminal_ops.py

@@ -10,6 +10,7 @@ import signal
 import re
 from pathlib import Path
 from typing import Dict, Optional, Tuple, TYPE_CHECKING
+from types import SimpleNamespace
 try:
     from config import (
         CODE_EXECUTION_TIMEOUT,
@@ -35,6 +36,7 @@ from modules.toolbox_container import ToolboxContainer
 
 if TYPE_CHECKING:
     from modules.user_container_manager import ContainerHandle
+    from modules.terminal_manager import TerminalManager
 
 class TerminalOperator:
     def __init__(self, project_path: str, container_session: Optional["ContainerHandle"] = None):
@@ -43,11 +45,13 @@ class TerminalOperator:
         # 自动检测Python命令，并记录虚拟环境变量（仅宿主机使用）
         self._python_env: Dict[str, str] = {}
         self.python_cmd = self._detect_python_runtime()
-        # Docker 容器内的 Python 命令（镜像内已将 /opt/agent-venv/bin 置于 PATH）
+        # Docker 容器内的 Python 命令（默认指向预装 venv）
-        self.container_python_cmd = os.environ.get("CONTAINER_PYTHON_CMD", "python3")
+        self.container_python_cmd = os.environ.get("CONTAINER_PYTHON_CMD", "/opt/agent-venv/bin/python3")
         print(f"{OUTPUT_FORMATS['info']} 检测到Python命令: {self.python_cmd}")
         self._toolbox: Optional[ToolboxContainer] = None
         self.container_session: Optional["ContainerHandle"] = container_session
+        # 记录 TerminalManager 引用，便于 CLI 场景复用同一容器
+        self._terminal_manager: Optional["TerminalManager"] = None
 
     def _reset_toolbox(self):
         """强制关闭并重建工具终端，保证每次命令/脚本运行独立环境。"""
@@ -73,6 +77,58 @@ class TerminalOperator:
 
         return self._detect_system_python()
 
+    @staticmethod
+    def _derive_pip_from_python(python_path: str) -> str:
+        """
+        根据 python 可执行文件推导匹配的 pip，可避免“python 在 venv、pip 却指向系统”。
+        若同目录下找不到 pip3/pip，则回退为 `<python> -m pip`。
+        """
+        try:
+            bin_dir = Path(python_path).resolve().parent
+            for name in ("pip3", "pip"):
+                cand = bin_dir / name
+                if cand.exists() and os.access(cand, os.X_OK):
+                    return str(cand)
+        except Exception:
+            pass
+        return f"{python_path} -m pip"
+
+    def attach_terminal_manager(self, manager: Optional["TerminalManager"]):
+        """由 MainTerminal/WebTerminal 注入 TerminalManager，便于共享终端容器。"""
+        self._terminal_manager = manager
+
+    def _resolve_active_container_session(self) -> Optional[SimpleNamespace]:
+        """
+        若已存在活动终端且在容器内运行，返回一个临时的容器句柄，
+        以便 run_command/run_python 复用同一个容器环境。
+        """
+        manager = self._terminal_manager
+        if not manager:
+            return None
+        active_name = getattr(manager, "active_terminal", None)
+        if not active_name:
+            return None
+        terminal = manager.terminals.get(active_name) if getattr(manager, "terminals", None) else None
+        if not terminal or not getattr(terminal, "using_container", False):
+            return None
+        container_name = getattr(terminal, "sandbox_container_name", None)
+        if not container_name:
+            return None
+        try:
+            mount_path = (terminal.sandbox_options.get("mount_path") or "/workspace").rstrip("/") or "/workspace"
+        except Exception:
+            mount_path = "/workspace"
+        return SimpleNamespace(mode="docker", container_name=container_name, mount_path=mount_path)
+
+    def _will_use_container(self, session_override: Optional["ContainerHandle"]) -> bool:
+        """根据会话/回退策略判断此次执行是否在容器中进行。"""
+        if session_override:
+            return getattr(session_override, "mode", None) == "docker"
+        if self.container_session:
+            return getattr(self.container_session, "mode", None) == "docker"
+        # 未绑定容器会话时会使用工具箱容器（同样是 Docker）
+        return True
+
     def _detect_preinstalled_python(self) -> Optional[str]:
         """尝试定位预装虚拟环境的 python 可执行文件。"""
         candidates = []
@@ -222,11 +278,20 @@ class TerminalOperator:
             }
         # 每次执行前重置工具容器（保持隔离），但下面改用一次性子进程执行，仍保留重置以兼容后续逻辑
         self._reset_toolbox()
-        execution_in_container = bool(self.container_session and getattr(self.container_session, "mode", None) == "docker")
+        # 尝试复用活动终端的容器（CLI 场景与 terminal_input 环境保持一致）
+        session_override = None
+        if not self.container_session:
+            session_override = self._resolve_active_container_session()
+        # 判断本次应在容器中执行：已绑定容器会话、复用终端容器或将使用工具箱容器
+        execution_in_container = self._will_use_container(session_override)
         python_rewrite = self.container_python_cmd if execution_in_container else self.python_cmd
+        pip_rewrite = self._derive_pip_from_python(python_rewrite)
         # 统一替换 python/python3，为保障虚拟环境命中只替换独立单词
         if re.search(r"\bpython3?\b", command):
             command = re.sub(r"\bpython3?\b", python_rewrite, command)
+        # 同步替换 pip/pip3，确保指向同一环境
+        if re.search(r"(?<![/.\w-])pip3?\b", command):
+            command = re.sub(r"(?<![/.\w-])pip3?\b", pip_rewrite, command)
         
         # 验证命令
         valid, error = self._validate_command(command)
@@ -254,10 +319,42 @@ class TerminalOperator:
         
         print(f"{OUTPUT_FORMATS['terminal']} 执行命令: {command}")
         print(f"{OUTPUT_FORMATS['info']} 工作目录: {work_path}")
-        
+
         start_ts = time.time()
+
+        # 优先在绑定的容器或活动终端的容器内执行，保证与实时终端环境一致
+        if self.container_session or session_override:
+            result_payload = await self._run_command_subprocess(
+                command,
+                work_path,
+                timeout,
+                session_override=session_override
+            )
+        else:
+            # 若未绑定用户容器，则使用工具箱容器（与终端相同镜像/预装包）
+            toolbox = self._get_toolbox()
+            payload = await toolbox.run(command, work_path, timeout)
+            result_payload = self._format_toolbox_output(payload)
+            # 追加耗时信息以对齐接口
+            result_payload["elapsed_ms"] = int((time.time() - start_ts) * 1000)
+            result_payload["timeout"] = timeout
+            # 字符数检查（与主流程一致）
+            if result_payload.get("success") and "output" in result_payload:
+                char_count = len(result_payload["output"])
+                if char_count > MAX_RUN_COMMAND_CHARS:
+                    return {
+                        "success": False,
+                        "error": f"结果内容过大，有{char_count}字符，请使用限制字符数的获取内容方式，根据程度选择10k以内的数",
+                        "char_count": char_count,
+                        "limit": MAX_RUN_COMMAND_CHARS,
+                        "command": command
+                    }
+            return result_payload
+
         # 改为一次性子进程执行，确保等待到超时或命令结束
-        result_payload = await self._run_command_subprocess(command, work_path, timeout)
+        result_payload = result_payload if result_payload is not None else await self._run_command_subprocess(
+            command, work_path, timeout
+        )
         
         # 字符数检查
         if result_payload.get("success") and "output" in result_payload:
@@ -312,17 +409,24 @@ class TerminalOperator:
             result["truncated"] = payload["truncated"]
         return result
     
-    async def _run_command_subprocess(self, command: str, work_path: Path, timeout: int) -> Dict:
+    async def _run_command_subprocess(
+        self,
+        command: str,
+        work_path: Path,
+        timeout: int,
+        session_override: Optional["ContainerHandle"] = None
+    ) -> Dict:
         start_ts = time.time()
         try:
             process = None
             exec_cmd = None
             use_shell = True
+            session = session_override or self.container_session
 
             # 如果存在容器会话且模式为docker，则在容器内执行
-            if self.container_session and getattr(self.container_session, "mode", None) == "docker":
+            if session and getattr(session, "mode", None) == "docker":
-                container_name = getattr(self.container_session, "container_name", None)
+                container_name = getattr(session, "container_name", None)
-                mount_path = getattr(self.container_session, "mount_path", "/workspace") or "/workspace"
+                mount_path = getattr(session, "mount_path", "/workspace") or "/workspace"
                 docker_bin = shutil.which("docker") or "docker"
                 try:
                     relative = work_path.relative_to(self.project_path).as_posix()
@@ -334,6 +438,10 @@ class TerminalOperator:
                 exec_cmd = [
                     docker_bin,
                     "exec",
+                    "-e",
+                    "PATH=/opt/agent-venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
+                    "-e",
+                    "VIRTUAL_ENV=/opt/agent-venv",
                     "-w",
                     container_workdir,
                     container_name,