refactor(sub_agent): 子智能体从 Node.js 子进程改为主进程内 Python 协程

- 重写子智能体执行核心，不再启动 easyagent Node.js 子进程
- 新增 modules/sub_agent/ 包集中管理子智能体逻辑
- 工具调用复用主进程 WebTerminal.handle_tool_call，自然经过沙箱/容器链路
- 子智能体模型独立读取 ~/.agents/<mode>/config/sub_agent_models.json
- 支持 8 个工具：read_file/write_file/edit_file(replacements+replace_all)/run_command/web_search/extract_webpage/search_workspace/read_mediafile
- 修复子智能体进度弹窗：标题颜色、write_file 显示、过滤非 progress 条目、统一滚动条样式
- 更新 AGENTS.md / CLAUDE.md 子智能体描述
- 新增 test/test_sub_agent_regression.py 回归测试

AGENTS.md

@@ -31,7 +31,8 @@
   - `docs/cli_slash_commands_spec.md`: CLI `/` 指令设计说明
 - **其他子项目/资源**
   - `android-webview-app/`: Android WebView 客户端工程
-  - `easyagent/`: 子智能体执行逻辑所在目录（当前所有子智能体执行均走这里；旧版 `sub_agent/` 已删除）
+  - `modules/sub_agent/`: 子智能体执行逻辑（主进程内 `asyncio.Task`，工具调用复用主进程链路）
+  - `easyagent/`: 旧版 Node.js 子智能体实现，暂时保留但已不再使用
   - `_experiments/`: 本地实验残留与历史文档归档目录（**不纳入 git**，见 §7）
 
 ## 1.5) 运行态数据目录与路径变量（2026-06 更新）

CLAUDE.md

@@ -38,7 +38,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 - `context_manager.py` - 上下文构建、token统计、对话管理
 - `memory_manager.py` - 长期记忆管理
 - `todo_manager.py` - 待办事项管理
-- `sub_agent_manager.py` - 子智能体任务调度
+- `modules/sub_agent/` - 子智能体任务调度与执行（主进程内协程，工具调用复用主 WebTerminal）
 - `user_manager.py` - 用户认证和工作区管理
 - `upload_security.py` - 上传文件隔离扫描
 
@@ -329,7 +329,8 @@ agents/
 ├── config/            # 配置拆分（api/limits/terminal/paths ...，由 __init__ 聚合）
 ├── static/            # Vue 3 + TS Web 前端
 ├── cli/               # React + Ink CLI 前端（重写中）
-├── easyagent/         # 子智能体执行逻辑（旧版 sub_agent/ 已删除）
+├── modules/sub_agent/ # 子智能体执行逻辑（主进程内协程）
+├── easyagent/         # 旧版 Node.js 子智能体实现，暂时保留但已不再使用
 ├── docker/            # 容器镜像
 ├── prompts/           # 系统提示词
 ├── scripts/           # 运维脚本（数据迁移 migrate_runtime_data.py 等）

config/sub_agent.py

