/** * master-session.js — Master Agent 会话管理器 * * Master Agent 是一个持久化的 Claude CLI 会话,具备完整对话能力。 * 当 Master 判断需要多 Agent 协作时,通过 [TOOL_CALL:] 调用系统工具。 * * 工作流程: * 1. 用户消息 → Master Claude CLI * 2. Master 直接回答 / 调用 orchestrate 工具生成计划 * 3. Master 按计划调用 call_sub_agent 执行各步骤 * 4. 收集 Sub Agent 结果 → 综合回复用户 */ const { EventEmitter } = require('events'); const fs = require('fs'); const path = require('path'); const AgentManager = require('./agent-manager'); const { ToolRegistry } = require('../tools/tool-registry'); const { createOrchestrateTool } = require('../tools/orchestrate-tool'); const { createCallAgentTool, createListAgentsTool } = require('../tools/call-agent-tool'); const { createReadCheckpointTool, createReadLogTool, createListRetrainJobsTool } = require('../tools/read-tools'); const { createSearchSkillsTool, createCallMarketplaceSkillTool } = require('../tools/marketplace-tools'); const { createTrainSkillTool } = require('../tools/train-skill-tool'); const { waitForGlobalLogManager } = require('./logging'); class MasterSession extends EventEmitter { /** * @param {object} config - 系统配置 (agents + backends) * @param {string} workDir - 工作目录 * @param {object} [options] */ constructor(config, workDir, options = {}) { super(); this.config = config; this.workDir = workDir; this.options = options; this.manager = null; this.toolRegistry = new ToolRegistry(); this._initialized = false; this._messageCount = 0; // 多轮对话计数 this._busy = false; // 并发控制 this._messageQueue = []; // 排队消息 } /** * 初始化 Master 会话 * - 创建 AgentManager * - 注册所有可用工具 */ async init() { if (this._initialized) return; await waitForGlobalLogManager(); this.manager = new AgentManager(this.config, this.workDir, { runRegistry: this.options.runRegistry || null, }); // 注册核心工具 this.toolRegistry.registerAll([ createOrchestrateTool(), createCallAgentTool(), createListAgentsTool(), // Retrain 监控工具 createReadCheckpointTool(), createReadLogTool(), createListRetrainJobsTool(), // 市场技能搜索 + 动态调用 createSearchSkillsTool(), createCallMarketplaceSkillTool(), // Innovator 训练流水线 createTrainSkillTool(), ]); this._initialized = true; this.emit('initialized'); } /** * 处理用户消息 * * 将消息发给 Master Claude CLI,解析输出中的工具调用, * 执行工具后将结果反馈给 Master,直到 Master 输出最终回复。 * * 支持多轮对话: * - 首次消息:注入完整 system prompt(工具描述 + Agent 列表) * - 后续消息:仅发送用户消息(ACP session 保持上下文) * * @param {string} message - 用户消息 * @param {object} [options] * @param {function} [options.onData] - 流式数据回调 * @param {function} [options.onToolCall] - 工具调用通知 * @param {number} [options.maxToolRounds] - 最大工具调用轮次,默认 10 * @returns {Promise} */ async handleMessage(message, options = {}) { // 如果 Master 忙碌,排队等待 if (this._busy) { console.log('[MasterSession] 用户消息排队等待中...'); return new Promise(resolve => { this._messageQueue.push({ message, options, resolve, isSystem: false }); }); } return this._executeMessage(message, options, false); } /** * 构建初始 Master prompt */ _buildMasterPrompt(userMessage, toolPrompt, agentList) { // 加载 Master SKILL.md 角色指令 let skillMd = ''; try { const skillPath = path.join(this.workDir, 'master', 'SKILL.md'); skillMd = fs.readFileSync(skillPath, 'utf-8'); } catch { } return `\ ${skillMd} ${toolPrompt} ## 可用 Sub Agent ${agentList} --- ## 用户消息 ${userMessage} --- ## ⚠️ 关键行为约束 1. **不要解释你的推理过程** — 直接行动 2. **需要调用工具时,立即输出 [TOOL_CALL:] 块** — 不要先写一段分析 3. **简单问题 → 直接回答(不超过 3 段)** 4. **需要编排 → 直接输出 [TOOL_CALL: call_sub_agent] 或 [TOOL_CALL: orchestrate]** 5. **你的文本输出会被解析 [TOOL_CALL:] 标记 — 如果你只是描述而不输出标记,工具不会被执行**`; } /** * 构建后续消息 prompt(多轮对话) * ACP session 已保持上下文,只需发新消息 */ _buildFollowUpPrompt(userMessage) { return `\ ## 用户后续消息 ${userMessage} --- 请根据用户消息决定如何回应(你仍然可以使用之前提供给你的工具)。`; } /** * 构建工具执行结果的反馈 prompt */ _buildToolResultPrompt(toolResults) { const blocks = toolResults.map(tr => { const resultStr = typeof tr.result === 'object' ? JSON.stringify(tr.result, null, 2) : String(tr.result); return `### 工具 ${tr.tool} 执行结果\n\n\`\`\`json\n${resultStr}\n\`\`\``; }).join('\n\n'); return `\ ## 工具执行结果 ${blocks} --- 请根据工具执行结果决定下一步: - 如果是 orchestrate(plan) 的结果 → 按计划逐步调用 call_sub_agent - 如果是 call_sub_agent 的结果 → 检查是否还有后续步骤 - 所有步骤完成 → 综合所有结果,给用户最终回复`; } /** * 构建 Agent + Skill 列表文本 * 仅列出核心系统 Agent(repo/skills/ 下的),不列出 data/skills/ 的市场技能 */ _buildAgentList() { const agents = this.config.agents || {}; const { CORE_SKILLS } = require('../../services/skill-dir-manager'); const lines = []; for (const [name, cfg] of Object.entries(agents)) { if (name === 'master' || name === 'gateway') continue; // 只列出核心系统 agent,跳过市场技能 if (!CORE_SKILLS.has(name)) continue; lines.push(`- **${name}** (${cfg.type}): ${cfg.description || ''}`); } return lines.join('\n'); } /** * 处理系统消息(巡检等) * * 与 handleMessage 类似,但: * - 不增加 _messageCount(不影响用户对话上下文判断) * - 使用独立的 prompt 前缀标识为系统消息 * - 等待 _busy 锁释放后再执行 * * @param {string} message - 系统消息 * @param {object} [options] * @returns {Promise} */ async handleSystemMessage(message, options = {}) { // 等待当前任务完成 if (this._busy) { console.log('[MasterSession] 系统消息排队等待中...'); await new Promise(resolve => { this._messageQueue.push({ message, options, resolve, isSystem: true }); }); return; // resolve 会在 _processQueue 中被调用 } return this._executeMessage(`[系统巡检消息]\n\n${message}`, options, true); } /** * 内部:执行消息处理(带 _busy 锁) */ async _executeMessage(prompt, options, isSystem = false) { if (!this._initialized) await this.init(); this._busy = true; try { const { onData, onToolCall, maxToolRounds = 10 } = options; let currentPrompt; if (this._messageCount === 0 && !isSystem) { const toolPrompt = this.toolRegistry.generateToolPrompt(); const agentList = this._buildAgentList(); currentPrompt = this._buildMasterPrompt(prompt, toolPrompt, agentList); } else { currentPrompt = this._buildFollowUpPrompt(prompt); } if (!isSystem) this._messageCount++; let allOutput = ''; let toolCallHistory = []; let totalApiMs = 0; for (let round = 0; round < maxToolRounds; round++) { this.emit('round', { round, prompt: currentPrompt.substring(0, 200) }); const roundStart = Date.now(); const result = await this.manager.callAgent('master', currentPrompt, { onData: onData || (() => { }), }); totalApiMs += Date.now() - roundStart; const masterOutput = result.output || ''; allOutput += masterOutput + '\n'; const toolCalls = ToolRegistry.parseToolCalls(masterOutput); if (toolCalls.length === 0) { return { type: isSystem ? 'system_reply' : 'reply', output: masterOutput, toolCalls: toolCallHistory, rounds: round + 1, durationApiMs: totalApiMs, }; } const toolResults = []; for (const tc of toolCalls) { if (onToolCall) onToolCall(tc); this.emit('tool_call', tc); const context = { config: this.config, workDir: this.workDir, agentManager: this.manager, onSubAgentData: options.onSubAgentData || null, userId: options.userId || null, }; const toolResult = await this.toolRegistry.execute(tc.tool, tc.params, context); toolResults.push({ ...tc, result: toolResult }); toolCallHistory.push({ ...tc, result: toolResult }); this.emit('tool_result', { tool: tc.tool, result: toolResult }); } currentPrompt = this._buildToolResultPrompt(toolResults); } return { type: 'max_rounds', output: allOutput, toolCalls: toolCallHistory, rounds: maxToolRounds, durationApiMs: totalApiMs, }; } finally { this._busy = false; this._processQueue(); } } /** * 处理排队的消息 */ async _processQueue() { if (this._messageQueue.length === 0) return; if (this._busy) return; const next = this._messageQueue.shift(); if (!next) return; try { const prompt = next.isSystem ? `[系统巡检消息]\n\n${next.message}` : next.message; const result = await this._executeMessage(prompt, next.options, next.isSystem); if (next.resolve) next.resolve(result); } catch (err) { console.error('[MasterSession] Queue message error:', err.message); if (next.resolve) next.resolve({ type: 'error', output: err.message }); } } /** * 关闭会话 */ async shutdown() { this._messageQueue = []; if (this.manager) { await this.manager.closeAllSessions(); } } } module.exports = { MasterSession };