From 960d087db0d612b1813de18a6db0018fa5f84a5a Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Wed, 10 Jun 2026 03:09:31 +0800 Subject: [PATCH] fix(sub-agent): stdout output, real-time progress, config externalization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - config/sub_agent: 子智能体模型配置走 deploy_config_path 直指，无回退 - easyagent/config: ensureConfig 文件不存在时 throw 而非自动创建 - easyagent/batch: 所有输出走 stdout JSONL (emit)，不写文件；conversation 走 emit 不写工作区 - sub_agent_manager: 新增 conversation_file；运行期间 select 非阻塞 drain stdout 即时写 progress --- config/sub_agent.py | 11 +- easyagent/src/batch/index.js | 311 +++++++++-------------------------- easyagent/src/config.js | 11 +- modules/sub_agent_manager.py | 143 ++++++++++++++-- 4 files changed, 217 insertions(+), 259 deletions(-) diff --git a/config/sub_agent.py b/config/sub_agent.py index 7eeecdc..c52893c 100644 --- a/config/sub_agent.py +++ b/config/sub_agent.py @@ -2,7 +2,12 @@ import os -from .paths import _resolve_repo_path, DATA_DIR, DEFAULT_PROJECT_PATH +from .paths import ( + _resolve_repo_path, + DATA_DIR, + DEFAULT_PROJECT_PATH, + deploy_config_path, +) # 子智能体服务 SUB_AGENT_SERVICE_BASE_URL = os.environ.get("SUB_AGENT_SERVICE_URL", "http://127.0.0.1:8092") @@ -17,6 +22,9 @@ SUB_AGENT_STATE_FILE = _resolve_repo_path(os.environ.get("SUB_AGENT_STATE_FILE", SUB_AGENT_PROJECT_RESULTS_DIR = _resolve_repo_path(os.environ.get("SUB_AGENT_PROJECT_RESULTS_DIR", ""), f"{DEFAULT_PROJECT_PATH}/sub_agent_results") SUB_AGENT_MAX_ACTIVE = int(os.environ.get("SUB_AGENT_MAX_ACTIVE", "5")) +# 子智能体模型配置文件（仅从部署目录读取，读不到就报错） +SUB_AGENT_MODELS_CONFIG_FILE = os.environ.get("SUB_AGENT_MODELS_CONFIG_FILE", "") or deploy_config_path("sub_agent_models.json") + __all__ = [ "SUB_AGENT_SERVICE_BASE_URL", "SUB_AGENT_DEFAULT_TIMEOUT", @@ -25,4 +33,5 @@ __all__ = [ "SUB_AGENT_PROJECT_RESULTS_DIR", "SUB_AGENT_STATE_FILE", "SUB_AGENT_MAX_ACTIVE", + "SUB_AGENT_MODELS_CONFIG_FILE", ] diff --git a/easyagent/src/batch/index.js b/easyagent/src/batch/index.js index 388a09c..60f8776 100644 --- a/easyagent/src/batch/index.js +++ b/easyagent/src/batch/index.js @@ -3,7 +3,8 @@ /** * easyagent 批处理模式 - * 用于子智能体执行，不需要交互式 CLI + * 所有输出走 stdout（JSONL），不写文件。 + * Python 侧捕获 stdout 后写入正式存储目录。 */ const fs = require('fs'); @@ -13,19 +14,23 @@ 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 }); + process.stdout.write(line + '\n'); +} + +// ── 解析命令行参数 ── function parseArgs() { const args = process.argv.slice(2); const config = { workspace: process.cwd(), taskFile: null, systemPromptFile: null, - outputFile: null, - statsFile: null, - progressFile: null, agentId: null, modelKey: null, thinkingMode: null, + configFile: null, timeout: 600, }; @@ -37,12 +42,6 @@ function parseArgs() { config.taskFile = args[++i]; } else if (arg === '--system-prompt-file' && i + 1 < args.length) { config.systemPromptFile = args[++i]; - } else if (arg === '--output-file' && i + 1 < args.length) { - config.outputFile = args[++i]; - } else if (arg === '--stats-file' && i + 1 < args.length) { - config.statsFile = args[++i]; - } else if (arg === '--progress-file' && i + 1 < args.length) { - config.progressFile = args[++i]; } else if (arg === '--agent-id' && i + 1 < args.length) { config.agentId = args[++i]; } else if (arg === '--model-key' && i + 1 < args.length) { @@ -51,13 +50,15 @@ function parseArgs() { 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'); @@ -67,80 +68,44 @@ function readFile(filePath) { } } -// 写入输出文件 -function writeOutput(filePath, data) { - try { - fs.writeFileSync(filePath, JSON.stringify(data, null, 2), 'utf8'); - } catch (err) { - console.error(`写入输出文件失败: ${filePath}`, err); - } -} - -// 更新统计文件 -function updateStats(statsFile, stats) { - try { - fs.writeFileSync(statsFile, JSON.stringify(stats, null, 2), 'utf8'); - } catch (err) { - console.error(`更新统计文件失败: ${statsFile}`, err); - } -} - -function appendProgress(progressFile, entry) { - if (!progressFile) return; - try { - const dir = path.dirname(progressFile); - if (!fs.existsSync(dir)) { - fs.mkdirSync(dir, { recursive: true }); - } - fs.appendFileSync(progressFile, `${JSON.stringify(entry)}\n`, 'utf8'); - } catch (err) { - console.error(`写入进度失败: ${progressFile}`, err); - } -} - -// 主函数 +// ── 主函数 ── async function main() { const config = parseArgs(); - if (!config.taskFile || !config.systemPromptFile || !config.outputFile) { - console.error('缺少必需参数: --task-file, --system-prompt-file, --output-file'); + 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'); - const ensuredConfig = modelConfig.ensureConfig(); + 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) { - writeOutput(config.outputFile, { - success: false, - summary: '未找到可用模型配置', - error: 'no_model_config', - }); + 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) { - writeOutput(config.outputFile, { - success: false, - summary: `未找到模型: ${modelKey}`, - error: 'model_not_found', - }); + emit('output', { success: false, summary: '未找到模型: ' + modelKey, error: 'model_not_found' }); process.exit(1); } - // 加载工具定义 + // 加载工具 const tools = JSON.parse(fs.readFileSync(path.join(__dirname, '../../doc/tools.json'), 'utf8')); - - // 添加 finish_task 工具 tools.push({ type: 'function', function: { @@ -149,14 +114,8 @@ async function main() { parameters: { type: 'object', properties: { - success: { - type: 'boolean', - description: '任务是否成功完成。true=成功完成所有要求；false=部分完成或遇到问题无法继续。', - }, - summary: { - type: 'string', - description: '任务完成摘要（50-200字），说明完成了什么工作、生成了哪些文件、关键发现或结果。如果失败，说明原因和已完成的部分。', - }, + success: { type: 'boolean', description: '任务是否成功完成。' }, + summary: { type: 'string', description: '任务完成摘要（50-200字）。' }, }, required: ['success', 'summary'], }, @@ -169,7 +128,7 @@ async function main() { { role: 'user', content: taskMessage }, ]; - // 统计信息 + // 统计 const stats = { runtime_start: Date.now(), runtime_seconds: 0, @@ -185,7 +144,7 @@ async function main() { const startTime = Date.now(); const timeoutMs = config.timeout * 1000; let turnCount = 0; - const maxTurns = 50; // 防止无限循环 + const maxTurns = 50; try { while (true) { @@ -193,48 +152,33 @@ async function main() { stats.api_calls += 1; stats.turn_count = turnCount; - // 检查超时 const elapsed = Date.now() - startTime; if (elapsed > timeoutMs) { - writeOutput(config.outputFile, { + emit('output', { success: false, summary: '任务超时未完成', timeout: true, - stats: { - ...stats, - runtime_seconds: Math.floor(elapsed / 1000), - }, + stats: { ...stats, runtime_seconds: Math.floor(elapsed / 1000) }, }); process.exit(1); } - // 检查轮次限制 if (turnCount > maxTurns) { - writeOutput(config.outputFile, { + emit('output', { success: false, - summary: `任务执行超过 ${maxTurns} 轮，可能陷入循环`, + summary: '任务执行超过 ' + maxTurns + ' 轮，可能陷入循环', max_turns_exceeded: true, - stats: { - ...stats, - runtime_seconds: Math.floor(elapsed / 1000), - }, + stats: { ...stats, runtime_seconds: Math.floor(elapsed / 1000) }, }); process.exit(1); } - // 更新统计文件 - if (config.statsFile) { - updateStats(config.statsFile, { - ...stats, - runtime_seconds: Math.floor(elapsed / 1000), - turn_count: turnCount, - }); - } + // 发送统计更新 + emit('stats', { ...stats, runtime_seconds: Math.floor(elapsed / 1000), turn_count: turnCount }); // 调用 API let assistantMessage = { role: 'assistant', content: '', tool_calls: [] }; let reasoningBuffer = ''; - let currentToolCall = null; let usage = null; try { @@ -251,37 +195,19 @@ async function main() { const delta = choice?.delta; if (!delta) continue; - // 处理内容 if (delta.content) { assistantMessage.content += delta.content; } - // 处理 reasoning_content（兼容 reasoning_details） - if (delta.reasoning_content || delta.reasoning_details || choice?.reasoning_details) { - let rc = ''; - if (delta.reasoning_content) { - rc = delta.reasoning_content; - } else if (delta.reasoning_details) { - if (Array.isArray(delta.reasoning_details)) { - rc = delta.reasoning_details.map((d) => d.text || '').join(''); - } else if (typeof delta.reasoning_details === 'string') { - rc = delta.reasoning_details; - } else if (delta.reasoning_details && typeof delta.reasoning_details.text === 'string') { - rc = delta.reasoning_details.text; - } - } else if (choice?.reasoning_details) { - if (Array.isArray(choice.reasoning_details)) { - rc = choice.reasoning_details.map((d) => d.text || '').join(''); - } else if (typeof choice.reasoning_details === 'string') { - rc = choice.reasoning_details; - } else if (choice.reasoning_details && typeof choice.reasoning_details.text === 'string') { - rc = choice.reasoning_details.text; - } - } - if (rc) reasoningBuffer += rc; + 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) { @@ -291,55 +217,37 @@ async function main() { type: 'function', function: { name: '', arguments: '' }, }; - currentToolCall = assistantMessage.tool_calls[tc.index]; - } else { - currentToolCall = assistantMessage.tool_calls[tc.index]; } - - if (tc.id) currentToolCall.id = tc.id; - if (tc.function?.name) currentToolCall.function.name += tc.function.name; - if (tc.function?.arguments) currentToolCall.function.arguments += tc.function.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; } } } - // 处理 usage if (chunk.usage) { usage = normalizeUsagePayload(chunk.usage); } } } catch (err) { - writeOutput(config.outputFile, { + emit('output', { success: false, - summary: `API 调用失败: ${err.message}`, + summary: 'API 调用失败: ' + err.message, error: 'api_error', - stats: { - ...stats, - runtime_seconds: Math.floor((Date.now() - startTime) / 1000), - }, + stats: { ...stats, runtime_seconds: Math.floor((Date.now() - startTime) / 1000) }, }); process.exit(1); } - // 更新 token 统计 - if (usage) { - applyUsage(stats.token_usage, usage); - } + if (usage) applyUsage(stats.token_usage, usage); - // 添加助手消息到历史（reasoning_content 放在 content 前） const finalAssistantMessage = reasoningBuffer - ? { - role: 'assistant', - reasoning_content: reasoningBuffer, - content: assistantMessage.content, - tool_calls: assistantMessage.tool_calls, - } + ? { role: 'assistant', reasoning_content: reasoningBuffer, content: assistantMessage.content, tool_calls: assistantMessage.tool_calls } : assistantMessage; messages.push(finalAssistantMessage); - // 如果没有工具调用，检查是否忘记调用 finish_task if (!assistantMessage.tool_calls || assistantMessage.tool_calls.length === 0) { - // 兜底机制：提醒调用 finish_task messages.push({ role: 'user', content: '如果你已经完成了任务，请调用 finish_task 工具提交完成报告。如果还没有完成，请继续执行任务。', @@ -347,95 +255,50 @@ async function main() { continue; } - // 执行工具调用 for (const toolCall of assistantMessage.tool_calls) { const toolName = toolCall.function.name; stats.tool_calls = (stats.tool_calls || 0) + 1; - // 检查是否是 finish_task if (toolName === 'finish_task') { let args = {}; try { args = JSON.parse(toolCall.function.arguments || '{}'); } catch (err) { - writeOutput(config.outputFile, { + emit('output', { success: false, summary: 'finish_task 参数解析失败', error: 'invalid_finish_args', - stats: { - ...stats, - runtime_seconds: Math.floor((Date.now() - startTime) / 1000), - }, + stats: { ...stats, runtime_seconds: Math.floor((Date.now() - startTime) / 1000) }, }); process.exit(1); } - // 保存对话记录到交付目录的 .subagent/ 子目录 - try { - // 从任务文件路径推断交付目录 - const taskFileDir = path.dirname(config.taskFile); - // 读取任务消息以获取交付目录路径 - const taskContent = fs.readFileSync(config.taskFile, 'utf8'); - const deliverablesDirMatch = taskContent.match(/\*\*交付目录\*\*：(.+)/); - if (deliverablesDirMatch) { - const deliverablesDirPath = deliverablesDirMatch[1].split('\n')[0].trim(); - const subagentDir = path.join(deliverablesDirPath, '.subagent'); + // 输出对话记录到 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) }, + }); - // 确保目录存在 - if (!fs.existsSync(subagentDir)) { - fs.mkdirSync(subagentDir, { recursive: true }); - } - - // 保存对话记录 - const conversationData = { - 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), - }, - }; - - const conversationFile = path.join(subagentDir, 'conversation.json'); - fs.writeFileSync(conversationFile, JSON.stringify(conversationData, null, 2), 'utf8'); - } - } catch (err) { - // 保存对话记录失败不影响任务完成 - console.error('保存对话记录失败:', err); - } - - // 写入输出并退出 - writeOutput(config.outputFile, { + // 最终输出 + emit('output', { success: args.success || false, summary: args.summary || '任务完成', - stats: { - ...stats, - runtime_seconds: Math.floor((Date.now() - startTime) / 1000), - }, + stats: { ...stats, runtime_seconds: Math.floor((Date.now() - startTime) / 1000) }, }); process.exit(0); } let progressArgs = {}; - try { - progressArgs = JSON.parse(toolCall.function.arguments || '{}'); - } catch (err) { - progressArgs = { _raw: toolCall.function.arguments || '' }; - } - const progressId = toolCall.id || `tool_${Date.now()}_${Math.random().toString(16).slice(2, 8)}`; - appendProgress(config.progressFile, { - id: progressId, - tool: toolName, - status: 'running', - args: progressArgs, - ts: Date.now(), - }); + 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, @@ -444,44 +307,28 @@ async function main() { abortSignal: null, }); - appendProgress(config.progressFile, { - id: progressId, - tool: toolName, - status: result && result.success ? 'completed' : 'failed', - args: progressArgs, - ts: Date.now(), - }); + 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 || '', - }); + messages.push({ role: 'tool', tool_call_id: toolCall.id, content: result.formatted || '' }); } } } catch (err) { - writeOutput(config.outputFile, { + emit('output', { success: false, - summary: `执行出错: ${err.message}`, + summary: '执行出错: ' + err.message, error: 'execution_error', - stats: { - ...stats, - runtime_seconds: Math.floor((Date.now() - startTime) / 1000), - }, + stats: { ...stats, runtime_seconds: Math.floor((Date.now() - startTime) / 1000) }, }); process.exit(1); } } -// 运行 main().catch((err) => { console.error('批处理模式执行失败:', err); process.exit(1); diff --git a/easyagent/src/config.js b/easyagent/src/config.js index 78538b9..9b08912 100644 --- a/easyagent/src/config.js +++ b/easyagent/src/config.js @@ -98,15 +98,10 @@ function buildConfig(raw, filePath) { }; } -function ensureConfig() { - const file = DEFAULT_CONFIG_PATH; +function ensureConfig(customPath) { + const file = customPath || DEFAULT_CONFIG_PATH; if (!fs.existsSync(file)) { - const template = { - tavily_api_key: '', - default_model: '', - models: [], - }; - fs.writeFileSync(file, JSON.stringify(template, null, 2), 'utf8'); + throw new Error(`子智能体模型配置文件不存在: ${file}`); } const content = fs.readFileSync(file, 'utf8'); const raw = JSON.parse(content); diff --git a/modules/sub_agent_manager.py b/modules/sub_agent_manager.py index 76856ff..32636f6 100644 --- a/modules/sub_agent_manager.py +++ b/modules/sub_agent_manager.py @@ -9,6 +9,7 @@ import shlex import platform import shutil import signal +import select from pathlib import Path, PurePosixPath from typing import Dict, List, Optional, Any, Tuple, TYPE_CHECKING @@ -16,6 +17,7 @@ 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, @@ -124,16 +126,13 @@ class SubAgentManager: except ValueError as exc: return {"success": False, "error": str(exc)} - # 创建.subagent目录用于存储对话历史 - subagent_dir = deliverables_path / ".subagent" - subagent_dir.mkdir(parents=True, exist_ok=True) - # 准备文件路径 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) @@ -152,9 +151,6 @@ class SubAgentManager: workspace_path=str(self.project_path), task_file=str(task_file), system_prompt_file=str(system_prompt_file), - output_file=str(output_file), - stats_file=str(stats_file), - progress_file=str(progress_file), agent_id=agent_id, timeout_seconds=timeout_seconds, model_key=model_key, @@ -206,6 +202,9 @@ class SubAgentManager: return {"success": False, "error": f"构建宿主机沙箱启动命令失败：{exc}"} try: + logger.debug(f"[SubAgent] 启动进程 task={task_id} execution_mode={execution_mode}") + logger.debug(f"[SubAgent] cmd: {self._join_shell_words(launch_cmd)[:500]}") + logger.debug(f"[SubAgent] cwd: {launch_cwd}") process = subprocess.Popen( launch_cmd, stdout=subprocess.PIPE, @@ -214,6 +213,7 @@ class SubAgentManager: env=env, pass_fds=pass_fds, ) + logger.debug(f"[SubAgent] 进程已启动 task={task_id} pid={process.pid}") except Exception as exc: return {"success": False, "error": f"启动子智能体失败: {exc}"} finally: @@ -231,7 +231,6 @@ class SubAgentManager: "task": task, "status": "running", "deliverables_dir": str(deliverables_path), - "subagent_dir": str(subagent_dir), "timeout_seconds": timeout_seconds, "thinking_mode": thinking_mode, "created_at": time.time(), @@ -241,6 +240,7 @@ class SubAgentManager: "output_file": str(output_file), "stats_file": str(stats_file), "progress_file": str(progress_file), + "conversation_file": str(conversation_file), "pid": process.pid, "execution_mode": execution_mode, "container_name": container_name, @@ -374,9 +374,6 @@ class SubAgentManager: workspace_path: str, task_file: str, system_prompt_file: str, - output_file: str, - stats_file: str, - progress_file: str, agent_id: int, timeout_seconds: int, model_key: Optional[str], @@ -388,11 +385,9 @@ class SubAgentManager: "--workspace", workspace_path, "--task-file", task_file, "--system-prompt-file", system_prompt_file, - "--output-file", output_file, - "--stats-file", stats_file, - "--progress-file", progress_file, "--agent-id", str(agent_id), "--timeout", str(timeout_seconds), + "--config-file", SUB_AGENT_MODELS_CONFIG_FILE, ] if model_key: cmd.extend(["--model-key", model_key]) @@ -417,19 +412,131 @@ class SubAgentManager: cmd = [str(fd_num) if token == "__SECCOMP_FD__" else token for token in plan_command] return cmd, (fd_num,), seccomp_fd def _check_task_status(self, task: Dict) -> Dict: - """检查任务状态，如果完成则解析输出。""" + """检查任务状态，如果完成则从 stdout 解析输出并写入正式目录。""" task_id = task["task_id"] process = self.processes.get(task_id) # 检查进程是否结束 if process: + # 先非阻塞读取 stdout 中已有的行（运行中和已结束都需要） + stdout_lines = task.get("_stdout_lines", []) + progress_file = Path(task.get("progress_file", "")) + new_lines_count = 0 + + while True: + try: + ready, _, _ = select.select([process.stdout], [], [], 0) + except (ValueError, OSError): + break # fd 已关闭 + if not ready: + break + line = process.stdout.readline() + if not line: + break + line_text = line.decode('utf-8', errors='replace') if isinstance(line, bytes) else line + line_text = line_text.strip() + if line_text: + stdout_lines.append(line_text) + new_lines_count += 1 + + if new_lines_count: + task["_stdout_lines"] = stdout_lines + # 即时写入所有 progress 行到文件（全量重写，保证与内存一致） + try: + progress_file.parent.mkdir(parents=True, exist_ok=True) + with open(str(progress_file), "w", encoding="utf-8") as f: + for line_text in stdout_lines: + try: + entry = json.loads(line_text) + if entry.get("type") == "progress": + f.write(json.dumps(entry, ensure_ascii=False) + "\n") + except Exception: + pass + except Exception: + pass + returncode = process.poll() if returncode is None: - # 进程还在运行 return {"status": "running", "task_id": task_id} - # 进程已结束，读取输出文件 + # 进程已结束，读取 stderr + stderr_text = "" + try: + stderr_data = process.stderr.read() + stderr_text = stderr_data.decode('utf-8', errors='replace') if isinstance(stderr_data, bytes) and stderr_data else "" + except Exception: + pass + + logger.debug( + f"[SubAgent] 进程结束 task={task_id} pid={process.pid} " + f"rc={returncode} stdout_lines={len(stdout_lines)} stderr_len={len(stderr_text)}" + ) + if stderr_text: + logger.debug(f"[SubAgent] stderr: {stderr_text[:2000]}") + + # 解析累积的 stdout 行，将输出写入正式目录 + output_data = None + progress_entries = [] + stats_data = None + conversation_data = None + for line_text in stdout_lines: + if not line_text: + continue + try: + entry = json.loads(line_text) + except Exception: + continue + entry_type = entry.get("type", "") + if entry_type == "output": + output_data = entry + elif entry_type == "progress": + progress_entries.append(entry) + elif entry_type == "stats": + stats_data = entry + elif entry_type == "conversation": + conversation_data = entry + + # 写入正式目录 + output_file = Path(task.get("output_file", "")) + stats_file = Path(task.get("stats_file", "")) + progress_file = Path(task.get("progress_file", "")) + conversation_file = Path(task.get("conversation_file", "")) + + if output_data: + try: + output_file.parent.mkdir(parents=True, exist_ok=True) + output_file.write_text(json.dumps(output_data, ensure_ascii=False), encoding="utf-8") + logger.debug(f"[SubAgent] output 已写入: {output_file}") + except Exception as exc: + logger.debug(f"[SubAgent] 写入 output 失败: {exc}") + + if stats_data: + try: + stats_file.parent.mkdir(parents=True, exist_ok=True) + stats_file.write_text(json.dumps(stats_data, ensure_ascii=False), encoding="utf-8") + except Exception: + pass + + if progress_entries: + try: + progress_file.parent.mkdir(parents=True, exist_ok=True) + with open(str(progress_file), "w", encoding="utf-8") as f: + for entry in progress_entries: + f.write(json.dumps(entry, ensure_ascii=False) + "\n") + except Exception: + pass + + if conversation_data: + try: + conversation_file.parent.mkdir(parents=True, exist_ok=True) + conversation_file.write_text(json.dumps(conversation_data, ensure_ascii=False), encoding="utf-8") + logger.debug(f"[SubAgent] conversation 已写入: {conversation_file}") + except Exception as exc: + logger.debug(f"[SubAgent] 写入 conversation 失败: {exc}") + + # 读取输出文件 output_file = Path(task.get("output_file", "")) + logger.debug(f"[SubAgent] 检查输出文件: {output_file} exists={output_file.exists()}") if not output_file.exists(): # 输出文件不存在，可能是异常退出 task["status"] = "failed" @@ -594,7 +701,7 @@ class SubAgentManager: {task} **交付目录**：{deliverables_path} -请将所有生成的文件保存到此目录。对话历史会自动保存到 {deliverables_path}/.subagent/ 目录。 +请将所有生成的文件保存到此目录。 **超时时间**：{timeout_seconds} 秒