/** * Agent Chat Route — 通用 Agent 对话 SSE 端点 * * 存储架构：conversations/agents/{agentId}/{conversationId}/ * ├── meta.json ← 对话元数据 * ├── {messageId}/messages.log ← 每轮对话日志 * └── history.json ← 有序的消息列表（供前端展示 + 上下文恢复） * * 多轮对话： * - ClaudeCliProtocol 持久进程 → 天然多轮 * - 服务器重启后自动恢复：从 history.json 加载历史，注入 system prompt + 历史摘要 * * 使用 @mas/acp-core 的 ClaudeCliProtocol。 */ const express = require('express'); const path = require('path'); const fs = require('fs'); const { randomUUID } = require('crypto'); const multer = require('multer'); const { ClaudeCliProtocol } = require('@mas/acp-core'); const { Parser } = require('../src/core/parser'); const { IdleTimer } = require('../src/engine/idle-timer'); const AgentManager = require('../src/core/agent-manager'); const locator = require('../services/agent-locator'); const { OkrService } = require('../services/okr-service'); const { saveSessionSummary, loadRecentMemory, retrieveRelevantPlans, autoReflect, consolidateMemory, generateDailyReport } = require('../services/memory-service'); // 「这家公司雇了谁」的唯一真相来源（28 §4.2）：公司侧不再有 agent 目录可扫 const { listCompanyAgents, CompanyService, canAccessCompany } = require('../services/company-service'); const companyService = new CompanyService(); // 交付物归宿只有一份推导（28 §4.4），公司环境上下文与执行链路的运行时坐标共用它 const { companyWorkspaceDir } = require('../src/core/prompt-builder'); const systemConfig = require('../config'); const { requireAuthOrInternal, INTERNAL_TOKEN, INTERNAL_TOKEN_HEADER } = require('../middleware/jwt-auth'); const router = express.Router(); // 38 号文档 Phase 1：全站要求登录，但这个文件同时承接三处服务端自环调用 // （company.js:82/506、task-settle.js:452 的 fetch('.../api/agent/...')）—— // 那几处拿不到用户 cookie，用 requireAuthOrInternal 而不是 requireAuth， // 放行带对内部令牌的自环请求，其余情况仍然要求真实登录。 router.use(requireAuthOrInternal); // 38 号文档 Phase 3：agent-chat.js 一直靠 :agentId 里的命名空间前缀（"{companyId}:{role}"， // locator.companyStateDir 解析）区分会话属于哪家公司，但 Phase 1/2 都没顾上这个文件的 // 归属校验——只要求登录，没查公司归不归你。核实过是真的能跨租户读会话（GET .../conversations // 对不是自己公司的 companyId:role 依然 200），这里补上。 // 没有 `:` 前缀的是全局 agent（master/org-innovator 等，落 _global），本来就是全系统共享， // 不属于任何公司，不做归属校验——跟 system/agents/ 本体全局共享是同一件事的两个层面。 // 内部自环调用（招聘/训练自动流水线）持有 req.user.internal，直接放行，不能对它做归属校验 // （自环调用没有"当前用户"这个概念，company.js/task-settle.js 那三处会主动打到任意公司名下）。 router.param('agentId', (req, res, next, agentId) => { if (req.user?.internal) return next(); const colonIdx = agentId.indexOf(':'); if (colonIdx === -1) return next(); // 全局 agent，无公司维度 const companyId = agentId.slice(0, colonIdx); if (!canAccessCompany(companyService.get(companyId), req.user)) { return res.status(404).json({ error: `Agent ${agentId} 不存在` }); } next(); }); // ── 自唤醒回调：端口取实际监听端口，测试环境整体短路 ──────────────────── // // 曾经这六处各自写死 `hostname:'localhost', port: systemConfig.port`（3010）。 // 单测把路由挂到 `app.listen(0)` 的临时端口上，回调却照样打进**线上 3010**： // 每跑一次 task-dispatch-store 的汇报用例，就给真 CEO 发一条 chat、起一个常驻 // claude CLI 子进程、在 system/agents/ceo/_global/ 下留一个 callback-T-* 目录。 // 症状是静默的（测试全绿），只有进程数和内存在涨 —— 同 docs/done/claude-cli-session-leak.md。 // 端口从 req.socket 取（收到这条请求的那个监听端口，天然就是"我自己"）。 const isTestEnv = () => !!process.env.VITEST || process.env.NODE_ENV === 'test'; function selfChat(req, agentId, message, conversationId, label = 'SelfChat') { if (isTestEnv()) { console.log(`[${label}] ⏭️ 测试环境，跳过对 ${agentId} 的自唤醒 (conv: ${conversationId})`); return; } const http = require('http'); const postData = JSON.stringify({ message, conversationId }); const chatReq = http.request({ hostname: req?.socket?.localAddress || '127.0.0.1', port: req?.socket?.localPort || systemConfig.port, path: `/api/agent/${agentId}/chat`, method: 'POST', // 本文件顶部 router.use(requireAuthOrInternal) 之后，这条自环 fetch 也要带内部令牌， // 否则 401——之前漏了这一处，TaskDispatch/TaskCallback/TaskTransition/TaskReject/ // HumanRequest 全部 7 个调用点静默失败（下面 chatRes 分支之前不看 statusCode， // 401 也照样打印"✅ 已唤醒"，症状跟 claude-cli-session-leak.md 那次一样是静默的）。 headers: { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(postData), [INTERNAL_TOKEN_HEADER]: INTERNAL_TOKEN, }, }, (chatRes) => { let body = ''; chatRes.on('data', (c) => { body += c; }); chatRes.on('end', () => { if (chatRes.statusCode >= 400) { console.error(`[${label}] ❌ 唤醒 ${agentId} 失败: HTTP ${chatRes.statusCode} ${body.slice(0, 200)} (conv: ${conversationId})`); } else { console.log(`[${label}] ✅ 已唤醒 ${agentId} (conv: ${conversationId})`); } }); }); chatReq.on('error', (err) => console.error(`[${label}] ❌ 回调失败: ${err.message}`)); chatReq.write(postData); chatReq.end(); } // ── 一个寻址名，三种语义（28 号设计 §4.1）────────────────────────────── // // 改造前这里只有一个 getAgentDir()，返回值同时被当成「CLI 的 cwd」「本体只读根」 // 「私有可写根」三样用 —— 而它算的是 bodyDir，**公司段被丢弃**。同一个函数扛三种 // 语义，正是两个越权 bug 的同一个根： // ① co-alpha 与 co-beta 派给 `ceo` 的任务行追加进同一份 system/agents/ceo/todo.md； // ② PUT/DELETE /:agentId/file/* 按 `co-alpha:ceo` 寻址就能覆写共享本体，co-beta 立刻生效。 // 两个 bug 都写在 gitignore 挡住的路径上，git status 一片干净，所以没人发现。 // // getBodyDir() 本体，只读 system/agents/{aid}/ agent.md / soul.md / config.json / .claude/skills // getStateDir() 公司态，可写 system/agents/{aid}/{cid}/ conversations / memory / todo.md / plan.md / uploads // getTaskDir() 任务态，cwd system/agents/{aid}/{cid}/{convId}/ function getBodyDir(id) { return locator.bodyDir(locator.parseAddress(id).agentId); } /** * Claude CLI 的 cwd —— 深度 3（裁决 A：与 agent-manager / graph 链路统一）。 * * agent-chat 链路没有 taskId，**用 convId 充当**：一次对话就是一个任务，同一轮对话的 * 多次调用共享同一个工作目录，语义与 graph 链路的 taskDir(aid, cid, taskId) 自洽。 * 深度统一的收益不是整齐好看，而是 agent.md 里的相对引用有唯一解： * 本体恒在 `../../../`、仓库根恒在 `../../../../../`；两套深度并存时这些引用必然有一半是错的， * 而读不到 soul.md 是**静默降级**成通用 Base Agent 文本，不报错。 */ function getTaskDir(id, convId) { const { companyId, agentId } = locator.parseAddress(id); return locator.taskDir(agentId, companyId, convId); } /** * 把相对路径钉死在 baseDir 内，越界返回 null。写法同 routes/company.js 的 resolveInside —— * 那里的注释记了为什么不能用 startsWith：`co-a` 是 `co-a-evil` 的字符串前缀， * startsWith 会把跨租户的 `../co-a-evil/x` 判成合法。 */ function resolveInside(baseDir, rel) { const base = path.resolve(baseDir); const abs = path.resolve(base, rel || ''); const r = path.relative(base, abs); if (r === '') return abs; if (r.startsWith('..') || path.isAbsolute(r)) return null; return abs; } const okrService = new OkrService(); // CONV_ROOT is now dynamically evaluated per-agent based on DATA_DIR. function loadSystemConfig() { const configDir = path.resolve(__dirname, '../config'); const backends = JSON.parse(fs.readFileSync(path.join(configDir, 'backends.json'), 'utf-8')); const { loadConfigFromSkills } = require('../src/core/skill-config-loader'); const skillsDir = systemConfig.skillsCodeDir; return loadConfigFromSkills(skillsDir, backends); } const fullSystemConfig = loadSystemConfig(); // 初始化 AgentManager 用于 Sub-Agent 调用 const agentManager = new AgentManager(fullSystemConfig, path.resolve(__dirname, '../../')); // ═══════════════════════════════════════ // meta_creator 硬编码执行器 // 保证 AgentScaffold 父类一定被调用 // ═══════════════════════════════════════ /** * 从 Innovator 的任务文本中解析 Agent 创建参数，直接调用 AgentScaffold。 * 不依赖 Claude 子进程拼 curl 命令，100% 可靠。 * * @param {string} task - Innovator 传来的任务描述 * @param {string} parentAgentId - 父 Agent ID (如 'innovator-001') * @param {string} convId - 当前对话 ID * @param {function} sendSSE - SSE 回调 * @param {string} callId - 调用 ID * @returns {Promise<{agent: string, output: string, status: string}>} */ async function executeMetaCreator(task, parentAgentId, convId, sendSSE, callId) { sendSSE('sub_agent', { event: 'chunk', callId, agent: 'meta_creator', chunk: '🏗️ [meta_creator] 开始通过 AgentScaffold 父类创建 Agent...\n' }); // Step 1: 从任务文本中解析 Agent 名称 // 支持多种格式: "data/agents/poetry_writer", "Agent: poetry_writer", "名称: xxx", "agent_name" 等 let agentName = null; let displayName = null; // 尝试匹配常见模式 const patterns = [ /data\/agents\/(\w+)/, /[Aa]gent[:\s]+(\w+)/, /名称[:\s]*(\S+)/, /创建[^"]*?[「"'"]([^"'」]+)['"」]/, /(?:id|ID)[:\s]*['"]*(\w+)['"]*/, ]; for (const pat of patterns) { const m = task.match(pat); if (m) { agentName = m[1].replace(/[^a-zA-Z0-9_-]/g, '_').toLowerCase(); break; } } // 如果没解析到，从领域关键词生成 if (!agentName) { const domainMatch = task.match(/领域[:\s]*([^\n,，]+)/); if (domainMatch) { agentName = domainMatch[1].trim() .replace(/[\u4e00-\u9fa5]+/g, (m) => { // 简单中文→拼音映射（仅常用词） const map = { '诗词': 'poetry', '小说': 'novel', '古典': 'classical', '创作': 'writer', '写诗': 'poetry', '写作': 'writing' }; return map[m] || m; }) .replace(/[^a-zA-Z0-9]/g, '_') .replace(/_+/g, '_') .replace(/^_|_$/g, '') .toLowerCase(); } } if (!agentName) agentName = `agent_${Date.now().toString(36)}`; // 解析显示名称 const displayMatch = task.match(/显示名[称:\s]*([^\n,，]+)/) || task.match(/名称[:\s]*([^\n,，]+)/); displayName = displayMatch ? displayMatch[1].trim() : agentName; sendSSE('sub_agent', { event: 'chunk', callId, agent: 'meta_creator', chunk: `📝 Agent ID: ${agentName}\n📝 显示名称: ${displayName}\n` }); // Step 2: 读取 knowhow 和 benchmark（如果有） const parentDir = getBodyDir(parentAgentId); const workspaceDir = path.join(parentDir, 'workspace', agentName); let knowhow = ''; let benchmark = ''; // 搜索可能的 knowhow/benchmark 位置 const searchDirs = [workspaceDir, path.join(parentDir, 'workspace'), path.join(parentDir, 'knowledge')]; for (const dir of searchDirs) { if (!knowhow) { const kp = path.join(dir, 'knowhow.md'); if (fs.existsSync(kp)) knowhow = fs.readFileSync(kp, 'utf-8'); } if (!benchmark) { const bp = path.join(dir, 'benchmark.md'); if (fs.existsSync(bp)) benchmark = fs.readFileSync(bp, 'utf-8'); } } sendSSE('sub_agent', { event: 'chunk', callId, agent: 'meta_creator', chunk: `📚 Knowhow: ${knowhow ? `${knowhow.length}字` : '未找到'}\n📊 Benchmark: ${benchmark ? `${benchmark.length}字` : '未找到'}\n` }); // Step 3: 生成基础 SKILL.md (用 knowhow + benchmark 合成) const skillMd = generateSkillMd(agentName, displayName, knowhow, benchmark, task); // Step 4: 调用 AgentScaffold 父类创建完整目录 // baseDir 走 agent-locator（28 批次 5-21）：全仓四个 AgentScaffold 实例同一个本体根 // // 已存在就**不许再 create**（同 company.js 的 409 前置判重）：`scaffold.create()` 是 // 无条件覆写 —— agent.md / soul.md / identity.md / tools.md / user.md / heartbeat.md / // config.json / todo.md / plan.md 一律换成通用默认模板。baseDir 统一到 system/agents // 之前，被覆写的是一家公司的私有副本；统一之后覆写的是**全系统唯一那一份本体**。 // agentName 是从 innovator 的自然语言任务里正则出来的，撞上 org-coder / ceo 这类现役 // id 完全可能，而覆写全程 200、日志还写「✅ 已创建」—— 盘上定义被抹平且没有任何一处报错。 if (fs.existsSync(locator.bodyDir(agentName))) { const msg = `本体 "${agentName}" 已存在，meta_creator 不覆写现役本体（要改定义请走 agent-creator / evolver）`; sendSSE('sub_agent', { event: 'chunk', callId, agent: 'meta_creator', chunk: `⛔ ${msg}\n` }); return { agent: 'meta_creator', output: msg, status: 'failed' }; } const { AgentScaffold } = await import('../../packages/agent-core/AgentScaffold.js'); const scaffold = new AgentScaffold(locator.AGENTS_ROOT); const agentMd = `# ${displayName}\n\n你是 **${displayName}**，专业的 AI Agent。\n\n## 执行准则\n你的完整操作手册硬编码在 main_skill/SKILL.md 中，请严格遵照执行。\n`; const agentDir = await scaffold.create({ id: agentName, type: 'skill', // 执行型 Agent name: displayName, parentId: parentAgentId, bootstrap: { agentMd } }); // 写入生成的 SKILL.md const skillPath = path.join(agentDir, '.claude', 'skills', 'main_skill', 'SKILL.md'); fs.writeFileSync(skillPath, skillMd, 'utf-8'); // agent.md + soul.md（脚手架写的）+ SKILL.md（上一行）齐了 —— 过一次转正闸门（28 批次 5-23） require('../services/graph-service').promoteDraftIfComplete(agentName); // 如果有 knowhow/benchmark，复制到新 Agent 的 knowledge/ 下 const knowledgeDir = path.join(agentDir, 'knowledge'); if (knowhow) fs.writeFileSync(path.join(knowledgeDir, 'knowhow.md'), knowhow, 'utf-8'); if (benchmark) fs.writeFileSync(path.join(knowledgeDir, 'benchmark.md'), benchmark, 'utf-8'); const output = `✅ Agent "${displayName}" (${agentName}) 已通过 AgentScaffold 父类创建完毕。 目录: data/agents/${agentName}/ 标准结构: ├── agent.md (身份定义) ├── soul.md (灵魂) ├── identity.md (标识) ├── config.json (配置,含 cli.toolPolicy) ├── .claude/skills/main_skill/SKILL.md (核心技能, ${skillMd.length}字) ├── memory/ (记忆) ├── knowledge/ (知识${knowhow ? ', 含 knowhow.md' : ''}${benchmark ? ', 含 benchmark.md' : ''}) ├── workspace/ (工作区) └── logs/ (日志) [RESULT: DONE]`; sendSSE('sub_agent', { event: 'chunk', callId, agent: 'meta_creator', chunk: output }); console.log(`[meta_creator] ✅ Agent "${displayName}" (${agentName}) 创建完毕: ${agentDir}`); return { agent: 'meta_creator', output, status: 'success' }; } /** * 根据 knowhow/benchmark/task 生成 SKILL.md 内容 */ function generateSkillMd(agentName, displayName, knowhow, benchmark, task) { // 从 task 中提取领域信息 const domainMatch = task.match(/领域[:\s]*([^\n,，]+)/); const domain = domainMatch ? domainMatch[1].trim() : displayName; let content = `--- name: ${agentName} description: > ${displayName} — 专业的 ${domain} Agent。 基于深度调研的方法论和标杆案例创建。 version: 1.0 domain: ${domain} --- # ${displayName} 你是 **${displayName}**，专业的 ${domain} 执行 Agent。 ## 核心方法论 `; if (knowhow) { // 提取 knowhow 中的关键部分（不超过 3000 字） content += knowhow.slice(0, 3000); } else { content += `（待补充 — 请通过 meta_evolving 迭代训练补充方法论）\n`; } content += `\n\n## 质量标准\n\n`; if (benchmark) { content += benchmark.slice(0, 1500); } else { content += `（待补充 — 请通过 meta_evolving 迭代训练补充评分标准）\n`; } content += `\n\n## 输出协议\n\n完成后**必须**以 \`[RESULT: DONE]\` 标记收尾。\n`; return content; } // ═══════════════════════════════════════ // Bootstrap 工具 // ═══════════════════════════════════════ // 💰 Token 防护：单文件注入上限 const MAX_PROMPT_FILE_SIZE = 15 * 1024; // 15KB per file const MAX_TOTAL_PROMPT_SIZE = 100 * 1024; // 100KB total system prompt function loadFile(agentDir, filename) { try { const content = fs.readFileSync(path.join(agentDir, filename), 'utf-8'); if (content.length > MAX_PROMPT_FILE_SIZE) { console.warn(`[TokenGuard] ⚠️ ${filename} 超过 ${MAX_PROMPT_FILE_SIZE/1024}KB (${(content.length/1024).toFixed(1)}KB)，已截断`); return content.substring(0, MAX_PROMPT_FILE_SIZE) + `\n\n... [⚠️ 文件过大已截断，原始 ${(content.length/1024).toFixed(1)}KB，请用 Read 工具按需读取完整内容]`; } return content; } catch { return ''; } } // ═══════════════════════════════════════ // 基类模板目录 — 运行时继承 // ═══════════════════════════════════════ const BASE_DIR = path.join(__dirname, '../../packages/agent-core/base'); // ── 启动时预加载 common + type 基类文件到内存（几乎不变的文件，避免每次请求读盘） ── const _baseFileCache = new Map(); function _loadBaseFile(filepath) { if (_baseFileCache.has(filepath)) return _baseFileCache.get(filepath); try { const content = fs.readFileSync(filepath, 'utf-8').trim(); _baseFileCache.set(filepath, content); return content; } catch { return ''; } } // 预热 common 文件 ['methodology.md', 'tools-base.md', 'capabilities.md'].forEach(f => { _loadBaseFile(path.join(BASE_DIR, 'common', f)); }); // 预热 type 基类 ['master', 'cxo', 'skill', 'orchestrator', 'worker'].forEach(t => { _loadBaseFile(path.join(BASE_DIR, t, 'agent-base.md')); _loadBaseFile(path.join(BASE_DIR, t, 'cron-base.json')); }); /** * 构建 system prompt — 三层继承 * * Layer 1: base/common/* (所有 Agent 通用的工具规范、能力说明) * Layer 2: base/{type}/*-base.md (类型专属的角色定义) * Layer 3: data/agents/{id}/* (个体专属的覆盖/扩展) * * 继承规则: * - agent.md: base/type/agent-base.md + agent/agent.md (追加) * - tools.md: base/common/tools-base.md + agent/tools.md (追加) * - 其他文件 (soul.md, identity.md 等): 只从 agent/ 读取（纯个性化） */ function buildSystemPrompt(agentDir, userMessage, addr, convId) { const config = loadAgentConfig(agentDir); const agentType = config?.type || 'skill'; const parts = []; // ── Layer 0: 公司环境上下文（交付物归宿，28 §4.4）── // 公司 id 取自**寻址名**，不再拿 agentDir 做路径正则猜：运行态下沉后 agent 的目录是 // system/agents/{aid}/{cid}/...，老正则（data/companies/{cid}/agents/{role}）匹配不上， // 于是公司 agent 静默拿不到这段声明、把交付物写进自己的私有目录，同事读不到也不报错。 const { companyId, agentId: roleName } = locator.parseAddress(addr || path.basename(agentDir)); // ── Layer 0': 运行时坐标（裁决 A / 28 §4.4）── // cwd 下沉到深度 3 之后，agent.md 里那些 `soul.md` / `../../workspace/x.md` 的相对引用 // 全部要重算。改漏一处不会报错 —— 读不到就是静默降级成通用文本。 // 所以这里把绝对路径直接摆出来兜底：改写漏了，agent 也能自己找到本体。 // // 相对量一律 path.relative 实算，**不写字面量**：cwd 是 `{aid}/{cid}/{taskId}`、本体是 `{aid}`， // 中间只隔了 `{cid}` / `{taskId}` 两层，所以真值是 `../..`。手写成 `../../../` 会指到 // `system/agents/` 本身 —— agent 照着去读 soul.md 只会读到「不存在」，而这条链路上 // 「读不到本体」不抛错，只是静默退化成通用 Base Agent 文本。 if (addr && convId) { const taskDir = getTaskDir(addr, convId); const relBody = path.relative(taskDir, getBodyDir(addr)) || '.'; const relState = path.relative(taskDir, getStateDir(addr)) || '.'; parts.push(`# 📍 运行时坐标（绝对路径，以此为准） - 你的工作目录（cwd）：\`${taskDir}\` —— 只放草稿和日志，会被清理 - 你的本体文件（agent.md / soul.md / .claude/）：\`${getBodyDir(addr)}\`（相对 cwd 是 \`${relBody}\`）—— 只读，不要改 - 你的公司态目录（记忆 / 会话 / todo.md / plan.md）：\`${getStateDir(addr)}\`（相对 cwd 是 \`${relState}\`）`); } if (companyId) { // 路径推导与 prompt-builder 的运行时坐标声明共用同一份实现（company-paths → companyWorkspaceDir）： // goal-contract.js 的注释已经记过一次「两份逐字同构的约定」失效的账，这里不再手拼第三份。 const companyWorkspace = companyWorkspaceDir(companyId); parts.push(`# 🏢 公司环境上下文 你隶属于公司 **${companyId}**，你的角色是 **${roleName}**。 ## ⚠️ 关键路径规范（必须遵守！） - **公司共享工作区**: \`${companyWorkspace}/\` - **你写入的所有文件（战略蓝图、Pipeline JSON、分析报告等）都必须放在 \`${companyWorkspace}/\` 目录下** - 🚫 **绝对禁止**写入全局的 \`workspace/\` 目录！那是旧版遗留路径！ - 🚫 **绝对禁止**写入 \`data/workspace/\` 目录！ - ✅ 正确路径示例: \`${companyWorkspace}/strategy.md\`、\`${companyWorkspace}/pipeline_design.json\` - 你的同事（CEO/CTO/COO/HR 等）共享同一个 \`${companyWorkspace}/\` 目录，用它来交换文件和数据。 - 你自己的工作目录只放草稿和日志，会被清理 —— 交付物写在那里等于没交付。`); } // ── Layer 1: 通用基类 ── const commonFiles = ['methodology.md', 'tools-base.md', 'capabilities.md']; for (const f of commonFiles) { const fp = path.join(BASE_DIR, 'common', f); const content = _loadBaseFile(fp); if (content) parts.push(content); } // ── Layer 2: 类型基类 ── const typeBaseAgent = path.join(BASE_DIR, agentType, 'agent-base.md'); const typeContent = _loadBaseFile(typeBaseAgent); if (typeContent) parts.push(typeContent); // ── Layer 3: Agent 专属文件 ── const agentListText = Object.entries(fullSystemConfig.agents || {}) .filter(([name]) => name !== 'master' && name !== 'org-innovator') .map(([name, cfg]) => { const typeLabel = cfg.type === 'orchestrator' ? '子编排' : cfg.type; return `- ${name} (${typeLabel}): ${cfg.description || ''}`; }) .join('\n'); // `user.md`（汇报关系）是**公司态**文件，不是本体（28 §4.1 的四层归属表）： // 写侧是 company-service._initCompanyState → system/agents/{aid}/{cid}/user.md， // 每雇一次按那家公司的 ceo / type 重写一份。这里若照本体目录读，写侧和读侧就不同源 —— // 而两边都不报错：本体根上确实躺着 28 份历史 user.md（`git ls-files` 可查），内容是 // 当年某一家公司的汇报关系，例如 system/agents/claude-youtube-skills/user.md 写着 // 「所属公司: test」。idearun 的同一个 agent 在 chat 里就会同时收到「你隶属于 idearun」 // （上面的公司环境上下文段）和「你的公司是 test」（这一段），而正确的那份 // system/agents/claude-youtube-skills/idearun/user.md 一个字都没进提示词。 const companyStateFiles = new Set(['user.md']); const agentFiles = ['agent.md', 'soul.md', 'identity.md', 'user.md', 'tools.md']; for (const f of agentFiles) { // 公司态文件只在有公司段时改从公司态目录读；无公司的全局 agent（master 等） // 仍回落本体那份 —— 它们的 `_global/` 下没有人会去写 user.md。 const srcDir = (companyStateFiles.has(f) && companyId) ? getStateDir(addr) : agentDir; let c = loadFile(srcDir, f); if (c.trim()) { if (c.includes('{{AVAILABLE_AGENTS}}')) { c = c.replace('{{AVAILABLE_AGENTS}}', agentListText); } parts.push(c.trim()); } } // heartbeat.md const heartbeat = loadFile(agentDir, 'heartbeat.md'); if (heartbeat.trim()) parts.push(heartbeat.trim()); // ── Phase 3: 注入 OKR 目标上下文 ── // 记忆一律按**寻址名**取（`cid:aid`），与写侧 saveSessionSummary(agentId, …) 对齐 —— // 拿 config.id 去取的话，公司 agent 会读到裸 agentId 那份（= 别家公司的记忆），且不报错。 const memoryAddr = addr || config?.id || path.basename(agentDir); const agentIdForMemory = config?.id || path.basename(agentDir); const stateDir = getStateDir(memoryAddr); try { const goalContext = okrService.getActiveContextForAgent(agentIdForMemory); if (goalContext) { parts.push(goalContext.trim()); } } catch (err) { console.error(`[GoalContext] ⚠️ 读取目标上下文失败: ${err.message}`); } // memory — 核心记忆索引 const memory = loadFile(stateDir, 'memory/memory.md'); if (memory.trim()) parts.push('---\n\n# 我的记忆\n\n' + memory.trim()); // short-term — 近期对话摘要 (Phase 1 MVP) const recentMemory = loadRecentMemory(memoryAddr); if (recentMemory) parts.push('---\n\n' + recentMemory); // hippocampus — 海马体检索相关经验 (Phase 2) // 将 userMessage 放在前，避免 todo.md 历史堆积导致当前任务的关键词被截断 // todo.md 读**公司态**目录（与 /task/dispatch 的写侧同源）：读本体那份的话， // 检索关键词来自别家公司的任务，检索结果自然也是别家的经验。 const todoContent = loadFile(stateDir, 'todo.md'); const searchText = [userMessage || '', todoContent].join('\n\n'); const hippocampusPlans = retrieveRelevantPlans(memoryAddr, searchText); if (hippocampusPlans) parts.push('---\n\n' + hippocampusPlans); // skills let mainSkill = loadFile(agentDir, '.claude/skills/main_skill/SKILL.md'); if (mainSkill.trim()) { if (mainSkill.includes('{{AVAILABLE_AGENTS}}')) { mainSkill = mainSkill.replace('{{AVAILABLE_AGENTS}}', agentListText); } parts.push('---\n\n# 我的核心主技能 (执行准则)\n\n' + mainSkill.trim()); } // ── Phase 1 / Micro-Verify: 当前局部任务的试错经验池 `_lessons.md` ── // 当 Agent 遇到棘手的坑或执行报错时，要求它总结并写入 _lessons.md。 // 通过在此处注入，它下次 retry 时便能带着前车之鉴，避免死循环。 const localLessons = loadFile(agentDir, 'workspace/_lessons.md'); if (localLessons.trim()) { parts.push('---\n\n## ⚠️ 本地试错防坑经验 (_lessons.md)\n不要重复踩下面的坑，仔细阅读并绕开：\n\n' + localLessons.trim()); } const skills = loadFile(agentDir, '.claude/skills/skills.md'); if (skills.trim()) parts.push('---\n\n' + skills.trim()); // ── 扫描所有 skill 子目录的 SKILL.md ── const skillsDir = path.join(agentDir, '.claude', 'skills'); if (fs.existsSync(skillsDir)) { const entries = fs.readdirSync(skillsDir, { withFileTypes: true }); for (const entry of entries) { if (!entry.isDirectory() || entry.name === 'main_skill') continue; const skillMdPath = path.join(skillsDir, entry.name, 'SKILL.md'); if (fs.existsSync(skillMdPath)) { const skillContent = fs.readFileSync(skillMdPath, 'utf-8').trim(); if (skillContent) { parts.push(`---\n\n## 依赖技能: ${entry.name}\n\n${skillContent}`); } } } } let result = parts.join('\n\n---\n\n'); // 💰 Token 防护：总 system prompt 上限 if (result.length > MAX_TOTAL_PROMPT_SIZE) { console.warn(`[TokenGuard] ⚠️ System prompt 超过 ${MAX_TOTAL_PROMPT_SIZE/1024}KB (${(result.length/1024).toFixed(1)}KB)，已截断`); result = result.substring(0, MAX_TOTAL_PROMPT_SIZE) + `\n\n... [⚠️ System prompt 过大已截断]`; } return result; } /** * 构建 cron 恢复指令 — 合并基类 + Agent 专属 cron * * Layer 1: base/{type}/cron-base.json (类型通用的定时任务) * Layer 2: data/agents/{id}/cron.json (Agent 专属追加) * * 合并策略: 合并两个数组，去重 (相同 schedule+prompt 只保留一个) */ function buildCronInstructions(agentDir) { const config = loadAgentConfig(agentDir); const agentType = config?.type || 'skill'; let allCrons = []; // Layer 1: 基类 cron const baseCronPath = path.join(BASE_DIR, agentType, 'cron-base.json'); const baseCronContent = _loadBaseFile(baseCronPath); if (baseCronContent) { try { const base = JSON.parse(baseCronContent); allCrons = allCrons.concat(base); } catch (err) { console.error(`[Cron] ⚠️ 解析 ${baseCronPath} 失败:`, err.message); } } // Layer 2: Agent 专属 cron (追加) const agentCronPath = path.join(agentDir, 'cron.json'); if (fs.existsSync(agentCronPath)) { try { const agent = JSON.parse(fs.readFileSync(agentCronPath, 'utf-8')); allCrons = allCrons.concat(agent); } catch (err) { console.error(`[Cron] ⚠️ 读取 ${agentCronPath} 失败:`, err.message); } } // 过滤已启用 + 按 schedule 去重（后加入的 Agent 专属 cron 覆盖基类） const enabled = allCrons.filter(c => c.enabled !== false); const scheduleMap = new Map(); for (const c of enabled) { // 同一 schedule 下，后者(Agent专属)覆盖前者(基类) scheduleMap.set(c.schedule, c); } const unique = Array.from(scheduleMap.values()); if (unique.length === 0) return ''; const lines = unique.map((c, i) => { const recur = c.recurring !== false ? '循环' : '单次'; return `${i + 1}. 调度: \`${c.schedule}\` (${recur})\n Prompt: "${c.prompt}"`; }).join('\n'); return `\n\n---\n\n## ⏰ 定时任务恢复\n\n**重要**: 请在处理用户消息之前，先用 CronCreate 工具创建以下定时任务（这些是你的持久化调度配置，每次新会话需重建）：\n\n${lines}\n\n请依次调用 CronCreate 创建上述任务，然后再回答用户。`; } function loadAgentConfig(agentDir) { try { return JSON.parse(fs.readFileSync(path.join(agentDir, 'config.json'), 'utf-8')); } catch { return null; } } function loadIdentity(agentDir) { const content = loadFile(agentDir, 'identity.md'); const r = { name: 'Agent', emoji: '🤖', greeting: '' }; const nm = content.match(/名称\**[：:]\s*(.+)/); if (nm) r.name = nm[1].trim(); // fallback: 如果 identity.md 没有名称，从 config.json 取 name if (r.name === 'Agent') { const cfg = loadAgentConfig(agentDir); if (cfg && cfg.name && cfg.name !== 'Agent') r.name = cfg.name; } const em = content.match(/Emoji\**[：:]\s*(.+)/); if (em) r.emoji = em[1].trim(); const gm = content.match(/## 问候语\n"?([^"]*)"?/s); if (gm) r.greeting = gm[1].trim().replace(/^"|"$/g, ''); return r; } function setupSSE(res) { res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', 'X-Accel-Buffering': 'no', }); return (event, data) => res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); } // ═══════════════════════════════════════ // 对话存储 — conversationId/messageId 两层 // ═══════════════════════════════════════ /** * 会话 / 记忆的落点 —— 公司态目录 `system/agents/{aid}/{cid}/`（28 §4.1，批次 3 第 13 条）。 * 与本体目录（agent.md / soul.md / todo.md / .claude/）分开：本体只读且全系统一份， * 而会话是 (agentId, companyId) 私有的。全局 agent（master 等）落 `_global`。 */ function getStateDir(id) { return locator.companyStateDir(id); } function getConvDir(agentId, convId) { const dir = path.join(getStateDir(agentId), 'conversations', convId); if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true }); return dir; } /** 读写 meta.json */ function loadMeta(agentId, convId) { try { return JSON.parse(fs.readFileSync(path.join(getStateDir(agentId), 'conversations', convId, 'meta.json'), 'utf-8')); } catch { return null; } } function saveMeta(agentId, convId, meta) { const dir = getConvDir(agentId, convId); meta.updatedAt = new Date().toISOString(); fs.writeFileSync(path.join(dir, 'meta.json'), JSON.stringify(meta, null, 2)); } /** 读写 history.json — 有序消息列表 */ function loadHistory(agentId, convId) { try { const historyPath = path.join(getStateDir(agentId), 'conversations', convId, 'history.json'); if (fs.existsSync(historyPath)) { return JSON.parse(fs.readFileSync(historyPath, 'utf-8')); } return []; } catch { return []; } } /** * 解析 messages.log 为消息数组 * 格式: [ISO_TIMESTAMP] [USER|AGENT]\n content...\n\n */ function parseMessagesLog(logContent) { const messages = []; // 按消息头分割: [timestamp] [USER] 或 [timestamp] [AGENT] const pattern = /\[(\d{4}-\d{2}-\d{2}T[\d:.]+Z)\]\s+\[(USER|AGENT)\]\n/g; const matches = []; let match; while ((match = pattern.exec(logContent)) !== null) { matches.push({ index: match.index, timestamp: match[1], role: match[2], headerLen: match[0].length }); } for (let i = 0; i < matches.length; i++) { const start = matches[i].index + matches[i].headerLen; const end = i + 1 < matches.length ? matches[i + 1].index : logContent.length; const content = logContent.substring(start, end).trim(); messages.push({ role: matches[i].role === 'USER' ? 'user' : 'assistant', content, timestamp: matches[i].timestamp, }); } return messages; } function appendHistory(agentId, convId, msg) { const history = loadHistory(agentId, convId); history.push(msg); const dir = getConvDir(agentId, convId); fs.writeFileSync(path.join(dir, 'history.json'), JSON.stringify(history, null, 2)); } /** 写日志（流式，带缓冲） */ const _logBuffers = new Map(); const LOG_BUFFER_SIZE = 4096; // 4KB const LOG_FLUSH_INTERVAL = 1000; // 1s function logMessageChunk(agentId, convId, content) { const key = `${agentId}::${convId}`; if (!_logBuffers.has(key)) { _logBuffers.set(key, { buf: '', timer: null, agentId, convId }); } const entry = _logBuffers.get(key); entry.buf += content; if (entry.buf.length >= LOG_BUFFER_SIZE) { _flushLogBuffer(key); } else if (!entry.timer) { entry.timer = setTimeout(() => _flushLogBuffer(key), LOG_FLUSH_INTERVAL); } } function _flushLogBuffer(key) { const entry = _logBuffers.get(key); if (!entry || !entry.buf) return; try { const dir = getConvDir(entry.agentId, entry.convId); fs.appendFileSync(path.join(dir, 'messages.log'), entry.buf, 'utf-8'); } catch (err) { console.error(`[LogBuffer] ⚠️ flush 失败: ${key}`, err.message); } entry.buf = ''; if (entry.timer) { clearTimeout(entry.timer); entry.timer = null; } } /** 强制刷出指定对话的日志缓冲 */ function flushLogBufferForConv(agentId, convId) { const key = `${agentId}::${convId}`; _flushLogBuffer(key); _logBuffers.delete(key); } /** 传统单次写日志 (为了兼容) */ function logMessage(agentId, convId, role, content) { const ts = new Date().toISOString(); const tag = role === 'user' ? 'USER' : 'AGENT'; const line = `[${ts}] [${tag}]\n${content}\n\n`; logMessageChunk(agentId, convId, line); } function listConversations(agentId) { const agentConvDir = path.join(getStateDir(agentId), 'conversations'); if (!fs.existsSync(agentConvDir)) return []; try { return fs.readdirSync(agentConvDir, { withFileTypes: true }) .filter(d => d.isDirectory()) .map(d => { const m = loadMeta(agentId, d.name); if (m) m.id = m.id || m.conversationId || d.name; return m; }) .filter(Boolean) .sort((a, b) => new Date(b.updatedAt).getTime() - new Date(a.updatedAt).getTime()); } catch { return []; } } // ═══════════════════════════════════════ // ACP Session 池 — 支持多轮 + 上下文恢复 // ═══════════════════════════════════════ // key: `${agentId}:${conversationId}` → { protocol, messageCount, idleTimer } const sessions = new Map(); // ── 会话回收 ── // sessions 里存的是**常驻**的 claude CLI 子进程（stream-json 模式，不是一次性的 `claude -p`）。 // 在此之前只有四条回收路径：协议 error / 子进程 disconnected / 手动删对话 / 手动 stop —— // 也就是说「正常跑完的对话」永远不回收。心跳每 5 分钟用一个**全新** convId 打进来 // （heartbeat-service.js:206），于是每次心跳都新起一个进程且永不释放，实测 2h40m 攒了 82 个 // 进程 / 17.5 GB。所以：心跳跑完立刻回收，普通对话按空闲 TTL 回收。 const SESSION_IDLE_MS = Number(process.env.AGENT_SESSION_IDLE_MS) || 30 * 60 * 1000; /** 只从池子里摘掉（子进程已经自己死了的场景），顺手停表 */ function dropSession(key) { const entry = sessions.get(key); entry?.idleTimer?.cancel(); sessions.delete(key); return entry; } /** 摘掉 + 杀子进程 */ function closeSession(key, reason) { const entry = dropSession(key); if (!entry) return false; try { entry.protocol.disconnect(); } catch { } console.log(`[AgentChat] 🧹 会话已回收 ${key} (${reason})`); return true; } /** * 后台调用的种类 → [对话标题, meta.type, 会话是否一次性] * * 在此之前后台五条调用（心跳巡检 / Pipeline 步骤 / 日程 / 告警 / 日报）无差别打 `isHeartbeat:true`， * 于是对话列表里全都显示成「⏱️ 自主心跳巡检」—— 四类被错标成第五类。 * * 「一次性」那一列不是标签，是**回收策略**，判据只有一条：convId 会不会被下一轮复用。 * - 心跳 / 日程 / 告警 / 日报：convId 带时间戳，用完即弃 → 跑完立刻回收 * - Pipeline 步骤：convId 是 `pipeline-{runId}-step-{i}`，**被驳回重做时会复用同一个**， * 留着会话能让重试直接接上下文（否则要走 buildPrompt 场景 C 冷启动重注入历史）。 * 它不会变成泄漏 —— 有空闲 TTL 兜底。 */ const CALL_KINDS = { heartbeat: ['⏱️ 自主心跳巡检', 'heartbeat', true], pipeline: ['🔧 流水线步骤', 'pipeline', false], schedule: ['📅 日程任务', 'schedule', true], alert: ['🚨 告警', 'alert', true], report: ['📊 日报', 'report', true], }; /** 没带 kind 的老调用方（heartbeat-service 只发 isHeartbeat）按心跳算，行为不变 */ function resolveKind(body) { return CALL_KINDS[body.kind] || (body.isHeartbeat ? CALL_KINDS.heartbeat : null); } /** * 请求结束后给会话上弦；请求开始时由 getOrCreateProtocol 停表。 * 计时器只在「没有在途请求」的间隙走，所以再长的单次生成也不会被 TTL 打断 * —— 这比在流式回调里 touch() 少一处需要维护的调用点。 */ function armSessionIdle(key) { const entry = sessions.get(key); if (!entry) return; entry.idleTimer?.cancel(); entry.idleTimer = new IdleTimer( SESSION_IDLE_MS, () => closeSession(key, `空闲超过 ${Math.round(SESSION_IDLE_MS / 60000)} 分钟`), { label: `agent-session:${key}` } ).start(); } async function getOrCreateProtocol(agentId, convId) { const key = `${agentId}:${convId}`; let entry = sessions.get(key); if (entry && entry.protocol.isConnected) { entry.idleTimer?.cancel(); // 请求在途期间不计空闲 return entry; } // 关闭旧的 if (entry) { try { entry.protocol.disconnect(); } catch {} } // 身份 / 工具策略读**本体**（只读，全系统一份）；跑起来的 cwd 是**任务态**目录，两者不是一回事。 const bodyDir = getBodyDir(agentId); const identity = loadIdentity(bodyDir); const agentConfig = loadAgentConfig(bodyDir); // 工具隔离策略：从 config.json 的 cli.toolPolicy 读取（由 AgentScaffold 父类生成） // 编排型 Agent 禁用所有原生工具，强制纯文本 [CALL:] 输出 const cliOptions = {}; const toolPolicy = agentConfig?.cli?.toolPolicy; if (toolPolicy?.mode === 'full') { // 执行型 Agent：完整工具权限（不设 allowedTools 限制） console.log(`[AgentChat] 🔓 ${identity.name} 工具策略: full (所有工具可用)`); } else if (toolPolicy && toolPolicy.allowedTools !== undefined) { // 使用 config.json 中声明的策略 (新架构) cliOptions.allowedTools = toolPolicy.allowedTools; if (toolPolicy.disallowedTools) cliOptions.disallowedTools = toolPolicy.disallowedTools; console.log(`[AgentChat] 🔒 ${identity.name} 工具策略 (from config): allowedTools=${JSON.stringify(cliOptions.allowedTools)}`); } else if (agentConfig && (agentConfig.type === 'orchestrator')) { // 兼容旧 config.json (没有 cli.toolPolicy 字段) — 仅 orchestrator 默认禁用 cliOptions.allowedTools = ''; console.log(`[AgentChat] 🔒 ${identity.name} 启用编排模式 (legacy fallback): 原生工具已全部禁用`); } // effort 级别控制 (low=禁用thinking | medium | high | max) // 优先从 config.json 的 cli.effort 读取，其次环境变量 CLAUDE_EFFORT if (agentConfig?.cli?.effort) { cliOptions.effort = agentConfig.cli.effort; console.log(`[AgentChat] 💡 ${identity.name} effort=${cliOptions.effort}`); } const protocol = new ClaudeCliProtocol({ name: identity.name }, cliOptions); const releaseLock = () => { const lock = agentChatLocks.get(agentId); if (lock && lock.convId === convId) { agentChatLocks.delete(agentId); console.log(`[AgentChat] 🔓 ${identity.name} 锁已释放 (conv: ${convId})`); } }; // 绑定错误监听器，防止 EventEmitter 抛出未捕获异常导致整个 Node 进程奔溃 (例如 spawn claude ENOENT) protocol.on('error', (err) => { console.error(`[AgentChat] 🚨 ${identity.name} 协议级异常:`, err.message); dropSession(key); // 下次提问将重建进程 releaseLock(); }); protocol.on('disconnected', ({ code, signal }) => { dropSession(key); releaseLock(); console.log(`[AgentChat] 💤 ${identity.name} 会话断开 (code=${code}, signal=${signal})`); }); // cwd = 任务态目录（裁决 A）。**不能**是本体目录：本体是只读且全公司共享的， // 让 CLI 落在那里等于两家公司在同一个目录里互相踩草稿，且 agent 顺手写下的文件 // 会变成第三家公司下次调用时读到的"历史遗留"。 const cwd = getTaskDir(agentId, convId); fs.mkdirSync(cwd, { recursive: true }); await protocol.connect(null, cwd); await protocol.initialize(); await protocol.newSession({ cwd }); const newEntry = { protocol, messageCount: 0, agentId, convId, idleTimer: null }; sessions.set(key, newEntry); console.log(`[AgentChat] 🧬 ${identity.name} ACP 会话已创建 (conv: ${convId})`); return newEntry; } /** * 构建 prompt — 根据是否首次消息 & 是否有历史上下文 * * 场景 A: 全新对话的第一条消息 → system prompt + 用户消息 * 场景 B: 同一进程中的后续消息 → 直接发消息（ACP 自带上下文） * 场景 C: 服务重启后恢复对话 → system prompt + 历史摘要 + 用户消息 */ // ── 文件上传相关工具 ── const TEXT_EXTENSIONS = new Set([ '.md', '.txt', '.log', '.js', '.ts', '.jsx', '.tsx', '.py', '.java', '.go', '.rs', '.sql', '.sh', '.css', '.html', '.vue', '.svelte', '.json', '.yaml', '.yml', '.toml', '.xml', '.csv', '.conf', '.ini', '.env', '.dockerfile', '.makefile', '.rb', '.php', '.c', '.cpp', '.h', '.hpp', '.swift', '.kt', '.scala', '.r', ]); const IMAGE_EXTENSIONS = new Set(['.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg', '.bmp', '.ico']); const MAX_INJECT_SIZE = 50 * 1024; // 50KB per file const MAX_TOTAL_INJECT = 500 * 1024; // 500KB total function isTextFile(ext) { return TEXT_EXTENSIONS.has(ext.toLowerCase()); } function isImageFile(ext) { return IMAGE_EXTENSIONS.has(ext.toLowerCase()); } function formatFileSize(bytes) { if (bytes < 1024) return bytes + 'B'; if (bytes < 1024 * 1024) return (bytes / 1024).toFixed(1) + 'KB'; return (bytes / (1024 * 1024)).toFixed(1) + 'MB'; } function getLanguageTag(ext) { const map = { '.js': 'javascript', '.ts': 'typescript', '.py': 'python', '.java': 'java', '.go': 'go', '.rs': 'rust', '.sql': 'sql', '.sh': 'bash', '.css': 'css', '.html': 'html', '.vue': 'vue', '.json': 'json', '.yaml': 'yaml', '.yml': 'yaml', '.xml': 'xml', '.csv': 'csv', '.md': 'markdown', '.rb': 'ruby', '.php': 'php', '.c': 'c', '.cpp': 'cpp', }; return map[ext.toLowerCase()] || ''; } function sanitizeFilename(name) { return name.replace(/[\/\\:*?"<>|]/g, '_').replace(/\.\./g, '_').slice(0, 200); } /** * 上传件落**公司态**目录（参数是寻址名，不是目录）——它跟 conversations 一样属于 * `(agentId, companyId)`：两家公司往同一个本体传同名文件，落在本体里就是互相覆盖。 */ function getUploadDir(id, convId) { return path.join(getStateDir(id), 'uploads', convId); } /** * 扫描 uploads/{convId}/ 目录，构建附件上下文块注入到用户消息前。 */ function buildUploadContext(id, convId) { const uploadDir = getUploadDir(id, convId); if (!fs.existsSync(uploadDir)) return ''; const files = fs.readdirSync(uploadDir).filter(f => !f.startsWith('.')); if (files.length === 0) return ''; let contextBlock = '\n\n---\n📎 **附件上下文** (以下文件已上传到当前会话):\n\n'; let totalInjected = 0; for (const file of files) { const ext = path.extname(file).toLowerCase(); const filePath = path.join(uploadDir, file); let stat; try { stat = fs.statSync(filePath); } catch { continue; } if (isTextFile(ext) && stat.size <= MAX_INJECT_SIZE && totalInjected + stat.size <= MAX_TOTAL_INJECT) { try { const content = fs.readFileSync(filePath, 'utf-8'); const lang = getLanguageTag(ext); contextBlock += `### 📄 ${file} (${formatFileSize(stat.size)})\n\`\`\`${lang}\n${content}\n\`\`\`\n\n`; totalInjected += stat.size; } catch { contextBlock += `### 📄 ${file} (${formatFileSize(stat.size)}) - 读取失败\n\n`; } } else if (isTextFile(ext) && stat.size > MAX_INJECT_SIZE) { // 大文本文件：只注入前 50KB try { const fd = fs.openSync(filePath, 'r'); const buf = Buffer.alloc(MAX_INJECT_SIZE); fs.readSync(fd, buf, 0, MAX_INJECT_SIZE, 0); fs.closeSync(fd); const lang = getLanguageTag(ext); contextBlock += `### 📄 ${file} (${formatFileSize(stat.size)}, 仅显示前50KB)\n\`\`\`${lang}\n${buf.toString('utf-8')}\n...（文件过大已截断，完整路径: ${filePath}）\n\`\`\`\n\n`; totalInjected += MAX_INJECT_SIZE; } catch { contextBlock += `### 📄 ${file} (${formatFileSize(stat.size)}) - 路径: ${filePath}\n\n`; } } else if (isImageFile(ext)) { contextBlock += `### 🖼️ ${file} (${formatFileSize(stat.size)})\n(图片文件，路径: ${filePath})\n\n`; } else { contextBlock += `### 📦 ${file} (${formatFileSize(stat.size)})\n(二进制文件，路径: ${filePath})\n\n`; } if (totalInjected >= MAX_TOTAL_INJECT) { contextBlock += '> ⚠️ 已达到注入上限(500KB)，剩余文件仅列出路径。\n\n'; break; } } contextBlock += '---\n'; return contextBlock; } function buildPrompt(agentDir, agentId, convId, userMessage, isNewSession) { // 注入上传文件上下文（按寻址名取，与上传端点的落点是同一份推导） const uploadContext = buildUploadContext(agentId, convId); const enrichedMessage = uploadContext ? uploadContext + '\n' + userMessage : userMessage; if (!isNewSession) { // 场景 B: ACP 进程还在，多轮上下文自动保持 return enrichedMessage; } const systemPrompt = buildSystemPrompt(agentDir, userMessage, agentId, convId); const cronInstructions = buildCronInstructions(agentDir); const history = loadHistory(agentId, convId); if (history.length === 0) { // 场景 A: 全新对话 return `${systemPrompt}${cronInstructions}\n\n---\n\n## 用户消息\n\n${enrichedMessage}`; } // 场景 C: 恢复上下文 — 注入历史摘要 const recentHistory = history.slice(-20); const historyText = recentHistory.map(m => { const role = m.role === 'user' ? 'Human' : 'Assistant'; const content = m.content.length > 500 ? m.content.slice(0, 500) + '...(截断)' : m.content; return `${role}: ${content}`; }).join('\n\n'); return `${systemPrompt}${cronInstructions}\n\n---\n\n## 之前的对话历史（供你参考以保持上下文连贯）\n\n${historyText}\n\n---\n\n## 用户新消息\n\n${enrichedMessage}`; } // ═══════════════════════════════════════ // 路由 // ═══════════════════════════════════════ router.get('/quota', (req, res) => { try { const DATA_ROOT = path.resolve(__dirname, '../../data'); const records = []; const daysBack = parseInt(String(req.query.days || '30')) || 30; // 默认读最近 30 天 const cutoff = new Date(Date.now() - daysBack * 86400000).toISOString(); // 读取 billing 日志（兼容旧 billing.log + 新 billing-YYYY-MM-DD.log） const readBilling = (dir) => { if (!fs.existsSync(dir)) return; const files = fs.readdirSync(dir).filter(f => f.startsWith('billing')); for (const file of files) { // 按日分片优化：跳过超出时间范围的文件 const dateMatch = file.match(/billing-(\d{4}-\d{2}-\d{2})\.log/); if (dateMatch && dateMatch[1] < cutoff.substring(0, 10)) continue; const p = path.join(dir, file); try { fs.readFileSync(p, 'utf-8').split('\n').forEach(line => { line = line.trim(); if (line.startsWith('{')) { try { const rec = JSON.parse(line); if (rec.timestamp >= cutoff) records.push(rec); } catch(e) {} } else if (line) { // 旧的纯文本兼容 try { const parts = line.split('|').map(p => p.trim()); const ts = parts[0]; if (ts < cutoff) return; const ai = parts[1].replace('Agent:', '').trim(); const ci = parts[2].replace('Conv:', '').trim(); const cost = parseFloat(parts[3].replace('Cost: $', '').trim()); const dur = parseInt(parts[4].replace('Duration:', '').replace('ms', '').trim()); records.push({ timestamp: ts, agentId: ai, convId: ci, costUsd: cost || 0, durationMs: dur || 0, usage: { input: 0, output: 0 } }); } catch(e) {} } }); } catch(e) {} } }; readBilling(path.resolve(__dirname, '../../system/agents/master/logs')); const compDir = path.join(DATA_ROOT, 'companies'); if (fs.existsSync(compDir)) { const comps = fs.readdirSync(compDir, { withFileTypes: true }); for (const c of comps) { if (c.isDirectory() && !c.name.startsWith('.')) { readBilling(path.join(compDir, c.name, 'logs')); } } } res.json(records); } catch (err) { res.status(500).json({ error: err.message }); } }); // ── Agent List 缓存 (5秒 TTL，避免每次请求全量扫描磁盘) ── let _agentListCache = null; let _agentListCacheTime = 0; const AGENT_LIST_CACHE_TTL = 5000; // 5 秒 router.get('/list', (req, res) => { try { // 缓存命中：5秒内直接返回 if (_agentListCache && Date.now() - _agentListCacheTime < AGENT_LIST_CACHE_TTL) { return res.json(_agentListCache); } const allAgents = [] // 辅助函数：读一个 agent 的本体定义 + 它在某家公司下的最近活跃 // // 本体目录恒为 system/agents/{bodyId}（28 批次 4：公司不再持有 agent 实体）， // 所以「扫哪个目录」这件事只对 system 侧成立；公司侧改成按名册逐个点名。 // roleName 与 bodyId 可以不同（同一本体在一家公司担两个角色），两者分开传。 const collectAgent = (bodyId, roleName, companyId) => { const agentDir = locator.bodyDir(bodyId) const config = loadAgentConfig(agentDir) if (!config) return const identity = loadIdentity(agentDir) let lastActiveAt = null let lastMessage = '' // 会话落在公司态目录（28 §4.1）：system 侧是 {aid}/_global/，公司侧是 {aid}/{cid}/ const convDir = path.join( locator.companyDir(bodyId, companyId === 'system' ? null : companyId), 'conversations' ) if (fs.existsSync(convDir)) { try { const convs = fs.readdirSync(convDir, { withFileTypes: true }).filter(d => d.isDirectory()) let latest = 0 let latestConvName = null for (const c of convs) { const metaPath = path.join(convDir, c.name, 'meta.json') if (fs.existsSync(metaPath)) { try { const meta = JSON.parse(fs.readFileSync(metaPath, 'utf-8')) const t = new Date(meta.updatedAt || meta.createdAt).getTime() if (t > latest) { latest = t latestConvName = c.name } } catch {} } } if (latest > 0) { lastActiveAt = new Date(latest).toISOString() if (latestConvName) { const historyPath = path.join(convDir, latestConvName, 'history.json') if (fs.existsSync(historyPath)) { try { const history = JSON.parse(fs.readFileSync(historyPath, 'utf-8')) if (history.length > 0) { const msg = history[history.length - 1] if (Array.isArray(msg.content)) { lastMessage = msg.content.map(c => c.text || '').join(' ') } else { lastMessage = msg.content || '' } if (msg.timestamp) lastActiveAt = msg.timestamp } } catch {} } } } } catch {} } // 截断一行 if (lastMessage) { lastMessage = lastMessage.replace(/[\r\n]+/g, ' ').trim() if (lastMessage.length > 50) lastMessage = lastMessage.substring(0, 50) + '...' } // System 级 Agent ID 保持原样，Company 级需要拼装 "companyId:roleName" // —— 公司侧用 roleName 而不是 config.id：本体的 config.id 是本体自己的名字， // 拿它拼地址会在「同一本体担两个角色」时把两行折叠成同一个 id。 const agentId = companyId === 'system' ? (config.id || bodyId) : `${companyId}:${roleName}` allAgents.push({ id: agentId, name: identity.name, emoji: identity.emoji, type: config.type, status: config.status, greeting: identity.greeting, lastActiveAt, lastMessage, companyId // 新增用于前端分组过滤 }) } // 1. 扫描 System Agents（system/agents/ 深度 1；`_` / `.` 开头是保留 id，不是可调用 agent） const sysAgentsDir = locator.AGENTS_ROOT if (fs.existsSync(sysAgentsDir)) { for (const e of fs.readdirSync(sysAgentsDir, { withFileTypes: true })) { if (!e.isDirectory() || e.name.startsWith('.') || e.name.startsWith('_')) continue collectAgent(e.name, e.name, 'system') } } // 2. Company Agents：名册查 company_agents（28 批次 4-19）。 // 公司侧已经没有 agent 目录可扫 —— 继续扫的结果是列表里一个公司 agent 都不剩。 for (const row of listCompanyAgents()) { collectAgent(row.system_agent_id, row.role_name, row.company_id) } // 排序: master 永远第一，其他按活跃倒序 allAgents.sort((a, b) => { if (a.type === 'master') return -1 if (b.type === 'master') return 1 const ta = a.lastActiveAt ? new Date(a.lastActiveAt).getTime() : 0 const tb = b.lastActiveAt ? new Date(b.lastActiveAt).getTime() : 0 return tb - ta }) // 更新缓存 _agentListCache = allAgents; _agentListCacheTime = Date.now(); res.json(allAgents) } catch (err) { res.status(500).json({ error: err.message }) } }); // ── Phase 2: 小脑方案固化 API ── router.post('/:agentId/memory/cerebellum', (req, res) => { const { saveCerebellumPlan } = require('../services/memory-service'); const { scenario, keywords, steps, conditions, notes, rawContent, confidence } = req.body; if (!scenario) return res.status(400).json({ error: 'scenario is required' }); const result = saveCerebellumPlan(req.params.agentId, { scenario, keywords, steps, conditions, notes, rawContent, confidence: confidence || 0.7, }); if (result) { res.json({ success: true, path: result }); } else { res.status(500).json({ error: 'Failed to save cerebellum plan' }); } }); // ── Phase 3: 记忆管理 API ── router.post('/:agentId/memory/consolidate', (req, res) => { const result = consolidateMemory(req.params.agentId); res.json(result); }); router.post('/:agentId/memory/daily-report', (req, res) => { const reportPath = generateDailyReport(req.params.agentId); res.json({ success: !!reportPath, path: reportPath }); }); router.get('/:agentId/info', (req, res) => { const agentDir = getBodyDir(req.params.agentId); if (!fs.existsSync(agentDir)) return res.status(404).json({ error: 'Agent not found' }); const config = loadAgentConfig(agentDir); const identity = loadIdentity(agentDir); // Read cron settings let cron = []; try { const cronPath = path.join(agentDir, 'cron.json'); if (fs.existsSync(cronPath)) { cron = JSON.parse(fs.readFileSync(cronPath, 'utf8')); } } catch (e) { console.error('Failed to parse cron.json:', e); } res.json({ id: config?.id || req.params.agentId, ...identity, type: config?.type, status: config?.status, // 心跳读**公司态优先**（28-followups §3.3）—— 与 _setupAgentHeartbeat 读的是同一个函数。 // 直接读本体 config.json 的话，(agent, 公司) 一旦设过公司态，这里显示的就不是实际在跑的值： // 界面写「30 分钟」而调度器按本体缺省的 15 分钟在跑，且两边都不报错。 heartbeat: require('../services/heartbeat-service').readHeartbeatConfig(req.params.agentId), cron: cron, bootstrapFiles: ['agent.md','soul.md','identity.md','user.md','tools.md','heartbeat.md', 'cron.json'].filter(f => loadFile(agentDir, f).trim().length > 0), }); }); /** * 心跳配置写入 —— **落公司态，不回写本体**（28-followups §3.3）。 * * 原来这里是 `getBodyDir(id)` → 整份重写本体 `config.json`，公司段在寻址第一步就丢了。 * 两个后果，都不报错： * ① 跨租户串写：`co-a:ceo` 改心跳，`co-b:ceo` 跟着变（同一份本体）； * ② 一旦该 (agent, 公司) 有了公司态文件，读侧优先取公司态 → **这个端点的写入静默无效**， * 界面照样收到 `{success:true}`，而实际在跑的还是旧值。 * * 现在读写都走 heartbeat-service 那**一对**函数，与心跳挂载读的是同一个答案。 * 入参校验也归它（intervalMinutes 直接进 setInterval，NaN / 0 / 负数会让定时器退化成 * 每 tick 触发一次）—— 非法值在写盘前抛，这里翻成 400。 */ router.put('/:agentId/heartbeat', (req, res) => { const agentId = req.params.agentId; try { const heartbeatService = require('../services/heartbeat-service'); // 存在性检查走服务自己的寻址：`cid:role` 里冒号后面是**角色名**，未必等于本体 id。 // 用 getBodyDir(id)（= 拿角色名当本体 id）判的话，凡是角色名 ≠ 本体 id 的都会被 // 判成「agent 不存在」，而写侧其实找得到 —— 判 404 和写入必须问同一个问题。 if (!fs.existsSync(heartbeatService.bodyDirOf(agentId))) { return res.status(404).json({ error: 'Agent not found' }); } let heartbeat; try { heartbeat = heartbeatService.writeHeartbeatConfig(agentId, { enabled: req.body.enabled || false, intervalMinutes: req.body.intervalMinutes, }); } catch (err) { return res.status(400).json({ error: err.message }); // 入参非法，不是服务器故障 } // 定时器跟着改：开着的重挂（_setupAgentHeartbeat 内部会先清旧的），关掉的摘掉 if (heartbeat.enabled) { heartbeatService._setupAgentHeartbeat(agentId); } else if (heartbeatService.timers.has(agentId)) { clearInterval(heartbeatService.timers.get(agentId)); heartbeatService.timers.delete(agentId); } res.json({ success: true, heartbeat }); } catch (err) { res.status(500).json({ error: err.message }); } }); router.put('/:agentId/cron', (req, res) => { try { const agentDir = getBodyDir(req.params.agentId); if (!fs.existsSync(agentDir)) return res.status(404).json({ error: 'Agent not found' }); const cronPath = path.join(agentDir, 'cron.json'); const cronList = Array.isArray(req.body.cron) ? req.body.cron : []; fs.writeFileSync(cronPath, JSON.stringify(cronList, null, 2), 'utf8'); // Check if the heartbeat service picks up the cron changes automatically. // Generally cron jobs execute within the heartbeat mechanism. res.json({ success: true, cron: cronList }); } catch (err) { res.status(500).json({ error: err.message }); } }); /** 列出对话 */ router.get('/:agentId/conversations', (req, res) => { res.json(listConversations(req.params.agentId)); }); /** 获取对话详情 (meta + history) */ router.get('/:agentId/conversations/:convId', (req, res) => { const { agentId, convId } = req.params; const meta = loadMeta(agentId, convId); if (!meta) return res.status(404).json({ error: 'Not found' }); const messages = loadHistory(agentId, convId); // 同时返回 sub-agent logs 列表 const subLogs = listSubAgentLogs(agentId, convId); res.json({ ...meta, messages, subLogs }); }); /** 获取子 Agent 日志列表 */ router.get('/:agentId/conversations/:convId/sub-logs', (req, res) => { res.json(listSubAgentLogs(req.params.agentId, req.params.convId)); }); /** 获取单个子 Agent 日志内容 */ router.get('/:agentId/conversations/:convId/sub-logs/:logFile', (req, res) => { const { agentId, convId, logFile } = req.params; const logPath = path.join(getConvDir(agentId, convId), logFile); if (!logFile.endsWith('.log') || !fs.existsSync(logPath)) { return res.status(404).json({ error: 'Log not found' }); } const content = fs.readFileSync(logPath, 'utf-8'); res.json({ file: logFile, content }); }); function listSubAgentLogs(agentId, convId) { const convDir = getConvDir(agentId, convId); if (!fs.existsSync(convDir)) return []; return fs.readdirSync(convDir) .filter(f => f.startsWith('sub-') && f.endsWith('.log')) .map(f => { const match = f.match(/^(sub-[a-f0-9]+)-(.+)\.log$/); const content = fs.readFileSync(path.join(convDir, f), 'utf-8'); const status = content.includes('STATUS: SUCCESS') ? 'success' : content.includes('STATUS: ERROR') ? 'error' : 'running'; return { file: f, callId: match ? match[1] : f, agent: match ? match[2] : 'unknown', status, size: content.length, content, // 直接内嵌内容，省一次 API 调用 }; }) .sort((a, b) => a.file.localeCompare(b.file)); } /** 删除对话 */ router.delete('/:agentId/conversations/:convId', (req, res) => { const { agentId, convId } = req.params; try { // 1. 删除对话目录（会话 / 上传件都在公司态目录下，见 getStateDir 注释） const convDir = path.join(getStateDir(agentId), 'conversations', convId); if (fs.existsSync(convDir)) fs.rmSync(convDir, { recursive: true, force: true }); // 2. 删除关联的日志目录/文件 —— 本体根上的 logs/ 是老布局的遗留，新日志落任务态目录， // 这里两处都清：任务态目录整个就是这次对话的（convId 充当 taskId），可以直接删。 const taskDir = getTaskDir(agentId, convId); if (fs.existsSync(taskDir)) fs.rmSync(taskDir, { recursive: true, force: true }); const logsDir = path.join(getBodyDir(agentId), 'logs'); if (fs.existsSync(logsDir)) { for (const f of fs.readdirSync(logsDir)) { if (f.includes(convId)) { const fp = path.join(logsDir, f); fs.rmSync(fp, { recursive: true, force: true }); } } } // 3. 删除上传文件 const uploadDir = getUploadDir(agentId, convId); if (fs.existsSync(uploadDir)) fs.rmSync(uploadDir, { recursive: true, force: true }); // 4. 断开会话 closeSession(`${agentId}:${convId}`, '对话被删除'); console.log(`[AgentChat] 🗑️ 已删除对话 ${agentId}/${convId} (含uploads)`); res.json({ success: true }); } catch (err) { res.status(500).json({ error: err.message }); } }); /** 停止当前正在生成的任务 */ // 40 号文档方案二（3.1.1）：中止后不能立刻同步返回——原来那个 /chat 请求的 handler // 是否已经走到 finally（释放 agentChatLocks、res.end()）还不确定，"打断并发送"要用这个 // 信号判断能不能安全发第二条。这里轮询等最多 3 秒，超时不算失败，只是 cleaned:false， // 交给前端决定要不要"强制发送"。 router.post('/:agentId/chat/:convId/stop', async (req, res) => { const { agentId, convId } = req.params; const key = `${agentId}:${convId}`; if (!closeSession(key, '用户手动中断')) { return res.json({ success: false, message: '未找到运行中的 Agent 进程' }); } const cleaned = await waitForLockRelease(agentId); res.json({ success: true, message: 'Agent 已中断', cleaned }); }); /** 读取 bootstrap file */ router.get('/:agentId/bootstrap/:filename', (req, res) => { const allowed = ['agent.md','soul.md','identity.md','user.md','tools.md','heartbeat.md']; if (!allowed.includes(req.params.filename)) return res.status(400).json({ error: 'Invalid' }); res.json({ filename: req.params.filename, content: loadFile(getBodyDir(req.params.agentId), req.params.filename) }); }); /** * v5: Agent 文件树 + 文件预览 + todo/plan */ // GET /api/agent/:agentId/files — 递归列出 Agent 目录文件树 router.get('/:agentId/files', (req, res) => { const agentDir = getBodyDir(req.params.agentId); if (!fs.existsSync(agentDir)) return res.status(404).json({ error: 'Agent not found' }); const scanDir = (dir, relativePath = '') => { const entries = fs.readdirSync(dir, { withFileTypes: true }); const result = []; for (const entry of entries) { const relPath = relativePath ? `${relativePath}/${entry.name}` : entry.name; if (entry.name.startsWith('.') && entry.name !== '.claude') continue; if (entry.isDirectory()) { const children = scanDir(path.join(dir, entry.name), relPath); result.push({ name: entry.name, path: relPath, type: 'dir', children, count: children.length }); } else { const stat = fs.statSync(path.join(dir, entry.name)); result.push({ name: entry.name, path: relPath, type: 'file', size: stat.size }); } } return result.sort((a, b) => { if (a.type !== b.type) return a.type === 'dir' ? -1 : 1; return a.name.localeCompare(b.name); }); }; try { const tree = scanDir(agentDir); const countFiles = (nodes) => { return nodes.reduce((sum, n) => sum + (n.type === 'file' ? 1 : countFiles(n.children || [])), 0); }; res.json({ tree, totalFiles: countFiles(tree) }); } catch (err) { res.status(500).json({ error: err.message }); } }); // GET /api/agent/:agentId/file/* — 读取 Agent 目录内的文件内容 router.get('/:agentId/file/*', (req, res) => { const filePath = req.params[0]; if (!filePath) return res.status(400).json({ error: 'file path required' }); // 读侧看本体（只读展示）；写侧看公司态目录，见下面 PUT/DELETE 的注释。 const fullPath = resolveInside(getBodyDir(req.params.agentId), filePath); if (!fullPath) return res.status(403).json({ error: 'Access denied' }); if (!fs.existsSync(fullPath)) return res.status(404).json({ error: 'File not found' }); try { const stat = fs.statSync(fullPath); if (stat.isDirectory()) return res.status(400).json({ error: 'Path is a directory' }); const ext = path.extname(fullPath).toLowerCase(); const isBinary = ['.png', '.jpg', '.jpeg', '.gif', '.ico', '.woff', '.woff2', '.ttf'].includes(ext); if (isBinary) return res.json({ path: filePath, size: stat.size, binary: true, content: '(binary file)' }); const content = fs.readFileSync(fullPath, 'utf-8'); res.json({ path: filePath, size: stat.size, content, ext }); } catch (err) { res.status(500).json({ error: err.message }); } }); // PUT /api/agent/:agentId/file/* — 保存文件内容 router.put('/:agentId/file/*', (req, res) => { const filePath = req.params[0]; if (!filePath) return res.status(400).json({ error: 'file path required' }); // 可写根是**公司态**目录，不是本体（28 §4.1：本体只读，只有 evolver / agent-creator 能改）。 // 用本体当可写根时，`co-alpha:ceo` 这个地址能覆写 system/agents/ceo/agent.md， // 而 co-beta 的 ceo 用的是同一份文件 —— 一次保存就是一次跨租户改写，且全程无报错。 const writeRoot = getStateDir(req.params.agentId); const fullPath = resolveInside(writeRoot, filePath); if (!fullPath) return res.status(403).json({ error: 'Access denied' }); try { // 确保目录存在 const dir = path.dirname(fullPath); if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true }); fs.writeFileSync(fullPath, req.body.content || '', 'utf-8'); console.log(`[AgentChat] 💾 已保存文件 ${req.params.agentId}/${filePath}`); res.json({ success: true, path: filePath }); } catch (err) { res.status(500).json({ error: err.message }); } }); // DELETE /api/agent/:agentId/file/* — 删除文件或文件夹 router.delete('/:agentId/file/*', (req, res) => { const filePath = req.params[0]; if (!filePath) return res.status(400).json({ error: 'file path required' }); // 同 PUT：删除的根也只能是公司态目录。`../../agent.md` 这类逃逸由 resolveInside 拦下 —— // path.join + startsWith 拦不住它，`..` 在 join 阶段就被折叠成了合法的本体内路径。 const fullPath = resolveInside(getStateDir(req.params.agentId), filePath); if (!fullPath) return res.status(403).json({ error: 'Access denied' }); if (!fs.existsSync(fullPath)) return res.status(404).json({ error: 'Not found' }); try { fs.rmSync(fullPath, { recursive: true, force: true }); console.log(`[AgentChat] 🗑️ 已删除 ${req.params.agentId}/${filePath}`); res.json({ success: true, path: filePath }); } catch (err) { res.status(500).json({ error: err.message }); } }); // GET/PUT /api/agent/:agentId/todo · /plan // // todo.md / plan.md 归 `(agentId, companyId)`，所以落公司态目录：它们记的是「谁派了什么活给我」， // 而派活的是公司。放本体上时，co-alpha 与 co-beta 派给 `ceo` 的任务行追加进同一份文件 —— // agent 自读时看到两家公司的任务混编、前端任务面板也是，而这两个文件被 gitignore 挡着， // git status 一片干净，症状只会以「ceo 去干了别家公司的活」的形式浮出来。 function readStateFile(id, name) { const fp = path.join(getStateDir(id), name); try { return fs.readFileSync(fp, 'utf-8'); } catch { return ''; } } function writeStateFile(id, name, content) { const dir = getStateDir(id); fs.mkdirSync(dir, { recursive: true }); // 公司态目录是懒创建的，首次派单时可能还不存在 fs.writeFileSync(path.join(dir, name), content, 'utf-8'); } router.get('/:agentId/todo', (req, res) => { res.json({ content: readStateFile(req.params.agentId, 'todo.md') }); }); router.put('/:agentId/todo', (req, res) => { if (!fs.existsSync(getBodyDir(req.params.agentId))) return res.status(404).json({ error: 'Agent not found' }); writeStateFile(req.params.agentId, 'todo.md', req.body.content || ''); res.json({ success: true }); }); router.get('/:agentId/plan', (req, res) => { res.json({ content: readStateFile(req.params.agentId, 'plan.md') }); }); router.put('/:agentId/plan', (req, res) => { if (!fs.existsSync(getBodyDir(req.params.agentId))) return res.status(404).json({ error: 'Agent not found' }); writeStateFile(req.params.agentId, 'plan.md', req.body.content || ''); res.json({ success: true }); }); // GET/PUT /api/agent/:agentId/cron — 定时任务声明 (cron.json) router.get('/:agentId/cron', (req, res) => { const fp = path.join(getBodyDir(req.params.agentId), 'cron.json'); if (!fs.existsSync(fp)) return res.json([]); try { res.json(JSON.parse(fs.readFileSync(fp, 'utf-8'))); } catch { res.json([]); } }); router.put('/:agentId/cron', (req, res) => { const agentDir = getBodyDir(req.params.agentId); if (!fs.existsSync(agentDir)) return res.status(404).json({ error: 'Agent not found' }); const crons = req.body; if (!Array.isArray(crons)) return res.status(400).json({ error: 'Body must be an array of cron objects' }); fs.writeFileSync(path.join(agentDir, 'cron.json'), JSON.stringify(crons, null, 2), 'utf-8'); console.log(`[Cron] 📝 ${req.params.agentId} cron.json 已更新 (${crons.length} tasks)`); res.json({ success: true, count: crons.length }); }); /** * POST /api/agent/scaffold — 通过 AgentScaffold 父类创建新 Agent 实例 * * body: { id, type, name, parentId?, bootstrap?: { agentMd?, soulMd?, skillMd? } } * * 这是 meta_creator 子 Agent 调用的核心接口，确保每个新 Agent 都 * 继承了标准目录结构（.claude/skills/main_skill, memory, workspace 等） */ router.post('/scaffold', async (req, res) => { try { const { id, type = 'skill', name, parentId, bootstrap = {} } = req.body; if (!id || !name) return res.status(400).json({ error: 'id and name are required' }); // 判重与 company.js 的 POST /templates 同一套（那里已经 409 了，这里原来没有）： // scaffold.create() 无条件覆写 agent.md / soul.md / identity.md / tools.md / user.md / // heartbeat.md / config.json / todo.md / plan.md。28 批次 5-21 把 baseDir 统一到 // system/agents 之后，这个端点等于「任何拿得到 HTTP 的 agent 都能把任意现役本体 // 重置成通用骨架」，而且返回 200 + success:true。本体只读是本次改造的核心不变量， // 唯一允许写它的是 evolver / agent-creator 那条显式进化链路，不是脚手架。 if (fs.existsSync(locator.bodyDir(id))) { return res.status(409).json({ error: `本体 "${id}" 已存在，不允许用脚手架覆写（本体只读，改定义走 agent-creator / evolver）` }); } const { AgentScaffold } = await import('../../packages/agent-core/AgentScaffold.js'); const scaffold = new AgentScaffold(locator.AGENTS_ROOT); const agentDir = await scaffold.create({ id, type, name, parentId, bootstrap }); if (bootstrap.skillMd) { const skillPath = path.join(agentDir, '.claude', 'skills', 'main_skill', 'SKILL.md'); fs.writeFileSync(skillPath, bootstrap.skillMd, 'utf-8'); } // 定义写完了就在这儿把闸门过一遍（28 批次 5-23）：本体如果正停在 draft、 // 三个定义文件又已齐全，这里就是它最早能转正的一刻。库里没有这行 = 这个 agent // 还不属于任何公司，函数直接空跑。 const { promoteDraftIfComplete } = require('../services/graph-service'); promoteDraftIfComplete(id); console.log(`[Scaffold] ✅ 新 Agent "${name}" (${id}) 已通过父类脚手架创建: ${agentDir}`); res.json({ success: true, agentId: id, agentDir, message: `Agent ${name} 已创建 (v5: +todo/plan/hippocampus)`, structure: [ '.claude/skills/main_skill/', 'memory/', 'memory/cerebellum/', 'knowledge/', 'workspace/', 'conversations/', 'logs/', 'agent.md', 'soul.md', 'identity.md', 'tools.md', 'todo.md', 'plan.md', 'cron.json', 'config.json' ] }); } catch (err) { console.error('[Scaffold] Error:', err.message); res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // v5: Task Dispatch — 跨 Agent 任务分发 // ═══════════════════════════════════════ /** * POST /api/agent/task/dispatch — 向目标 Agent 分发任务 * * body: { * from: 'master-001', // 发送者 Agent ID * to: 'agent-creator', // 目标 Agent ID * task: '创建一个诗人 Agent', // 任务描述 * priority: 'P0', // 优先级 * planId: 'P-001', // 在发送者 plan.md 中的编号 (可选) * autoExecute: true, // 是否自动异步触发执行 * } */ // ── Task 元数据：存储换成 tasks 表（27-P3C 批 5 §十二）───────────────── // // **本文件除了这两个函数的内脏，其余一行都没动。** `/task/dispatch` 与 `/task/report` 的 // URL、请求体、响应体形态全部原样 —— 它们印在 packages/agent-core/base/common/tools-base.md // 里，是**所有 agent 的工具说明**，改一个字等于同时改所有 agent 的提示词。 // // 原来的存储是 system/tasklog/{taskId}.json 一任务一文件。换掉它的三个理由： // · 去重要「扫整个目录 + 逐个 JSON.parse」，任务一多就是 O(n) 次磁盘读，且没有索引可用； // · 派单任务在库里看不见 —— 控制台的任务列表读的是 tasks 表，派单那套完全是平行世界； // · 同一个 T- 前缀两套 id 体系，读日志时分不清。 // // **meta 的形状不变**（调用方按字段名取值），只是落点从文件换成表。 const DISPATCH_COMPANY = '_global'; // 派单没有 company 这一维，全局 agent 的保留 id（28 §5.1） /** meta.status ↔ tasks.status。两边取值域不同，来回都要过一次翻译 */ const META_TO_TASK_STATUS = { dispatched: 'running', done: 'completed', failed: 'failed', blocked: 'blocked', cancelled: 'cancelled' }; const TASK_TO_META_STATUS = { running: 'dispatched', completed: 'done', queued: 'dispatched', draft: 'dispatched' }; /** 正文首行当标题；派单正文把换行写成
，先还原再取 */ function dispatchTitle(task) { const line = String(task || '').replace(//gi, '\n').split('\n')[0].trim(); if (!line) return '(无标题派单)'; return line.length > 60 ? `${line.slice(0, 60)}…` : line; } function saveTaskMeta(taskId, meta) { const wdb = require('../services/work-graph-db'); const db = wdb.getDb(); const now = new Date().toISOString(); const status = META_TO_TASK_STATUS[meta.status] || 'running'; const existing = db.prepare('SELECT id, goal_id FROM tasks WHERE id = ?').get(taskId); if (existing) { // 更新路径：/task/report 的写法是「loadTaskMeta → 改几个字段 → 整个对象写回」， // 所以这里是全量 upsert 而不是增量 patch。 db.prepare(` UPDATE tasks SET status = @status, summary = @summary, finished_at = @done, description = @desc, priority = @prio, updated_at = @now WHERE id = @id `).run({ id: taskId, status, summary: meta.summary || null, done: meta.completedAt || null, desc: String(meta.task || ''), prio: meta.priority || 'P1', now, }); return; } // 新建路径：每条派单自带一个 goal —— tasks.goal_id 是 NOT NULL 真外键，而派单没有 // 「人提的需求」那一层。与 §六.3 给 POST /:cid/tasks 设计的兼容路径同一条， // 不给 goal_id 开可空的口子。 const goalId = wdb.genGoalId(); const createdAt = meta.createdAt || now; db.prepare(` INSERT INTO goals (id, company_id, title, description, status, priority, okr_node_id, created_by, report_to, created_at, updated_at, dispatched_at) VALUES (@id, @cid, @title, @desc, 'active', @prio, @okr, @by, @by, @at, @at, @at) `).run({ id: goalId, cid: DISPATCH_COMPANY, title: `来自 ${meta.from || '?'} 的派单`, desc: String(meta.task || ''), prio: meta.priority || 'P1', okr: meta.nodeId || null, by: meta.from || null, at: createdAt, }); db.prepare(` INSERT INTO tasks (id, goal_id, company_id, seq, title, description, status, execution_mode, depends_on, attempt, priority, created_by, executor, assignee_agent_id, parent_task_id, summary, execution_conv_id, created_at, updated_at, queued_at, started_at) VALUES (@id, @gid, @cid, 1, @title, @desc, @status, 'manual', '[]', 1, @prio, @by, 'agent', @to, @parent, @summary, @conv, @at, @at, @at, @at) `).run({ id: taskId, gid: goalId, cid: DISPATCH_COMPANY, title: dispatchTitle(meta.task), desc: String(meta.task || ''), status, prio: meta.priority || 'P1', by: meta.from || null, to: meta.to || null, parent: meta.parentTaskId || null, summary: meta.summary || null, conv: meta.executionConvId || null, at: createdAt, }); } function loadTaskMeta(taskId) { const wdb = require('../services/work-graph-db'); const row = wdb.getDb().prepare(` SELECT t.*, g.okr_node_id FROM tasks t JOIN goals g ON g.id = t.goal_id WHERE t.id = ? `).get(taskId); if (!row) return null; // objectiveId 不单独存：它恒等于 okr node id 的首段（OkrService.parseNodeId 与 // company.js:reportRunOutcome 都是这个口径），而且它只在与 nodeId 一起用时才有意义 // （两处读取方都是 `if (objId && meta.nodeId)`）。 const nodeId = row.okr_node_id || null; const objectiveId = nodeId ? String(nodeId).split('.')[0] : null; return { taskId: row.id, from: row.created_by, to: row.assignee_agent_id, task: row.description, priority: row.priority, objectiveId, goalId: objectiveId, // 旧字段名，保留供存量读取方兼容（27-NAMING §六） nodeId, parentTaskId: row.parent_task_id, executionConvId: row.execution_conv_id, createdAt: row.created_at, status: TASK_TO_META_STATUS[row.status] || row.status, completedAt: row.finished_at, summary: row.summary, }; } router.post('/task/dispatch', async (req, res) => { try { const { from, to, task, priority = 'P1', planId, nodeId, autoExecute = true, parentTaskId: explicitParent } = req.body; // 27-NAMING §六：字段正名 goalId → objectiveId，两者都接受，旧字段保留兼容 const objectiveId = req.body.objectiveId || req.body.goalId || null; if (!from || !to || !task) return res.status(400).json({ error: 'from, to, task are required' }); // 存在性问的是**本体**（这个 agent 有没有），任务行写的是**公司态**（谁派的活）—— // 两者不是同一个目录，合成一个就是把公司段丢掉，见下面 todo.md / plan.md 的落点。 if (!fs.existsSync(getBodyDir(to))) return res.status(404).json({ error: `Target agent "${to}" not found` }); if (!fs.existsSync(getBodyDir(from))) return res.status(404).json({ error: `Sender agent "${from}" not found` }); const taskId = `T-${Date.now().toString(36)}`; const now = new Date().toISOString().slice(0, 16).replace('T', ' '); // ── 任务去重检查：相同接收方 30 分钟内是否有相似任务已分发 ── // 批 5：从「readdirSync 整个 tasklog + 逐个 JSON.parse」换成一条带索引的 SQL // （idx_tasks_assignee = company_id + assignee_agent_id + created_at）。 // **判据本身一个字没改** —— 仍是「同一接收方 + 30 分钟内 + 仍在执行 + 正文互相包含」， // 换的只是从哪儿把候选捞出来。响应体形态也原样（agent 的提示词里印着它）。 const DUPLICATE_WINDOW_MS = 30 * 60 * 1000; try { const wdb = require('../services/work-graph-db'); const since = new Date(Date.now() - DUPLICATE_WINDOW_MS).toISOString(); const rows = wdb.getDb().prepare(` SELECT id, description FROM tasks WHERE company_id = @cid AND assignee_agent_id = @to AND executor = 'agent' AND status = 'running' AND created_at > @since AND archived_at IS NULL ORDER BY created_at DESC `).all({ cid: DISPATCH_COMPANY, to, since }); const normNew = task.toLowerCase().replace(/[\s\n\r]+/g, ' ').trim(); for (const r of rows) { const normExist = String(r.description || '').toLowerCase() .replace(/
/g, ' ').replace(/[\s\n\r]+/g, ' ').trim(); if (normNew === normExist || normNew.includes(normExist) || normExist.includes(normNew)) { console.log(`[TaskDispatch] 🚫 去重拦截: ${from} → ${to} 与 ${r.id} 重复`); return res.status(409).json({ success: false, error: 'Duplicate task detected', existingTaskId: r.id, existingTask: r.description, message: `去重拦截：${to} 在 30 分钟内已有相似任务 ${r.id} 处于进行中，请勿重复分发。`, }); } } } catch (e) { // 去重查不动不该让派单整个失败，但**必须留声** —— 静默跳过等于去重形同虚设 console.error(`[TaskDispatch] ⚠️ 去重检查失败，本次放行: ${e.message}`); } // 净化多行文本和竖线，防止破坏 Markdown 表格格式 const safeTask = task.replace(/\n/g, '
').replace(/\|/g, '|'); // ── 自动推断 parentTaskId ── // 如果调用者没有显式传 parentTaskId，则尝试从 .tasks/ 目录下 // 查找 from agent 正在执行的（status=dispatched 且 to=from）的 task let parentTaskId = explicitParent || null; if (!parentTaskId) { // 批 5：同样从扫目录换成一条 SQL，判据不变 ——「派给 from 且仍在执行」里最近的一条， // 那就是 from 此刻正在干的活，新派的这条挂在它下面。排序交给 SQL（原来是把全部候选 // 读进内存再 sort，目录一大就是白读一遍）。 try { const wdb = require('../services/work-graph-db'); const parent = wdb.getDb().prepare(` SELECT id FROM tasks WHERE company_id = @cid AND assignee_agent_id = @from AND executor = 'agent' AND status = 'running' AND archived_at IS NULL ORDER BY created_at DESC LIMIT 1 `).get({ cid: DISPATCH_COMPANY, from }); if (parent) { parentTaskId = parent.id; console.log(`[TaskDispatch] 🔗 Auto-detected parentTaskId: ${parentTaskId} (${from} is executing it)`); } } catch (e) { console.error(`[TaskDispatch] ⚠️ 父任务推断失败，本次不挂父: ${e.message}`); } } // 1. 写入目标 Agent 的 todo.md（公司态：`co-alpha:ceo` 与 `co-beta:ceo` 各写各的） let todoContent = readStateFile(to, 'todo.md'); // 移除空占位行 todoContent = todoContent.replace(/\| *\| *（暂无待办）.*\n?/g, ''); const newTodoRow = `| ${taskId} | ${safeTask} | ⬚ 未开始 | ${from} | ${priority} | ${now} | - |\n`; writeStateFile(to, 'todo.md', todoContent.trimEnd() + '\n' + newTodoRow); // 2. 更新发送者的 plan.md (如果有 planId) if (planId) { let planContent = readStateFile(from, 'plan.md'); planContent = planContent.replace(/\| *\| *（暂无计划）.*\n?/g, ''); const newPlanRow = `| ${planId} | ${safeTask} | ${to} | 🔄 已分发 | ${now} | taskId: ${taskId} |\n`; writeStateFile(from, 'plan.md', planContent.trimEnd() + '\n' + newPlanRow); } // 3. 更新 OKR 节点状态 (如果有 objectiveId) if (objectiveId && nodeId) { try { okrService.updateNode(objectiveId, nodeId, { status: 'in_progress', taskId, owner: to, }); console.log(`[TaskDispatch] 🎯 Objective ${objectiveId} 节点 ${nodeId} → in_progress`); } catch (err) { console.warn(`[TaskDispatch] ⚠️ Goal 更新失败: ${err.message}`); } } console.log(`[TaskDispatch] 📨 ${from} → ${to}: "${task}" (${taskId}, ${priority}${objectiveId ? `, objective: ${objectiveId}` : ''}${parentTaskId ? `, parent: ${parentTaskId}` : ''})`); // 4. 持久化 Task 元数据 (用于 task/report 回调) const executionConvId = autoExecute ? `task-${taskId}` : null; saveTaskMeta(taskId, { taskId, from, to, task: safeTask, priority, objectiveId: objectiveId || null, goalId: objectiveId || null, // 旧字段名，保留供存量读取方兼容 nodeId: nodeId || null, parentTaskId: parentTaskId || null, executionConvId, createdAt: new Date().toISOString(), status: 'dispatched', }); // 5. 异步自动触发执行 (发送 chat 消息给目标 Agent) if (autoExecute) { // 异步触发，不阻塞当前响应 setImmediate(async () => { try { const goalInfo = objectiveId ? `\n目标: ${objectiveId} → ${nodeId}` : ''; const taskMessage = `[TASK from ${from}]\n任务编号: ${taskId}\n优先级: ${priority}${goalInfo}\n\n${task}\n\n请立即执行此任务。执行完成后，如果你的 agent.md 规定了后续步骤，请在同一轮对话中继续执行，不要停下来等待上级催促！\n全部完成后使用以下命令汇报:\nnode bin/lemon.js task report --taskId ${taskId} --from ${to} --to ${from} --status done --summary "(你的完成摘要)"`; // 内部触发聊天 (通过 HTTP 自调用) selfChat(req, to, taskMessage, executionConvId, 'TaskDispatch'); } catch (err) { console.error(`[TaskDispatch] ❌ 异步触发错误: ${err.message}`); } }); } res.json({ success: true, taskId, from, to, task, priority, autoExecute, executionConvId, message: `任务已分发给 ${to}` + (autoExecute ? '，正在异步执行' : ''), }); } catch (err) { console.error('[TaskDispatch] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * POST /api/agent/task/report — Agent 汇报任务完成 * * body: { * taskId: 'T-xxx', * from: 'agent-creator', // 执行者 * to: 'master-001', // 上级（任务发送者） * status: 'done' | 'failed' | 'blocked', * summary: '已成功创建 poet Agent', * outputs: ['poet/agent.md', 'poet/soul.md'], // 产出文件 (可选) * } */ router.post('/task/report', (req, res) => { try { const { taskId, from, to, status = 'done', summary = '', outputs = [], planId, nodeId } = req.body; const objectiveId = req.body.objectiveId || req.body.goalId || null; if (!taskId || !from || !to) return res.status(400).json({ error: 'taskId, from, to are required' }); const now = new Date().toISOString().slice(0, 16).replace('T', ' '); const statusEmoji = status === 'done' ? '✅ 完成' : status === 'failed' ? '❌ 失败' : '⏸ 阻塞'; const safeSummary = (summary || '').replace(/\n/g, '
').replace(/\|/g, '|'); // 1. 更新执行者的 todo.md (标记完成) —— 落点必须与 dispatch 写侧同一份推导， // 否则回报会去改另一个文件里同名的行（或者什么都改不到），而且不报错。 let todoContent = readStateFile(from, 'todo.md'); if (todoContent) { // 找到 taskId 所在行，更新状态和检查时间 todoContent = todoContent.replace( new RegExp(`(\\| *${taskId} *\\|[^|]*\\|) *[^|]* *(\\|.*)$`, 'm'), `$1 ${statusEmoji} $2`.replace(/\| *- *\|$/, `| ${now} |`) ); writeStateFile(from, 'todo.md', todoContent); } // 2. 更新上级的 plan.md (标记完成) — 使用行级匹配，兼容各种格式 { const content = readStateFile(to, 'plan.md'); if (content) { const lines = content.split('\n'); let updated = false; for (let i = 0; i < lines.length; i++) { const line = lines[i]; // 匹配包含 taskId 的行（无论格式如何） if (line.includes(taskId) || (planId && line.includes(planId))) { // 替换状态列：找到第4个 | 分隔的列（状态列），替换为新状态 const cells = line.split('|').map(c => c.trim()); if (cells.length >= 5) { // cells: ['', '#', '子任务', '责任人', '状态', '上次检查', '备注', ''] cells[4] = ` ${statusEmoji} `; cells[5] = ` ${now} `; if (safeSummary && cells.length >= 7) { cells[6] = ` ${safeSummary} `; } lines[i] = cells.join(' | ').replace(/\*\*/g, ''); // 去掉粗体 updated = true; console.log(`[TaskReport] 📝 更新 plan.md 第 ${i + 1} 行: ${statusEmoji}`); } } } if (updated) { writeStateFile(to, 'plan.md', lines.join('\n')); } else { console.warn(`[TaskReport] ⚠️ 未在 ${to}/plan.md 中找到 taskId=${taskId} 的行`); } } } // 3. 更新 OKR 节点状态 (如果有 objectiveId) if (objectiveId && nodeId) { try { const goalStatus = status === 'done' ? 'done' : status === 'failed' ? 'failed' : 'blocked'; okrService.updateNode(objectiveId, nodeId, { status: goalStatus, summary, output: outputs.join(', '), }); console.log(`[TaskReport] 🎯 Objective ${objectiveId} 节点 ${nodeId} → ${goalStatus}`); } catch (err) { console.warn(`[TaskReport] ⚠️ Goal 更新失败: ${err.message}`); } } console.log(`[TaskReport] 📋 ${from} → ${to}: ${taskId} = ${status} (${summary})`); // 4. Pipeline↔Task 桥接：自动推进 Pipeline 步骤 let pipelineAdvanced = false; try { const DATA_ROOT = path.resolve(__dirname, '../../data'); const { PipelineService } = require('../services/pipeline-service'); // 扫描所有公司的 Pipeline Runs const companyDirs = fs.readdirSync(DATA_ROOT, { withFileTypes: true }) .filter(d => d.isDirectory() && !['agents', 'skills', 'templates'].includes(d.name) && fs.existsSync(path.join(DATA_ROOT, d.name, 'company.json'))); for (const compDir of companyDirs) { const pSvc = new PipelineService(compDir.name); // from 可能是 "companyId:agentId" 格式，Pipeline step.agent 可能只是 "agentId" const agentShortId = from.includes(':') ? from.split(':')[1] : from; const activeStep = pSvc.findActiveRunStep(agentShortId) || pSvc.findActiveRunStep(from); if (activeStep) { const advanceStatus = status === 'done' ? 'done' : 'failed'; pSvc.advanceStep(activeStep.runId, activeStep.stepIndex, { output: { summary, outputs: outputs || [], taskId }, status: advanceStatus, }); pipelineAdvanced = true; console.log(`[TaskReport→Pipeline] 🔗 自动推进 ${compDir.name} Pipeline Run ${activeStep.runId} 步骤 ${activeStep.stepIndex} → ${advanceStatus}`); break; } } } catch (pipeErr) { console.warn(`[TaskReport→Pipeline] ⚠️ Pipeline 桥接失败: ${pipeErr.message}`); } // 5. 自动回调：通知上级 Agent 任务已完成，唤醒其继续执行后续步骤 if (status === 'done' || status === 'failed') { setImmediate(async () => { try { // 读取 task 元数据获取上下文 const taskMeta = loadTaskMeta(taskId); // 💰 Token 优化：尝试复用目标 Agent 的已有 session，避免创建新 CLI 进程 let callbackConvId = `callback-${taskId}`; for (const [key, entry] of sessions.entries()) { if (key.startsWith(`${to}:`) && entry.protocol.isConnected) { callbackConvId = entry.convId; console.log(`[TaskCallback] ♻️ 复用 ${to} 已有 session: ${callbackConvId}`); break; } } const statusText = status === 'done' ? '✅ 已完成' : '❌ 失败'; const callbackMessage = `[TASK_CALLBACK]\n任务编号: ${taskId}\n执行者: ${from}\n状态: ${statusText}\n摘要: ${summary || '(无摘要)'}\n产出文件: ${outputs.length > 0 ? outputs.join(', ') : '(无)'}\n${taskMeta ? `原始任务: ${taskMeta.task}` : ''}\n\n**去重提醒**：在 dispatch 任何新任务前，请务必先检查目标 Agent 的 todo.md 和全局任务状态。如果同一目标已有相似任务在进行中，直接跳过即可，绝对禁止重复分发。\n\n请检查上述任务的执行结果，然后根据你的 plan.md 继续执行下一步计划。`; selfChat(req, to, callbackMessage, callbackConvId, 'TaskCallback'); // 更新 task 元数据 if (taskMeta) { taskMeta.status = status; taskMeta.completedAt = new Date().toISOString(); taskMeta.summary = summary; saveTaskMeta(taskId, taskMeta); } } catch (err) { console.error(`[TaskCallback] ❌ 回调触发错误: ${err.message}`); } }); } res.json({ success: true, taskId, status, objectiveId, goalId: objectiveId, nodeId, message: `任务 ${taskId} 已汇报为 ${status}` }); } catch (err) { console.error('[TaskReport] Error:', err.message); res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // POST /api/agent/:agentId/task/transition — 强制流转与记忆断舍离 (Context Flush) // ═══════════════════════════════════════ router.post('/:agentId/task/transition', (req, res) => { try { const { phase, artifact } = req.body; const agentId = req.params.agentId; const agentDir = getBodyDir(agentId); if (!fs.existsSync(agentDir)) return res.status(404).json({ error: 'Agent not found' }); console.log(`[TaskTransition] 🔄 ${agentId} 正在跨越阶段至: ${phase}`); // 1. 查找当前的活动会话 ID let activeConvId = null; let sessionKey = null; for (const [k, v] of sessions.entries()) { if (v.agentId === agentId) { activeConvId = v.convId; sessionKey = k; break; } } // 2. Phase 3 核心: 事前阻断阀 (Human-in-the-loop) // 拦截 P0 任务由 plan -> execute 的跳转，强制挂起并申请 Master 审批 if (phase === 'execute' && activeConvId && activeConvId.startsWith('task-')) { const taskId = activeConvId.substring(5); const taskMeta = loadTaskMeta(taskId); if (taskMeta && taskMeta.priority === 'P0') { console.log(`[TaskTransition] 🛑 拦截 P0 任务进入 Execute，挂起等待人类审批 (taskId: ${taskId})`); // 更新 OKR 节点状态为 pending_approval const tmObjId = taskMeta.objectiveId || taskMeta.goalId; if (tmObjId && taskMeta.nodeId) { try { okrService.updateNode(tmObjId, taskMeta.nodeId, { status: 'pending_approval' }); } catch (err) {} } // 通知 Master 审批 setImmediate(() => { const masterMsg = `[⚠️ 事前审批请求]\n你的下属 ${agentId} 已经完成了 P0 级任务 ${taskId} 的规划阶段。\n现在它正准备加载大威力执行权限 (Execute) 落地方案。\n\n请你阅读其产生的 ${artifact || '_plan.md'}，并在确认无误后通过指令手动放行它。`; selfChat(req, 'master', masterMsg, `approval-${taskId}`, 'TaskTransition'); }); // 掐断当前进程使其休眠，等待重新唤醒 const v = sessions.get(sessionKey); if (v && v.protocol) v.protocol.disconnect(); sessions.delete(sessionKey); return res.json({ success: true, message: '已拦截。该任务为 P0 级别，必须等待 Master 审批放行后才能进入执行环节！' }); } } // 若未被阻断，更新 OKR 为进行中子状态 (如 researching, executing等) if (activeConvId && activeConvId.startsWith('task-')) { const taskId = activeConvId.substring(5); const taskMeta = loadTaskMeta(taskId); const tmObjId2 = taskMeta && (taskMeta.objectiveId || taskMeta.goalId); if (tmObjId2 && taskMeta.nodeId) { try { const subStatus = phase === 'research' ? 'researching' : phase === 'plan' ? 'planning' : phase === 'execute' ? 'executing' : phase === 'verify' ? 'verifying' : 'in_progress'; okrService.updateNode(tmObjId2, taskMeta.nodeId, { status: subStatus }); } catch (err) {} } } // 3. Hot Swap 动态换脑: 把 main_skill 替换为 meta-* 技能 const sourceSkill = path.resolve(__dirname, `../../system/skills/meta-${phase}/SKILL.md`); const targetSkill = path.join(agentDir, '.claude', 'skills', 'main_skill', 'SKILL.md'); if (fs.existsSync(sourceSkill)) { if (!fs.existsSync(path.dirname(targetSkill))) fs.mkdirSync(path.dirname(targetSkill), { recursive: true }); fs.copyFileSync(sourceSkill, targetSkill); console.log(`[TaskTransition] 🧠 ${agentId} 动态技能树已切换为 meta-${phase}`); } else { // 27-P3B W11：`system/skills/` 整个目录不在仓库里，这个 existsSync 恒 false。 // 以前这里没有 else 分支 —— 换装静默不发生，连日志都不打，于是 // 27-P3/P4 §九bis「EXECUTE 阶段才拿破坏性权限」看起来像在生效，其实一次都没生效过。 // 本轮按 W11 方案 (b) 如实降级：不补技能包，但把空转喊出来。 // 需要强调的是：**补上技能包也不会关上权限那道闸** —— 工具权限来自 // config.json 的 cli.toolPolicy（见本文件 getOrCreateProtocol），与 phase 无关， // Hot Swap 换的只是拼进提示词的那段文本。详见 §九bis。 console.warn( `[TaskTransition] ⚠️ ${agentId} 流转到 ${phase}，但元技能包不存在（${sourceSkill}）——` + '本次流转**未换装**，四段式当前只有 Context Flush 生效；' + '工具权限仍由 config.json 的 cli.toolPolicy 决定，与阶段无关（27-P3B W11）', ); } // 4. Context Flush 记忆断舍离: 掐断当前运行中的 ACP 进程，清空历史消息 if (sessionKey) { const v = sessions.get(sessionKey); if (v && v.protocol) v.protocol.disconnect(); sessions.delete(sessionKey); } if (activeConvId) { const dir = getConvDir(agentId, activeConvId); // 清空长周期的幻觉来源 fs.writeFileSync(path.join(dir, 'history.json'), '[]', 'utf-8'); logMessageChunk(agentId, activeConvId, `\n\n[SYSTEM] 🛑 Context Flush 触发！记忆已断开，阶段切换至 ${phase}\n\n`); // 3. 构建起步上下文: 告诉 Agent 它的前序成就 let artifactContent = ''; if (artifact) { // artifact 可能带有 'workspace/' 前缀，尝试解析。 // 任务态目录排在最前：cwd 下沉之后（裁决 A），agent 上一阶段留下的草稿就写在那里； // 只找本体目录的话这里恒读到空串，而空串会被上面的模板静默吞掉、不报错。 // artifact 来自请求体，所以每个候选都过 resolveInside：不然 `../../../.env` // 就是一条「把仓库里任意文件读进提示词」的通路。 const flushTaskDir = getTaskDir(agentId, activeConvId); const possiblePaths = [ resolveInside(flushTaskDir, artifact), resolveInside(flushTaskDir, path.basename(artifact)), resolveInside(agentDir, artifact), resolveInside(agentDir, path.basename(artifact)), resolveInside(agentDir, path.join('workspace', path.basename(artifact))), ].filter(Boolean); for (const p of possiblePaths) { if (fs.existsSync(p)) { artifactContent = fs.readFileSync(p, 'utf-8'); break; } } } const flushMsg = `【系统指令：强制状态切换】 你的生命周期已成功切入新阶段：**${phase.toUpperCase()}**。 你之前的对话记忆已被完全抹除防范幻觉，你的系统技能已获得 ${phase} 的授权限制。 ${artifactContent ? `这是上一阶段你或你的前置节点留下的交付物（Artifact），你必须严格参照此文件行动，不再另行发散：\n\n\`\`\`\n${artifactContent}\n\`\`\`` : ''} 请开始你的新阶段工作！`; // 写入一条用户指令开启新的清爽对话 appendHistory(agentId, activeConvId, { role: 'user', content: flushMsg, messageId: `msg-${Date.now()}-flush`, timestamp: new Date().toISOString() }); // 4. 发送唤醒指令让他立刻基于新脑子行动 setImmediate(() => { selfChat(req, agentId, '请确认你已收到系统的新阶段挂载指令，并立即执行你当前阶段本分的任务！', activeConvId, 'TaskTransition'); }); } res.json({ success: true, message: `已成功进行 Context Flush 并切换至 ${phase} 阶段` }); } catch (err) { console.error('[TaskTransition] Error:', err.message); res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // POST /api/agent/task/reject — COO 质量门控：驳回任务 // ═══════════════════════════════════════ /** * POST /api/agent/task/reject — 驳回 Pipeline 步骤，要求 Agent 重做 * * body: { * companyId: 'lemonv', * runId: 'R-xxx', * stepIndex: 2, * reason: '文章质量不达标，缺少数据支撑', * reviewedBy: 'coo', * } */ router.post('/task/reject', (req, res) => { try { const { companyId, runId, stepIndex, reason = '', reviewedBy = 'coo' } = req.body; if (!companyId || !runId || stepIndex === undefined) { return res.status(400).json({ error: 'companyId, runId, stepIndex are required' }); } const { PipelineService } = require('../services/pipeline-service'); const pSvc = new PipelineService(companyId); const run = pSvc.rejectStep(runId, Number(stepIndex), { reason, reviewedBy }); const step = run.steps[Number(stepIndex)]; const agentId = step.agent; // 自动触发 Agent 重新执行（如果有 agent） if (agentId) { setImmediate(() => { const rejectMessage = [ `[PIPELINE TASK - REDO]`, `⚠️ 你的上次产出被 ${reviewedBy} 驳回。`, `驳回原因: ${reason}`, `这是第 ${step.rejections?.length || 1} 次重做。`, ``, `请认真阅读以上反馈，改进你的输出质量后重新提交。`, `完成后调用:`, `node bin/lemon.js task report --taskId redo-${runId}-${stepIndex} --from ${agentId} --to ${reviewedBy} --status done --summary "(改进后的完成摘要)"`, ].join('\n'); selfChat(req, agentId, rejectMessage, `redo-${runId}-step-${stepIndex}`, 'TaskReject'); }); } console.log(`[TaskReject] 🔙 ${companyId} Run ${runId} 步骤 ${stepIndex} 被 ${reviewedBy} 驳回: ${reason}`); res.json({ success: true, runId, stepIndex, reason, reviewedBy, rejectionCount: step.rejections?.length || 1, message: `步骤 ${stepIndex} 已驳回，Agent ${agentId} 将重新执行`, }); } catch (err) { console.error('[TaskReject] Error:', err.message); res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // POST /api/agent/human-request — Human-in-the-loop: API Key 申请等 // ═══════════════════════════════════════ router.post('/human-request', (req, res) => { try { const { from, type, key, reason } = req.body; if (!from || !type || !key) { return res.status(400).json({ error: 'from, type, key are required' }); } console.log(`[HumanRequest] 🙋 ${from} 遇到需要人类介入的请求: ${type} - ${key} (${reason})`); // 1. 同步写入根目录下的 .env 以备人类填入 const envPath = path.resolve(__dirname, '../../.env'); let envContent = ''; try { envContent = fs.readFileSync(envPath, 'utf-8'); } catch {} // 如果还没包含这个 key，追加进去 if (!envContent.includes(key + '=')) { const appendStr = `\n# [${new Date().toISOString().substring(0, 10)}] Requested by ${from}: ${reason}\n${key}=\n`; fs.appendFileSync(envPath, appendStr, 'utf-8'); console.log(`[HumanRequest] 🔑 已将 ${key} 追加到 .env 等待配置`); } // 2. 写入 master 的 To-Do List const masterWorkspace = path.resolve(__dirname, '../../system/agents/master/workspace'); if (!fs.existsSync(masterWorkspace)) fs.mkdirSync(masterWorkspace, { recursive: true }); const todoPath = path.join(masterWorkspace, 'human_todo.md'); const todoStr = `- [ ] **${type.toUpperCase()}**: Agent \`${from}\` 需要 \`${key}\` (${reason}) — ` + `_请在根目录 \`.env\` 中配置后打钩_\n`; fs.appendFileSync(todoPath, todoStr, 'utf-8'); // 3. 通知 Master (非阻塞) setImmediate(() => { const masterMsg = [ `[🙋 HUMAN INTERVENTION REQUIRED]`, `你的下属 ${from} 遇到了系统级的阻碍。`, `请求类型: ${type}`, `资源名: ${key}`, `原因: ${reason}`, ``, `我已经将其记录到了 workspace/human_todo.md 和根目录的 .env 文件中。`, `请敦促人类 Master 提供该 Key，或者你评估是否可以使用备用方案。` ].join('\n'); selfChat(req, 'master', masterMsg, `human-req-${Date.now().toString(36)}`, 'HumanRequest'); }); res.json({ success: true, message: `已将请求上报给 Master 并记录至 Human To-Do List` }); } catch (err) { console.error('[HumanRequest] Error:', err.message); res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // 文件上传 API // ═══════════════════════════════════════ const uploadStorage = multer.diskStorage({ destination: (req, file, cb) => { const convId = req.body.conversationId || `conv-${Date.now().toString(36)}`; req._uploadConvId = convId; // 传递给后续逻辑 const uploadDir = getUploadDir(req.params.agentId, convId); fs.mkdirSync(uploadDir, { recursive: true }); cb(null, uploadDir); }, filename: (req, file, cb) => { const safe = sanitizeFilename(file.originalname); cb(null, safe); }, }); const upload = multer({ storage: uploadStorage, limits: { fileSize: 10 * 1024 * 1024, files: 20 }, // 10MB per file, 20 files max }); // POST /api/agent/:agentId/upload — 上传文件 router.post('/:agentId/upload', upload.array('files', 20), (req, res) => { try { const agentId = req.params.agentId; const agentDir = getBodyDir(agentId); if (!fs.existsSync(agentDir)) return res.status(404).json({ error: 'Agent not found' }); const convId = req._uploadConvId || req.body.conversationId; const uploaded = (req.files || []).map(f => { const ext = path.extname(f.originalname).toLowerCase(); return { name: f.filename, originalName: f.originalname, size: f.size, type: isTextFile(ext) ? 'text' : isImageFile(ext) ? 'image' : 'binary', path: `uploads/${convId}/${f.filename}`, }; }); console.log(`[Upload] 📎 ${agentId}/${convId}: ${uploaded.length} file(s) uploaded`); res.json({ conversationId: convId, uploaded }); } catch (err) { res.status(500).json({ error: err.message }); } }); // GET /api/agent/:agentId/uploads/:convId — 列出已上传文件 router.get('/:agentId/uploads/:convId', (req, res) => { try { const uploadDir = getUploadDir(req.params.agentId, req.params.convId); if (!fs.existsSync(uploadDir)) return res.json([]); const files = fs.readdirSync(uploadDir).filter(f => !f.startsWith('.')).map(f => { const fp = path.join(uploadDir, f); const stat = fs.statSync(fp); const ext = path.extname(f).toLowerCase(); return { name: f, size: stat.size, type: isTextFile(ext) ? 'text' : isImageFile(ext) ? 'image' : 'binary', uploadedAt: stat.mtime.toISOString(), }; }); res.json(files); } catch (err) { res.status(500).json({ error: err.message }); } }); // DELETE /api/agent/:agentId/uploads/:convId/:filename — 删除单个文件 router.delete('/:agentId/uploads/:convId/:filename', (req, res) => { try { const filePath = path.join(getUploadDir(req.params.agentId, req.params.convId), sanitizeFilename(req.params.filename)); if (fs.existsSync(filePath)) fs.unlinkSync(filePath); res.json({ success: true }); } catch (err) { res.status(500).json({ error: err.message }); } }); // GET /api/agent/:agentId/uploads/:convId/:filename — 下载/预览文件 router.get('/:agentId/uploads/:convId/:filename', (req, res) => { try { const filePath = path.join(getUploadDir(req.params.agentId, req.params.convId), sanitizeFilename(req.params.filename)); if (!fs.existsSync(filePath)) return res.status(404).json({ error: 'File not found' }); res.sendFile(filePath); } catch (err) { res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // POST /api/agent/:agentId/chat — SSE 流式对话 // ═══════════════════════════════════════ // 🔒 Agent 级并发互斥锁 — 同一 Agent 同时只允许一个活跃 chat session // 同一 convId 的多轮对话不受限制（ACP 进程复用） // 32 号文档方案 B：这把锁挪到了共享模块 agent-session-lock.js，company.js 的招聘装配 // 重试逻辑也要用同一把——两条路径各自维护一份的话，各自都以为自己没有并发。 const { agentChatLocks, AGENT_LOCK_TIMEOUT, waitForLockRelease } = require('../src/core/agent-session-lock'); // 40 号文档方案一：Claude CLI 一次 agent_message_chunk 可能带几十到几百字符，SSE 原样 // 转发就是"一坨一坨蹦出来"，没有打字机效果。这里按固定步长切成小片再各发一条 SSE 事件—— // 只影响推给前端的粒度，`logMessageChunk` 落盘的仍是整段，不受影响。 const TYPEWRITER_CHUNK_SIZE = 10; function sliceForTypewriter(text, size = TYPEWRITER_CHUNK_SIZE) { const pieces = []; for (let i = 0; i < text.length; i += size) pieces.push(text.slice(i, i + size)); return pieces; } router.post('/:agentId/chat', async (req, res) => { const { message, conversationId: reqConvId } = req.body; if (!message) return res.status(400).json({ error: 'message is required' }); const agentId = req.params.agentId; const agentDir = getBodyDir(agentId); if (!fs.existsSync(agentDir)) return res.status(404).json({ error: 'Agent not found' }); const convId = reqConvId || `conv-${Date.now().toString(36)}`; // 🔒 并发互斥检查：同一 Agent 的并发请求被拒绝——无论 convId 是否相同。 // 40 号文档风险①：这里原来只挡"不同 convId"，同一 convId 的第二个请求会直接 // 覆盖锁往下跑，等于两个循环同时喂同一个 ACP 会话（未定义行为）。此前一直靠前端 // textarea 的 disabled 属性挡住，从没被服务端真正堵过；"打断并发送"要解锁输入框， // 这条路径就变得可达了，必须在这里把闸门补上。 const existingLock = agentChatLocks.get(agentId); if (existingLock) { const elapsed = Date.now() - existingLock.startedAt; if (elapsed < AGENT_LOCK_TIMEOUT) { console.log(`[AgentChat] 🔒 拒绝 ${agentId} 并发请求 — 已有活跃会话 ${existingLock.convId} (${(elapsed / 1000).toFixed(0)}s)`); return res.status(429).json({ error: `Agent ${agentId} 正在执行另一个会话 (${existingLock.convId})，请等待完成`, activeConvId: existingLock.convId, elapsedMs: elapsed, sameConversation: existingLock.convId === convId, }); } // 超时释放 console.warn(`[AgentChat] ⚠️ 强制释放超时锁: ${agentId} / ${existingLock.convId}`); agentChatLocks.delete(agentId); } agentChatLocks.set(agentId, { convId, startedAt: Date.now() }); const msgId = `msg-${Date.now()}-${randomUUID().substring(0, 8)}`; const sendSSE = setupSSE(res); const identity = loadIdentity(agentDir); const kind = resolveKind(req.body); // 确保 meta.json 存在 let meta = loadMeta(agentId, convId); if (!meta) { meta = { conversationId: convId, agentId, title: kind ? kind[0] : message.slice(0, 50), type: kind ? kind[1] : 'chat', createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), status: 'running', stats: { messageCount: 0, toolCallCount: 0 }, }; } if (!meta.stats) meta.stats = { messageCount: 0, toolCallCount: 0 }; meta.stats.messageCount = (meta.stats.messageCount || 0) + 1; meta.status = 'running'; saveMeta(agentId, convId, meta); // 写用户消息到两层存储 logMessage(agentId, convId, 'user', message); appendHistory(agentId, convId, { role: 'user', content: message, messageId: msgId, timestamp: new Date().toISOString() }); sendSSE('session', { conversationId: convId, agentId, messageId: msgId }); sendSSE('thinking', { message: `${identity.emoji} ${identity.name} 思考中...` }); let fullOutput = ''; // 记录 Agent 开始回复的标识头 logMessageChunk(agentId, convId, `[${new Date().toISOString()}] [AGENT]\n`); try { // 获取或创建 ACP 会话（per conversation） const entry = await getOrCreateProtocol(agentId, convId); const isNewSession = entry.messageCount === 0; // 构建 prompt（自动处理首次/后续/恢复三种场景） let prompt = buildPrompt(agentDir, agentId, convId, message, isNewSession); entry.messageCount++; let loopCount = 0; while (loopCount < 10) { loopCount++; fullOutput = ''; // Reset fullOutput for this turn // 监听流式事件 let currentToolInput = ''; const onUpdate = (payload) => { const update = payload.update; if (update?.sessionUpdate === 'agent_message_chunk' && update.content?.text) { console.log(`[SSE-DEBUG] 📤 agent_output (${update.content.text.length} chars)`); fullOutput += update.content.text; logMessageChunk(agentId, convId, update.content.text); for (const piece of sliceForTypewriter(update.content.text)) { sendSSE('agent_output', { agent: agentId, message: piece }); } } else if (update?.sessionUpdate === 'agent_thought_chunk' && update.content?.text) { logMessageChunk(agentId, convId, update.content.text); sendSSE('thinking', { message: update.content.text }); } }; const onAcpEvent = (event) => { // 透传所有 ACP 事件给前端（完整可观测性） sendSSE('acp_raw', { method: event.method, params: event.params, ts: event.timestamp }); if (event.method === 'claude/tool_use_start') { currentToolInput = ''; const toolName = event.params.name; const toolMsg = `\n🔧 [TOOL_CALL] ${toolName} (ID: ${event.params.tool_use_id || 'unknown'})\n`; fullOutput += toolMsg; logMessageChunk(agentId, convId, toolMsg); sendSSE('tool_call', { tool: toolName, toolId: event.params.tool_use_id, message: `🔧 调用工具: ${toolName}` }); meta.stats.toolCallCount = (meta.stats.toolCallCount || 0) + 1; } else if (event.method === 'claude/stream/content_block_stop') { // 工具调用结束 — 输出积攒的输入参数 if (currentToolInput) { const inputLog = ` 输入: ${currentToolInput.slice(0, 500)}${currentToolInput.length > 500 ? '...' : ''}\n`; fullOutput += inputLog; logMessageChunk(agentId, convId, inputLog); sendSSE('tool_input', { input: currentToolInput }); currentToolInput = ''; } } else if (event.method === 'session/result') { if (event.params?.duration_ms) { sendSSE('session_cost', { cost: event.params.cost_usd, duration: event.params.duration_ms, usage: event.params.usage }); // Phase 5: Economic Audit - 记录账单 (Token Profiling) JSON lines try { let billingDir; if (agentId && agentId.includes(':')) { const compId = agentId.split(':')[0]; billingDir = path.resolve(__dirname, `../../data/companies/${compId}/logs`); } else { billingDir = path.resolve(__dirname, '../../system/agents/master/logs'); } if (!fs.existsSync(billingDir)) fs.mkdirSync(billingDir, { recursive: true }); const logEntry = { timestamp: new Date().toISOString(), agentId, convId, model: event.params.model || 'unknown', costUsd: event.params.cost_usd || 0, durationMs: event.params.duration_ms, usage: event.params.usage || { input: 0, output: 0 } }; const billingDate = new Date().toISOString().substring(0, 10); fs.appendFileSync(path.join(billingDir, `billing-${billingDate}.log`), JSON.stringify(logEntry) + '\n', 'utf-8'); } catch (err) {} } if (!fullOutput && event.params?.result) { fullOutput = event.params.result; sendSSE('agent_output', { agent: agentId, message: event.params.result }); } } else if (event.method?.startsWith('claude/')) { // 捕获其他 claude 事件（tool_result 等） const innerEvent = event.params; // input_json_delta — 工具的 JSON 输入增量 if (innerEvent?.delta?.type === 'input_json_delta' && innerEvent.delta.partial_json) { currentToolInput += innerEvent.delta.partial_json; } // tool_result — 工具执行结果 if (innerEvent?.type === 'tool_result' || innerEvent?.content_block?.type === 'tool_result') { const result = innerEvent.content || innerEvent.content_block?.content || ''; const resultStr = typeof result === 'string' ? result : JSON.stringify(result); const truncated = resultStr.slice(0, 1000) + (resultStr.length > 1000 ? `...(${resultStr.length} chars)` : ''); const resultLog = ` 结果: ${truncated}\n`; fullOutput += `\n` + resultLog; logMessageChunk(agentId, convId, resultLog); sendSSE('tool_result', { result: truncated }); } } }; entry.protocol.on('session_update', onUpdate); entry.protocol.on('acp_event', onAcpEvent); // 监听 stderr / error / disconnect — 将 Claude CLI 错误实时推送到前端 const onStderr = (msg) => { console.warn(`[AgentChat] stderr from ${agentId}:`, msg.substring(0, 300)); sendSSE('acp_raw', { method: 'claude/stderr', params: { message: msg }, ts: Date.now() }); // 如果包含严重错误关键词，额外发一条 error 事件 if (/error|rate.?limit|429|403|timeout|ECONNREFUSED|failed/i.test(msg)) { sendSSE('error', { message: `⚠️ Claude CLI: ${msg.substring(0, 500)}` }); } }; const onDisconnected = ({ code, signal }) => { if (code !== 0) { sendSSE('error', { message: `Claude CLI 异常退出 (code=${code}, signal=${signal})` }); } }; entry.protocol.on('stderr', onStderr); entry.protocol.on('disconnected', onDisconnected); // 发出状态以便前端展示不卡壳 sendSSE('agent_output', { agent: agentId, message: `_[请求已发送，等待底层大模型响应...]_\\\n\\\n` }); // 发送 prompt 并等待完成 await entry.protocol.sendPrompt(prompt); // 清理监听 entry.protocol.off('session_update', onUpdate); entry.protocol.off('acp_event', onAcpEvent); entry.protocol.off('stderr', onStderr); entry.protocol.off('disconnected', onDisconnected); // 持久化当前轮次的回复 if (fullOutput) { logMessageChunk(agentId, convId, '\n\n'); appendHistory(agentId, convId, { role: 'assistant', content: fullOutput, messageId: `msg-${Date.now()}-${loopCount}`, timestamp: new Date().toISOString() }); } // 解析是否带有 [CALL: sub_agent] 标记 const calls = Parser.parseCalls(fullOutput, { allowParallel: true, agents: fullSystemConfig.agents }); if (calls.length === 0) { break; // 没有后续调用，当前回合结束 } // 存在 Sub-Agent 嵌套调用 logMessageChunk(agentId, convId, `\n\n[System] 执行子 Agent 编排调度（并行）...\n`); sendSSE('thinking', { message: `🔄 分析到系统编排指令，正在并行启动 ${calls.length} 个子代理...` }); const resultsText = []; const MAX_CONCURRENCY = 2; // Checkpoint: 保存当前调度计划 const checkpoint = { loopCount, calls: calls.map(c => ({ agent: c.agent, task: c.task })), completedCalls: [], timestamp: new Date().toISOString() }; const checkpointPath = path.join(getConvDir(agentId, convId), 'checkpoint.json'); fs.writeFileSync(checkpointPath, JSON.stringify(checkpoint, null, 2)); // 分批并发执行 (最多同时跑 MAX_CONCURRENCY 个子进程) for (let i = 0; i < calls.length; i += MAX_CONCURRENCY) { const batch = calls.slice(i, i + MAX_CONCURRENCY); await Promise.all(batch.map(async (call) => { const callId = `sub-${randomUUID().substring(0, 6)}`; const subLogPath = path.join(getConvDir(agentId, convId), `${callId}-${call.agent}.log`); // 初始化子 Agent 独立日志文件 const subLogHeader = `[${new Date().toISOString()}] [SUB_AGENT: ${call.agent}] [CALL_ID: ${callId}]\n` + `任务:\n${call.task}\n\n--- 执行输出 ---\n`; fs.writeFileSync(subLogPath, subLogHeader); sendSSE('sub_agent', { event: 'start', callId, agent: call.agent, message: `🚀 开始执行: ${call.agent}...` }); logMessageChunk(agentId, convId, `\n--- 调度 [${callId}]: ${call.agent} ---\n任务: ${call.task}\n\n`); try { let res; // ═══ 硬编码拦截: meta_creator → 直接调用 AgentScaffold 父类 ═══ if (call.agent === 'meta_creator') { res = await executeMetaCreator(call.task, agentId, convId, sendSSE, callId); // 写入子 Agent 日志 fs.appendFileSync(subLogPath, res.output + '\n'); } else { res = await agentManager.callAgent(call.agent, call.task, { metadata: { parentConvId: convId, layer: 'sub' }, onData: (chunk) => { sendSSE('sub_agent', { event: 'chunk', callId, agent: call.agent, chunk }); // 实时追加到子 Agent 独立日志 fs.appendFileSync(subLogPath, chunk); } }); } // 写入最终结果到子 Agent 日志 fs.appendFileSync(subLogPath, `\n\n--- 执行结果 ---\n[${new Date().toISOString()}] STATUS: SUCCESS\n输出长度: ${(res.output || '').length} 字符\n`); resultsText.push(`[Sub Agent ${call.agent} (${callId}) 返回结果]\n${res.output}\n[/Sub Agent ${call.agent} (${callId}) 返回结果]`); logMessageChunk(agentId, convId, `✅ 子 Agent ${call.agent} [${callId}] 执行成功。\n`); sendSSE('sub_agent', { event: 'done', callId, agent: call.agent, status: 'success', message: `✅ [${call.agent}] 执行完毕.` }); // Checkpoint: 标记为完毕 checkpoint.completedCalls.push({ callId, agent: call.agent, status: 'success' }); } catch (subErr) { // 写入错误到子 Agent 日志 fs.appendFileSync(subLogPath, `\n\n--- 执行结果 ---\n[${new Date().toISOString()}] STATUS: ERROR\n错误: ${subErr.message}\n${subErr.stack || ''}\n`); resultsText.push(`[Sub Agent ${call.agent} (${callId}) 执行失败]\n${subErr.message}\n[/Sub Agent ${call.agent} (${callId}) 执行失败]`); logMessageChunk(agentId, convId, `❌ 子 Agent ${call.agent} [${callId}] 失败: ${subErr.message}\n`); sendSSE('sub_agent', { event: 'done', callId, agent: call.agent, status: 'error', message: `❌ [${call.agent}] 抛出异常: ${subErr.message}` }); checkpoint.completedCalls.push({ callId, agent: call.agent, status: 'error', error: subErr.message }); } // 每批次后更新 checkpoint fs.writeFileSync(checkpointPath, JSON.stringify(checkpoint, null, 2)); })); } // 构造下一次反思与后续动作的 Prompt prompt = ` ## Sub Agent 执行结果反馈 ${resultsText.join('\n\n')} --- 请对本次执行结果进行 Reflection 反思，并在确定所有步骤完成时，记得在对话结尾加上 [RESULT: DONE]。 `; // 更新消息计数和记录 User 视角的 system 反馈消息 entry.messageCount++; logMessageChunk(agentId, convId, `\n[${new Date().toISOString()}] [SYSTEM FEEDBACK]\n${prompt}\n\n`); appendHistory(agentId, convId, { role: 'user', content: prompt, messageId: `sys-${Date.now()}`, timestamp: new Date().toISOString() }); // 循环继续，下一次将会把这个反馈 prompt 发给 Innovator logMessageChunk(agentId, convId, `[${new Date().toISOString()}] [AGENT]\n`); } meta.status = 'completed'; flushLogBufferForConv(agentId, convId); // 刷出日志缓冲 saveMeta(agentId, convId, meta); // ── Phase 1 MVP: 自动保存 Session 摘要到记忆 ── try { saveSessionSummary(agentId, convId, meta, fullOutput); } catch (memErr) { console.error('[Memory] Session 摘要保存失败:', memErr.message); } // ── Phase 3: 自动反思 (toolCallCount>=3 或 duration>=60s 时触发) ── try { autoReflect(agentId, convId, meta, fullOutput); } catch (refErr) { console.error('[Memory] 反思失败:', refErr.message); } sendSSE('result', { conversationId: convId, agentId, messageId: msgId, output: fullOutput }); } catch (err) { console.error('[AgentChat] Error:', err.message); logMessageChunk(agentId, convId, `\n\n[ERROR]\n${err.message}\n\n`); flushLogBufferForConv(agentId, convId); // 刷出日志缓冲 meta.status = 'failed'; meta.error = err.message; saveMeta(agentId, convId, meta); // 出错的会话不留给下一轮复用 —— 但必须**连子进程一起杀**。 // 这里原本是 dropSession（只摘出池子、不杀进程），而 finally 里的 closeSession // 查不到条目就直接返回 false：子进程于是活着、却再没有任何引用指向它。 // 而且错的通常是 API（529 / DNS / 证书），CLI 子进程好端端的，一个都不会自己退。 // 实测 23 小时漏 38 个心跳进程，且失败的心跳**无一例外**是这条路径（status 全是 failed）。 closeSession(`${agentId}:${convId}`, '会话出错'); sendSSE('error', { message: err.message }); console.log(`[SSE-DEBUG] 🚨 sendSSE('error') called:`, err.message.substring(0, 200)); } finally { agentChatLocks.delete(agentId); res.end(); // convId 用完即弃的那几类立刻回收（留着就是纯泄漏）；其余保留会话复用能力，只上空闲 TTL。 const sessionKey = `${agentId}:${convId}`; if (kind?.[2]) closeSession(sessionKey, `${kind[1]} 结束`); else armSessionIdle(sessionKey); } }); module.exports = router; // 供 open-api.js 的 invoke 端点复用——外部 API 调用也要写进同一份会话存储， // 否则网页 AgentChat 看不到 API 触发的对话（见 docs/done/open-api-manual.md 的 invoke 端点）。 module.exports.loadMeta = loadMeta; module.exports.saveMeta = saveMeta; module.exports.appendHistory = appendHistory; module.exports.logMessage = logMessage; module.exports.flushLogBufferForConv = flushLogBufferForConv;