@@ -14,11 +14,6 @@ SUB_AGENT_SERVICE_BASE_URL = os.environ.get("SUB_AGENT_SERVICE_URL", "http://127
 SUB_AGENT_DEFAULT_TIMEOUT = int(os.environ.get("SUB_AGENT_DEFAULT_TIMEOUT", "180"))  # 秒
 SUB_AGENT_STATUS_POLL_INTERVAL = float(os.environ.get("SUB_AGENT_STATUS_POLL_INTERVAL", "2.0"))
 
-# 子智能体进程自身的网络权限（与工具级 HOST_SANDBOX_NETWORK_PERMISSION 独立）。
-# 子智能体需要调用模型 API，因此默认给予完整网络；工具级网络限制仍由
-# HOST_SANDBOX_NETWORK_PERMISSION 控制，并由主进程对命令单独加沙箱。
-SUB_AGENT_HOST_NETWORK_PERMISSION = os.environ.get("SUB_AGENT_HOST_NETWORK_PERMISSION", "full")
-
 # 存储与并发限制
 # 子智能体任务目录与状态文件属于运行态数据，跟随 DATA_DIR（默认 ~/.agents/<mode>/data）。
 SUB_AGENT_TASKS_BASE_DIR = _resolve_repo_path(os.environ.get("SUB_AGENT_TASKS_BASE_DIR", ""), f"{DATA_DIR}/sub_agent_tasks")
@@ -39,5 +34,4 @@ __all__ = [
     "SUB_AGENT_STATE_FILE",
     "SUB_AGENT_MAX_ACTIVE",
     "SUB_AGENT_MODELS_CONFIG_FILE",
-    "SUB_AGENT_HOST_NETWORK_PERMISSION",
 ]

core/main_terminal.py

@@ -48,7 +48,7 @@ from modules.terminal_ops import TerminalOperator
 from modules.memory_manager import MemoryManager
 from modules.terminal_manager import TerminalManager
 from modules.todo_manager import TodoManager
-from modules.sub_agent_manager import SubAgentManager
+from modules.sub_agent import SubAgentManager
 from modules.background_command_manager import BackgroundCommandManager
 from modules.ocr_client import OCRClient
 from modules.easter_egg_manager import EasterEggManager
@@ -144,6 +144,7 @@ class MainTerminal(MainTerminalCommandMixin, MainTerminalContextMixin, MainTermi
                 data_dir=str(self.data_dir),
                 container_session=container_session,
             )
+            self.sub_agent_manager.set_terminal(self)
             self.background_command_manager = BackgroundCommandManager(
                 project_path=self.project_path,
             )

core/main_terminal_parts/commands.py

@@ -47,7 +47,7 @@ from modules.terminal_ops import TerminalOperator
 from modules.memory_manager import MemoryManager
 from modules.terminal_manager import TerminalManager
 from modules.todo_manager import TodoManager
-from modules.sub_agent_manager import SubAgentManager
+from modules.sub_agent import SubAgentManager
 from modules.webpage_extractor import extract_webpage_content, tavily_extract
 from modules.ocr_client import OCRClient
 from modules.easter_egg_manager import EasterEggManager

core/main_terminal_parts/context.py

@@ -46,7 +46,7 @@ from modules.terminal_ops import TerminalOperator
 from modules.memory_manager import MemoryManager
 from modules.terminal_manager import TerminalManager
 from modules.todo_manager import TodoManager
-from modules.sub_agent_manager import SubAgentManager
+from modules.sub_agent import SubAgentManager
 from modules.webpage_extractor import extract_webpage_content, tavily_extract
 from modules.ocr_client import OCRClient
 from modules.easter_egg_manager import EasterEggManager

core/main_terminal_parts/tools_definition.py

@@ -47,7 +47,7 @@ from modules.terminal_ops import TerminalOperator
 from modules.memory_manager import MemoryManager
 from modules.terminal_manager import TerminalManager
 from modules.todo_manager import TodoManager
-from modules.sub_agent_manager import SubAgentManager
+from modules.sub_agent import SubAgentManager
 from modules.webpage_extractor import extract_webpage_content, tavily_extract
 from modules.ocr_client import OCRClient
 from modules.easter_egg_manager import EasterEggManager

core/main_terminal_parts/tools_execution.py

@@ -57,7 +57,7 @@ from modules.terminal_ops import TerminalOperator
 from modules.memory_manager import MemoryManager
 from modules.terminal_manager import TerminalManager
 from modules.todo_manager import TodoManager
-from modules.sub_agent_manager import SubAgentManager
+from modules.sub_agent import SubAgentManager
 from modules.webpage_extractor import extract_webpage_content, tavily_extract
 from modules.ocr_client import OCRClient
 from modules.easter_egg_manager import EasterEggManager

core/main_terminal_parts/tools_policy.py

@@ -47,7 +47,7 @@ from modules.terminal_ops import TerminalOperator
 from modules.memory_manager import MemoryManager
 from modules.terminal_manager import TerminalManager
 from modules.todo_manager import TodoManager
-from modules.sub_agent_manager import SubAgentManager
+from modules.sub_agent import SubAgentManager
 from modules.webpage_extractor import extract_webpage_content, tavily_extract
 from modules.ocr_client import OCRClient
 from modules.easter_egg_manager import EasterEggManager

core/main_terminal_parts/tools_read.py

@@ -47,7 +47,7 @@ from modules.terminal_ops import TerminalOperator
 from modules.memory_manager import MemoryManager
 from modules.terminal_manager import TerminalManager
 from modules.todo_manager import TodoManager
-from modules.sub_agent_manager import SubAgentManager
+from modules.sub_agent import SubAgentManager
 from modules.webpage_extractor import extract_webpage_content, tavily_extract
 from modules.ocr_client import OCRClient
 from modules.easter_egg_manager import EasterEggManager

easyagent/src/batch/index.js

@@ -1,340 +0,0 @@
-#!/usr/bin/env node
-'use strict';
-
-/**
- * easyagent 批处理模式
- * 所有输出走 stdout（JSONL），不写文件。
- * Python 侧捕获 stdout 后写入正式存储目录。
- */
-
-const fs = require('fs');
-const path = require('path');
-const { streamChat } = require('../model/client');
-const { executeTool } = require('../tools/dispatcher');
-const { getModelByKey } = require('../config');
-const { applyUsage, normalizeUsagePayload } = require('../utils/token_usage');
-
-// ── stdout 输出 ──
-function emit(type, data) {
-  const line = JSON.stringify({ type, ...data });
-  try {
-    // 使用同步写并刷新，避免管道缓冲导致主进程不能实时读到进度
-    fs.writeSync(1, line + '\n');
-  } catch (err) {
-    process.stdout.write(line + '\n');
-  }
-}
-
-// ── 解析命令行参数 ──
-function parseArgs() {
-  const args = process.argv.slice(2);
-  const config = {
-    workspace: process.cwd(),
-    taskFile: null,
-    systemPromptFile: null,
-    agentId: null,
-    modelKey: null,
-    thinkingMode: null,
-    configFile: null,
-    timeout: 600,
-  };
-
-  for (let i = 0; i < args.length; i++) {
-    const arg = args[i];
-    if (arg === '--workspace' && i + 1 < args.length) {
-      config.workspace = args[++i];
-    } else if (arg === '--task-file' && i + 1 < args.length) {
-      config.taskFile = args[++i];
-    } else if (arg === '--system-prompt-file' && i + 1 < args.length) {
-      config.systemPromptFile = args[++i];
-    } else if (arg === '--agent-id' && i + 1 < args.length) {
-      config.agentId = args[++i];
-    } else if (arg === '--model-key' && i + 1 < args.length) {
-      config.modelKey = args[++i];
-    } else if (arg === '--thinking-mode' && i + 1 < args.length) {
-      config.thinkingMode = String(args[++i] || '').trim().toLowerCase();
-    } else if (arg === '--timeout' && i + 1 < args.length) {
-      config.timeout = parseInt(args[++i], 10);
-    } else if (arg === '--config-file' && i + 1 < args.length) {
-      config.configFile = args[++i];
-    }
-  }
-
-  return config;
-}
-
-// ── 读取文件 ──
-function readFile(filePath) {
-  try {
-    return fs.readFileSync(filePath, 'utf8');
-  } catch (err) {
-    console.error(`读取文件失败: ${filePath}`, err);
-    process.exit(1);
-  }
-}
-
-// ── 主函数 ──
-async function main() {
-  const config = parseArgs();
-
-  if (!config.taskFile || !config.systemPromptFile) {
-    console.error('缺少必需参数: --task-file, --system-prompt-file');
-    process.exit(1);
-  }
-
-  // 读取任务
-  const taskMessage = readFile(config.taskFile);
-  const systemPrompt = readFile(config.systemPromptFile);
-
-  // 加载模型配置
-  const modelConfig = require('../config');
-  let ensuredConfig;
-  try {
-    ensuredConfig = modelConfig.ensureConfig(config.configFile);
-  } catch (e) {
-    emit('output', { success: false, summary: '模型配置加载失败: ' + e.message, error: 'config_load_error' });
-    process.exit(1);
-  }
-
-  if (!ensuredConfig.valid_models || ensuredConfig.valid_models.length === 0) {
-    emit('output', { success: false, summary: '未找到可用模型配置', error: 'no_model_config' });
-    process.exit(1);
-  }
-
-  // 选择模型
-  const modelKey = config.modelKey || ensuredConfig.default_model_key;
-  const model = getModelByKey(ensuredConfig, modelKey);
-  if (!model) {
-    emit('output', { success: false, summary: '未找到模型: ' + modelKey, error: 'model_not_found' });
-    process.exit(1);
-  }
-
-  // 加载工具
-  const tools = JSON.parse(fs.readFileSync(path.join(__dirname, '../tools/tools.json'), 'utf8'));
-  tools.push({
-    type: 'function',
-    function: {
-      name: 'finish_task',
-      description: '完成当前任务并退出。调用此工具表示你已经完成了分配的任务，所有交付文件已准备好。',
-      parameters: {
-        type: 'object',
-        properties: {
-          success: { type: 'boolean', description: '任务是否成功完成。' },
-          summary: { type: 'string', description: '任务完成摘要（50-200字）。' },
-        },
-        required: ['success', 'summary'],
-      },
-    },
-  });
-
-  // 初始化对话
-  const messages = [
-    { role: 'system', content: systemPrompt },
-    { role: 'user', content: taskMessage },
-  ];
-
-  // 统计
-  const stats = {
-    runtime_start: Date.now(),
-    runtime_seconds: 0,
-    files_read: 0,
-    edit_files: 0,
-    searches: 0,
-    web_pages: 0,
-    commands: 0,
-    api_calls: 0,
-    token_usage: { prompt: 0, completion: 0, total: 0 },
-  };
-
-  const startTime = Date.now();
-  const timeoutMs = config.timeout * 1000;
-  let turnCount = 0;
-  const maxTurns = 50;
-
-  try {
-    while (true) {
-      turnCount++;
-      stats.api_calls += 1;
-      stats.turn_count = turnCount;
-
-      const elapsed = Date.now() - startTime;
-      if (elapsed > timeoutMs) {
-        emit('output', {
-          success: false,
-          summary: '任务超时未完成',
-          timeout: true,
-          stats: { ...stats, runtime_seconds: Math.floor(elapsed / 1000) },
-        });
-        process.exit(1);
-      }
-
-      if (turnCount > maxTurns) {
-        emit('output', {
-          success: false,
-          summary: '任务执行超过 ' + maxTurns + ' 轮，可能陷入循环',
-          max_turns_exceeded: true,
-          stats: { ...stats, runtime_seconds: Math.floor(elapsed / 1000) },
-        });
-        process.exit(1);
-      }
-
-      // 发送统计更新
-      emit('stats', { ...stats, runtime_seconds: Math.floor(elapsed / 1000), turn_count: turnCount });
-
-      // 调用 API
-      let assistantMessage = { role: 'assistant', content: '', tool_calls: [] };
-      let reasoningBuffer = '';
-      let usage = null;
-
-      try {
-        for await (const chunk of streamChat({
-          config: ensuredConfig,
-          modelKey,
-          messages,
-          tools,
-          thinkingMode: config.thinkingMode === 'thinking',
-          currentContextTokens: 0,
-          abortSignal: null,
-        })) {
-          const choice = chunk.choices?.[0];
-          const delta = choice?.delta;
-          if (!delta) continue;
-
-          if (delta.content) {
-            assistantMessage.content += delta.content;
-          }
-
-          if (delta.reasoning_content) {
-            reasoningBuffer += delta.reasoning_content;
-          } else if (delta.reasoning_details) {
-            const rd = delta.reasoning_details;
-            if (Array.isArray(rd)) reasoningBuffer += rd.map(d => d.text || '').join('');
-            else if (typeof rd === 'string') reasoningBuffer += rd;
-            else if (rd && typeof rd.text === 'string') reasoningBuffer += rd.text;
-          }
-
-          if (delta.tool_calls) {
-            for (const tc of delta.tool_calls) {
-              if (tc.index !== undefined) {
-                if (!assistantMessage.tool_calls[tc.index]) {
-                  assistantMessage.tool_calls[tc.index] = {
-                    id: tc.id || '',
-                    type: 'function',
-                    function: { name: '', arguments: '' },
-                  };
-                }
-                const ct = assistantMessage.tool_calls[tc.index];
-                if (tc.id) ct.id = tc.id;
-                if (tc.function?.name) ct.function.name += tc.function.name;
-                if (tc.function?.arguments) ct.function.arguments += tc.function.arguments;
-              }
-            }
-          }
-
-          if (chunk.usage) {
-            usage = normalizeUsagePayload(chunk.usage);
-          }
-        }
-      } catch (err) {
-        emit('output', {
-          success: false,
-          summary: 'API 调用失败: ' + err.message,
-          error: 'api_error',
-          stats: { ...stats, runtime_seconds: Math.floor((Date.now() - startTime) / 1000) },
-        });
-        process.exit(1);
-      }
-
-      if (usage) applyUsage(stats.token_usage, usage);
-
-      const finalAssistantMessage = reasoningBuffer
-        ? { role: 'assistant', reasoning_content: reasoningBuffer, content: assistantMessage.content, tool_calls: assistantMessage.tool_calls }
-        : assistantMessage;
-      messages.push(finalAssistantMessage);
-
-      if (!assistantMessage.tool_calls || assistantMessage.tool_calls.length === 0) {
-        messages.push({
-          role: 'user',
-          content: '如果你已经完成了任务，请调用 finish_task 工具提交完成报告。如果还没有完成，请继续执行任务。',
-        });
-        continue;
-      }
-
-      for (const toolCall of assistantMessage.tool_calls) {
-        const toolName = toolCall.function.name;
-        stats.tool_calls = (stats.tool_calls || 0) + 1;
-
-        if (toolName === 'finish_task') {
-          let args = {};
-          try {
-            args = JSON.parse(toolCall.function.arguments || '{}');
-          } catch (err) {
-            emit('output', {
-              success: false,
-              summary: 'finish_task 参数解析失败',
-              error: 'invalid_finish_args',
-              stats: { ...stats, runtime_seconds: Math.floor((Date.now() - startTime) / 1000) },
-            });
-            process.exit(1);
-          }
-
-          // 输出对话记录到 stdout，由 Python 管理程序写入正式目录
-          emit('conversation', {
-            agent_id: config.agentId,
-            created_at: new Date(startTime).toISOString(),
-            completed_at: new Date().toISOString(),
-            success: args.success,
-            summary: args.summary,
-            messages: messages,
-            stats: { ...stats, runtime_seconds: Math.floor((Date.now() - startTime) / 1000) },
-          });
-
-          // 最终输出
-          emit('output', {
-            success: args.success || false,
-            summary: args.summary || '任务完成',
-            stats: { ...stats, runtime_seconds: Math.floor((Date.now() - startTime) / 1000) },
-          });
-          process.exit(0);
-        }
-
-        let progressArgs = {};
-        try { progressArgs = JSON.parse(toolCall.function.arguments || '{}'); } catch (e) { progressArgs = { _raw: toolCall.function.arguments || '' }; }
-        const progressId = toolCall.id || 'tool_' + Date.now() + '_' + Math.random().toString(16).slice(2, 8);
-
-        emit('progress', { id: progressId, tool: toolName, status: 'running', args: progressArgs, ts: Date.now() });
-
-        const result = await executeTool({
-          workspace: config.workspace,
-          config: ensuredConfig,
-          allowMode: 'full_access',
-          toolCall,
-          abortSignal: null,
-        });
-
-        emit('progress', { id: progressId, tool: toolName, status: result && result.success ? 'completed' : 'failed', args: progressArgs, ts: Date.now() });
-
-        if (toolName === 'read_file') stats.files_read++;
-        else if (toolName === 'edit_file') stats.edit_files++;
-        else if (toolName === 'search_workspace') stats.searches++;
-        else if (toolName === 'web_search' || toolName === 'extract_webpage') stats.web_pages++;
-        else if (toolName === 'run_command') stats.commands++;
-
-        messages.push({ role: 'tool', tool_call_id: toolCall.id, content: result.formatted || '' });
-      }
-    }
-  } catch (err) {
-    emit('output', {
-      success: false,
-      summary: '执行出错: ' + err.message,
-      error: 'execution_error',
-      stats: { ...stats, runtime_seconds: Math.floor((Date.now() - startTime) / 1000) },
-    });
-    process.exit(1);
-  }
-}
-
-main().catch((err) => {
-  console.error('批处理模式执行失败:', err);
-  process.exit(1);
-});

modules/sub_agent/__init__.py

@@ -0,0 +1,9 @@
+"""子智能体（Sub-Agent）模块包。
+
+子智能体现在作为主进程内的 asyncio.Task 运行，所有实际工具调用都通过
+主 WebTerminal 执行，因此自然复用主进程的宿主机沙箱 / Docker 容器链路。
+"""
+
+from modules.sub_agent.manager import SubAgentManager
+
+__all__ = ["SubAgentManager"]

modules/sub_agent/creation.py

@@ -0,0 +1,82 @@
+"""子智能体创建、查询与会话槽位管理工具。"""
+
+from __future__ import annotations
+
+import time
+import uuid
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from config import SUB_AGENT_DEFAULT_TIMEOUT, SUB_AGENT_MAX_ACTIVE
+
+
+class SubAgentCreationMixin:
+    """提供子智能体创建参数校验、任务ID生成、交付目录解析与槽位管理能力。"""
+
+    tasks: Dict[str, Dict[str, Any]]
+    conversation_agents: Dict[str, List[int]]
+    project_path: Path
+
+    def _select_task(self, task_id: Optional[str], agent_id: Optional[int]) -> Optional[Dict]:
+        self.reconcile_task_states()
+        if task_id:
+            return self.tasks.get(task_id)
+        if agent_id is None:
+            return None
+        candidates = [
+            task for task in self.tasks.values()
+            if task.get("agent_id") == agent_id and task.get("status") in {"pending", "running"}
+        ]
+        if candidates:
+            candidates.sort(key=lambda item: item.get("created_at", 0), reverse=True)
+            return candidates[0]
+        return None
+
+    def _active_task_count(self, conversation_id: Optional[str] = None) -> int:
+        self.reconcile_task_states(conversation_id=conversation_id)
+        active = [t for t in self.tasks.values() if t.get("status") in {"pending", "running"}]
+        if conversation_id:
+            active = [t for t in active if t.get("conversation_id") == conversation_id]
+        return len(active)
+
+    def _ensure_agent_slot_available(self, conversation_id: str, agent_id: int) -> bool:
+        used = self.conversation_agents.setdefault(conversation_id, [])
+        return agent_id not in used
+
+    def _mark_agent_id_used(self, conversation_id: str, agent_id: int):
+        used = self.conversation_agents.setdefault(conversation_id, [])
+        if agent_id not in used:
+            used.append(agent_id)
+
+    def _validate_create_params(self, agent_id: Optional[int], summary: str, task: str, target_dir: str) -> Optional[str]:
+        if agent_id is None:
+            return "子智能体代号不能为空"
+        try:
+            agent_id = int(agent_id)
+        except ValueError:
+            return "子智能体代号必须是整数"
+        if agent_id <= 0:
+            return "子智能体代号必须为正整数"
+        if not summary or not summary.strip():
+            return "任务摘要不能为空"
+        if not task or not task.strip():
+            return "任务详情不能为空"
+        if target_dir is None:
+            return "指定文件夹不能为空"
+        return None
+
+    def _generate_task_id(self, agent_id: int) -> str:
+        suffix = uuid.uuid4().hex[:6]
+        return f"sub_{agent_id}_{int(time.time())}_{suffix}"
+
+    def _resolve_deliverables_dir(self, relative_dir: str) -> Path:
+        relative_dir = relative_dir.strip() if relative_dir else ""
+        if not relative_dir:
+            raise ValueError("交付目录不能为空，必须指定")
+        deliverables_path = (self.project_path / relative_dir).resolve()
+        if not str(deliverables_path).startswith(str(self.project_path)):
+            raise ValueError("交付目录必须位于项目目录内")
+        if deliverables_path.exists():
+            raise ValueError("交付目录必须为不存在的新目录")
+        deliverables_path.mkdir(parents=True, exist_ok=True)
+        return deliverables_path

modules/sub_agent/manager.py

@@ -0,0 +1,467 @@
+"""子智能体任务管理（主进程内协程模式）。
+
+子智能体不再作为独立子进程启动，而是作为 SubAgentManager 所在事件循环中的
+asyncio.Task 运行。所有实际工具调用都通过主 WebTerminal 执行，因此自然复用
+主进程的宿主机沙箱 / Docker 容器链路。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import threading
+import time
+from pathlib import Path, PurePosixPath
+from typing import Any, Dict, List, Optional, TYPE_CHECKING
+
+from config import (
+    OUTPUT_FORMATS,
+    SUB_AGENT_DEFAULT_TIMEOUT,
+    SUB_AGENT_MAX_ACTIVE,
+    SUB_AGENT_MODELS_CONFIG_FILE,
+    SUB_AGENT_STATE_FILE,
+    SUB_AGENT_STATUS_POLL_INTERVAL,
+    SUB_AGENT_TASKS_BASE_DIR,
+)
+from utils.logger import setup_logger
+from modules.sub_agent.task import SubAgentTask
+from modules.sub_agent.prompts import build_user_message, build_system_prompt
+from modules.sub_agent.tools import handle_search_workspace, handle_read_mediafile
+from modules.sub_agent.state import SubAgentStateMixin
+from modules.sub_agent.stats import SubAgentStatsMixin
+from modules.sub_agent.creation import SubAgentCreationMixin
+
+if TYPE_CHECKING:
+    from core.web_terminal import WebTerminal
+    from modules.user_container_manager import ContainerHandle
+
+logger = setup_logger(__name__)
+TERMINAL_STATUSES = {"completed", "failed", "timeout"}
+
+
+class SubAgentManager(SubAgentStateMixin, SubAgentStatsMixin, SubAgentCreationMixin):
+    """负责主智能体与子智能体的任务调度（协程模式）。"""
+
+    def __init__(
+        self,
+        project_path: str,
+        data_dir: str,
+        container_session: Optional["ContainerHandle"] = None,
+    ):
+        self.project_path = Path(project_path).resolve()
+        self.data_dir = Path(data_dir).resolve()
+        self.base_dir = Path(SUB_AGENT_TASKS_BASE_DIR).resolve()
+        self.state_file = Path(SUB_AGENT_STATE_FILE).resolve()
+        self.models_config_file = SUB_AGENT_MODELS_CONFIG_FILE
+        self.container_session: Optional["ContainerHandle"] = container_session
+        self.host_execution_mode: str = "sandbox"
+        self.terminal: Optional["WebTerminal"] = None
+
+        self.base_dir.mkdir(parents=True, exist_ok=True)
+        self.state_file.parent.mkdir(parents=True, exist_ok=True)
+
+        self.tasks: Dict[str, Dict[str, Any]] = {}
+        self.conversation_agents: Dict[str, List[int]] = {}
+        self._running_tasks: Dict[str, asyncio.Task] = {}
+        self._event_loop: Optional[asyncio.AbstractEventLoop] = None
+        self._loop_thread: Optional[threading.Thread] = None
+        self._state_lock = threading.Lock()
+
+        self._load_state()
+        try:
+            self.reconcile_task_states()
+        except Exception:
+            pass
+
+    # ------------------------------------------------------------------
+    # 生命周期与事件循环
+    # ------------------------------------------------------------------
+    def _ensure_event_loop(self) -> asyncio.AbstractEventLoop:
+        """确保有一个独立的后台事件循环供子智能体使用。"""
+        if self._event_loop is not None and not self._event_loop.is_closed():
+            return self._event_loop
+
+        loop = asyncio.new_event_loop()
+        self._event_loop = loop
+
+        def run_loop():
+            asyncio.set_event_loop(loop)
+            try:
+                loop.run_forever()
+            finally:
+                try:
+                    loop.close()
+                except Exception:
+                    pass
+
+        thread = threading.Thread(target=run_loop, name="sub-agent-loop", daemon=True)
+        thread.start()
+        self._loop_thread = thread
+        return loop
+
+    async def _create_task(self, coro):
+        """在事件循环内部把协程包装为 Task。"""
+        return asyncio.create_task(coro)
+
+    def _run_coro(self, coro):
+        """在后台事件循环中调度一个协程并返回 asyncio.Task。"""
+        loop = self._ensure_event_loop()
+        # 先提交创建 Task 的协程，阻塞等待拿到 Task 句柄
+        future = asyncio.run_coroutine_threadsafe(self._create_task(coro), loop)
+        return future.result(timeout=10)
+
+    def set_terminal(self, terminal: "WebTerminal") -> None:
+        """注入主终端引用，用于工具执行代理。"""
+        self.terminal = terminal
+
+    def set_container_session(self, session: Optional["ContainerHandle"]):
+        """更新容器会话信息。"""
+        self.container_session = session
+
+    def set_host_execution_mode(self, mode: str) -> None:
+        normalized = str(mode or "").strip().lower()
+        self.host_execution_mode = "direct" if normalized == "direct" else "sandbox"
+
+    # ------------------------------------------------------------------
+    # 公共方法
+    # ------------------------------------------------------------------
+    def create_sub_agent(
+        self,
+        *,
+        agent_id: int,
+        summary: str,
+        task: str,
+        deliverables_dir: str,
+        timeout_seconds: Optional[int] = None,
+        conversation_id: Optional[str] = None,
+        run_in_background: bool = False,
+        model_key: Optional[str] = None,
+        thinking_mode: Optional[str] = None,
+    ) -> Dict:
+        """创建子智能体任务并启动协程。"""
+        validation_error = self._validate_create_params(agent_id, summary, task, deliverables_dir)
+        if validation_error:
+            return {"success": False, "error": validation_error}
+
+        if not thinking_mode:
+            return {"success": False, "error": "缺少 thinking_mode 参数，必须指定 fast 或 thinking"}
+        if thinking_mode not in {"fast", "thinking"}:
+            return {"success": False, "error": "thinking_mode 仅支持 fast 或 thinking"}
+
+        if not conversation_id:
+            return {"success": False, "error": "缺少对话ID，无法创建子智能体"}
+
+        if not self._ensure_agent_slot_available(conversation_id, agent_id):
+            return {
+                "success": False,
+                "error": f"该对话已使用过编号 {agent_id}，请更换新的子智能体代号。"
+            }
+
+        if self._active_task_count(conversation_id) >= SUB_AGENT_MAX_ACTIVE:
+            return {
+                "success": False,
+                "error": f"该对话已存在 {SUB_AGENT_MAX_ACTIVE} 个运行中的子智能体，请稍后再试。",
+            }
+
+        task_id = self._generate_task_id(agent_id)
+        task_root = self.base_dir / task_id
+        task_root.mkdir(parents=True, exist_ok=True)
+
+        try:
+            deliverables_path = self._resolve_deliverables_dir(deliverables_dir)
+        except ValueError as exc:
+            return {"success": False, "error": str(exc)}
+
+        task_file = task_root / "task.txt"
+        system_prompt_file = task_root / "system_prompt.txt"
+        output_file = task_root / "output.json"
+        stats_file = task_root / "stats.json"
+        progress_file = task_root / "progress.jsonl"
+        conversation_file = task_root / "conversation.json"
+
+        prompt_workspace = self._get_runtime_path(self.project_path)
+        deliverables_display = self._get_runtime_path(deliverables_path)
+        user_message = build_user_message(agent_id, summary, task, deliverables_display, timeout_seconds or SUB_AGENT_DEFAULT_TIMEOUT)
+        task_file.write_text(user_message, encoding="utf-8")
+
+        system_prompt = build_system_prompt(prompt_workspace)
+        system_prompt_file.write_text(system_prompt, encoding="utf-8")
+
+        timeout_seconds = timeout_seconds or SUB_AGENT_DEFAULT_TIMEOUT
+
+        task_record = {
+            "task_id": task_id,
+            "agent_id": agent_id,
+            "summary": summary,
+            "task": task,
+            "status": "running",
+            "deliverables_dir": str(deliverables_path),
+            "timeout_seconds": timeout_seconds,
+            "thinking_mode": thinking_mode,
+            "created_at": time.time(),
+            "updated_at": time.time(),
+            "conversation_id": conversation_id,
+            "run_in_background": run_in_background,
+            "task_root": str(task_root),
+            "output_file": str(output_file),
+            "stats_file": str(stats_file),
+            "progress_file": str(progress_file),
+            "conversation_file": str(conversation_file),
+            "execution_mode": "in_process",
+            "container_name": None,
+        }
+        self.tasks[task_id] = task_record
+        self._mark_agent_id_used(conversation_id, agent_id)
+        self._save_state()
+
+        sub_agent = SubAgentTask(
+            manager=self,
+            task_record=task_record,
+            task_message=user_message,
+            system_prompt=system_prompt,
+            model_key=model_key,
+            thinking_mode=thinking_mode,
+        )
+        task_coro = sub_agent.run()
+        asyncio_task = self._run_coro(task_coro)
+        sub_agent._task = asyncio_task
+        self._running_tasks[task_id] = asyncio_task
+
+        def _on_done(fut):
+            self._running_tasks.pop(task_id, None)
+            self.reconcile_task_states(conversation_id=conversation_id)
+
+        asyncio_task.add_done_callback(_on_done)
+
+        message = f"子智能体{agent_id} 已创建，任务ID: {task_id}"
+        print(f"{OUTPUT_FORMATS['info']} {message}")
+
+        return {
+            "success": True,
+            "task_id": task_id,
+            "agent_id": agent_id,
+            "status": "running",
+            "message": message,
+            "deliverables_dir": str(deliverables_path),
+            "run_in_background": run_in_background,
+        }
+
+    def wait_for_completion(
+        self,
+        *,
+        task_id: Optional[str] = None,
+        agent_id: Optional[int] = None,
+        timeout_seconds: Optional[int] = None,
+    ) -> Dict:
+        """阻塞等待子智能体完成或超时。"""
+        task = self._select_task(task_id, agent_id)
+        if not task:
+            return {"success": False, "error": "未找到对应的子智能体任务"}
+
+        if task.get("status") in TERMINAL_STATUSES or task.get("status") == "terminated":
+            if task.get("final_result"):
+                return task["final_result"]
+            return {"success": False, "status": task.get("status"), "message": "子智能体已结束。"}
+
+        real_task_id = task["task_id"]
+        running_task = self._running_tasks.get(real_task_id)
+        deadline = time.time() + (timeout_seconds or task.get("timeout_seconds") or SUB_AGENT_DEFAULT_TIMEOUT)
+
+        while time.time() < deadline:
+            self.reconcile_task_states()
+            if task.get("status") in TERMINAL_STATUSES:
+                return task.get("final_result") or {"success": False, "status": task.get("status")}
+            if running_task and running_task.done():
+                self.reconcile_task_states()
+                return task.get("final_result") or {"success": False, "status": task.get("status")}
+            time.sleep(SUB_AGENT_STATUS_POLL_INTERVAL)
+
+        return self._handle_timeout(task)
+
+    def terminate_sub_agent(
+        self,
+        *,
+        task_id: Optional[str] = None,
+        agent_id: Optional[int] = None,
+    ) -> Dict:
+        """强制关闭指定子智能体。"""
+        task = self._select_task(task_id, agent_id)
+        if not task:
+            return {"success": False, "error": "未找到对应的子智能体任务"}
+
+        task_id = task["task_id"]
+        running_task = self._running_tasks.pop(task_id, None)
+        if running_task and not running_task.done():
+            running_task.cancel()
+            deadline = time.time() + 5
+            while not running_task.done() and time.time() < deadline:
+                time.sleep(0.05)
+
+        self._mark_task_terminated(
+            task,
+            message="子智能体已被强制关闭。",
+            system_message=f"🛑 子智能体{task.get('agent_id')} 已被手动关闭。",
+            notified=True,
+        )
+        self._save_state()
+
+        return {
+            "success": True,
+            "task_id": task_id,
+            "message": "子智能体已被强制关闭。",
+            "system_message": f"🛑 子智能体{task.get('agent_id')} 已被手动关闭。",
+        }
+
+    def get_sub_agent_status(
+        self,
+        *,
+        agent_ids: Optional[List[int]] = None,
+    ) -> Dict:
+        """获取指定子智能体的详细状态。"""
+        if not agent_ids:
+            return {"success": False, "error": "必须指定至少一个agent_id"}
+
+        results = []
+        for agent_id in agent_ids:
+            task = self._select_task(None, agent_id)
+            if not task:
+                results.append({
+                    "agent_id": agent_id,
+                    "found": False,
+                    "error": "未找到对应的子智能体任务"
+                })
+                continue
+
+            if task.get("status") not in TERMINAL_STATUSES.union({"terminated"}):
+                self._check_task_status(task)
+
+            stats = {}
+            stats_file = Path(task.get("stats_file", ""))
+            if stats_file.exists():
+                try:
+                    stats = json.loads(stats_file.read_text(encoding="utf-8"))
+                except Exception:
+                    pass
+            stats_summary = self._build_stats_summary(stats)
+
+            results.append({
+                "agent_id": agent_id,
+                "found": True,
+                "task_id": task["task_id"],
+                "status": task["status"],
+                "summary": task.get("summary"),
+                "created_at": task.get("created_at"),
+                "updated_at": task.get("updated_at"),
+                "deliverables_dir": task.get("deliverables_dir"),
+                "stats": stats,
+                "stats_summary": stats_summary,
+                "final_result": task.get("final_result"),
+            })
+
+        return {"success": True, "results": results}
+
+    def poll_updates(self) -> List[Dict]:
+        """检查运行中的子智能体任务，返回新完成的结果。"""
+        updates: List[Dict] = []
+        self.reconcile_task_states()
+        pending_tasks = [
+            task for task in self.tasks.values()
+            if task.get("status") not in TERMINAL_STATUSES.union({"terminated"})
+        ]
+        if not pending_tasks:
+            return updates
+
+        state_changed = False
+        for task in pending_tasks:
+            result = self._check_task_status(task)
+            if result["status"] in TERMINAL_STATUSES:
+                updates.append(result)
+                state_changed = True
+
+        if state_changed:
+            self._save_state()
+        return updates
+
+    def lookup_task(self, *, task_id: Optional[str] = None, agent_id: Optional[int] = None) -> Optional[Dict]:
+        """只读查询任务信息。"""
+        task = self._select_task(task_id, agent_id)
+        if not task:
+            return None
+        return {
+            "task_id": task.get("task_id"),
+            "agent_id": task.get("agent_id"),
+            "status": task.get("status"),
+            "timeout_seconds": task.get("timeout_seconds"),
+            "conversation_id": task.get("conversation_id"),
+        }
+
+    def get_overview(self, conversation_id: Optional[str] = None) -> List[Dict[str, Any]]:
+        """返回子智能体任务概览，用于前端展示。"""
+        self.reconcile_task_states(conversation_id=conversation_id)
+        overview: List[Dict[str, Any]] = []
+        for task_id, task in self.tasks.items():
+            if conversation_id and task.get("conversation_id") != conversation_id:
+                continue
+
+            snapshot = {
+                "task_id": task_id,
+                "agent_id": task.get("agent_id"),
+                "summary": task.get("summary"),
+                "status": task.get("status"),
+                "created_at": task.get("created_at"),
+                "updated_at": task.get("updated_at"),
+                "target_dir": task.get("target_project_dir"),
+                "last_tool": task.get("last_tool"),
+                "deliverables_dir": task.get("deliverables_dir"),
+                "copied_path": task.get("copied_path"),
+                "conversation_id": task.get("conversation_id"),
+                "sub_conversation_id": task.get("sub_conversation_id"),
+            }
+
+            if snapshot["status"] in TERMINAL_STATUSES or snapshot["status"] == "terminated":
+                final_result = task.get("final_result") or {}
+                snapshot["final_message"] = final_result.get("system_message") or final_result.get("message")
+                snapshot["success"] = final_result.get("success")
+
+            overview.append(snapshot)
+
+        overview.sort(key=lambda item: item.get("created_at") or 0, reverse=True)
+        return overview
+
+    # ------------------------------------------------------------------
+    # 工具执行代理
+    # ------------------------------------------------------------------
+    async def execute_tool_for_sub_agent(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
+        """代表子智能体在主进程中执行工具。"""
+        if not self.terminal:
+            return {"success": False, "error": "子智能体管理器未绑定终端，无法执行工具"}
+
+        try:
+            if tool_name == "search_workspace":
+                return await handle_search_workspace(self.project_path, self.terminal, arguments)
+            if tool_name == "read_mediafile":
+                return await handle_read_mediafile(self.project_path, arguments)
+
+            # 其余工具直接走主进程 handle_tool_call，自然经过沙箱/容器/权限链路
+            result_text = await self.terminal.handle_tool_call(tool_name, arguments)
+            try:
+                return json.loads(result_text)
+            except Exception:
+                return {"success": True, "output": result_text}
+        except Exception as exc:
+            logger.exception(f"[SubAgent] 工具执行异常: {tool_name}")
+            return {"success": False, "error": f"工具执行异常: {exc}"}
+
+    def _get_runtime_path(self, host_path: Path) -> str:
+        """将宿主机路径映射为容器内路径（仅用于提示展示）。"""
+        if not self.container_session or getattr(self.container_session, "mode", None) != "docker":
+            return str(host_path)
+        mount_path = (getattr(self.container_session, "mount_path", None) or "/workspace").rstrip("/") or "/workspace"
+        try:
+            relative = host_path.resolve().relative_to(self.project_path)
+        except Exception:
+            return mount_path
+        if str(relative) in {"", "."}:
+            return mount_path
+        return str(PurePosixPath(mount_path) / PurePosixPath(relative.as_posix()))
+

modules/sub_agent/prompts.py

@@ -0,0 +1,114 @@
+"""子智能体提示词构建。"""
+
+from __future__ import annotations
+
+import platform
+from datetime import datetime
+from typing import Optional
+
+
+def build_user_message(
+    agent_id: int,
+    summary: str,
+    task: str,
+    deliverables_path: str,
+    timeout_seconds: int,
+) -> str:
+    """构建发送给子智能体的用户消息。"""
+    return f"""你是子智能体 #{agent_id}，负责完成以下任务：
+
+**任务摘要**：{summary}
+
+**任务详情**：
+{task}
+
+**交付目录**：{deliverables_path}
+请将所有生成的文件保存到此目录。
+
+**超时时间**：{timeout_seconds} 秒
+
+完成任务后，请调用 finish_task 工具提交完成报告。"""
+
+
+def build_system_prompt(workspace_path: str) -> str:
+    """构建子智能体的系统提示。"""
+    system_info = f"{platform.system()} {platform.release()}"
+    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+    return f"""你是一个专注的子智能体，负责独立完成分配的任务。
+
+# 身份定位
+
+你是主智能体创建的子智能体，拥有完整的工具能力（读写文件、执行命令、搜索网页等）。你的职责是专注完成分配的单一任务，不要偏离任务目标。
+
+# 工作流程
+
+1. **理解任务**：仔细阅读任务描述，明确目标和要求
+2. **制定计划**：规划完成任务的步骤
+3. **执行任务**：使用工具完成各个步骤
+4. **生成交付**：将所有结果文件放到指定的交付目录
+5. **提交报告**：使用 finish_task 工具提交完成报告并退出
+
+# 工作原则
+
+## 专注性
+- 只完成分配的任务，不要做额外的工作
+- 不要尝试与用户对话或询问问题
+- 遇到问题时，在能力范围内解决或在报告中说明
+
+## 独立性
+- 你与主智能体共享工作区，可以访问所有文件
+- 你的工作范围应该与其他子智能体不重叠
+- 不要修改任务描述之外的文件
+
+## 效率性
+- 直接开始工作，不要过度解释
+- 合理使用工具，避免重复操作
+- 注意超时限制，在时间内完成核心工作
+
+## 完整性
+- 确保交付目录中的文件完整可用
+- 生成的文档要清晰、格式正确
+- 代码要包含必要的注释和说明
+
+# 交付要求
+
+所有结果文件必须放在指定的交付目录中，包括：
+- 主要成果文件（文档、代码、报告等）
+- 支持文件（数据、配置、示例等）
+- 不要在交付目录外创建文件
+
+# 完成任务
+
+任务完成后，必须调用 finish_task 工具：
+- success: 是否成功完成
+- summary: 完成摘要（说明做了什么、生成了什么）
+
+调用 finish_task 后，你会立即退出，无法继续工作。
+
+# 工具使用
+
+你拥有以下工具能力：
+- read_file: 读取文件内容
+- write_file / edit_file: 创建或修改文件
+- search_workspace: 搜索文件和代码
+- run_command: 执行终端命令
+- web_search / extract_webpage: 搜索和提取网页内容
+- read_mediafile: 读取图片/视频文件
+- finish_task: 完成任务并退出（必须调用）
+
+# 注意事项
+
+1. **结果传达**：你在运行期间产生的记录与输出不会被直接传递给主智能体。务必把所有需要传达的信息写进 `finish_task` 工具的 `summary` 字段，以及交付目录中的落盘文件里。
+2. **不要无限循环**：如果任务无法完成，说明原因并提交报告
+3. **不要超出范围**：只操作任务描述中指定的文件/目录
+4. **不要等待输入**：你是自主运行的，不会收到用户的进一步指令
+5. **注意时间限制**：超时会被强制终止，优先完成核心工作
+
+# 当前环境
+
+- 工作区路径: {workspace_path}
+- 系统: {system_info}
+- 当前时间: {current_time}
+
+现在开始执行任务。"""

modules/sub_agent/state.py

@@ -0,0 +1,324 @@
+"""子智能体状态管理 Mixin。"""
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from utils.logger import setup_logger
+
+logger = setup_logger(__name__)
+TERMINAL_STATUSES = {"completed", "failed", "timeout"}
+
+
+class SubAgentStateMixin:
+    """提供子智能体任务状态加载、保存、刷新与结果落盘能力。"""
+
+    state_file: Path
+    tasks: Dict[str, Dict[str, Any]]
+    conversation_agents: Dict[str, List[int]]
+    _running_tasks: Dict[str, Any]
+    _state_lock: Any
+
+    def _load_state(self):
+        with self._state_lock:
+            if not self.state_file.exists():
+                self.tasks = {}
+                self.conversation_agents = {}
+                return
+            try:
+                data = json.loads(self.state_file.read_text(encoding="utf-8"))
+                loaded_tasks = data.get("tasks", {})
+                loaded_agents = data.get("conversation_agents", {})
+            except json.JSONDecodeError:
+                logger.warning("子智能体状态文件损坏，已忽略。")
+                self.tasks = {}
+                self.conversation_agents = {}
+                return
+
+            runtime_only_keys = {"_stdout_lines"}
+            merged_tasks: Dict[str, Dict] = {}
+            for task_id, task in loaded_tasks.items():
+                existing = self.tasks.get(task_id)
+                if existing:
+                    for key in runtime_only_keys:
+                        if key in existing:
+                            task[key] = existing[key]
+                merged_tasks[task_id] = task
+            self.tasks = merged_tasks
+            self.conversation_agents = loaded_agents
+
+            if self.tasks:
+                migrated = False
+                for task in self.tasks.values():
+                    if task.get("parent_conversation_id"):
+                        continue
+                    candidate = task.get("conversation_id") or (task.get("service_payload") or {}).get("parent_conversation_id")
+                    if candidate:
+                        task["parent_conversation_id"] = candidate
+                        migrated = True
+                if migrated:
+                    self._save_state_unsafe()
+
+    def _save_state(self):
+        with self._state_lock:
+            self._save_state_unsafe()
+
+    def _save_state_unsafe(self):
+        payload = {
+            "tasks": self.tasks,
+            "conversation_agents": self.conversation_agents,
+        }
+        try:
+            self.state_file.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+        except Exception as exc:
+            logger.warning(f"保存子智能体状态失败: {exc}")
+
+    def _check_task_status(self, task: Dict) -> Dict:
+        """检查任务状态，读取输出文件。"""
+        task_id = task["task_id"]
+        output_file = Path(task.get("output_file", ""))
+        if not output_file.exists():
+            running_task = self._running_tasks.get(task_id)
+            if running_task and running_task.done():
+                try:
+                    running_task.result(timeout=0)
+                except Exception:
+                    pass
+                task["status"] = "failed"
+                task["updated_at"] = time.time()
+            return {"status": task.get("status", "running"), "task_id": task_id}
+
+        try:
+            output = json.loads(output_file.read_text(encoding="utf-8"))
+        except Exception as exc:
+            task["status"] = "failed"
+            task["updated_at"] = time.time()
+            return {"success": False, "status": "failed", "task_id": task_id, "message": f"输出文件解析失败: {exc}"}
+
+        success = output.get("success", False)
+        summary = output.get("summary", "")
+        stats = output.get("stats", {})
+        elapsed_seconds = self._compute_elapsed_seconds(task)
+
+        if output.get("timeout"):
+            status = "timeout"
+        elif output.get("max_turns_exceeded"):
+            status = "failed"
+            summary = f"任务执行超过最大轮次限制。{summary}"
+        elif success:
+            status = "completed"
+        else:
+            status = "failed"
+
+        task["status"] = status
+        task["updated_at"] = time.time()
+        if status == "completed" and elapsed_seconds is not None:
+            task["elapsed_seconds"] = elapsed_seconds
+            task["runtime_seconds"] = elapsed_seconds
+
+        agent_id = task.get("agent_id")
+        task_summary = task.get("summary")
+        deliverables_dir = task.get("deliverables_dir")
+        stats_summary = self._build_stats_summary(stats)
+
+        if status == "completed":
+            system_message = self._compose_sub_agent_message(
+                prefix=f"✅ 子智能体{agent_id} 任务摘要：{task_summary} 已完成。",
+                stats_summary=stats_summary,
+                summary=summary,
+                deliverables_dir=deliverables_dir,
+                duration_seconds=elapsed_seconds,
+            )
+        elif status == "timeout":
+            system_message = self._compose_sub_agent_message(
+                prefix=f"⏱️ 子智能体{agent_id} 任务摘要：{task_summary} 超时未完成。",
+                stats_summary=stats_summary,
+                summary=summary,
+            )
+        else:
+            system_message = self._compose_sub_agent_message(
+                prefix=f"❌ 子智能体{agent_id} 任务摘要：{task_summary} 执行失败。",
+                stats_summary=stats_summary,
+                summary=summary,
+            )
+
+        result = {
+            "success": success,
+            "status": status,
+            "task_id": task_id,
+            "agent_id": agent_id,
+            "message": summary,
+            "deliverables_dir": deliverables_dir,
+            "stats": stats,
+            "stats_summary": stats_summary,
+            "system_message": system_message,
+        }
+        if status == "completed" and elapsed_seconds is not None:
+            result["elapsed_seconds"] = elapsed_seconds
+            result["runtime_seconds"] = elapsed_seconds
+        task["final_result"] = result
+        return result
+
+    def _handle_timeout(self, task: Dict) -> Dict:
+        """处理任务超时。"""
+        task_id = task["task_id"]
+        running_task = self._running_tasks.pop(task_id, None)
+        if running_task and not running_task.done():
+            running_task.cancel()
+            deadline = time.time() + 5
+            while not running_task.done() and time.time() < deadline:
+                time.sleep(0.05)
+
+        task["status"] = "timeout"
+        task["updated_at"] = time.time()
+
+        stats = {}
+        stats_file = Path(task.get("stats_file", ""))
+        if stats_file.exists():
+            try:
+                stats = json.loads(stats_file.read_text(encoding="utf-8"))
+            except Exception:
+                stats = {}
+        stats_summary = self._build_stats_summary(stats)
+        system_message = self._compose_sub_agent_message(
+            prefix=f"⏱️ 子智能体{task.get('agent_id')} 任务摘要：{task.get('summary')} 超时未完成。",
+            stats_summary=stats_summary,
+            summary="等待超时，子智能体已被终止。",
+        )
+
+        result = {
+            "success": False,
+            "status": "timeout",
+            "task_id": task_id,
+            "agent_id": task.get("agent_id"),
+            "message": "等待超时，子智能体已被终止。",
+            "stats": stats,
+            "stats_summary": stats_summary,
+            "system_message": system_message,
+        }
+        task["final_result"] = result
+        self._save_state()
+        return result
+
+    def _mark_task_terminated(
+        self,
+        task: Dict[str, Any],
+        *,
+        message: str,
+        system_message: Optional[str] = None,
+        notified: bool = False,
+    ) -> Dict[str, Any]:
+        task["status"] = "terminated"
+        task["updated_at"] = time.time()
+        if notified:
+            task["notified"] = True
+        result = {
+            "success": False,
+            "status": "terminated",
+            "task_id": task.get("task_id"),
+            "agent_id": task.get("agent_id"),
+            "message": message,
+            "system_message": system_message or message,
+        }
+        task["final_result"] = result
+        return result
+
+    def _mark_task_done(
+        self,
+        task_id: str,
+        success: bool,
+        summary: str,
+        runtime_seconds: int,
+    ) -> None:
+        task = self.tasks.get(task_id)
+        if not task:
+            return
+        status = "completed" if success else "failed"
+        task["status"] = status
+        task["updated_at"] = time.time()
+        task["runtime_seconds"] = runtime_seconds
+        self._check_task_status(task)
+        self._save_state()
+
+    def _refresh_task_runtime_state(self, task: Dict[str, Any]) -> Dict[str, Any]:
+        """刷新单个任务运行态。"""
+        status = task.get("status")
+        if status in TERMINAL_STATUSES.union({"terminated"}):
+            return {"status": status, "task_id": task.get("task_id")}
+
+        task_id = task.get("task_id")
+        running_task = self._running_tasks.get(task_id) if task_id else None
+        if running_task:
+            if running_task.done():
+                try:
+                    running_task.result(timeout=0)
+                except Exception:
+                    pass
+                return self._check_task_status(task)
+            return {"status": "running", "task_id": task_id}
+
+        output_file = Path(task.get("output_file", ""))
+        if output_file.exists():
+            return self._check_task_status(task)
+
+        if self._should_force_cleanup_stale_task(task):
+            return self._mark_task_terminated(
+                task,
+                message="子智能体疑似僵尸任务，已超时自动清理运行状态。",
+                system_message="⚠️ 子智能体长时间未结束，系统已自动清理运行状态。",
+                notified=True,
+            )
+
+        return self._mark_task_terminated(
+            task,
+            message="检测到子智能体任务已退出，已自动清理运行状态。",
+            system_message="⚠️ 子智能体任务异常退出，系统已自动清理运行状态。",
+            notified=True,
+        )
+
+    def reconcile_task_states(self, conversation_id: Optional[str] = None) -> int:
+        """修正运行态任务状态，返回修正条目数。"""
+        changed = 0
+        for task in self.tasks.values():
+            if not isinstance(task, dict):
+                continue
+            if conversation_id and task.get("conversation_id") != conversation_id:
+                continue
+            before_status = task.get("status")
+            before_notified = task.get("notified")
+            self._refresh_task_runtime_state(task)
+            if task.get("status") != before_status or task.get("notified") != before_notified:
+                changed += 1
+        if changed:
+            self._save_state()
+        return changed
+
+    def _should_force_cleanup_stale_task(self, task: Dict[str, Any]) -> bool:
+        try:
+            created_at = float(task.get("created_at") or 0)
+        except (TypeError, ValueError):
+            created_at = 0
+        if created_at <= 0:
+            return False
+        from config import SUB_AGENT_DEFAULT_TIMEOUT
+        timeout_seconds = int(task.get("timeout_seconds") or SUB_AGENT_DEFAULT_TIMEOUT or 0)
+        timeout_seconds = max(timeout_seconds, 1)
+        grace_seconds = 120
+        elapsed = time.time() - created_at
+        if elapsed <= (timeout_seconds + grace_seconds):
+            return False
+        output_file = Path(task.get("output_file", ""))
+        if output_file.exists():
+            return False
+        progress_file = Path(task.get("progress_file", ""))
+        if progress_file.exists():
+            try:
+                stale_span = time.time() - progress_file.stat().st_mtime
+                if stale_span <= grace_seconds:
+                    return False
+            except Exception:
+                pass
+        return True

modules/sub_agent/stats.py

@@ -0,0 +1,69 @@
+"""子智能体统计摘要与消息构建工具。"""
+
+from __future__ import annotations
+
+import time
+from typing import Any, Dict, Optional
+
+
+class SubAgentStatsMixin:
+    """提供子智能体任务耗时、统计摘要与结果消息组装能力。"""
+
+    @staticmethod
+    def _compute_elapsed_seconds(task: Dict) -> Optional[int]:
+        try:
+            created_at = float(task.get("created_at") or 0)
+            updated_at = float(task.get("updated_at") or time.time())
+        except (TypeError, ValueError):
+            return None
+        if created_at <= 0:
+            return None
+        return int(round(max(0.0, updated_at - created_at)))
+
+    @staticmethod
+    def _coerce_stat_int(value: Any) -> int:
+        try:
+            return max(0, int(value))
+        except (TypeError, ValueError):
+            return 0
+
+    def _build_stats_summary(self, stats: Optional[Dict[str, Any]]) -> str:
+        if not isinstance(stats, dict):
+            stats = {}
+        api_calls = self._coerce_stat_int(stats.get("api_calls") or stats.get("turn_count"))
+        files_read = self._coerce_stat_int(stats.get("files_read"))
+        write_files = self._coerce_stat_int(stats.get("write_files"))
+        edit_files = self._coerce_stat_int(stats.get("edit_files"))
+        searches = self._coerce_stat_int(stats.get("searches"))
+        web_pages = self._coerce_stat_int(stats.get("web_pages"))
+        commands = self._coerce_stat_int(stats.get("commands"))
+        lines = [
+            f"调用了{api_calls}次",
+            f"阅读了{files_read}次文件",
+            f"写入了{write_files}次文件",
+            f"编辑了{edit_files}次文件",
+            f"搜索了{searches}次内容",
+            f"查看了{web_pages}个网页",
+            f"运行了{commands}个指令",
+        ]
+        return "\n".join(lines)
+
+    def _compose_sub_agent_message(
+        self,
+        *,
+        prefix: str,
+        stats_summary: str,
+        summary: str,
+        deliverables_dir: Optional[str] = None,
+        duration_seconds: Optional[int] = None,
+    ) -> str:
+        parts = [prefix]
+        if stats_summary:
+            parts.append(stats_summary)
+        if duration_seconds is not None:
+            parts.append(f"运行了{duration_seconds}秒")
+        if summary:
+            parts.append(summary)
+        if deliverables_dir:
+            parts.append(f"交付目录：{deliverables_dir}")
+        return "\n\n".join(parts)

modules/sub_agent/task.py

@@ -0,0 +1,358 @@
+from __future__ import annotations
+
+import asyncio
+import base64
+import json
+import mimetypes
+import time
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional, TYPE_CHECKING
+
+from modules.sub_agent.toolkit import (
+    SUB_AGENT_TOOLS,
+    FINISH_TOOL,
+    _format_tool_result,
+    _build_sub_agent_profile,
+)
+from utils.api_client import DeepSeekClient
+from utils.logger import setup_logger
+
+if TYPE_CHECKING:
+    from modules.sub_agent.manager import SubAgentManager
+
+logger = setup_logger(__name__)
+
+
+class SubAgentTask:
+    """单个后台子智能体任务。"""
+
+    def __init__(
+        self,
+        manager: "SubAgentManager",
+        task_record: Dict[str, Any],
+        task_message: str,
+        system_prompt: str,
+        model_key: Optional[str],
+        thinking_mode: Optional[str],
+    ):
+        self.manager = manager
+        self.task_record = task_record
+        self.task_message = task_message
+        self.system_prompt = system_prompt
+        self.model_key = model_key
+        self.thinking_mode = thinking_mode or "fast"
+        self.task_id = task_record["task_id"]
+        self.agent_id = task_record["agent_id"]
+        self.timeout_seconds = int(task_record.get("timeout_seconds") or 180)
+        self.deliverables_dir = Path(task_record["deliverables_dir"])
+
+        self.output_file = Path(task_record["output_file"])
+        self.stats_file = Path(task_record["stats_file"])
+        self.progress_file = Path(task_record["progress_file"])
+        self.conversation_file = Path(task_record["conversation_file"])
+
+        self.messages: List[Dict[str, Any]] = [
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": task_message},
+        ]
+        self.stats = {
+            "runtime_start": time.time() * 1000,
+            "runtime_seconds": 0,
+            "files_read": 0,
+            "write_files": 0,
+            "edit_files": 0,
+            "searches": 0,
+            "web_pages": 0,
+            "commands": 0,
+            "api_calls": 0,
+            "token_usage": {"prompt": 0, "completion": 0, "total": 0},
+        }
+        self._stdout_lines: List[str] = []
+        self._cancelled = False
+        self._task: Optional[asyncio.Task] = None
+
+    def emit(self, type_: str, data: Dict[str, Any]) -> None:
+        """输出一行 JSONL 到 progress 文件并缓存。"""
+        line = json.dumps({"type": type_, **data}, ensure_ascii=False)
+        self._stdout_lines.append(line)
+        try:
+            self.progress_file.parent.mkdir(parents=True, exist_ok=True)
+            with open(self.progress_file, "a", encoding="utf-8") as f:
+                f.write(line + "\n")
+        except Exception:
+            pass
+
+    async def run(self) -> None:
+        """主 LLM 循环。"""
+        try:
+            await self._run_loop()
+        except asyncio.CancelledError:
+            self._cancelled = True
+            logger.debug(f"[SubAgent] task={self.task_id} 被取消")
+            # shield 避免取消信号中断最终状态落盘
+            await asyncio.shield(self._write_failure("子智能体被手动终止"))
+            raise
+        except Exception as exc:
+            logger.exception(f"[SubAgent] task={self.task_id} 执行异常")
+            await self._write_failure(f"执行异常: {exc}")
+
+    async def _run_loop(self) -> None:
+        client, model_key = self._build_client()
+        tools = list(SUB_AGENT_TOOLS)
+        tools.append(FINISH_TOOL)
+
+        start_time = time.time()
+        max_turns = 50
+
+        for turn in range(1, max_turns + 1):
+            if self._cancelled:
+                break
+            elapsed = time.time() - start_time
+            if elapsed > self.timeout_seconds:
+                await self._write_timeout(elapsed)
+                return
+
+            self.stats["api_calls"] += 1
+            self.stats["turn_count"] = turn
+            self.stats["runtime_seconds"] = int(elapsed)
+            self.emit("stats", {**self.stats, "turn_count": turn})
+
+            assistant_message, reasoning, tool_calls, usage = await self._call_model(client, model_key, tools)
+            if usage:
+                self._apply_usage(usage)
+
+            final_message: Dict[str, Any] = {"role": "assistant", "content": assistant_message}
+            if reasoning:
+                final_message["reasoning_content"] = reasoning
+            if tool_calls:
+                final_message["tool_calls"] = tool_calls
+            self.messages.append(final_message)
+
+            if not tool_calls:
+                self.messages.append({
+                    "role": "user",
+                    "content": "如果你已经完成了任务，请调用 finish_task 工具提交完成报告。如果还没有完成，请继续执行任务。",
+                })
+                continue
+
+            for tool_call in tool_calls:
+                if self._cancelled:
+                    break
+                name = tool_call.get("function", {}).get("name", "")
+                args = self._parse_args(tool_call)
+                progress_id = tool_call.get("id") or f"tool_{int(time.time() * 1000)}_{uuid.uuid4().hex[:6]}"
+
+                if name == "finish_task":
+                    await self._write_finish(args, elapsed)
+                    return
+
+                self.emit("progress", {"id": progress_id, "tool": name, "status": "running", "args": args, "ts": int(time.time() * 1000)})
+                result = await self._execute_tool(name, args)
+                self.emit("progress", {"id": progress_id, "tool": name, "status": "completed" if result.get("success") else "failed", "args": args, "ts": int(time.time() * 1000)})
+
+                self._update_stats(name)
+
+                content = _format_tool_result(name, result)
+                if name == "read_mediafile" and result.get("success"):
+                    content = self._build_media_tool_content(result) or content
+                self.messages.append({
+                    "role": "tool",
+                    "tool_call_id": tool_call.get("id", progress_id),
+                    "content": content,
+                })
+
+        await self._write_failure("任务执行超过最大轮次限制", max_turns_exceeded=True)
+
+    def _build_client(self) -> tuple:
+        """加载模型配置并初始化 DeepSeekClient。"""
+        config_path = self.manager.models_config_file
+        models: List[Dict[str, Any]] = []
+        default_key = ""
+        if Path(config_path).exists():
+            try:
+                raw = json.loads(Path(config_path).read_text(encoding="utf-8"))
+                models = raw.get("models", []) if isinstance(raw, dict) else (raw if isinstance(raw, list) else [])
+                default_key = str(raw.get("default_model", "")) if isinstance(raw, dict) else ""
+            except Exception as exc:
+                logger.error(f"[SubAgent] 加载模型配置失败: {exc}")
+
+        model_map = {}
+        valid_models = []
+        for item in models:
+            profile = _build_sub_agent_profile(item)
+            if profile:
+                key = profile["name"]
+                model_map[key] = profile
+                valid_models.append(key)
+
+        chosen_key = self.model_key or default_key
+        if chosen_key not in model_map and valid_models:
+            chosen_key = valid_models[0]
+        if chosen_key not in model_map:
+            raise RuntimeError(f"未找到可用子智能体模型配置: {config_path}")
+
+        client = DeepSeekClient(thinking_mode=(self.thinking_mode == "thinking"), web_mode=True)
+        client.model_key = chosen_key
+        client.project_path = str(self.manager.project_path)
+        if self.thinking_mode == "thinking":
+            # 子智能体的 thinking 模式应全程使用思考模型
+            client.deep_thinking_session = True
+        client.apply_profile(model_map[chosen_key])
+        return client, chosen_key
+
+    async def _call_model(
+        self,
+        client: DeepSeekClient,
+        model_key: str,
+        tools: List[Dict[str, Any]],
+    ) -> tuple:
+        """调用模型并解析 assistant 消息。"""
+        assistant_message = ""
+        reasoning = ""
+        tool_calls: List[Dict[str, Any]] = []
+        usage = None
+
+        async for chunk in client.chat(self.messages, tools=tools, stream=True):
+            if self._cancelled:
+                break
+            if chunk.get("error"):
+                raise RuntimeError(f"API 调用失败: {chunk.get('error')}")
+            choice = (chunk.get("choices") or [{}])[0]
+            delta = choice.get("delta") or {}
+            if delta.get("content"):
+                assistant_message += delta["content"]
+            if delta.get("reasoning_content"):
+                reasoning += delta["reasoning_content"]
+            elif delta.get("reasoning_details"):
+                rd = delta["reasoning_details"]
+                if isinstance(rd, list):
+                    reasoning += "".join(str(d.get("text") or "") for d in rd)
+                elif isinstance(rd, str):
+                    reasoning += rd
+                elif isinstance(rd, dict):
+                    reasoning += str(rd.get("text") or "")
+
+            for tc in delta.get("tool_calls") or []:
+                idx = tc.get("index")
+                if idx is None:
+                    continue
+                while len(tool_calls) <= idx:
+                    tool_calls.append({"id": "", "type": "function", "function": {"name": "", "arguments": ""}})
+                existing = tool_calls[idx]
+                if tc.get("id"):
+                    existing["id"] = tc["id"]
+                fn = tc.get("function") or {}
+                if fn.get("name"):
+                    existing["function"]["name"] += fn["name"]
+                if fn.get("arguments"):
+                    existing["function"]["arguments"] += fn["arguments"]
+
+            if chunk.get("usage"):
+                usage = chunk["usage"]
+
+        return assistant_message, reasoning, tool_calls, usage
+
+    def _parse_args(self, tool_call: Dict[str, Any]) -> Dict[str, Any]:
+        raw = tool_call.get("function", {}).get("arguments") or "{}"
+        try:
+            return json.loads(raw)
+        except Exception:
+            return {"_raw": raw}
+
+    async def _execute_tool(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
+        """通过 manager 调用主进程执行工具。"""
+        return await self.manager.execute_tool_for_sub_agent(name, args)
+
+    def _update_stats(self, name: str) -> None:
+        if name == "read_file":
+            self.stats["files_read"] += 1
+        elif name == "write_file":
+            self.stats["write_files"] += 1
+        elif name == "edit_file":
+            self.stats["edit_files"] += 1
+        elif name == "search_workspace":
+            self.stats["searches"] += 1
+        elif name in ("web_search", "extract_webpage"):
+            self.stats["web_pages"] += 1
+        elif name == "run_command":
+            self.stats["commands"] += 1
+
+    def _apply_usage(self, usage: Any) -> None:
+        try:
+            if isinstance(usage, dict):
+                prompt = usage.get("prompt_tokens") or usage.get("prompt") or 0
+                completion = usage.get("completion_tokens") or usage.get("completion") or 0
+                total = usage.get("total_tokens") or usage.get("total") or (prompt + completion)
+                self.stats["token_usage"]["prompt"] += int(prompt)
+                self.stats["token_usage"]["completion"] += int(completion)
+                self.stats["token_usage"]["total"] += int(total)
+        except Exception:
+            pass
+
+    def _build_media_tool_content(self, result: Dict[str, Any]) -> Any:
+        """把 read_mediafile 结果转成 OpenAI 多模态 content。"""
+        b64 = result.get("b64")
+        mime = result.get("mime")
+        file_type = result.get("type")
+        if not b64 or not mime:
+            return None
+        if file_type == "image":
+            return [
+                {"type": "text", "text": f"已附加图片: {result.get('path')}"},
+                {"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}},
+            ]
+        if file_type == "video":
+            return [
+                {"type": "text", "text": f"已附加视频: {result.get('path')}"},
+                {"type": "video_url", "video_url": {"url": f"data:{mime};base64,{b64}"}},
+            ]
+        return None
+
+    async def _write_finish(self, args: Dict[str, Any], elapsed: float) -> None:
+        success = bool(args.get("success", False))
+        summary = str(args.get("summary") or "").strip()
+        self._finalize_task(success, summary, elapsed)
+
+    async def _write_timeout(self, elapsed: float) -> None:
+        self._finalize_task(False, "任务超时未完成", elapsed, timeout=True)
+
+    async def _write_failure(self, message: str, *, max_turns_exceeded: bool = False, timeout: bool = False) -> None:
+        elapsed = time.time() - (self.stats["runtime_start"] / 1000)
+        self._finalize_task(False, message, elapsed, max_turns_exceeded=max_turns_exceeded, timeout=timeout)
+
+    def _finalize_task(self, success: bool, summary: str, elapsed: float, *, max_turns_exceeded: bool = False, timeout: bool = False) -> None:
+        runtime_seconds = int(elapsed)
+        output_data = {
+            "success": success,
+            "summary": summary,
+            "timeout": timeout,
+            "max_turns_exceeded": max_turns_exceeded,
+            "stats": {**self.stats, "runtime_seconds": runtime_seconds, "turn_count": self.stats.get("turn_count", 0)},
+        }
+        conversation_data = {
+            "agent_id": self.agent_id,
+            "created_at": datetime.fromtimestamp(self.stats["runtime_start"] / 1000).isoformat(),
+            "completed_at": datetime.now().isoformat(),
+            "success": success,
+            "summary": summary,
+            "messages": self.messages,
+            "stats": output_data["stats"],
+        }
+        stats_data = {**self.stats, "runtime_seconds": runtime_seconds, "turn_count": self.stats.get("turn_count", 0)}
+
+        self.output_file.parent.mkdir(parents=True, exist_ok=True)
+        self.output_file.write_text(json.dumps(output_data, ensure_ascii=False), encoding="utf-8")
+        self.stats_file.write_text(json.dumps(stats_data, ensure_ascii=False), encoding="utf-8")
+        self.conversation_file.write_text(json.dumps(conversation_data, ensure_ascii=False), encoding="utf-8")
+
+        self.emit("output", output_data)
+        self.emit("conversation", conversation_data)
+
+        self.manager._mark_task_done(self.task_id, success, summary, runtime_seconds)
+
+    def cancel(self) -> None:
+        self._cancelled = True
+        if self._task and not self._task.done():
+            self._task.cancel()

modules/sub_agent/toolkit.py

@@ -0,0 +1,336 @@
+"""子智能体工具定义、结果格式化与模型配置解析。"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List, Optional
+
+# 子智能体可用工具定义（与前端进度展示兼容）
+SUB_AGENT_TOOLS: List[Dict[str, Any]] = [
+    {
+        "type": "function",
+        "function": {
+            "name": "read_file",
+            "description": "读取/搜索/抽取 UTF-8 文本文件。type=read/search/extract，仅单文件操作。",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "文件路径：工作区内请用相对路径，工作区外可用绝对路径。"},
+                    "type": {"type": "string", "enum": ["read", "search", "extract"], "description": "读取模式。"},
+                    "max_chars": {"type": "integer", "description": "返回内容最大字符数，超出将截断。"},
+                    "start_line": {"type": "integer", "description": "[read] 起始行号（1-based）。"},
+                    "end_line": {"type": "integer", "description": "[read] 结束行号（>= start_line）。"},
+                    "query": {"type": "string", "description": "[search] 搜索关键词。"},
+                    "max_matches": {"type": "integer", "description": "[search] 最多返回多少条命中窗口。"},
+                    "context_before": {"type": "integer", "description": "[search] 命中行向上追加行数。"},
+                    "context_after": {"type": "integer", "description": "[search] 命中行向下追加行数。"},
+                    "case_sensitive": {"type": "boolean", "description": "[search] 是否区分大小写。"},
+                    "segments": {
+                        "type": "array",
+                        "description": "[extract] 需要抽取的行区间数组。",
+                        "items": {
+                            "type": "object",
+                            "properties": {
+                                "label": {"type": "string"},
+                                "start_line": {"type": "integer"},
+                                "end_line": {"type": "integer"},
+                            },
+                            "required": ["start_line", "end_line"],
+                        },
+                        "minItems": 1,
+                    },
+                },
+                "required": ["path", "type"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "write_file",
+            "description": "创建或覆盖文件；append=true 时追加内容。",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "file_path": {"type": "string", "description": "文件路径：工作区内请用相对路径。"},
+                    "content": {"type": "string", "description": "要写入的内容。"},
+                    "append": {"type": "boolean", "default": False, "description": "是否追加到文件末尾。"},
+                },
+                "required": ["file_path", "content"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "edit_file",
+            "description": "在文件中执行一处或多处精确字符串替换；建议先用 read_file 精确复制 old_string。",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "file_path": {"type": "string", "description": "文件路径：工作区内请用相对路径。"},
+                    "replacements": {
+                        "type": "array",
+                        "description": "替换组列表，每组包含 old_string/new_string/replace_all。",
+                        "items": {
+                            "type": "object",
+                            "properties": {
+                                "old_string": {"type": "string"},
+                                "new_string": {"type": "string"},
+                                "replace_all": {"type": "boolean", "default": False, "description": "是否替换所有匹配内容。"},
+                            },
+                            "required": ["old_string", "new_string"],
+                        },
+                        "minItems": 1,
+                    },
+                },
+                "required": ["file_path", "replacements"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "run_command",
+            "description": "在当前终端环境执行一次性命令（非交互式）。",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "command": {"type": "string", "description": "要执行的命令。"},
+                    "timeout": {"type": "number", "description": "超时时间（秒），必填。"},
+                    "working_dir": {"type": "string", "description": "工作目录：工作区内请用相对路径。"},
+                },
+                "required": ["command", "timeout"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "web_search",
+            "description": "网络搜索（Tavily）。用于外部资料或最新信息检索。",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "query": {"type": "string"},
+                    "max_results": {"type": "integer"},
+                    "topic": {"type": "string", "enum": ["general", "news", "finance"]},
+                    "time_range": {"type": "string", "enum": ["day", "week", "month", "year", "d", "w", "m", "y"]},
+                    "days": {"type": "integer"},
+                    "start_date": {"type": "string"},
+                    "end_date": {"type": "string"},
+                    "country": {"type": "string"},
+                    "include_domains": {"type": "array", "items": {"type": "string"}},
+                },
+                "required": ["query"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "extract_webpage",
+            "description": "网页内容提取（Tavily）。mode=read 直接返回内容，mode=save 保存为文件。",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "mode": {"type": "string", "enum": ["read", "save"]},
+                    "url": {"type": "string"},
+                    "target_path": {"type": "string", "description": "[save] 保存路径，必须是 .md 文件。"},
+                },
+                "required": ["mode", "url"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "search_workspace",
+            "description": "在本地目录内搜索文件名或文件内容（跨文件）。仅返回摘要。",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "mode": {"type": "string", "enum": ["file", "content"]},
+                    "query": {"type": "string"},
+                    "root": {"type": "string", "description": "搜索起点目录，默认 '.'。"},
+                    "use_regex": {"type": "boolean"},
+                    "case_sensitive": {"type": "boolean"},
+                    "max_results": {"type": "integer"},
+                    "max_matches_per_file": {"type": "integer"},
+                    "include_glob": {"type": "array", "items": {"type": "string"}},
+                    "exclude_glob": {"type": "array", "items": {"type": "string"}},
+                    "max_file_size": {"type": "integer"},
+                },
+                "required": ["mode", "query"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "read_mediafile",
+            "description": "读取图片或视频文件，以 base64 形式返回给模型。非媒体文件将拒绝。",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string"},
+                },
+                "required": ["path"],
+            },
+        },
+    },
+]
+
+FINISH_TOOL: Dict[str, Any] = {
+    "type": "function",
+    "function": {
+        "name": "finish_task",
+        "description": "完成当前任务并退出。调用此工具表示你已经完成了分配的任务，所有交付文件已准备好。",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "success": {"type": "boolean"},
+                "summary": {"type": "string", "description": "任务完成摘要（50-200字）。"},
+            },
+            "required": ["success", "summary"],
+        },
+    },
+}
+
+
+def _format_tool_result(name: str, raw: Any) -> str:
+    """把主进程返回的工具结果格式化为文本，供 LLM 消费。"""
+    if not isinstance(raw, dict):
+        return str(raw) if raw is not None else ""
+    if raw.get("success") is False:
+        return raw.get("error") or raw.get("message") or "失败"
+    if name == "read_file":
+        if raw.get("type") == "read":
+            return raw.get("content") or ""
+        if raw.get("type") == "search":
+            parts = []
+            for match in raw.get("matches") or []:
+                hits = ",".join(str(h) for h in (match.get("hits") or []))
+                parts.append(f"[{match.get('id')}] L{match.get('line_start')}-{match.get('line_end')} hits:{hits}")
+                parts.append(match.get("snippet") or "")
+                parts.append("")
+            return "\n".join(parts).strip()
+        if raw.get("type") == "extract":
+            parts = []
+            for seg in raw.get("segments") or []:
+                label = seg.get("label") or "segment"
+                parts.append(f"[{label}] L{seg.get('line_start')}-{seg.get('line_end')}")
+                parts.append(seg.get("content") or "")
+                parts.append("")
+            return "\n".join(parts).strip()
+        return ""
+    if name == "write_file":
+        return raw.get("message") or "写入成功"
+    if name == "edit_file":
+        count = raw.get("replacements")
+        msg = raw.get("message") or ""
+        return f"已替换 {count} 处: {raw.get('path')}{'，' + msg if msg else ''}"
+    if name == "run_command":
+        return raw.get("output") or ""
+    if name == "web_search":
+        lines = [f"🔍 搜索查询: {raw.get('query')}", f"📅 搜索时间: {raw.get('searched_at') or datetime.now().isoformat()}", "", "📝 AI摘要:", raw.get("summary") or "", "", "---", "", "📊 搜索结果:"]
+        for idx, item in enumerate(raw.get("results") or [], 1):
+            lines.append(f"{idx}. {item.get('title') or ''}")
+            lines.append(f"   🔗 {item.get('url') or ''}")
+            lines.append(f"   📄 {item.get('content') or ''}")
+        return "\n".join(lines).strip()
+    if name == "extract_webpage":
+        if raw.get("mode") == "save":
+            return f"已保存: {raw.get('path')}"
+        return f"URL: {raw.get('url')}\n{raw.get('content') or ''}"
+    if name == "search_workspace":
+        if raw.get("mode") == "file":
+            lines = [f"命中文件({len(raw.get('matches') or [])}):"]
+            for idx, p in enumerate(raw.get("matches") or [], 1):
+                lines.append(f"{idx}) {p}")
+            return "\n".join(lines)
+        if raw.get("mode") == "content":
+            parts = []
+            for item in raw.get("results") or []:
+                parts.append(item.get("file") or "")
+                for m in item.get("matches") or []:
+                    parts.append(f"- L{m.get('line')}: {m.get('snippet')}")
+                parts.append("")
+            return "\n".join(parts).strip()
+        return ""
+    if name == "read_mediafile":
+        return raw.get("message") or "媒体已读取"
+    return json.dumps(raw, ensure_ascii=False)
+
+
+def _build_sub_agent_profile(model_raw: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    """把 sub_agent_models.json 中的模型条目转成 DeepSeekClient.apply_profile 所需格式。"""
+    name = str(model_raw.get("name") or model_raw.get("model_name") or model_raw.get("model") or "").strip()
+    url = str(model_raw.get("url") or model_raw.get("base_url") or "").strip()
+    api_key = str(model_raw.get("apikey") or model_raw.get("api_key") or "").strip()
+    if not name or not url or not api_key:
+        return None
+
+    modes_text = str(model_raw.get("modes") or model_raw.get("mode") or model_raw.get("supported_modes") or "").lower()
+    supports_thinking = "thinking" in modes_text
+    fast_only = modes_text == "fast"
+
+    multimodal_text = str(model_raw.get("multimodal") or model_raw.get("multi_modal") or model_raw.get("multi") or "none").lower()
+    multimodal = "none"
+    if "video" in multimodal_text:
+        multimodal = "image+video"
+    elif "image" in multimodal_text:
+        multimodal = "image"
+
+    max_output = _to_int(model_raw.get("max_output") or model_raw.get("max_tokens") or model_raw.get("max_output_tokens"))
+    max_context = _to_int(model_raw.get("max_context") or model_raw.get("context_window") or model_raw.get("max_context_tokens"))
+
+    model_id = str(model_raw.get("model_id") or name).strip()
+    extra = _pick_dict(model_raw, ["extra_parameter", "extra_params", "extra"])
+    fast_extra = _pick_dict(model_raw, ["fast_extra_parameter", "fast_extra_params", "fast_extra"])
+    thinking_extra = _pick_dict(model_raw, ["thinking_extra_parameter", "thinking_extra_params", "thinking_extra"])
+
+    profile: Dict[str, Any] = {
+        "name": name,
+        "multimodal": multimodal,
+        "context_window": max_context,
+        "supports_thinking": supports_thinking,
+        "fast_only": fast_only,
+        "fast": {
+            "base_url": url.rstrip("/"),
+            "api_key": api_key,
+            "model_id": model_id,
+            "max_tokens": max_output,
+            "context_window": max_context,
+            "extra_params": {**extra, **fast_extra},
+        },
+    }
+    if supports_thinking:
+        profile["thinking"] = {
+            "base_url": url.rstrip("/"),
+            "api_key": api_key,
+            "model_id": model_id,
+            "max_tokens": max_output,
+            "context_window": max_context,
+            "extra_params": {**extra, **thinking_extra},
+        }
+    else:
+        profile["thinking"] = None
+    return profile
+
+
+def _to_int(value: Any) -> Optional[int]:
+    try:
+        num = int(value)
+        return num if num > 0 else None
+    except Exception:
+        return None
+
+
+def _pick_dict(source: Dict[str, Any], keys: List[str]) -> Dict[str, Any]:
+    for key in keys:
+        value = source.get(key)
+        if isinstance(value, dict):
+            return value
+    return {}
+

modules/sub_agent/tools.py

@@ -0,0 +1,255 @@
+"""子智能体特有工具实现（search_workspace / read_mediafile）。
+
+这些工具在主进程中没有同名工具，因此由子智能体管理器本地实现。
+search_workspace 优先复用主进程的 run_command 调用 rg，回退到 Python 遍历；
+read_mediafile 直接读取项目内媒体文件并返回 base64。
+"""
+
+from __future__ import annotations
+
+import base64
+import fnmatch
+import json
+import mimetypes
+import re
+import shlex
+import shutil
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+
+async def handle_search_workspace(
+    project_path: Path,
+    terminal: Any,
+    arguments: Dict[str, Any],
+) -> Dict[str, Any]:
+    """search_workspace 实现。"""
+    mode = arguments.get("mode")
+    query = arguments.get("query") or ""
+    root = arguments.get("root") or "."
+    use_regex = bool(arguments.get("use_regex", False))
+    case_sensitive = bool(arguments.get("case_sensitive", False))
+    max_results = int(arguments.get("max_results", 20))
+    max_matches_per_file = int(arguments.get("max_matches_per_file", 3))
+    include_glob = arguments.get("include_glob") or ["**/*"]
+    exclude_glob = arguments.get("exclude_glob") or []
+    max_file_size = arguments.get("max_file_size")
+
+    if mode not in {"file", "content"}:
+        return {"success": False, "error": "search_workspace mode 必须是 file 或 content"}
+
+    root_path = (project_path / root).resolve()
+    try:
+        root_path.relative_to(project_path)
+    except Exception:
+        return {"success": False, "error": "搜索根目录必须位于项目目录内"}
+
+    if mode == "file":
+        return _search_workspace_file(project_path, root_path, query, use_regex, case_sensitive, max_results, include_glob, exclude_glob)
+
+    rg_result = await _search_workspace_content_rg(
+        project_path, root_path, query, use_regex, case_sensitive, max_results, max_matches_per_file,
+        include_glob, exclude_glob, max_file_size, terminal
+    )
+    if rg_result is not None:
+        return rg_result
+
+    return _search_workspace_content_python(
+        project_path, root_path, query, use_regex, case_sensitive, max_results, max_matches_per_file,
+        include_glob, exclude_glob, max_file_size
+    )
+
+
+def _search_workspace_file(
+    project_path: Path,
+    root_path: Path,
+    query: str,
+    use_regex: bool,
+    case_sensitive: bool,
+    max_results: int,
+    include_glob: List[str],
+    exclude_glob: List[str],
+) -> Dict[str, Any]:
+    try:
+        pattern = re.compile(query, 0 if case_sensitive else re.IGNORECASE) if use_regex else None
+        needle = query if case_sensitive else query.lower()
+        matches = []
+        for item in _iter_files(project_path, root_path, include_glob, exclude_glob):
+            name = item.name
+            hay = name if case_sensitive else name.lower()
+            ok = pattern.search(name) if pattern else needle in hay
+            if ok:
+                matches.append(str(item.relative_to(project_path)))
+                if len(matches) >= max_results:
+                    break
+        return {
+            "success": True,
+            "mode": "file",
+            "root": str(root_path.relative_to(project_path)),
+            "query": query,
+            "matches": matches,
+        }
+    except Exception as exc:
+        return {"success": False, "error": str(exc)}
+
+
+async def _search_workspace_content_rg(
+    project_path: Path,
+    root_path: Path,
+    query: str,
+    use_regex: bool,
+    case_sensitive: bool,
+    max_results: int,
+    max_matches_per_file: int,
+    include_glob: List[str],
+    exclude_glob: List[str],
+    max_file_size: Optional[int],
+    terminal: Any,
+) -> Optional[Dict[str, Any]]:
+    if not shutil.which("rg"):
+        return None
+    try:
+        args = ["rg", "--line-number", "--no-heading", "--color=never"]
+        if not use_regex:
+            args.append("--fixed-strings")
+        if not case_sensitive:
+            args.append("-i")
+        if max_matches_per_file:
+            args.append(f"--max-count={max_matches_per_file}")
+        if max_file_size:
+            args.append(f"--max-filesize={max_file_size}")
+        for g in include_glob:
+            args.extend(["-g", g])
+        for g in exclude_glob:
+            args.extend(["-g", f"!{g}"])
+        args.extend([query, str(root_path)])
+
+        result_text = await terminal.handle_tool_call("run_command", {"command": shlex.join(args), "timeout": 60})
+        parsed = json.loads(result_text) if isinstance(result_text, str) else result_text
+        if not parsed.get("success"):
+            return None
+
+        files_map: Dict[str, List[Dict[str, Any]]] = {}
+        for line in (parsed.get("output") or "").splitlines():
+            if not line.strip():
+                continue
+            parts = line.split(":", 2)
+            if len(parts) < 3:
+                continue
+            file_path, line_num, snippet = parts[0], parts[1], parts[2]
+            try:
+                line_no = int(line_num)
+            except ValueError:
+                continue
+            files_map.setdefault(file_path, [])
+            if len(files_map[file_path]) < max_matches_per_file:
+                files_map[file_path].append({"line": line_no, "snippet": snippet.strip()})
+            if len(files_map) >= max_results and len(files_map[file_path]) >= max_matches_per_file:
+                break
+
+        results = [{"file": f, "matches": ms} for f, ms in list(files_map.items())[:max_results]]
+        return {
+            "success": True,
+            "mode": "content",
+            "root": str(root_path.relative_to(project_path)),
+            "query": query,
+            "results": results,
+        }
+    except Exception:
+        return None
+
+
+def _search_workspace_content_python(
+    project_path: Path,
+    root_path: Path,
+    query: str,
+    use_regex: bool,
+    case_sensitive: bool,
+    max_results: int,
+    max_matches_per_file: int,
+    include_glob: List[str],
+    exclude_glob: List[str],
+    max_file_size: Optional[int],
+) -> Dict[str, Any]:
+    try:
+        pattern = re.compile(query, 0 if case_sensitive else re.IGNORECASE) if use_regex else None
+        needle = query if case_sensitive else query.lower()
+        results = []
+        for file_path in _iter_files(project_path, root_path, include_glob, exclude_glob):
+            if max_file_size:
+                try:
+                    if file_path.stat().st_size > max_file_size:
+                        continue
+                except Exception:
+                    continue
+            try:
+                text = file_path.read_text(encoding="utf-8", errors="ignore")
+            except Exception:
+                continue
+            lines = text.splitlines()
+            matches = []
+            for idx, line in enumerate(lines, 1):
+                hay = line if case_sensitive else line.lower()
+                ok = pattern.search(line) if pattern else needle in hay
+                if ok:
+                    matches.append({"line": idx, "snippet": line.strip()})
+                    if len(matches) >= max_matches_per_file:
+                        break
+            if matches:
+                results.append({"file": str(file_path.relative_to(project_path)), "matches": matches})
+                if len(results) >= max_results:
+                    break
+        return {
+            "success": True,
+            "mode": "content",
+            "root": str(root_path.relative_to(project_path)),
+            "query": query,
+            "results": results,
+        }
+    except Exception as exc:
+        return {"success": False, "error": str(exc)}
+
+
+def _iter_files(project_path: Path, root_path: Path, include_glob: List[str], exclude_glob: List[str]):
+    """按 glob 规则遍历文件。"""
+    for item in root_path.rglob("*"):
+        if not item.is_file():
+            continue
+        try:
+            rel = item.relative_to(project_path)
+        except Exception:
+            continue
+        rel_str = str(rel)
+        included = any(fnmatch.fnmatch(rel_str, g) for g in include_glob)
+        if not included:
+            continue
+        if any(fnmatch.fnmatch(rel_str, g) for g in exclude_glob):
+            continue
+        yield item
+
+
+async def handle_read_mediafile(project_path: Path, arguments: Dict[str, Any]) -> Dict[str, Any]:
+    """read_mediafile 实现：直接读取项目内媒体文件并返回 base64。"""
+    path = arguments.get("path")
+    if not path:
+        return {"success": False, "error": "path 不能为空"}
+    try:
+        abs_path = (project_path / path).resolve()
+        abs_path.relative_to(project_path)
+    except Exception:
+        return {"success": False, "error": "非法路径，超出项目根目录"}
+
+    if not abs_path.exists() or not abs_path.is_file():
+        return {"success": False, "error": f"文件不存在: {path}"}
+
+    mime, _ = mimetypes.guess_type(str(abs_path))
+    if not mime or (not mime.startswith("image/") and not mime.startswith("video/")):
+        return {"success": False, "error": "禁止的文件类型"}
+
+    try:
+        data = abs_path.read_bytes()
+        b64 = base64.b64encode(data).decode("utf-8")
+        file_type = "image" if mime.startswith("image/") else "video"
+        return {"success": True, "path": path, "mime": mime, "type": file_type, "b64": b64}
+    except Exception as exc:
+        return {"success": False, "error": str(exc)}

server/app_legacy.py

@@ -195,7 +195,7 @@ for noisy_logger in ('engineio.server', 'socketio.server'):
     logging.getLogger(noisy_logger).setLevel(logging.ERROR)
     logging.getLogger(noisy_logger).disabled = True
 # 静音子智能体模块错误日志（交由 brief_log 或前端提示处理）
-sub_agent_logger = logging.getLogger('modules.sub_agent_manager')
+sub_agent_logger = logging.getLogger('modules.sub_agent.manager')
 sub_agent_logger.setLevel(logging.CRITICAL)
 sub_agent_logger.disabled = True
 sub_agent_logger.propagate = False

server/chat_flow_task_main.py

@@ -44,7 +44,7 @@ from modules.skill_hint_manager import SkillHintManager
 from modules.upload_security import UploadSecurityError
 from modules.user_manager import UserWorkspace
 from modules.usage_tracker import QUOTA_DEFAULTS
-from modules.sub_agent_manager import TERMINAL_STATUSES
+from modules.sub_agent import TERMINAL_STATUSES
 from modules.versioning_manager import ConversationVersioningManager, VersioningError
 from core.web_terminal import WebTerminal
 from utils.tool_result_formatter import format_tool_result_for_context

server/chat_flow_task_support.py

@@ -5,7 +5,7 @@ import time
 from datetime import datetime
 from typing import Any, Callable, Dict, List, Optional
 
-from modules.sub_agent_manager import TERMINAL_STATUSES
+from modules.sub_agent import TERMINAL_STATUSES
 
 
 _VALID_SOURCES = {

server/conversation.py

@@ -46,7 +46,7 @@ from modules.personalization_manager import (
 from modules.upload_security import UploadSecurityError
 from modules.user_manager import UserWorkspace
 from modules.usage_tracker import QUOTA_DEFAULTS
-from modules.sub_agent_manager import TERMINAL_STATUSES
+from modules.sub_agent import TERMINAL_STATUSES
 from modules.versioning_manager import ConversationVersioningManager, VersioningError
 from core.web_terminal import WebTerminal
 from utils.tool_result_formatter import format_tool_result_for_context

static/src/components/overlay/SubAgentActivityDialog.vue

@@ -80,6 +80,10 @@ const buildText = (entry: ActivityEntry) => {
     const path = args.path || args.file_path || '';
     return `阅读 ${path}`;
   }
+  if (tool === 'write_file') {
+    const path = args.file_path || args.path || '';
+    return `写入文件 ${path}`;
+  }
   if (tool === 'read_skill') {
     const skillName = args.skill_name || '';
     return `阅读技能 ${skillName}`;
@@ -143,7 +147,7 @@ const displayItems = computed(() => {
 
   for (let i = 0; i < entries.length; i++) {
     const entry = entries[i];
-    if (!entry) continue;
+    if (!entry || entry.type !== 'progress') continue;
 
     const baseKey = entry.id || `${entry.tool || 'tool'}-${entry.ts || i}`;
     const lastGroup = groups[groups.length - 1];

static/src/styles/components/overlays/_overlays.scss

@@ -2085,6 +2085,7 @@
 .subagent-activity-title {
   font-size: 16px;
   font-weight: 600;
+  color: var(--text-primary);
 }
 
 .subagent-activity-close {
@@ -2167,6 +2168,20 @@
   flex: 1;
   overflow-y: auto;
   padding: 8px 0;
+  scrollbar-width: thin;
+}
+
+.subagent-activity-body::-webkit-scrollbar {
+  width: 8px;
+}
+
+.subagent-activity-body::-webkit-scrollbar-track {
+  background: transparent;
+}
+
+.subagent-activity-body::-webkit-scrollbar-thumb {
+  background: var(--text-muted);
+  border-radius: 8px;
 }
 
 .subagent-activity-empty,

test/test_sub_agent_regression.py

@@ -0,0 +1,369 @@
+"""子智能体 Python 重写回归测试（mock 模型）。"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import shutil
+import tempfile
+import time
+from pathlib import Path
+from typing import Any, Dict, List
+
+import sys
+# 测试脚本在 test/ 目录下，需要把项目根目录加入路径
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+# 把子智能体运行态数据重定向到临时目录，避免写入受保护的 ~/.agents 路径
+_test_tmp = Path(tempfile.mkdtemp(prefix="sub_agent_cfg_"))
+os.environ["SUB_AGENT_TASKS_BASE_DIR"] = str(_test_tmp / "tasks")
+os.environ["SUB_AGENT_STATE_FILE"] = str(_test_tmp / "state.json")
+os.environ["SUB_AGENT_PROJECT_RESULTS_DIR"] = str(_test_tmp / "results")
+
+from modules.sub_agent import SubAgentManager
+from utils.api_client import DeepSeekClient
+
+
+class FakeTerminal:
+    """模拟 WebTerminal，只实现 handle_tool_call。"""
+
+    def __init__(self, project_path: Path):
+        self.project_path = project_path
+
+    async def handle_tool_call(self, name: str, arguments: Dict[str, Any]) -> str:
+        if name == "write_file":
+            path = self.project_path / arguments["file_path"]
+            path.parent.mkdir(parents=True, exist_ok=True)
+            mode = "a" if arguments.get("append") else "w"
+            with open(path, mode, encoding="utf-8") as f:
+                f.write(arguments["content"])
+            return json.dumps({"success": True, "message": "写入成功", "path": str(path)})
+
+        if name == "edit_file":
+            path = self.project_path / arguments["file_path"]
+            content = path.read_text(encoding="utf-8")
+            count = 0
+            for repl in arguments.get("replacements", []):
+                old = repl["old_string"]
+                new = repl["new_string"]
+                if repl.get("replace_all"):
+                    n = content.count(old)
+                    content = content.replace(old, new)
+                    count += n
+                else:
+                    content = content.replace(old, new, 1)
+                    count += 1
+            path.write_text(content, encoding="utf-8")
+            return json.dumps({"success": True, "path": str(path), "replacements": count})
+
+        if name == "read_file":
+            path = self.project_path / arguments["path"]
+            if not path.exists():
+                return json.dumps({"success": False, "error": f"文件不存在: {path}"})
+            text = path.read_text(encoding="utf-8")
+            return json.dumps({"success": True, "type": "read", "content": text})
+
+        if name == "run_command":
+            import subprocess
+            cmd = arguments["command"]
+            timeout = arguments.get("timeout", 30)
+            cwd = arguments.get("working_dir")
+            if cwd:
+                cwd = self.project_path / cwd
+            # mock 中不真正执行长时间 sleep，避免阻塞事件循环
+            if cmd.strip().startswith("sleep "):
+                return json.dumps({"success": True, "output": f"mock: {cmd}"})
+            try:
+                proc = subprocess.run(cmd, shell=True, cwd=cwd, capture_output=True, text=True, timeout=timeout)
+                output = (proc.stdout or "") + (proc.stderr or "")
+                return json.dumps({"success": proc.returncode == 0, "output": output})
+            except Exception as exc:
+                return json.dumps({"success": False, "output": str(exc)})
+
+        return json.dumps({"success": False, "error": f"未实现的工具: {name}"})
+
+
+class FakeModel:
+    """按脚本驱动 DeepSeekClient.chat 的返回。"""
+
+    def __init__(self, responses: List[List[Dict[str, Any]]]):
+        self.responses = responses
+        self.call_index = 0
+
+    async def chat(self, messages, *, tools=None, stream=True, **kwargs):
+        if self.call_index >= len(self.responses):
+            chunks = [{"choices": [{"delta": {"content": ""}}]}]
+        else:
+            chunks = self.responses[self.call_index]
+        self.call_index += 1
+        for chunk in chunks:
+            yield chunk
+
+
+def make_tool_chunk(tool_calls: List[Dict[str, Any]]) -> Dict[str, Any]:
+    return {
+        "choices": [{
+            "delta": {
+                "tool_calls": [
+                    {"index": i, "id": f"call_{i}", "function": {"name": tc["name"], "arguments": tc["args"]}}
+                    for i, tc in enumerate(tool_calls)
+                ]
+            }
+        }]
+    }
+
+
+def make_finish_chunk() -> Dict[str, Any]:
+    return {
+        "choices": [{"delta": {"content": ""}}]
+    }
+
+
+def install_fake_model(responses: List[List[Dict[str, Any]]]):
+    fake = FakeModel(responses)
+    original = DeepSeekClient.chat
+    DeepSeekClient.chat = fake.chat
+    return fake, original
+
+
+def restore_model(original):
+    DeepSeekClient.chat = original
+
+
+def test_write_file_and_edit():
+    """测试 write_file 与 edit_file（多处替换 + replace_all）。"""
+    tmp = Path(tempfile.mkdtemp(prefix="sub_agent_test_"))
+    try:
+        manager = SubAgentManager(str(tmp), str(tmp / "data"))
+        terminal = FakeTerminal(tmp)
+        manager.set_terminal(terminal)
+
+        # 模型调用序列：先 write_file，再 edit_file 多处替换，最后 finish
+        responses = [
+            [make_tool_chunk([{
+                "name": "write_file",
+                "args": json.dumps({"file_path": "demo.txt", "content": "aaa bbb aaa ccc\n"})
+            }])],
+            [make_tool_chunk([{
+                "name": "edit_file",
+                "args": json.dumps({
+                    "file_path": "demo.txt",
+                    "replacements": [
+                        {"old_string": "bbb", "new_string": "BBB"},
+                        {"old_string": "aaa", "new_string": "AAA", "replace_all": True},
+                    ]
+                })
+            }])],
+            [make_tool_chunk([{
+                "name": "finish_task",
+                "args": json.dumps({"success": True, "summary": "已完成写入与编辑测试"})
+            }])],
+        ]
+        _, original = install_fake_model(responses)
+        try:
+            result = manager.create_sub_agent(
+                agent_id=1,
+                summary="写入编辑测试",
+                task="测试 write_file 与 edit_file",
+                deliverables_dir="deliverables",
+                timeout_seconds=30,
+                conversation_id="conv-test-1",
+                thinking_mode="fast",
+            )
+            assert result["success"], result
+            task_id = result["task_id"]
+            final = manager.wait_for_completion(task_id=task_id, timeout_seconds=10)
+            assert final["status"] == "completed", final
+
+            content = (tmp / "demo.txt").read_text(encoding="utf-8")
+            assert content == "AAA BBB AAA ccc\n", f"内容不符: {content!r}"
+            assert (tmp / "deliverables").exists()
+        finally:
+            restore_model(original)
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+def test_search_workspace():
+    """测试 search_workspace 走 run_command rg 回退。"""
+    tmp = Path(tempfile.mkdtemp(prefix="sub_agent_test_"))
+    try:
+        manager = SubAgentManager(str(tmp), str(tmp / "data"))
+        terminal = FakeTerminal(tmp)
+        manager.set_terminal(terminal)
+
+        (tmp / "findme.txt").write_text("hello world unique string", encoding="utf-8")
+
+        async def fake_search(project_path, terminal_, arguments):
+            query = arguments["query"]
+            if arguments.get("mode") == "file":
+                matches = [p.name for p in project_path.rglob("*") if p.is_file() and query in p.name]
+                return {"success": True, "mode": "file", "matches": matches}
+            matches = []
+            for p in project_path.rglob("*.txt"):
+                text = p.read_text(encoding="utf-8")
+                if query in text:
+                    matches.append({"file": str(p.relative_to(project_path)), "matches": [{"line": 1, "snippet": text}]})
+            return {"success": True, "mode": "content", "results": matches}
+
+        import modules.sub_agent.tools as sat
+        original_search = sat.handle_search_workspace
+        sat.handle_search_workspace = fake_search
+
+        responses = [
+            [make_tool_chunk([{
+                "name": "search_workspace",
+                "args": json.dumps({"mode": "content", "query": "unique string"})
+            }])],
+            [make_tool_chunk([{
+                "name": "finish_task",
+                "args": json.dumps({"success": True, "summary": "搜索完成"})
+            }])],
+        ]
+        _, original_model = install_fake_model(responses)
+        try:
+            result = manager.create_sub_agent(
+                agent_id=2,
+                summary="搜索测试",
+                task="测试 search_workspace",
+                deliverables_dir="deliverables2",
+                timeout_seconds=30,
+                conversation_id="conv-test-2",
+                thinking_mode="fast",
+            )
+            assert result["success"], result
+            final = manager.wait_for_completion(task_id=result["task_id"], timeout_seconds=10)
+            assert final["status"] == "completed", final
+        finally:
+            restore_model(original_model)
+            sat.handle_search_workspace = original_search
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+def test_read_mediafile():
+    """测试 read_mediafile 返回 base64。"""
+    tmp = Path(tempfile.mkdtemp(prefix="sub_agent_test_"))
+    try:
+        manager = SubAgentManager(str(tmp), str(tmp / "data"))
+        terminal = FakeTerminal(tmp)
+        manager.set_terminal(terminal)
+
+        # 创建一个最小 PNG：1x1 透明像素
+        png_bytes = bytes.fromhex(
+            "89504e470d0a1a0a0000000d49484452"
+            "000000010000000108060000001f15c4"
+            "890000000d4944415478da6300180000"
+            "03010005fe027f0000000049454e44ae"
+            "426082"
+        )
+        (tmp / "pixel.png").write_bytes(png_bytes)
+
+        responses = [
+            [make_tool_chunk([{
+                "name": "read_mediafile",
+                "args": json.dumps({"path": "pixel.png"})
+            }])],
+            [make_tool_chunk([{
+                "name": "finish_task",
+                "args": json.dumps({"success": True, "summary": "媒体读取完成"})
+            }])],
+        ]
+        _, original = install_fake_model(responses)
+        try:
+            result = manager.create_sub_agent(
+                agent_id=3,
+                summary="媒体测试",
+                task="测试 read_mediafile",
+                deliverables_dir="deliverables3",
+                timeout_seconds=30,
+                conversation_id="conv-test-3",
+                thinking_mode="fast",
+            )
+            assert result["success"], result
+            final = manager.wait_for_completion(task_id=result["task_id"], timeout_seconds=10)
+            assert final["status"] == "completed", final
+        finally:
+            restore_model(original)
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+def test_terminate():
+    """测试手动终止子智能体。"""
+    tmp = Path(tempfile.mkdtemp(prefix="sub_agent_test_"))
+    try:
+        manager = SubAgentManager(str(tmp), str(tmp / "data"))
+        terminal = FakeTerminal(tmp)
+        manager.set_terminal(terminal)
+
+        # 模型先调用一次 run_command，之后返回空内容，使任务持续运行
+        responses = [
+            [make_tool_chunk([{
+                "name": "run_command",
+                "args": json.dumps({"command": "sleep 1000", "timeout": 1200})
+            }])],
+            [make_finish_chunk()],
+            [make_finish_chunk()],
+            [make_finish_chunk()],
+            [make_finish_chunk()],
+        ]
+        fake, original = install_fake_model(responses)
+        try:
+            result = manager.create_sub_agent(
+                agent_id=4,
+                summary="终止测试",
+                task="测试终止功能",
+                deliverables_dir="deliverables4",
+                timeout_seconds=30,
+                conversation_id="conv-test-4",
+                thinking_mode="fast",
+                run_in_background=True,
+            )
+            assert result["success"], result
+            time.sleep(0.5)
+            term_result = manager.terminate_sub_agent(task_id=result["task_id"])
+            assert term_result["success"], term_result
+            # 通过 task_id 查询最终状态
+            task = manager.lookup_task(task_id=result["task_id"])
+            assert task is not None
+            assert task["status"] == "terminated"
+        finally:
+            restore_model(original)
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+def run_all():
+    print("=" * 60)
+    print("SubAgent Python rewrite regression tests")
+    print("=" * 60)
+    tests = [
+        ("write_file + edit_file", test_write_file_and_edit),
+        ("search_workspace", test_search_workspace),
+        ("read_mediafile", test_read_mediafile),
+        ("terminate", test_terminate),
+    ]
+    passed = 0
+    failed = 0
+    for name, fn in tests:
+        print(f"\nRunning: {name} ...")
+        try:
+            fn()
+            print(f"  ✅ PASSED: {name}")
+            passed += 1
+        except Exception as exc:
+            print(f"  ❌ FAILED: {name} -> {exc}")
+            import traceback
+            traceback.print_exc()
+            failed += 1
+    print("\n" + "=" * 60)
+    print(f"RESULT: {passed} passed, {failed} failed")
+    print("=" * 60)
+    return failed == 0
+
+
+if __name__ == "__main__":
+    ok = run_all()
+    sys.exit(0 if ok else 1)