/** * Company Routes — 公司管理 API * * 挂载: /api/company */ const express = require('express'); const path = require('path'); const fs = require('fs'); const router = express.Router(); const { CompanyService, listCompanyAgents, buildAgentDefinitionBrief } = require('../services/company-service'); const { ScheduleService } = require('../services/schedule-service'); const orgGraphService = require('../services/org-graph-service'); // 28 批次 4:公司不再持有 agent 实体 —— 名册查 company_agents,目录一律经 agent-locator 推导 const locator = require('../services/agent-locator'); // 32 号文档方案 B:招聘装配(agent-creator 填 agent.md/soul.md)改走同进程调用 + 互斥锁 + // 重试,不再是裸 fetch 回环自己。 const AgentManager = require('../src/core/agent-manager'); const { fillAgentBodyReliably } = require('../services/agent-body-filler'); const companyService = new CompanyService(); const scheduleService = new ScheduleService(); /** * 数据根 / 公司根 —— 本文件里这两个路径此前在 9 个 handler 里各拼一份 * (6 份 `../../data/companies` + 3 份同义异名的 `../../data`),全是模块内的复制品。 * * 收敛的理由和 28 号把 13 处 `system/agents` 收进 `locator.AGENTS_ROOT` 是同一条: * **同一个事实只能有一处持有者 —— 路径也是事实**。副本的危害不是不好看,是下一次搬目录时 * 改 8 处漏 1 处,而本仓库读这类路径几乎都包在 `fs.existsSync()` 守卫里 —— 目录没了不报错, * 只是静默跳过。(这也是「按名字 grep 找不到同义异名副本」的现场:只搜 `DATA_DIR` 这个名字 * 会把它当成一个常量,其实它在本文件里指着两个不同的目录。) */ const REPO_ROOT = path.resolve(__dirname, '../..'); const DATA_ROOT = path.join(REPO_ROOT, 'data'); const COMPANIES_DIR = path.join(DATA_ROOT, 'companies'); /** * 把调用方给的相对路径解析到 baseDir 之内;越界返回 null。 * * **不能写成 `abs.startsWith(base)`** —— 那是本文件里六处文件读写检查原先的写法, * 它在「一个目录名是另一个的前缀」时会放行: * base = .../data/companies/foo * path = ../foo-evil/company.json → abs = .../data/companies/foo-evil/company.json * abs.startsWith(base) === true → 公司 foo 读到了公司 foo-evil 的文件 * PUT 那几处更糟:同样的缝隙上是跨公司**写**。 * * path.relative 才是真的在问「abs 在不在 base 里面」:越界时结果以 `..` 开头, * 指向别的盘/绝对路径时结果是绝对路径。绝对路径入参也被这一条挡住 * (path.resolve(base, '/etc/passwd') === '/etc/passwd' → relative 以 .. 开头)。 */ 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; // 指向 base 自身 if (r.startsWith('..') || path.isAbsolute(r)) return null; return abs; } /** * fire-and-forget 触发 orggraph-planning 规划这家公司的 workflow(27-P2 §4.4)。 * 不 await —— 一轮真实 Claude CLI 对话是分钟级的,调用方必须立即返回。 * 写法与本文件 /hire 里触发 agent-creator / innovator 的自调用保持一致。 */ function triggerOrgGraphPlanning(company, tag = 'OrgGraph') { const config = require('../config'); const PORT = config.port || 3000; const { id, name, goal, description } = company || {}; // ⚠️ 必须用 `${companyId}:orggraph-planning` 这个公司限定地址。 // 本体虽然只有一份,但公司态(company-context.md / 记忆 / 会话)是按公司分的, // 裸名 /api/agent/orggraph-planning/chat 会落到 `_global` 那份上,规划时看不到公司语境。 const namespacedId = `${id}:orggraph-planning`; fetch(`http://127.0.0.1:${PORT}/api/agent/${encodeURIComponent(namespacedId)}/chat`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ message: `公司 ${id}(${name})已创建。\n公司目标(company-goal):${goal || '(未填写)'}\n公司描述:${description || '(未填写)'}\n请按 orggraph-planning 工作流规划并建立这家公司的 workflow:读 system/org-graph-templates/ 候选骨架 + 可招角色清单,产出执行流图后调用 POST /api/company/${id}/org-graph 落库,再按拓扑序完成招聘。`, conversationId: `orggraph-bootstrap-${id}-${Date.now().toString(36)}`, }), }) // fetch 不会因为 4xx/5xx 而 reject,只在网络层出错时 reject。 // 不显式检查 res.ok 的话,地址写错这类问题会被完全静默吞掉(实测踩过)。 .then(res => { if (!res.ok) { console.error(`[${tag}] orggraph-planning 触发失败: HTTP ${res.status} @ ${namespacedId}`); } else { console.log(`[${tag}] ✅ 已触发 ${namespacedId} 规划 org graph`); } }) .catch(e => console.error(`[${tag}] orggraph-planning 触发失败(网络):`, e.message)); } // ═══════════════════════════════════════ // Company CRUD // ═══════════════════════════════════════ // POST /api/company/create router.post('/create', async (req, res) => { try { const { id, name, goal, description } = req.body; if (!id || !name) return res.status(400).json({ error: 'id 和 name 为必填项' }); // 校验 id 格式 (小写字母、数字、-) if (!/^[a-z0-9][a-z0-9-]*[a-z0-9]$/.test(id) && id.length < 2) { return res.status(400).json({ error: 'id 格式无效 (只允许小写字母、数字、-)' }); } const company = companyService.create({ id, name, goal, description }); // ── P2 T6:自动招聘 orggraph-planning 并异步触发组织结构规划 ── // 降级约定(27-P2 §4.4):这一步任何失败都不能影响公司创建本身, // 最差退化成一家没有 workflow 的空壳公司(等同 P2 上线前的现状)。 try { const bootstrap = companyService.hire(id, 'orggraph-planning', {}); // 只有拿到真模板(或目录已在)才值得起一轮对话:模板缺失时 hire 会退化成 base 空骨架, // 那个骨架里没有 §4.1 的八步工作流,触发了也规划不出东西。 if (bootstrap.status === 'hired_from_template' || bootstrap.status === 'already_exists') { triggerOrgGraphPlanning(company, 'Create'); } else { console.warn(`[Create] orggraph-planning 招聘未走模板分支 (status=${bootstrap.status}),跳过自动规划;模板就绪后可调 POST /api/company/${id}/org-graph/replan 补跑`); } } catch (e) { console.error('[Create] orggraph-planning 自动招聘失败(不影响公司创建):', e.message); } res.json({ success: true, company, message: '公司已创建,正在由 orggraph-planning 自动规划组织结构(异步进行中)。' }); } catch (err) { res.status(400).json({ error: err.message }); } }); // GET /api/company/list router.get('/list', (req, res) => { res.json(companyService.list()); }); // GET /api/company/templates router.get('/templates', (req, res) => { res.json(companyService.listTemplates()); }); // GET /api/company/:companyId router.get('/:companyId', (req, res) => { const company = companyService.get(req.params.companyId); if (!company) return res.status(404).json({ error: '公司不存在' }); res.json(company); }); // PUT /api/company/:companyId router.put('/:companyId', (req, res) => { const company = companyService.update(req.params.companyId, req.body); if (!company) return res.status(404).json({ error: '公司不存在' }); res.json({ success: true, company }); }); // DELETE /api/company/:companyId router.delete('/:companyId', (req, res) => { const company = companyService.delete(req.params.companyId); if (!company) return res.status(404).json({ error: '公司不存在' }); res.json({ success: true, company }); }); // POST /api/company/:companyId/shutdown router.post('/:companyId/shutdown', (req, res) => { const company = companyService.shutdown(req.params.companyId); if (!company) return res.status(404).json({ error: '公司不存在' }); // 停止该公司下所有 Agent 的心跳 try { const heartbeatService = require('../services/heartbeat-service'); const agents = companyService.listAgents(req.params.companyId); for (const agent of agents) { const namespacedId = `${req.params.companyId}:${agent.id}`; if (heartbeatService.timers.has(namespacedId)) { clearInterval(heartbeatService.timers.get(namespacedId)); heartbeatService.timers.delete(namespacedId); console.log(`[Company] 🛑 已停止 ${namespacedId} 的心跳`); } } } catch (err) { console.warn(`[Company] ⚠️ 停止心跳时出错: ${err.message}`); } res.json({ success: true, company }); }); // POST /api/company/:companyId/start — 启动公司 router.post('/:companyId/start', (req, res) => { try { const company = companyService.get(req.params.companyId); if (!company) return res.status(404).json({ error: '公司不存在' }); // 更新状态 const updated = companyService.update(req.params.companyId, { status: 'running', heartbeat: { enabled: true }, }); // 激活该公司下所有 Agent 的心跳 try { const heartbeatService = require('../services/heartbeat-service'); const agents = companyService.listAgents(req.params.companyId); for (const agent of agents) { const namespacedId = `${req.params.companyId}:${agent.id}`; heartbeatService._setupAgentHeartbeat(namespacedId); } console.log(`[Company] ✅ ${req.params.companyId} 已启动,${agents.length} 个 Agent 心跳已激活`); } catch (err) { console.warn(`[Company] ⚠️ 激活心跳时出错: ${err.message}`); } res.json({ success: true, company: updated }); } catch (err) { res.status(500).json({ error: err.message }); } }); // GET /api/company/:companyId/status — 获取公司运行状态(含 Agent 进程信息) router.get('/:companyId/status', (req, res) => { try { const company = companyService.get(req.params.companyId); if (!company) return res.status(404).json({ error: '公司不存在' }); const agents = companyService.listAgents(req.params.companyId); const heartbeatService = require('../services/heartbeat-service'); // 计算运行时长 const uptimeMs = company.status === 'running' && company.startedAt ? Date.now() - new Date(company.startedAt).getTime() : 0; const agentStatuses = agents.map(agent => { const namespacedId = `${req.params.companyId}:${agent.id}`; const hasHeartbeat = heartbeatService.timers.has(namespacedId); // 心跳配置按 `(agent, 公司)` 读:公司态优先,没设置过才回落本体的缺省。 // 走 heartbeatService 的同一个函数 —— 控制台显示的和定时器真正用的必须是同一个答案。 let heartbeatConfig = { enabled: false, intervalMinutes: 15 }; try { heartbeatConfig = heartbeatService.readHeartbeatConfig(namespacedId); } catch {} return { id: agent.id, name: agent.name, type: agent.type, status: agent.status, heartbeatActive: hasHeartbeat, heartbeatEnabled: heartbeatConfig.enabled || false, heartbeatInterval: heartbeatConfig.intervalMinutes || 15, }; }); res.json({ id: company.id, name: company.name, description: company.description, status: company.status, createdAt: company.createdAt, updatedAt: company.updatedAt, startedAt: company.startedAt || null, uptimeMs, heartbeat: company.heartbeat, agents: agentStatuses, agentCount: agents.length, activeAgentCount: agentStatuses.filter(a => a.heartbeatActive).length, }); } catch (err) { res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // Per-Agent 心跳控制 // ═══════════════════════════════════════ /** * 这家公司雇过这个角色没有(28 批次 4:判据从「公司私有目录在不在」改成「company_agents 有没有行」, * 那个目录整体删掉了,继续判目录的话下面三个心跳端点会无条件 404)。 * * 原来这里返回的是**本体** config.json 的路径,三个端点拿着它读改写 —— 心跳因此是本体级的: * A 公司调成 30 分钟,雇了同一个本体的 B 公司跟着变;更糟的是「读的时候补默认值再整份回写」, * 没人点过开关也能把 `system/agents/{aid}/config.json` 写脏(28 号 E2E 实测)。 * 现在心跳状态落公司态目录,落点与读写口径都收在 heartbeat-service 里,这里只剩「雇没雇」这一问。 */ function isHired(companyId, roleName) { return listCompanyAgents(companyId).some((r) => r.role_name === roleName); } // PUT /api/company/:companyId/agents/:agentId/heartbeat — 更新 Agent 心跳配置 router.put('/:companyId/agents/:agentId/heartbeat', (req, res) => { try { const { intervalMinutes, enabled } = req.body; // 非正数 / NaN 会让 setInterval 退化成每 tick 触发一次 —— 拦在入口,别让它落盘(`> 0` 对 NaN 为 false) if (intervalMinutes !== undefined && !(Number(intervalMinutes) > 0)) { return res.status(400).json({ error: 'intervalMinutes 必须是正数' }); } if (!isHired(req.params.companyId, req.params.agentId)) { return res.status(404).json({ error: `Agent ${req.params.agentId} 不存在` }); } const heartbeatService = require('../services/heartbeat-service'); const namespacedId = `${req.params.companyId}:${req.params.agentId}`; // 只写公司态(`system/agents/{aid}/{cid}/heartbeat.json`),本体一个字节都不动 const heartbeat = heartbeatService.writeHeartbeatConfig(namespacedId, { intervalMinutes, enabled }); // 如果正在运行,重新挂载心跳 if (heartbeat.enabled) { heartbeatService._setupAgentHeartbeat(namespacedId); } else if (heartbeatService.timers.has(namespacedId)) { clearInterval(heartbeatService.timers.get(namespacedId)); heartbeatService.timers.delete(namespacedId); } console.log(`[Company] ⚙️ ${namespacedId} 心跳配置更新: interval=${heartbeat.intervalMinutes}min, enabled=${heartbeat.enabled}`); res.json({ success: true, heartbeat }); } catch (err) { res.status(500).json({ error: err.message }); } }); // POST /api/company/:companyId/agents/:agentId/heartbeat/start — 启动单个 Agent 心跳 router.post('/:companyId/agents/:agentId/heartbeat/start', (req, res) => { try { if (!isHired(req.params.companyId, req.params.agentId)) { return res.status(404).json({ error: `Agent ${req.params.agentId} 不存在` }); } const heartbeatService = require('../services/heartbeat-service'); const namespacedId = `${req.params.companyId}:${req.params.agentId}`; heartbeatService.writeHeartbeatConfig(namespacedId, { enabled: true }); // 激活心跳 heartbeatService._setupAgentHeartbeat(namespacedId); console.log(`[Company] 💓 ${namespacedId} 心跳已启动`); res.json({ success: true, message: `${namespacedId} 心跳已启动` }); } catch (err) { res.status(500).json({ error: err.message }); } }); // POST /api/company/:companyId/agents/:agentId/heartbeat/stop — 停止单个 Agent 心跳 router.post('/:companyId/agents/:agentId/heartbeat/stop', (req, res) => { try { if (!isHired(req.params.companyId, req.params.agentId)) { return res.status(404).json({ error: `Agent ${req.params.agentId} 不存在` }); } const heartbeatService = require('../services/heartbeat-service'); const namespacedId = `${req.params.companyId}:${req.params.agentId}`; heartbeatService.writeHeartbeatConfig(namespacedId, { enabled: false }); // 停止心跳 if (heartbeatService.timers.has(namespacedId)) { clearInterval(heartbeatService.timers.get(namespacedId)); heartbeatService.timers.delete(namespacedId); } console.log(`[Company] 🛑 ${namespacedId} 心跳已停止`); res.json({ success: true, message: `${namespacedId} 心跳已停止` }); } catch (err) { res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // Agent 技能查询 // ═══════════════════════════════════════ // GET /api/company/:companyId/agents/:agentId/skills — 获取 Agent 技能列表 router.get('/:companyId/agents/:agentId/skills', (req, res) => { try { const path = require('path'); const fs = require('fs'); // Skill 挂在本体的 `.claude/skills/`(agent-manager 的 scoped skill 遍历找的就是这里), // 不是公司私有目录下那个从来没被 CLI 发现过的 `skills/`。 const hired = listCompanyAgents(req.params.companyId).find((r) => r.role_name === req.params.agentId); if (!hired) return res.json([]); const skillsDir = path.join(locator.bodyDir(hired.system_agent_id), '.claude', 'skills'); if (!fs.existsSync(skillsDir)) { return res.json([]); } const skills = fs.readdirSync(skillsDir, { withFileTypes: true }) .filter(d => d.isDirectory()) .map(d => d.name); res.json(skills); } catch (err) { res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // Agent 招聘 // ═══════════════════════════════════════ // POST /api/company/:companyId/hire router.post('/:companyId/hire', async (req, res) => { try { const { role, type, name, description, customAgent, customSoul, customUser } = req.body; if (!role) return res.status(400).json({ error: 'role 为必填项' }); const result = await companyService.hireWithTraining(req.params.companyId, role, { type, name: name || role, description, customAgent, customSoul, customUser, }); // ── 核心逻辑扩展:自动补充定义 + 搜索注入 3 个初始 Skill + Innovator 训练 ── if (result.status === 'training' || result.needsTraining) { (async () => { try { console.log(`[Hire] 🚀 自动流水线启动: 为新员工 ${role} 安排 [定义 → 技能注入 → 训练]`); const fs = require('fs'); const path = require('path'); const config = require('../config'); const skillDb = require('../services/skill-db'); const namespacedId = `${req.params.companyId}:${role}`; // result.agentDir 是**公司态**目录;定义文件与 skill 都写本体(28 批次 4: // 公司不再持有 agent 实体,往公司侧写等于刚清完又造一个新的死角)。 const bodyPath = locator.bodyDir(role); const PORT = config.port || 3000; // 1. 发起 Agent-Creator 编写骨架定义(32 号文档方案 B:同进程调用 + 互斥锁 + // 有限重试,替掉原来那发裸 fetch。原注释里"异步不等待"已经不准确—— // 这里 await 了,下面第 2/3 步因此天然拿到"agent-creator 真的写完了"这个 // 保证,不再是"assume 5 秒内写完"(第 3 步 setTimeout(5000) 那条注释)。 // ⚠️ 给的是**绝对路径**:agent-creator 的 cwd 是它自己的任务态目录 // (system/agents/agent-creator/_global/{taskId}/),相对路径从那里数出去 // 指向的是一个不存在的位置 —— 它会照着写,且不报错。 const fillResult = await fillAgentBodyReliably({ agentManager: new AgentManager({}, REPO_ROOT), targetAgentId: role, // 产出契约(28 批次 5-23)逐字点名三个必填文件:本体填不满就停在 // status='draft'、别家公司雇不走。以前只写「补充完整的定义文件(…等)」, // 「等」里那份 SKILL.md 谁也没保证过 —— org-coder 就是这么变成空定义的。 // 契约文本(含运行期布局那半段)单点定义在 company-service,见其注释。 message: buildAgentDefinitionBrief({ agentId: role, bodyPath, description: description || name || role, intro: `新员工 ${namespacedId} 已由脚手架创建空白目录在 \`${bodyPath}\`。请你按照标准模板,使用 Write 工具为该员工补充定义文件。写完后请答复。`, }), }); if (!fillResult.ok) { // 重试耗尽——不是把招聘流程一起判失败(目录/skill 注入照常做,参见 32 号文档 // §三「不做保证自动跑到底」),只是把这个失败从"一行日志"变成可查询的哨兵文件 // (agent-body-filler.js 已经写好),这里再喊一嗓子留个显眼的日志。 console.error(`[Hire] ⚠️ agent-creator 装配 ${role} 失败(重试 ${fillResult.attempts} 次):${fillResult.error}`); } // 2. 去网上/Arena搜索并注入 3 个基础 Skill。 // 落 `.claude/skills/` 而不是 `skills/`:只有前者会被 CLI(和 agent-manager // 的 scoped skill 遍历)发现,后者是一个从来没人读过的目录。 const skills = skillDb.searchSkills(role, { limit: 3 }); const skillsDir = path.join(bodyPath, '.claude', 'skills'); if (skills && skills.length > 0) { if (!fs.existsSync(skillsDir)) fs.mkdirSync(skillsDir, { recursive: true }); const copyDir = (src, dest) => { fs.mkdirSync(dest, { recursive: true }); const entries = fs.readdirSync(src, { withFileTypes: true }); for (const e of entries) { if (e.isDirectory()) copyDir(path.join(src, e.name), path.join(dest, e.name)); else fs.copyFileSync(path.join(src, e.name), path.join(dest, e.name)); } }; for (const skill of skills) { if (skill.directory && fs.existsSync(skill.directory)) { const targetPath = path.join(skillsDir, skill.skillName); if (!fs.existsSync(targetPath)) { try { copyDir(skill.directory, targetPath); console.log(`[Hire] 💉 成功为其注入基础技能: ${skill.skillName}`); } catch (e) { console.error(`[Hire] 注入技能 ${skill.skillName} 失败:`, e.message); } } } } } else { console.log(`[Hire] ⚠️ 网上/Arena 未找到该角色的匹配 Skill,将交由 Innovator 凭空生成`); } // 3. 延时 5 秒触发 Innovator 进行评估与训练 (等 creator 写入完毕) // 同上:`innovator` 改名 `org-innovator` 之后没同步改这里,招聘后自动训练 // 静默 404 失效(36 号文档阶段二实测发现,见 train 路由那处同款注释)。 setTimeout(() => { fetch(`http://127.0.0.1:${PORT}/api/agent/org-innovator/chat`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ message: `公司 ${req.params.companyId} 的员工 ${role} 骨架和基础技能已注入完毕。请你使用 FDE pipeline 对该 Agent ${namespacedId} (特别是它的技能)进行评测和进化训练,直到 Benchmark 达标 (Score ≥ 95)。训练过程包括 Hunt(验证数据), Digest, Evolver 等。请立即开始评估和训练,并保持向我汇报。`, conversationId: `train-${req.params.companyId}-${role}-${Date.now().toString(36)}` }) }).catch(e => console.error('[Hire] Innovator 请求错误:', e.message)); }, 5000); } catch (err) { console.error(`[Hire Orchestration] 自动流水线编排异常: ${err.message}`); } })(); } // ── P0 FIX: 动态心跳挂载 ── // 招聘成功后,立即为这个新 Agent 注册心跳定时器(如果它的 config.json 里 heartbeat.enabled = true) try { const heartbeatService = require('../services/heartbeat-service'); const namespacedId = `${req.params.companyId}:${role}`; heartbeatService._setupAgentHeartbeat(namespacedId); console.log(`[Hire] 💓 已为 ${namespacedId} 动态挂载心跳`); } catch (hbErr) { console.warn(`[Hire] ⚠️ 心跳挂载跳过: ${hbErr.message}`); } res.json(result); } catch (err) { res.status(400).json({ error: err.message }); } }); // POST /api/company/:companyId/fire — 辞退 Agent(招留管退之"退") router.post('/:companyId/fire', (req, res) => { try { // archive 参数保留在协议里(前端还在传),但已无实际作用 —— 见下方注释 const { role } = req.body; if (!role) return res.status(400).json({ error: 'role 为必填项' }); const companyId = req.params.companyId; const namespacedId = `${companyId}:${role}`; // 1. 解雇 = 把 company_agents 那一行标成 removed,同时把 company.json.agents[] 同步掉 // (companyService.fire() 两件事都做)。**不动本体、也不动公司态目录**: // 本体是全系统共享的,删了别家公司的 agent 就没了;公司态目录里是记忆, // 28 §5.6 明确「fire 只软删,记忆不随解雇删除」。 // archive 参数因此退化成无意义 —— 没有目录可搬了,原地保留就是归档。 const fired = companyService.fire(companyId, role); if (!fired) { return res.status(404).json({ error: `Agent ${role} 不存在于公司 ${companyId}` }); } // 2. 停止心跳 const heartbeatService = require('../services/heartbeat-service'); if (heartbeatService.timers.has(namespacedId)) { clearInterval(heartbeatService.timers.get(namespacedId)); heartbeatService.timers.delete(namespacedId); } // 3. P2 T7:org_graph 里对应节点软删除(级联失效它的进出边) try { orgGraphService.removeNode(companyId, role); } catch (e) { console.warn('[Fire] org_graph 清理失败(不影响辞退结果):', e.message); } console.log(`[Fire] 🔴 ${namespacedId} 已辞退`); // archived 恒为 true:公司态目录(记忆 / 会话)原地保留,本来就等于归档 res.json({ success: true, message: `${namespacedId} 已辞退`, archived: true }); } catch (err) { res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // Org Graph — 公司级 workflow 图(P2 / 27-P2 §4.3) // ═══════════════════════════════════════ // POST /api/company/:companyId/org-graph — 整体写入(规划产物落库) // body: { nodes: [{roleName, systemAgentId?}], edges: [{from, to, condition, edgeKind?}] }(边的 from/to 是 roleName) router.post('/:companyId/org-graph', (req, res) => { try { const { nodes, edges } = req.body || {}; const result = orgGraphService.writeGraph(req.params.companyId, { nodes, edges }); // 校验失败返回 400 + errors[{rule, detail}],供 orggraph-planning 自行修正后重试 if (!result.ok) return res.status(400).json({ error: '图不合法', errors: result.errors }); res.json({ success: true, ...result }); } catch (err) { res.status(500).json({ error: err.message }); } }); // PATCH /api/company/:companyId/org-graph — 增量修改(改完整张图仍须合法,否则整体回滚) // body: { addNodes?, removeNodes?, addEdges?, removeEdges? } router.patch('/:companyId/org-graph', (req, res) => { try { const result = orgGraphService.patchGraph(req.params.companyId, req.body || {}); if (!result.ok) return res.status(400).json({ error: '图不合法', errors: result.errors }); res.json({ success: true }); } catch (err) { res.status(500).json({ error: err.message }); } }); // GET /api/company/:companyId/org-graph — 只读;公司无图时返回空数组,不报错 // 返回 { nodes: [{id, roleName, systemAgentId, status}], edges: [{from, to, condition, edgeKind}] }(边的 from/to 是节点 id) router.get('/:companyId/org-graph', (req, res) => { try { res.json(orgGraphService.getGraph(req.params.companyId)); } catch (err) { res.status(500).json({ error: err.message }); } }); // POST /api/company/:companyId/org-graph/replan — 对已有公司手动重跑规划(存量公司补图入口) // 异步:立即返回,agent 跑完后自己调上面的 POST 写入新图 router.post('/:companyId/org-graph/replan', (req, res) => { try { const company = companyService.get(req.params.companyId); if (!company) return res.status(404).json({ error: `公司 ${req.params.companyId} 不存在` }); try { // 幂等:已经雇过就是 already_exists,不重复建目录 companyService.hire(company.id, 'orggraph-planning', {}); } catch (e) { console.warn('[Replan] orggraph-planning 招聘跳过:', e.message); } triggerOrgGraphPlanning(company, 'Replan'); res.json({ status: 'replanning', companyId: company.id }); } catch (err) { res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // 任务入口 — 用户 → 公司(T12 / 27-P3P4 §五 T12、27-E2E §三 阶段 3) // ═══════════════════════════════════════ // // 为什么必须新开一个入口:既有 POST /api/agent/task/dispatch 的 from / to 都要求是 // 已存在的 agent 目录,**只支持 agent → agent**(27-E2E §三 阶段 3 的 ❌ 缺口)。 // 用户把一句自然语言任务交给「一家公司」时没有 from agent,走不通那条路。 // 收敛侧的 POST /api/agent/task/report 则已经实现了完整回调链路,本入口复用它 —— // 但只在 **goal 粒度**复用一次,见 `services/task-settle.js` 与下方缺陷 3 那段注释。 /** * 「这条 root run 占着执行位」的取值域(27-P3P4 §一非目标;批 4 起同时能占几个由 * `company.json.max_parallel_tasks` 定,缺省 1 —— 取值域本身没变)。 * * **取值域只有一个持有者**:`work-graph-db.ACTIVE_ROOT_RUN_STATUS` —— 它是准入事务数位子的 * 判据、是调度器判「还有没有空位」的判据(批 4 之前它还兼任唯一索引 * `idx_graph_runs_one_active_root` 的 WHERE,那条索引已 DROP,判据只剩准入事务这一处)。 * 本文件以前抄了一份字面量在这里,而 work-graph-db 的注释白纸黑字写着「这里是全仓库 * 唯一的定义处」—— **注释是对的、代码没兑现**,正是 §十四.3 第 3 条记着的那个形状。 * 现在改成取过来。惰性取值(模块加载期 require 会与路由的加载顺序纠缠)。 */ const activeRootRunStatus = () => require('../services/work-graph-db').ACTIVE_ROOT_RUN_STATUS; /** * 「这条 run 还没撒手」的取值域 —— 归档守卫用(27-P3C 第二轮复核 缺陷 1 的连带)。 * * = 上面那个活跃集 **+ `blocked`**,而且必须由它派生、不许再抄一份字面量:§九.1 记着的教训就是 * `ACTIVE_ROOT_RUN_STATUS` 在两个文件里逐字重复,改一处漏一处。 * * 为什么 `blocked` 不直接并进 `ACTIVE_ROOT_RUN_STATUS`:那个常量是**准入事务数位子**的判据 * (批 4 之前还兼任唯一索引 `idx_graph_runs_one_active_root` 的 WHERE,索引里就没有 blocked), * 动它等于悄悄改「一条 run 占不占执行位」的口径 —— blocked 的 run 不占位,可以再起别的任务。 * 但对归档来说 blocked 恰恰**不能**放行:run 转 blocked 会把 task 也写成 blocked, * 人加完预算 `POST /api/graph-runs/:id/start` 就接着烧钱,而 task 行仍停在 blocked —— * 只看 `task.status === 'running'` 的旧守卫会放行,用户就把一条正在跑的 run 从界面上藏掉了。 */ const runHoldingStatus = () => [...activeRootRunStatus(), 'blocked']; /** * 这条 task 现在能不能被归档掉?两条判据只要中一条就是「不能」: * · task 行本身处于 running / blocked(blocked = 挂起等人,随时可能被恢复); * · 它的 current_run_id 指的 run 还在 runHoldingStatus() 里 —— 恢复后 run 已经 running、 * task 行却还留在 blocked(`RUN_TO_TASK_STATUS` 没有 running 这一档,回写只在收敛时发生), * 光看 task 行会漏掉这半截。 */ function taskHoldsRun(task) { if (task.status === 'running' || task.status === 'blocked') return true; if (!task.current_run_id) return false; const run = require('../services/work-graph-db').getRun(task.current_run_id); return !!run && runHoldingStatus().includes(run.run_status); } /** * 「现在该怎么脱身」—— 给 4xx 一句**真的打得通**的下一步。 * * 为什么非要一个函数:几处 409 此前都硬写着 `POST /api/graph-runs/${task.current_run_id}/stop`, * 而这条路对两种很常见的状态根本不存在: * · `current_run_id` 是空的(依赖推导出来的 blocked、入队做了一半的 task 都可能没有 run) * → URL 里出现字面量 `null`,打过去 404; * · run 还在 `queued`(排队中的 task,以及 blocked:upstream_failed:* 的 task 手上那条) * → `/stop` 的可用条件里没有 queued(见 graph-run.js 的 STOPPABLE_RUN_STATUS),打过去 400。 * 「指向不存在行为的提示」比不给提示更糟:用户照着做一次、失败一次,然后不知道还能做什么。 * * 取值域不在这里复述,从 `routes/graph-run.js` 取那唯一一份。 */ function escapeHatchHint(companyId, task) { const { STOPPABLE_RUN_STATUS } = require('./graph-run'); const run = task.current_run_id ? require('../services/work-graph-db').getRun(task.current_run_id) : null; const cancel = `POST /api/company/${companyId}/tasks/${task.id}/cancel`; if (run && STOPPABLE_RUN_STATUS.includes(run.run_status)) { return `先停它:POST /api/graph-runs/${run.id}/stop(软信号,到节点边界才落 stopped),停稳了再来`; } if (run) return `它的 run 还在 ${run.run_status}(停不掉,那条路只认 ${STOPPABLE_RUN_STATUS.join(' / ')})——撤下它:${cancel}`; return `这条任务手上没有活的 run,没有什么可停的——直接取消它:${cancel}`; } /** 任务级总预算默认值(27-P3P4 §四.A.3:中等任务粗估 60-100 次调用、1-2 小时) */ const DEFAULT_TASK_BUDGET = { total_agent_calls: 100, total_tokens: 5000000, wallclock_ms: 7200000, spent: { agent_calls: 0, tokens: 0 }, }; /** * run 目录 —— **`graph-engine.runRootDir` 的镜像**(27-P3C §六 缺陷 A)。 * * 为什么这里非有一份不可:`task-goal.json` 的写侧在本文件、读侧在 `graph-engine.describeTask` * 与 `goal-contract.deriveRunDir`,而读侧走 `runRootDir`(认 `GRAPH_RUNS_DIR`)、写侧原先 * 自己拼 `data/companies/…`(不认)。env 一设上两边就分叉,`describeTask` 拿到 null 后 * **静默退回兜底文案「执行图节点任务」** —— agent 从此收不到真实任务描述,而没有任何报错。 * * ponytail: 这是逐字同构的第二份实现,上限就是「改一处漏一处」这个老病。 * 不直接调 `graph-engine.runRootDir` 是因为它没导出、且入参是 run 行对象, * 而这里只有 companyId + rootRunId。收敛路径:在 `company-paths.js` 里加一个 * `runRootDirById(companyId, rootRunId)`,让 engine 的 `runRootDir` 与本函数同时改调它。 * * `GRAPH_RUNS_DIR` **每次调用时读**,照 `company-paths.js` 文件头那条教训: * 模块加载时快照会让「先 require 再设 env」的调用方算出另一个根。 */ function runRootDirById(companyId, rootRunId) { const runsRoot = process.env.GRAPH_RUNS_DIR || null; return runsRoot ? path.join(runsRoot, String(rootRunId)) : path.join(COMPANIES_DIR, String(companyId), 'runs', String(rootRunId)); } /** * 任务描述落盘到 root run 的 artifacts/(§四.A.5 的任务级共享产物区,跨 org node 只读)。 * graph_runs 上没有描述列,而 advanceOne 的任务描述是调用时透传的 opts.task; * 手动模式下每次 /next 由控制台发起,需要能把原始 task-goal 读回来。 */ function writeTaskGoalArtifact(companyId, rootRunId, payload) { const dir = path.join(runRootDirById(companyId, rootRunId), 'artifacts'); fs.mkdirSync(dir, { recursive: true }); const fp = path.join(dir, 'task-goal.json'); fs.writeFileSync(fp, JSON.stringify(payload, null, 2), 'utf-8'); return fp; } /** * ── run 级汇报为什么被摘掉了(27-P3C 第二轮复核 缺陷 3)── * * 这里原先住着 `terminalOrgAgent` + `reportRunOutcome`:auto driver 跑完后往 * `POST /api/agent/task/report` 打一条 run 粒度的汇报。它和 `task-settle.reportGoalOutcome` * 的 goal 粒度汇报会**同时**命中那个端点,而该端点第 5 步会 `setImmediate` 自调 * `/api/agent/{to}/chat` 唤醒上级 agent —— 批 1 的唯一形态是 1 goal : 1 task * (`POST /:cid/tasks` 自动建 goal,`createTask` 继承 `goal.report_to`),于是同一件事 * 把同一个上级叫醒两次、白烧一轮 LLM,其中一次带的还是它从没见过的 `G-` id。 * * 留下的是 goal 级那一条,理由不是「随便留一条」: * · run 级只挂得上 auto 模式的 driver,manual / stop / blocked 全不触发,本来就是半条链路; * · goal 级挂在唯一漏斗 `graph-run-sse.transitionRun` 的收敛钩子上,四种收敛路径全覆盖; * · 汇报的语义本来就是「人提的这个需求算不算做到了」,那是 goal 的粒度,不是 run 的。 * * 交付者身份(汇报的 `from`)随函数一起搬到了 `org-graph-service.terminalAgentId`, * 读侧改成 task-settle —— 判据只留一份,别在两个文件里各存一个末节点算法。 */ // ═══════════════════════════════════════ // 目标 / 任务两级模型(27-P3C §六) // ═══════════════════════════════════════ // // goal(人提的一个需求,description 就是验收边界)→ task(一件能独立跑的活)→ run(一次执行)。 // 这不是新概念,是给 graph_runs.goal_id / task_id 这对**从 P3 起就一直在写、却没有任何表接得住** // 的悬空外键补上它们的表:此前「一个需求拆成几件活」在库里根本不存在,界面上也就画不出来。 // // 本批(批 1)只落地同步的那一半:建行 / 改行 / 归档 / 下达即起 run。 // **排队不在本批**(run_status='queued'、调度器、depends_on 的出队判据都是批 3), // **自动拆分也不在**(task-planning 是批 2)。所以批 1 一次只起得来一条 run,起不来就是 409 —— // 与当时那条唯一索引 idx_graph_runs_one_active_root 的语义一致,绝不假装「已排上队」: // 仓库里还没有出队的 worker,排进去的任务不会被消费,那是一条会静默丢任务的路。 // (批 3 有了调度器之后下达改成入队、不再 409;批 4 DROP 了那条索引,执行位数改为 per-company。) /** * 数据层的两个专有错误 → HTTP(§四.4 描述冻结 / §六.3 依赖校验)。 * 不是这两个就返回 null,由调用方原样抛出去走 500 —— 把未知错误也翻成 4xx 会让真 bug 变成 * 「用户输入有问题」,那是最难被发现的一种掩盖。 */ function sendDomainError(res, err) { // 判定用 instanceof 而不是 `err.name === '…'`(缺陷 F10): // 字符串比较对**任何**把 name 设成同样字面量的错误都成立(包括 agent 生成代码里随手 // `err.name = 'DependsOnError'` 的那种),而且它让数据层导出的这两个类没有任何调用方 —— // 「导出了、零人使用」正是 §九.4quater 立案的惯犯。全仓 work-graph-db 只有相对路径 require, // 是同一个模块实例,instanceof 成立。 const { FrozenFieldError, DependsOnError } = require('../services/work-graph-db'); if (err instanceof FrozenFieldError) { return res.status(409).json({ error: err.message, frozenFields: err.fields, hint: '描述已被 agent 消费过(写进了 task-goal.json),要改就取消后新建,不要就地改', }); } if (err instanceof DependsOnError) { return res.status(400).json({ error: err.message }); } return null; } /** * mode 的信任边界校验,**三个入口共用这一份**(POST /goals、POST /tasks、PATCH /tasks)。 * * PATCH 侧原先没有这道校验,非法值直送 DB 撞 CHECK 约束 → 500 且把 SQLite 原文回给调用方 * (缺陷 F7)。500 的含义是「服务端出错了」,用户会去重试;正确答案是 400「你传错了」。 * * 返回 undefined 表示**没传**(调用方据此跳过这个字段),null 表示**传了但非法**—— * 两者必须分得开:`{"mode": null}` 走 `!== undefined` 的过滤会被当成没传, * 于是接口回 200 而什么都没改,那是最难被发现的一种半生效。 */ function normalizeMode(...candidates) { const raw = candidates.find((v) => v !== undefined); if (raw === undefined) return undefined; return raw === 'manual' || raw === 'auto' ? raw : null; } /** budget_json 列(TEXT)→ 对象;空 / 脏行返回 null,由调用方退回默认值 */ function parseBudget(raw) { if (!raw) return null; try { const v = JSON.parse(raw); return v && typeof v === 'object' ? v : null; } catch { return null; // 一条脏行不该让下达整个失败:退回 DEFAULT_TASK_BUDGET 至少能跑 } } /** * 一句自然语言 → 列表上的短标题。 * 截断丢信息无所谓:title 只是列表上的一个标签,不进任何 agent 的上下文(§四.4), * 验收边界一律看 description。 */ function deriveTitle(text) { const line = String(text || '').split('\n').find((l) => l.trim()) || ''; const t = line.trim(); return t.length > 60 ? `${t.slice(0, 60)}…` : t; } /** * 这家公司此刻的执行位:`{ max, used, free, holders }`。 * * **形状与判据都不在这里**,取 `routes/graph-run.js` 那唯一一份(它同时供 409 用)。 * 批 4 之前这里是 `activeRootRun(companyId)`:`listCompanyRuns(...).find(活跃)`, * 两个调用方各自误用它的一半,max 一放开两处都错: * · `startNextIfSlotFree` 拿它当「满没满」——它的语义是「有没有」, * 3 个位空着 2 个时 manual 任务永远起不来,端点还照常回「排队中」(写不进去还不报错); * · `/tasks/:id/start` 的 409 拿它当「谁挡着我」—— 它取的是 created_at **DESC** 的第一条, * 而 `RunSlotBusyError.heldBy` 是 **ASC**,同一次拒绝的两个字段会指向不同的占位者。 */ function slotsOf(companyId) { return require('./graph-run').slotState(companyId); } /** * 列表端点上给界面的那份执行位快照:`{ max, used, free }`。 * * 为什么落在 `GET /goals` 与 `GET /tasks` 上而不是 `GET /graph-runs`:控制台就是靠这两个 * 端点轮询左栏的,多一个请求就多一个「两份数据不同时刻」的窗口;而 graph-runs 那边只是列 run, * 在它上面再数一遍 root run 就是第三个持有者(批 4 把它的 canDispatch 删掉正是为此)。 * * **holders 不给**:列表页只回答「还能不能下达 / 要不要等」,不需要点名谁占着; * 真被拒绝时 409 会带着 holders 回来(同一个 `slotState` 产出,形状逐字相同)。 * 前端 `goal-task-model.readSlots` 也只读这三个数,且**不自己做 `max - used`** —— * free 带着「max 被改小时截断到 0」的语义,减法算不出来。 */ function slotsView(companyId) { const { max, used, free } = slotsOf(companyId); return { max, used, free }; } /** goal 行 → 对外形状。库里是 snake_case,API 一律 camelCase(同 getRunState 口径) */ function goalView(g, tasks = null) { return { goalId: g.id, companyId: g.company_id, title: g.title, description: g.description, status: g.status, priority: g.priority, okrNodeId: g.okr_node_id, createdBy: g.created_by, reportTo: g.report_to, // 验收窗口里 verdict 挂的是内部认领 token(含进程随机段与时间戳),不该出 API 面。 // 对外只说「验收进行中」,真判词落库后才透出去。判定走 task-settle 的 isPendingVerdict, // 别在这里比字符串 —— 取值域的持有者只有那一处。 verdict: require('../services/task-settle').isPendingVerdict(g.verdict) ? '(验收进行中)' : g.verdict, // 非空 = task-planning 三轮都没拆出合法 DAG,已降级成 1 条同描述的 task(§四.9 约束 2)。 // 界面拿它标注「自动拆分失败,已按单任务处理」——没有这个字段,降级后的 goal 与 // 「本来就只有一条 task」的 goal 完全无从区分,用户看到的是一次静悄悄变质的下达。 planningError: g.planning_error || null, // 拆分 / 验收各自烧掉的 token。goal 层的花费不挂在任何 run 的预算池上(那个池按 root run 算), // 所以它只记账不设闸门 —— 记不记这一笔,决定的是这半边花费看不看得见。 planningTokens: g.planning_tokens, validationTokens: g.validation_tokens, createdAt: g.created_at, updatedAt: g.updated_at, dispatchedAt: g.dispatched_at, archivedAt: g.archived_at, ...(tasks ? { tasks } : {}), }; } /** * 一家公司的队列位次表(taskId → 1-based 位次)。 * * **口径不在这里**:排序(priority → seq → created_at)只归 `task-scheduler.listQueue` 一处, * 这里只是把它的下标翻成 Map。路由自己再排一遍就是第二个持有者 —— canDispatch 的取值域被 * 前端复述了一遍还多写一个值、导致公司失败一次后再也下不了任务,就是这么来的。 * * 一次查询建全表,taskView 逐行 get —— 不是每行问一次调度器(那是列表页的 N+1)。 */ function queuePositions(companyId) { const { listQueue } = require('../services/task-scheduler'); const map = new Map(); listQueue(companyId).forEach((t, i) => map.set(t.id, i + 1)); return map; } /** * task 行 → 对外形状。progress 由 taskProgress 一条 GROUP BY 批量算好后传进来, * positions 由 queuePositions 一次算好后传进来。 * * `positions` 传 null 时 queuePosition 恒为 null —— 排队中的任务在界面上会显示成 * 「排队中(服务端没给位次)」。所以**凡是返回可能处于 queued 的 task 的端点都要传它**: * 少传一次不会报错,只会让那一行的位次凭空消失。 */ function taskView(t, progress = null, positions = null) { let dependsOn = []; try { const v = JSON.parse(t.depends_on || '[]'); if (Array.isArray(v)) dependsOn = v.map(String); } catch { dependsOn = []; // 一条脏行不该让整个列表炸掉 } return { taskId: t.id, goalId: t.goal_id, companyId: t.company_id, seq: t.seq, title: t.title, description: t.description, status: t.status, mode: t.execution_mode, priority: t.priority, // 回出来才看得见:budget 此前只有写侧(缺陷 F3),用户填了什么、最后生效的是不是它, // 在界面上一个字都查不到,直到任务跑到一半 blocked budget: parseBudget(t.budget_json), dependsOn, // 没有 run 或 run 还没有节点行时是 null,前端据此画「未开始」而不是 0/0 progress: progress || null, runId: t.current_run_id, attempt: t.attempt, // 排队位次(1-based),不在队里就是 null。前端拿它显示「排在第 N 位」—— // 那是「为什么点了下达却什么都没发生」的唯一答案,缺了它排队就是一个黑盒。 queuePosition: (positions && positions.get(t.id)) || null, blockedReason: t.blocked_reason, reportTo: t.report_to, createdAt: t.created_at, queuedAt: t.queued_at, startedAt: t.started_at, finishedAt: t.finished_at, archivedAt: t.archived_at, }; } /** 一批 task 的进度:一条 GROUP BY 出全部(验收 10 点名要挡的就是这里的 N+1) */ function progressOf(tasks) { const { taskProgress } = require('../services/work-graph-db'); return taskProgress(tasks.map((t) => t.current_run_id)); } /** 首个 task 下达时给 goal 盖上 dispatched_at —— 它就是 goal 级的描述冻结点(§四.4) */ function markGoalDispatched(goal) { const { updateGoal } = require('../services/work-graph-db'); if (goal.dispatched_at) return goal; return updateGoal(goal.id, { status: 'active', dispatchedAt: new Date().toISOString() }) || goal; } /** * 一条 task 身上「还没做完」的上游(= 除 completed 以外的任何状态)。 * * **查不到的行也算没做完**:依赖的 task 被归档删号时宁可挡下也不要放行 —— 放行等于把 * 「上游产物已就绪」这个前提凭空当真。口径与 `task-scheduler.dependenciesSatisfied` 一致 * (那边是私有函数,没导出;这里多的是「挡我的是谁」这份名单,报错文案要点名)。 */ function blockersOf(task) { const { getTask } = require('../services/work-graph-db'); let deps = []; try { const v = JSON.parse(task.depends_on || '[]'); if (Array.isArray(v)) deps = v.map(String); } catch { deps = []; } return deps.filter((id) => (getTask(id) || {}).status !== 'completed'); } /** * 「这次没抢到坑」—— 竞争的正常结局,不是故障。把它当异常抛给用户,会让「你排在第 2 位」 * 显示成 500。 * * **判据不在这里**:唯一持有者是 `work-graph-db.isSlotRaceError`。这里和 * `task-scheduler.isRaceError` 以前各写了一份,注释都声称「口径同对方」,而实际一份按 * `err.code`、一份按报错文本 —— 两个持有者从落地那天起就是分叉的。 */ function isSlotRace(err) { return require('../services/work-graph-db').isSlotRaceError(err); } /** * 入队:给一条 task 建一条 **`run_status='queued'`** 的 root run + 写 task-goal.json, * 并把 task 推到 `queued`(27-P3C 批 3 §五.1)。 * * 与批 1/批 2 的 `launchTaskRun` 的差别就是「不起跑」: * · `createRun({runStatus:'queued'})` **不占执行位**(占位的取值域只认 ready/running/paused, * 见 `activeRootRunStatus`),所以队列可以任意长,入队永不 409; * · 起跑那一跳(queued → ready → running)只有两个执行者:auto 归 `task-scheduler.tick`, * manual 归 `promoteQueuedTask`(人点了「下达 / 入队 / 重跑」才走)。 * * **顺序是有讲究的**:run 先建、artifact 先写,最后一条 UPDATE 才把 task 翻成 queued 并 * 同时写上 current_run_id(`updateTask` 是单条 UPDATE,两个字段原子落库)。 * 反过来先翻 task 的话,中间窗口里的 tick 会看到一条「queued 却没有 run」的任务, * 按 §五.1 第 5a 步把它标成 `enqueue_incomplete` 的 blocked —— 一次正常入队被判成坏数据。 * * `queued_at` 就是 task 级的描述冻结点(§四.4):从这一刻起 description / mode / budget 不可改, * 因为 task-goal.json 已经写出去、agent 随时会读它。重跑时**不刷新**它(描述没变过)。 * * @returns {{runId, run, artifact, task}} */ function enqueueTaskRun({ companyId, taskRow, budget }) { const wdb = require('../services/work-graph-db'); const created = wdb.createRun({ graphKind: 'org', graphId: companyId, // kind='org' 时 graph_id 就是 company_id companyId, goalId: taskRow.goal_id, taskId: taskRow.id, // §四.7 禁令一:**不传 okr_node_id**。okr_node_id 只存在 goals 表上。 // 一个 goal 挂一个 OKR 节点、下面 N 个 task 各带一份,收敛汇报就会把同一个节点写 N 次 // 且没有合并逻辑 —— 最后一个收敛的 task 说了算,一条 failed 的尾巴能把已经 done 的 // 节点翻成 failed。OKR 由 goal 收敛(全部 task 终态)时写一次,见 services/task-settle.js。 executionMode: taskRow.execution_mode, runStatus: 'queued', budget, }); const runId = created.id; const now = new Date().toISOString(); let artifact; try { artifact = writeTaskGoalArtifact(companyId, runId, { runId, companyId, taskId: taskRow.id, goalId: taskRow.goal_id, // artifact 里同样不写 okrNodeId:agent 侧读它只为拿任务描述,多一个会被误当可写的度量目标 task: taskRow.description, createdAt: now, }); const task = wdb.updateTask(taskRow.id, { status: 'queued', currentRunId: runId, queuedAt: taskRow.queued_at || now, // 重跑进来的行身上还留着上一轮的痕迹,不清就会在界面上显示成「排队中,已耗时 3 小时」 startedAt: null, finishedAt: null, blockedReason: null, // 人类任务从这一刻起被 graph-engine 接管跑图,不再是「无人认领」(T-mt5dkvjm): // 没有真实派单对象就写占位哨兵,COO 心跳巡检靠 to 字段判断时才不会把它当孤儿误报。 // 只在还没有真实值时才写,避免盖掉别的派单流程已经写好的 assignee。 assigneeAgentId: taskRow.assignee_agent_id || wdb.GRAPH_TAKEOVER_ASSIGNEE, }); return { runId, run: wdb.getRun(runId) || {}, artifact, task }; } catch (err) { // 入队做了一半:run 建出来了但 task 没翻过去。留着它就是一条**永远不会被出队**的 // 幽灵 run(没有任何 task 指向它),而且不报错。就地收掉,异常照抛给调用方翻 5xx。 try { require('../services/graph-engine').stopRun(runId); } catch (e) { console.error(`[TaskEntry] 入队回滚 run=${runId} 也失败了: ${e.message}`); } throw err; } } /** * 出队并起跑一条**排队中**的 task —— `task-scheduler.tickInner` 第 5 步的**人工版**。 * * 为什么这里非有一份不可:调度器**故意**不自动出队 manual 的任务(§五.2bis —— manual 的 run * 一 running 就占着坑停在那儿等人点「下一步」,没人点就把后面所有任务无限期堵死)。 * 那 manual 的任务由谁起跑?只能是人:`dispatch` / `enqueue` / `rerun` 三个端点都是人点的, * 「点一下 = 我现在就要它跑」。所以这条路的执行者是本函数,不是 tick。 * * ⚠️ 四步的**顺序与失败处理必须与 tickInner 5b-5e 保持一致**,改一处要改两处 * (ponytail: 等 services/ 可动时,把这段提成 `task-scheduler.promote(taskId)` 一处持有, * 本函数退化成一次调用;现在拆成两份是为了不与正在写 services/ 的会话打架): * b. `syncRunNodesToDef` —— 排队期间 org graph 可能被 replan 过。节点行是入队那一刻的快照、 * 定义是读时取最新,不对齐的话新增节点永远点不亮,而且全程无报错; * c. `transitionRun(queued → ready)` —— 唯一的出队原语。条件更新 + 准入事务, * 返回 null / 抛 RunSlotBusyError 都表示「没抢到」,不是错误; * d. `claimQueuedTask` —— 条件更新认领 task(写进去的 'running' 绝不在 WHERE 里)。 * 抢不到就把 run 收回去,否则会留下一条 ready 却没有任务的 run 白占着坑; * e. `startRun` —— 它内部回读确认真的转成了 running,没起来会抛(不再是无条件 ok)。 * * @returns {{runId, run, task}|null} null = 这轮没抢到(坑被占 / 被别人推走 / 认领失败) */ function promoteQueuedTask(taskRow) { const wdb = require('../services/work-graph-db'); const engine = require('../services/graph-engine'); const sse = require('../services/graph-run-sse'); const runId = taskRow.current_run_id; // 队里的每一条都必须有 run(入队时一起建的)。没有 = 入队做了一半, // 交给调度器去标 `enqueue_incomplete` 的 blocked,这里不假装起跑成功。 if (!runId || !wdb.getRun(runId)) return null; try { wdb.syncRunNodesToDef(runId); if (!sse.transitionRun(runId, 'queued', 'ready', {}, { summary: `人工出队:queued → ready(task ${taskRow.id})`, task_id: taskRow.id, })) return null; if (!wdb.claimQueuedTask(taskRow.id, runId)) { abandonRun(runId, `task=${taskRow.id} 认领失败(已不是 queued 或 current_run_id 已改指)`); return null; } // mode 从 task 行读回,不从请求闭包读:排队之后那个闭包早就不在了(§四.7 禁令二) const { driver } = engine.startRun(runId, { mode: taskRow.execution_mode, task: taskRow.description, }); // driver 只剩「别让 rejection 变成 unhandled」这一个用途:收敛后的对外汇报不挂在它身上, // 状态回写走 transitionRun 的钩子(services/task-settle.js),两者都与 driver 无关。 if (driver) driver.catch((e) => console.error(`[TaskEntry] auto driver 异常: ${e.message}`)); return { runId, run: wdb.getRun(runId) || {}, task: wdb.getTask(taskRow.id) }; } catch (err) { if (isSlotRace(err)) return null; // 没抢到,继续排队 // 到这儿说明可能已经出了队却没能起跑:task 标着 running 而没有活的 run, // 正是要消灭的那类「写了但没生效还不报错」。收回 run,让 task 落到可见的终态。 if ((wdb.getRun(runId) || {}).run_status === 'ready') { abandonRun(runId, `startRun 失败: ${err.message}`); } throw err; } } /** * 出队失败后的收尾:把 run 收成 stopped。 * 复用 `engine.stopRun` 而不是自己写状态转换 —— 它顺带把没开跑的节点批量 skipped、落 timeline、 * 并经由漏斗钩子让 task-settle 把 task 收成 cancelled。于是「起不来」在界面上是一条有终态、 * 有时间线的任务,而不是一条标着 running 却没有活 run 的幽灵。 */ function abandonRun(runId, why) { console.error(`[TaskEntry] 出队回滚 run=${runId}: ${why}`); try { require('../services/graph-engine').stopRun(runId); } catch (err) { console.error(`[TaskEntry] 回滚 run=${runId} 也失败了: ${err.message}`); } } /** * 人点了一下「下达 / 入队 / 重跑」之后,把队伍往前推一格。 * * 两个执行者各管一半,**次序不能反**: * 1. `scheduler.tick(companyId)` —— 同步跑一次调度:依赖推导(queued ⇄ blocked 双向重算) * + auto 任务出队。放在前面是因为推导结果决定了下一步还有谁够格出队。 * 它挂在漏斗上的那份是 `setImmediate` 的(避免在状态转换的栈里再转状态); * 这里是全新的请求栈,同步调没有递归问题,而且响应里要带上这一次的结果。 * 2. 坑还空着 → 由**人**把队首那条 manual 起起来(理由见 promoteQueuedTask)。 * * 候选顺序取自 `scheduler.listQueue` —— 队列位次(priority → seq → created_at)全仓库 * 只有那一个持有者,这里再排一遍就是第二个持有者(canDispatch 那次事故的形状)。 * 队首可能属于**别的 goal**:队列是公司级的,「我下达的是 goal A,起跑的却是 goal B 的活」 * 正是排队本来的含义,响应里的 started 会如实点名是哪几条。 * * 批 4:一次可能起跑**不止一条**(公司有 N 个执行位),所以返回数组。批 3 那两处 * 「只报第一条 / 抢不到就不再试下一条」在 max=1 时等价,max>1 时一个漏说、一个漏起。 * * @returns {Array<{runId, run, task}>} 本次真起跑了哪几条(空数组 = 一条都没起) */ function startNextIfSlotFree(companyId) { const scheduler = require('../services/task-scheduler'); const wdb = require('../services/work-graph-db'); const ticked = scheduler.tick(companyId); const started = []; // **tick 起走的那几条同样要报出来。** 少了这一段,auto 模式下 dispatch 明明起跑了任务, // 响应里却没有 `started` —— 前端 `dispatchedTaskId` 于是退回 `queued[0]`,选中态跳到 // 一条还在排队的任务上,用户点了「全部下达」却看不到正在跑的那张图,而接口回的是 201。 // 「做了但不说」和「说了没做」是同一类撒谎,都在本批要消灭的名单上。 // 批 4 从 `started[0]` 改成全部:一次 tick 就能填满剩余的 N 个位,只报第一条是同一句谎。 for (const s of ticked.started || []) { const task = wdb.getTask(s.taskId); if (task) started.push({ runId: s.runId, run: wdb.getRun(s.runId) || {}, task }); } for (const t of scheduler.listQueue(companyId)) { // auto 归调度器:这里再抢一次会绕开它「起不来就把 task 收回可见终态」的整套善后 if (t.execution_mode !== 'manual') continue; if (blockersOf(t).length) continue; // 上游没做完,依赖是硬约束 // 到这儿才问「还有位吗」,且每一轮都**重算**:刚起跑一条就少一个,而且别人也在抢。 // 批 4 之前这里是循环外一句 `if (activeRootRun(companyId)) return null;` —— // 「有没有」当成「满没满」,3 个位空着 2 个时 manual 任务永远起不来,端点却照常回「排队中」。 // ponytail: 每轮两次读(活跃 root run 一条查询 + company.json 一次 readFile)。 // 两条便宜的过滤排在它前面,所以只有真候选才付这个钱;长队列真成问题时把 free 缓存进循环、 // 按起跑数递减即可,别把判据挪到别处去。 if (slotsOf(companyId).free <= 0) break; const promoted = promoteQueuedTask(t); if (promoted) started.push(promoted); // 没抢到**不等于没位了**:promoteQueuedTask 把「位满」「被别人推走」「认领失败」都归成 null, // 而后两种下一条照样起得来。真满了由上面那句 free<=0 在下一轮 break —— 判据只留一处。 } return started; } /** 起跑了哪条 → 对外形状(三个端点此前各拼了一份,字段还不完全一样) */ function startedView(s) { return { taskId: s.task.id, goalId: s.task.goal_id, runId: s.runId, runStatus: (s.run || {}).run_status }; } /** * 响应里「本次起跑了什么」的两个字段。 * * `started` 仍是**单数**——前端 `dispatchedTaskId` 认的就是它,形状与批 3 逐字相同, * 不是破坏性变更;`startedTasks` 是本次起跑的全部。 * 两者**同源**:`started = startedTasks[0]`,不是另算一遍,所以分叉不出来。 */ function startedFields(list) { const arr = (Array.isArray(list) ? list : [list]).filter(Boolean); if (!arr.length) return {}; const views = arr.map(startedView); return { started: views[0], startedTasks: views }; } /** * 撤下一条排队中的 run(queued → stopped)。 * * 走 `transitionRun` 这个唯一漏斗而不是裸 UPDATE:它同时落 timeline、推 SSE,并触发 * `task-settle` 把 task 收成 cancelled(`RUN_TO_TASK_STATUS.stopped = 'cancelled'`)—— * 于是「取消排队」这件事在库里、时间线上、界面上是同一件事,不用在这里再写一遍 task 状态。 * * ⚠️ **这里原本写着「不用 engine.stopRun:对一条还没开跑、没有子 run 的排队 run 是多余的一圈」, * 那个前提被批 3 自己的 `requeueBlockedRootRun` 改没了。** 恢复类转换撞坑满会把 run 退回队列, * 所以 `run_status='queued'` 的 run **可以是跑过一半的**:有 success 节点、有 ready 节点、 * 还可能挂着活着的子 run(预算撞顶时引擎只把 root 置 blocked,子 run 要到节点边界才落 paused, * 也就是父 blocked 时子 run 可以仍在 running)。 * * 探针实测过后果:只发一次 `queued → stopped` 的话,root 落 stopped、task 显示「已取消」, * 而那条 running 的子 run **既没收到 stop_requested、也没被 settle,继续调 agent 烧 token**。 * * 所以改成无条件走 `engine.stopRun`:它置整棵树的 `stop_requested`、自底向上收敛、 * 把 pending/ready 节点 skip 掉。对一条真·空白的排队 run,多出来的那一圈只是几条命中 0 行的 * 条件更新 —— **拿这点开销换「绝不留下还在烧钱的子 run」是划算的**。 * * @returns {boolean} true = 本次由我撤下的;false = 它已经不在队里了(被抢跑 / 已撤过) */ function cancelQueuedRun(task) { const wdb = require('../services/work-graph-db'); const runId = task.current_run_id; if (!runId) return false; const run = wdb.getRun(runId); if (!run || run.run_status !== 'queued') return false; // stopRun 内部逐层走 transitionRun(同一个漏斗),所以 timeline / SSE / task-settle 的 // 收敛回写照旧;这里只负责把「整棵树」交给它,不再自己发那一条转换。 require('../services/graph-engine').stopRun(runId); return Boolean(wdb.getRun(runId) && wdb.getRun(runId).run_status === 'stopped'); } /** * 入队 / 下达 / 重跑三个端点的统一响应体。 * * `started` 只在**本次请求真的把某条 task 起跑了**时出现,且它可能不是 body 里那条 * (队首属于别的 goal 时)—— 前端 `dispatchedTaskId` 认的就是它。 * 批 4 起一次可能起跑多条(N 个执行位),全部在 `startedTasks` 里,`started` = 它的第一条。 * `queuePosition` 一律现算(`task-scheduler.queuePosition`),起跑了的那条不在队里 → null。 * * @param {Array|object|null} started `startNextIfSlotFree` 的返回值(数组), * 或 `promoteQueuedTask` 的单条返回值 —— `startedFields` 两种都收。 */ function queueReply(companyId, task, started) { const { queuePosition } = require('../services/task-scheduler'); const fresh = require('../services/work-graph-db').getTask(task.id) || task; const runId = fresh.current_run_id; return { success: true, taskId: fresh.id, goalId: fresh.goal_id, companyId, runId, status: fresh.status, queuePosition: queuePosition(companyId, fresh.id) || null, ...startedFields(started), ...(runId ? { links: runLinks(runId) } : {}), }; } /** run 的三条自描述链接:状态 / 手动下一步 / SSE 流 */ function runLinks(runId) { return { state: `/api/graph-runs/${runId}`, next: `/api/graph-runs/${runId}/next`, stream: `/api/graph-runs/${runId}/stream`, }; } /** * 一份 `tasks[]` 申报(人写的请求体,或 task-planning 拆出来的 specs)→ 可直接建行的 plans[]。 * * **在任何 INSERT 之前把整份 payload 校验完**(缺陷 F4):原先是边校验边插,第 k 条不合法时 * 0..k-1 条已各自 INSERT 且 goal 也已提交,400 之后库里躺着半个 goal + 半批 task。 * * 三条路共用这一份:同步建 goal(带 tasks)、批 2 的自动拆分产出、以及拆分失败的单条降级。 * 降级路径**必须**落到同一段代码上,否则它会自己长出第二套语义(默认 mode / 预算继承 / seq 编号 * 各差一点点),而那条路只有在 LLM 失败时才走到,差异要很久以后才会被人看见。 * * **seq 是「goal 内的号」,不是「本次 payload 的下标」**(缺陷 4):`createTask` 只在 seq 非整数时 * 才 `MAX(seq)+1`,这里显式给号,而 `idx_tasks_goal` 不是唯一索引 —— 往一个已经有 task 的 goal 里 * 插 1..N 不报错,只会让同一 goal 下出现两条 `seq=1`。所以调用方要为非空 goal 传 `seqBase` * (= 该 goal 现有的 MAX(seq)),本次产出的号从 `seqBase+1` 起。 * `depends_on` 里的**纯数字**引用是 payload 内的 seq,必须跟着同样偏移,否则翻译表对不上; * 非数字的原样透传(那是既有 task 的真 id,task id 恒以 `T-` 开头,不会与数字撞)。 * * @returns {{plans: object[]} | {error: string}} error 非空时调用方翻 400 */ function buildTaskPlans(specs, { priority, budget, mode, seqBase = 0 } = {}) { const base = Number.isInteger(seqBase) && seqBase > 0 ? seqBase : 0; const plans = []; for (let i = 0; i < specs.length; i += 1) { const s = specs[i] || {}; const d = String(s.description || s.task || '').trim(); if (!d) return { error: `tasks[${i}].description 为必填项` }; const m = normalizeMode(s.mode, s.execution_mode, mode); if (m === null) { return { error: `tasks[${i}].mode 只能是 'manual' 或 'auto',收到: ${JSON.stringify(s.mode ?? s.execution_mode)}` }; } const rawDeps = s.depends_on !== undefined ? s.depends_on : s.dependsOn; if (rawDeps !== undefined && rawDeps !== null && !Array.isArray(rawDeps)) { return { error: `tasks[${i}].depends_on 必须是数组` }; } plans.push({ seq: base + i + 1, // seq 显式给,别靠 MAX+1 —— depends_on 的 seq 引用要能对得上服务端分配的号 title: String(s.title || '').trim() || deriveTitle(d), description: d, // 优先级:task 自己的 mode > goal 级 mode(第三个候选)> 'manual'。 // 「继承 goal」这一层是批 2 加的,对显式传 tasks[] 的调用方是行为变化(此前恒 manual), // 保留它是因为反过来更荒谬:用户在表单上把整个 goal 选成 auto,拆出来的 task 却全是 // manual、一条都不会自己跑 —— 那显然不是他要的。goal 级 mode 本就是「这批活怎么跑」 // 的声明,只对拆分产出生效、对显式列出的 task 无效,等于同一个字段两套语义。 // 这条约定已写进 POST /:companyId/goals 的 JSDoc,并由用例钉住(company-goal-task-routes)。 mode: m || 'manual', // undefined = 三级都没给 → 默认 manual;null 在上面已经 400 掉了 priority: s.priority || priority || 'P1', budget: s.budget !== undefined ? s.budget : budget, // 每条可单独给,否则继承 goal 级默认 // 纯数字 = payload 内的 seq 引用,跟着一起偏移;其它是真 id,原样透传 deps: (rawDeps || []).map(String).map((dep) => (/^\d+$/.test(dep) ? String(base + Number(dep)) : dep)), }); } return { plans }; } /** * plans[] → tasks 行,并把 `depends_on` 里的 **seq 引用**翻成真 id。 * * **调用方必须自己开事务**:环检测要等兄弟行都在库里才跑得了(它查的就是同 goal 的 tasks), * 一旦抛错整批要能连 goal 一起回滚,不留孤儿。 * * 走 updateTask 而不是在这里自己判环:环检测的唯一持有者是数据层的 validateDependsOn, * 路由再抄一份就是第二个持有者。 */ function insertTasksForGoal(wdb, goal, plans, createdBy) { const created = plans.map((p) => wdb.createTask({ goalId: goal.id, companyId: goal.company_id, seq: p.seq, title: p.title, description: p.description, executionMode: p.mode, priority: p.priority, budgetJson: p.budget, // 建行时就存下来:草稿路径上请求体活不到 dispatch(缺陷 F3) createdBy, }) ); const bySeq = new Map(created.map((t) => [String(t.seq), t.id])); return created.map((row, i) => { const deps = plans[i].deps; if (!deps.length) return row; const real = deps.map((d) => bySeq.get(d) || d); // 认不出的原样透传,由数据层报「不存在」 return wdb.updateTask(row.id, { dependsOn: real }); }); } /** * 一次自动拆分的**总时限**(缺陷 6)。默认 5 分钟,`GOAL_SPLIT_TIMEOUT_MS` 可覆盖。 * * 为什么必须有:`manager.callAgent` 只有 idle timeout(默认 30 分钟,而且只在**完全没有流数据** * 时才触发),三轮最坏 90 分钟。而前端轮询 3 分钟就放弃 —— 也就是说验收 20 的 * 「绝不卡在 planning」此前只由客户端保障,用户换个浏览器 / 直接查库看到的仍是一个转了一个半小时 * 的 planning。时限要挂在服务端,那才是状态的持有者。 * * 为什么是 5 分钟:比前端那 3 分钟窗口大(真拆到第 3 轮的 goal 不该只因为前端不看了就被判降级), * 又只有最坏值 90 分钟的 1/18。前端放弃后用户再刷新,绝大多数情况 goal 已经落地了。 */ function goalSplitTimeoutMs() { const raw = Number(process.env.GOAL_SPLIT_TIMEOUT_MS); return Number.isFinite(raw) && raw > 0 ? raw : 5 * 60 * 1000; } /** * 后台跑一次自动拆分(§四.9)。**这个函数绝不 throw** —— 它跑在 setImmediate 里, * 没有任何调用方接得住异常,抛出去就是一个 unhandled rejection + 一个永远停在 `planning` 的 goal。 * * 四条出口。前三条把 goal 推离 `planning`,第四条是「已经有人替我推走了」: * ① 拆出合法 DAG → 落 N 条 draft task,goal 回 `draft` * ② 三轮耗尽 / 炸了 / 超时 → 降级成 1 条同描述 task,goal 回 `draft` + 写 planning_error * ③ 连降级都落库失败 → goal 回 `draft` + 写 planning_error(零 task,人可在界面上自己加) * ④ 收工时 goal 已不在 planning → **什么都不插**,只补记 planning_tokens + 一行日志 * * ④ 是缺陷 1 / 5:「推离 planning」的前提是它**还在** planning。拆分要跑几十秒,这期间 * 用户完全可能点了表单自己推荐的逃生口(「跳过,按单任务下达」→ `POST /:cid/tasks`, * 那条路会 `markGoalDispatched` 把 goal 写成 active 并起 run),或者在左栏归档了它。 * 所以落点一律走数据层的条件更新 `settleGoalPlanning`,命中才准插 task。 * * ⚠️ **goal 的落点是 `draft` 不是 `active`**,这一处与 §四.9 约束 2 / §八 验收 20 的字面 * (「goal 转 active」)不同,理由是那两处与本文档自己的另外两条硬规矩冲突: * · §四.9 与 §六.2 都写着「产出一律 draft,**不自动入队**,入队走 dispatch」; * · `active` 的语义由批 1 的 `markGoalDispatched` 钉死 = 「已下达」(连带盖 dispatched_at, * 那是 goal 级的描述冻结点)。拆分失败时把 goal 标成 active 却一条 task 都没起, * 界面上就是一个「进行中」但什么都没在跑的 goal —— 正是本文档反复点名的「界面撒谎」。 * 硬要求「绝不能卡在 planning」两种落点都满足;两者取其一时选不撒谎的那个。 * 真要恢复「失败即下达」,正确做法是在这里显式调 dispatch,而不是只改状态字面量。 */ function runGoalSplit(companyId, goalId, defaults) { const wdb = require('../services/work-graph-db'); const { planTasks } = require('../services/task-planning'); return (async () => { const goal = wdb.getGoal(goalId); if (!goal) return; // 拆分期间被删了:没有可推的对象,直接收工 let specs = null; let tokens = 0; let planningError = null; let timer = null; try { // ponytail: 超时只是**不再等**,planTasks 里那个 CLI 子进程还会自己跑到底, // 它之后烧掉的 token 记不上(拿不到一个还没 resolve 的 promise 里的计数)。 // 上限就是这一条;要精确记账得让 task-planning 收 AbortSignal,中断时把已烧量抛出来。 const planning = planTasks({ companyId, goal }); planning.catch(() => {}); // 超时走掉之后它才 reject 的话,那就是一个 unhandled rejection const budgetMs = goalSplitTimeoutMs(); const r = await Promise.race([ planning, new Promise((_, reject) => { timer = setTimeout( () => reject(new Error(`自动拆分超过总时限 ${Math.round(budgetMs / 1000)}s,已降级成单任务`)), budgetMs ); // 这个定时器不该把进程钉在事件循环里:拆分是后台副业,没有它进程该退就退 if (timer.unref) timer.unref(); }), ]); specs = Array.isArray(r && r.specs) && r.specs.length ? r.specs : null; tokens = (r && r.tokens) || 0; if (!specs) planningError = 'task-planning 返回了空的任务列表'; } catch (e) { // 已经烧掉的 token 挂在 error 上(task-planning 的两条抛出路径都会挂),照样记账: // 失败的那几轮是真花了钱的,不记 = 这半边花费永远看不见。 tokens = e.tokens || 0; planningError = e.message; } finally { if (timer) clearTimeout(timer); } // 降级:1 条同描述的 task。created_by 是 'human' 而不是 'task-planning' —— // 这条根本不是它拆出来的,记成它的产出会让「拆分质量」这类统计从一开始就是脏的。 // **但这条 task 的描述留到事务里现取**,不在这里定死,理由见下面 `degradeSpec` 的调用处。 const degraded = !specs; if (degraded) specs = [{ seq: 1, title: goal.title, description: goal.description, depends_on: [] }]; try { // **先判后插**(缺陷 1 / 缺陷 5):条件更新是唯一的准入判据 —— 命中才说明这个 goal // 在这几十秒里没被人推走(逃生口下达 / 归档)。没命中还照插,插进去的就是一批孤儿: // goal 已经在跑别的(或已归档),这些 draft task 谁都不会来收, // 而 settleGoalIfAllTasksDone 见到非终态就永不收敛。 // // seq 也在这个事务里现取(缺陷 4):逃生口可能已经占掉了 seq=1,1..N 硬编会撞号。 // better-sqlite3 是同步单写者,同事务内取到的 MAX(seq) 到插入之间不会被别的请求插队。 const outcome = wdb.getDb().transaction(() => { const settle = wdb.settleGoalPlanning(goalId, { status: 'draft', planningTokens: tokens, planningError }); if (!settle.settled) return { settled: false }; // **降级那条 task 的描述在这里现取**:函数顶部读到的是几十秒前的快照,而这段时间里 // `PATCH /goals/:goalId` 改 description 是被允许的(goal 还在 planning,dispatched_at // 还空,冻结规则没生效)。用旧快照落库,结果是 goal 上写着新需求、它唯一的那条 task // 写着旧需求 —— 而两段文字都是验收边界(org-validator 判的就是这段描述), // 一份目标两个口径,验收从此无解,且没有任何报错提示。 // 位置卡在 CAS 之后、同一个事务内:CAS 命中才说明这个 goal 这一刻仍归拆分所有; // better-sqlite3 是同步单写者,同事务内读到的这一行就是 task 将要引用的最终版本。 // 只有降级路径需要重读 —— 拆分成功时 specs 是 LLM 基于旧描述产出的,重读也换不掉它们。 const fresh = degraded ? wdb.getGoal(goalId) || goal : null; const effectiveSpecs = fresh ? [{ seq: 1, title: fresh.title, description: fresh.description, depends_on: [] }] : specs; const built = buildTaskPlans(effectiveSpecs, { ...defaults, seqBase: wdb.maxTaskSeq(goalId) }); if (built.error) throw new Error(`拆分产出无法建行:${built.error}`); insertTasksForGoal(wdb, goal, built.plans, planningError ? 'human' : 'task-planning'); return { settled: true, count: built.plans.length }; })(); if (!outcome.settled) { // 无声放弃是这一批要消灭的症状本身,所以这里必须留一行:花掉的 token 已经补记(见 // settleGoalPlanning),丢掉的是拆分产出。人看到这行才知道「我点了跳过」和「拆分白跑了」是同一件事。 console.warn( `[TaskEntry] goal ${goalId} 拆分收工时已不在 planning(多半是用户点了「跳过,按单任务下达」或归档了它),` + `放弃落库 ${specs.length} 条 task,只补记 ${tokens} tokens` ); } } catch (e) { console.error(`[TaskEntry] goal ${goalId} 拆分产物落库失败: ${e.message}`); // 落库失败也必须把 goal 推离 planning,否则前一条 catch 保住的东西在这里又丢了。 // 同样走条件更新:这条路上 goal 一样可能已经被人推走。 try { wdb.settleGoalPlanning(goalId, { status: 'draft', planningTokens: tokens, planningError: `${planningError ? `${planningError};` : ''}拆分产物落库失败:${e.message}`, }); } catch (e2) { console.error(`[TaskEntry] goal ${goalId} 状态回写也失败了: ${e2.message}`); } } })(); } /** 公司存在性 + goal 归属校验,返回 goal 行;不合法时已经 res 过了,调用方直接 return */ function requireGoal(req, res) { const { getGoal } = require('../services/work-graph-db'); const goal = getGoal(req.params.goalId); if (!goal || goal.company_id !== req.params.companyId) { res.status(404).json({ error: `goal ${req.params.goalId} 不存在于公司 ${req.params.companyId}` }); return null; } return goal; } /** 同上,task 版 */ function requireTask(req, res) { const { getTask } = require('../services/work-graph-db'); const task = getTask(req.params.taskId); if (!task || task.company_id !== req.params.companyId) { res.status(404).json({ error: `task ${req.params.taskId} 不存在于公司 ${req.params.companyId}` }); return null; } return task; } /** * GET /api/company/:companyId/goals — goal 列表 * query: status / includeArchived / withTasks=1(带出每个 goal 的 task 摘要) * * withTasks 存在的理由就是**不让前端 N+1**:一次把全公司的 task 拉回来在内存里归组, * 进度另走一条 GROUP BY(验收 10)。 */ router.get('/:companyId/goals', (req, res) => { try { const wdb = require('../services/work-graph-db'); const companyId = req.params.companyId; // 公司不存在 → 404(缺陷 F9)。不查的话打错一个字母就回 200 + 空列表, // 与「这家公司确实还没有目标」完全无法区分 —— 排查会从「为什么目标丢了」开始, // 而真相是 URL 里的 id 拼错了。本文件其它端点都有这道兜底,两个新 GET 漏了。 if (!companyService.get(companyId)) { return res.status(404).json({ error: `公司 ${companyId} 不存在` }); } const includeArchived = req.query.includeArchived === '1' || req.query.includeArchived === 'true'; const goals = wdb.listCompanyGoals(companyId, { status: req.query.status || null, includeArchived }); if (req.query.withTasks !== '1' && req.query.withTasks !== 'true') { return res.json({ goals: goals.map((g) => goalView(g)), slots: slotsView(companyId) }); } const tasks = wdb.listCompanyTasks(companyId, { includeArchived }); const progress = progressOf(tasks); // 位次一次算全公司(一条查询),下面逐行 get —— 不是每行问一次调度器(那是列表页的 N+1) const positions = queuePositions(companyId); const byGoal = new Map(); for (const t of tasks) { if (!byGoal.has(t.goal_id)) byGoal.set(t.goal_id, []); byGoal.get(t.goal_id).push(t); } res.json({ goals: goals.map((g) => goalView( g, // 两级树的显示顺序 = seq(listGoalTasks 的口径)。队列位次是另一回事,找调度器要。 (byGoal.get(g.id) || []) .sort((a, b) => a.seq - b.seq || a.created_at.localeCompare(b.created_at)) .map((t) => taskView(t, progress.get(t.current_run_id), positions)) ) ), slots: slotsView(companyId), }); } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * POST /api/company/:companyId/goals — 新建 goal * * body: { * title?, description | task, priority?, okr_node_id?, report_to?, budget?, * tasks?: [{ title?, description, mode?, priority?, budget?, depends_on? }] * } * * **两条路,响应码不同,调用方靠 `split` 字段分辨**(§四.9): * * · **同步(`split: false`,或显式传了 `tasks`)** → `201`,body 里就有全部 task。 * 显式传了 tasks 还去调 agent 是纯浪费:调用方已经把拆分结果给出来了。 * 存量调用方 `POST /:cid/tasks` 走的也是这一条 —— **它们不该因为这次改动突然多花一次 LLM 调用**。 * * · **异步(默认 `split: true`)** → `202`,goal 停在 `planning`、`tasks: []`, * 后台调 task-planning 拆分,几十秒后 task 才出现(前端轮询 GET /goals/:goalId)。 * `planning` 是个**真会持续几十秒**的状态,但它有出口:runGoalSplit 的三条路径全都会把它推走, * 失败时降级成 1 条同描述的 task 并写 `planningError`。绝不留一个永远转圈的 goal。 * * 两条路的产出都是 `draft`,**都不自动下达**(下达走 POST .../dispatch)。 * * **`mode` 是三级继承,goal 级对显式 tasks[] 同样生效**:`tasks[i].mode` > body 顶层的 * `mode`/`execution_mode` > `'manual'`。也就是说 `{mode:'auto', tasks:[{description:'…'}]}` * 建出来的每条 task 都是 auto,不是 manual。写在这里是因为它是一条**明写的约定**而不是实现细节: * 顶层 mode 声明的是「这批活怎么跑」,只对拆分产出生效、对手写 tasks[] 无效的话,同一个字段 * 就有了两套语义。要让某条 task 反过来是 manual,在那条 task 上显式写 `mode:'manual'`。 * * **`tasks[].depends_on` 的两种引用形态**(缺陷 F5):同一次 POST 里兄弟 task 的 id 由服务端 * 生成,调用方在写请求体时根本无从引用 —— 只认真 id 的话这个入参结构上只有传空数组才可能成功, * 而批 2 的 task-planning 拆分产出的正是「第 2 条依赖第 1 条」这种形状(§四.9 的产出契约就写着 * `depends_on 只引用本次产出的 seq`)。所以这里收两种: * - **数字**(或纯数字字符串)→ 本次 payload 内的 `seq`,1-based,即 `tasks[]` 的下标 + 1; * - **其它字符串** → 既有 task 的真 id(本端点建的是新 goal,同 goal 内没有旧 task, * 所以这条路只会走到数据层的「只能引用同 goal 内已存在的 task」错误上;留着它是为了 * 和 `PATCH /tasks/:id` 的入参形态一致,调用方不必按端点切换写法)。 * 响应里的 `tasks[].dependsOn` 一律是**翻译后的真 id**。 */ router.post('/:companyId/goals', (req, res) => { const companyId = req.params.companyId; try { const body = req.body || {}; const description = String(body.description || body.task || '').trim(); if (!description) return res.status(400).json({ error: 'description(需求原文 = 验收边界)为必填项' }); if (!companyService.get(companyId)) { return res.status(404).json({ error: `公司 ${companyId} 不存在` }); } const wdb = require('../services/work-graph-db'); const goalTitle = String(body.title || '').trim() || deriveTitle(description); const priority = body.priority || 'P1'; const createdBy = body.created_by || 'human'; const goalFields = { companyId, title: goalTitle, description, priority, okrNodeId: body.okr_node_id || null, createdBy, reportTo: body.report_to || null, }; // goal 级 mode 在这里就过一次信任边界:拆分产出的 task 没有自己的 mode,全靠继承它, // 而校验推迟到 buildTaskPlans 会把「你 goal 上的 mode 写错了」报成「tasks[0].mode 收到 undefined」。 const goalMode = normalizeMode(body.mode, body.execution_mode); if (goalMode === null) { return res.status(400).json({ error: `mode 只能是 'manual' 或 'auto',收到: ${JSON.stringify(body.mode ?? body.execution_mode)}` }); } // 拆分产出的每条 task 要继承的 goal 级默认值。**只用值不用请求对象**: // 后台拆分跑在响应发出之后,把 req.body 整个带进闭包等于让一份可变的外部输入活过请求生命周期。 const defaults = { priority, budget: body.budget, mode: goalMode }; // 显式给了 tasks 就是「拆分结果已经在请求体里了」,再调一次 agent 是纯浪费。 const hasExplicitTasks = Array.isArray(body.tasks) && body.tasks.length > 0; const wantSplit = body.split !== false && !hasExplicitTasks; // ── 异步路径:先建 goal(planning)+ 立刻 202,拆分在后台跑 ────────────── if (wantSplit) { let goal; wdb.getDb().transaction(() => { // createGoal 的 status 恒为 'draft'(写死在 INSERT 里,建不出 planning 的 goal), // 所以是「建完立刻改」。两句裹一个事务:中间崩掉会留下一个 draft 且零 task 的 goal, // 那种 goal 界面上是「草稿,等你加任务」,没人会去等一个不存在的拆分。 goal = wdb.createGoal(goalFields); goal = wdb.updateGoal(goal.id, { status: 'planning' }) || goal; })(); // ponytail: 拆分只活在本进程里(setImmediate + 内存里的 promise)。进程在拆分中途被杀, // 这个 goal 会永久停在 'planning' —— 没有任何人会重启它。上限就是这一条: // 起服务时扫一遍 status='planning' 的 goal 并重推,属批 3 的中断恢复,与调度器一起做。 setImmediate(() => { runGoalSplit(companyId, goal.id, defaults).catch((e) => // runGoalSplit 内部已经全兜了;这里只防「兜底代码自己炸了」变成 unhandled rejection console.error(`[TaskEntry] goal ${goal.id} 拆分兜底失败: ${e.message}`) ); }); return res.status(202).json({ success: true, split: true, ...goalView(goal, []), // tasks: [] —— 拆分还没产出,别让调用方以为这个 goal 是空的 links: { self: `/api/company/${companyId}/goals/${goal.id}` }, }); } // ── 同步路径:没给 tasks 就退化成 1 条同描述的 task ───────────────────── const specs = hasExplicitTasks ? body.tasks : [{ title: goalTitle, description }]; const built = buildTaskPlans(specs, defaults); if (built.error) return res.status(400).json({ error: built.error }); // 建行整段一个事务 —— depends_on 的环检测要等兄弟行都在库里才跑得了(它查的就是同 goal // 的 tasks),一旦抛错整批连 goal 一起回滚,不留孤儿。 let goal; let created = []; try { wdb.getDb().transaction(() => { goal = wdb.createGoal(goalFields); created = insertTasksForGoal(wdb, goal, built.plans, createdBy); })(); } catch (err) { const handled = sendDomainError(res, err); if (handled) return; throw err; } res.status(201).json({ success: true, split: false, ...goalView(goal, created.map((t) => taskView(t))), links: { dispatch: `/api/company/${companyId}/goals/${goal.id}/dispatch` }, }); } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** GET /api/company/:companyId/goals/:goalId — goal + 全部 task + 每条 task 的当前 run 摘要 */ router.get('/:companyId/goals/:goalId', (req, res) => { try { const wdb = require('../services/work-graph-db'); const goal = requireGoal(req, res); if (!goal) return; const tasks = wdb.listGoalTasks(goal.id, { includeArchived: req.query.includeArchived === '1' || req.query.includeArchived === 'true', }); const progress = progressOf(tasks); const positions = queuePositions(goal.company_id); res.json( goalView( goal, tasks.map((t) => { const view = taskView(t, progress.get(t.current_run_id), positions); // run 摘要按主键逐条查:一个 goal 内的 task 是个位数,这里的 N 有硬上限, // 不是列表页那种随公司规模长的 N+1(进度仍然只走一条 GROUP BY)。 const run = t.current_run_id ? wdb.getRun(t.current_run_id) : null; view.run = run ? { runId: run.id, runStatus: run.run_status, executionMode: run.execution_mode, updatedAt: run.updated_at } : null; return view; }) ) ); } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * PATCH /api/company/:companyId/goals/:goalId — 改 title / priority / okr_node_id * dispatched_at 非空时带 description 一律 409(§四.4)。 */ router.patch('/:companyId/goals/:goalId', (req, res) => { try { const { updateGoal } = require('../services/work-graph-db'); const goal = requireGoal(req, res); if (!goal) return; const body = req.body || {}; const patch = {}; if (body.title !== undefined) patch.title = String(body.title); if (body.priority !== undefined) patch.priority = String(body.priority); if (body.okr_node_id !== undefined) patch.okrNodeId = body.okr_node_id || null; // description 照传给数据层,由它按冻结规则决定放行还是抛 —— 冻结判据只留数据层一处, // 路由这边再抄一遍 `if (goal.dispatched_at)` 就是第二个持有者。 if (body.description !== undefined) patch.description = String(body.description); try { res.json(goalView(updateGoal(goal.id, patch))); } catch (err) { const handled = sendDomainError(res, err); if (handled) return; throw err; } } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * POST /api/company/:companyId/goals/:goalId/dispatch — 下达这个 goal * * **现在的语义(批 3 起)= 下达即入队**:给 goal 盖上 dispatched_at(描述从此冻结), * 把它下面**所有** draft task 入队,再推一次队伍(见下方 startNextIfSlotFree); * 起跑了哪几条如实写在响应的 `started` / `startedTasks` 里。 * * 批 1 的语义是「只起第一条上游就绪的,其余原地不动」,理由是:当时一家公司同时只有一个 * active root run(唯一索引 idx_graph_runs_one_active_root),而出队的 worker 是批 3 才有的 * 东西 —— 把 N 条都标成「已入队」而没有任何东西会把它们推走,就是把「静默丢任务」写进数据。 * 批 3 有了调度器,这条理由消失;批 4 DROP 了那条索引,执行位数改为 per-company。 * * **依赖是硬约束**(缺陷 F6 + 第二轮复核 缺陷 4):绝不起一条上游没跑完的 task —— * 「接口收下了 depends_on、返回成功、随后无声地违反它」比不支持依赖更糟:用户设了 * 「部署依赖构建」,dispatch 却起了 seq 最小的部署,图跑完了他也不会知道顺序错了。 * 但「挡」不等于「拒绝整个 goal」:有别的 draft 跑得动就起那一条(见下方挑选逻辑), * 一条都跑不动才 409。批 1 的自救路径:draft 状态可以 `PATCH /tasks/:id` 改 depends_on 或 seq * (两条都真的通,seq 是缺陷 4 一起补上的 —— hint 里写着的自救路径必须存在)。 */ router.post('/:companyId/goals/:goalId/dispatch', (req, res) => { const companyId = req.params.companyId; try { const wdb = require('../services/work-graph-db'); const goal = requireGoal(req, res); if (!goal) return; const drafts = wdb.listGoalTasks(goal.id).filter((t) => t.status === 'draft'); if (!drafts.length) { return res.status(409).json({ error: `goal ${goal.id} 下没有待下达的 task`, hint: `先 POST /api/company/${companyId}/tasks 往这个 goal 里加一条(带 goal_id)`, }); } const graph = orgGraphService.getGraph(companyId); if (!graph.nodes || graph.nodes.length === 0) { return res.status(409).json({ error: `公司 ${companyId} 还没有 org graph,无法接任务`, hint: `先调 POST /api/company/${companyId}/org-graph/replan 让 orggraph-planning 规划组织结构`, }); } // ⚠️ **这里以前有一道「公司已有 active root run → 409」的预检查,批 3 删掉了它** —— // 那正是本批要消灭的东西:下达不再因为「有别的任务在跑」被拒绝,而是排队。 // 并发下真正的守卫仍在(`updateRunStatus` 的准入事务;批 4 DROP 掉唯一索引之后它是唯一那道), // 只是它守的是**出队**那一跳,不是入队。入队永不因位满失败:queued 的 run 不占执行位。 // // 同样删掉的还有「挑第一条跑得动的 draft、全都跑不动就 409」那一段。§六.2 写的是 // 「把该 goal 下**所有** draft task 入队」—— 有了队列,依赖没就绪不再是拒绝的理由, // 而是「先排着,上游 completed 了调度器自己会挑它出来」。顺带消灭了那条死结: // `validateDependsOn` 只拒环不拒正向引用,`tasks[0].depends_on:[tasks[1]]` 建得出来, // 而旧 dispatch 恒 409,这个 goal 通过 API 永远下达不了。 const enqueued = []; for (const d of drafts) { enqueued.push(enqueueTaskRun({ companyId, taskRow: d, // 优先级:本次 dispatch 的 body > 建 task 时存下的 budget_json > 默认值(缺陷 F3)。 // 中间那一环原先根本没读,于是「409 已存为草稿 → 稍后 dispatch」这条被 hint 明确推荐的 // 路径上,用户提交时填的预算被静默丢弃、run 拿着 DEFAULT_TASK_BUDGET 跑到一半 blocked。 // body 覆盖后不回写 task 行:budget 在 queued_at 之后是冻结字段(§四.4), // 草稿行留的是「当初申报的额度」,本次实际用量以 run 的 budget_json 为准。 budget: (req.body && req.body.budget) || parseBudget(d.budget_json) || DEFAULT_TASK_BUDGET, })); } const dispatchedGoal = markGoalDispatched(goal); // 入队完才推队伍:tick 要看到这一批已经躺在队里,否则本次下达的第一条要等下一次唤醒。 const started = startNextIfSlotFree(companyId); const positions = queuePositions(companyId); res.status(201).json({ success: true, goalId: dispatchedGoal.id, goalStatus: dispatchedGoal.status, // 本次入队的全部 task 及其位次。起跑了的那条不在队里 → position 为 null(不是 0) queued: enqueued.map((e) => ({ taskId: e.task.id, runId: e.runId, queuePosition: positions.get(e.task.id) || null })), // `started` / `startedTasks` 只在本次请求真的把 task 起跑了时出现,**且未必是本次入队的那几条**: // 队列是公司级的,位子空时顶上来的可能是别的 goal 早就排着的活。前端 dispatchedTaskId // 认的就是 `started` —— 如实点名比假装「你下达的那条起跑了」有用。 // 批 4:公司有 N 个执行位时一次能起跑多条,全在 `startedTasks` 里(`started` = 它的第一条)。 ...startedFields(started), ...(started.length ? { links: runLinks(started[0].runId) } : {}), }); } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * DELETE /api/company/:companyId/goals/:goalId — 归档(连带归档其下 task) * * 归档只是隐藏:行还在,`includeArchived=1` 随时查得回来。有 task 还攥着 run 就先 409 * (判据见 `taskHoldsRun`:running / blocked / run 仍在活跃取值域)—— * 把一条正在烧钱的 run 的 task 藏起来,等于让它在界面上消失却继续跑。 */ router.delete('/:companyId/goals/:goalId', (req, res) => { try { const wdb = require('../services/work-graph-db'); const goal = requireGoal(req, res); if (!goal) return; // 正在自动拆分(缺陷 5):此刻的 task 列表恒为空,下面那道 409 必过 —— 归档只盖 archived_at、 // 不动 status,几十秒后拆分照样往这个已经不可见的 goal 里插 N 条活 task。 // 现在 settleGoalPlanning 的 `archived_at IS NULL` 已经挡住了插入,但那是**静默丢掉**一次 // 真花了钱的拆分。在这里直接拒绝,用户才知道自己踩的是什么,并拿到两条明确的出路。 if (goal.status === 'planning' && !goal.archived_at) { return res.status(409).json({ error: `goal ${goal.id} 正在自动拆分中(status=planning),拆完才能归档`, hint: `等它拆完(通常几十秒,最长 ${Math.round(goalSplitTimeoutMs() / 1000)}s)再归档;` + `等不了就先用逃生口按单任务下达:POST /api/company/${goal.company_id}/tasks {"goal_id":"${goal.id}", ...}`, }); } const tasks = wdb.listGoalTasks(goal.id); const running = tasks.filter(taskHoldsRun); if (running.length) { return res.status(409).json({ error: `goal ${goal.id} 下还有正在执行或挂起的 task`, runningTasks: running.map((t) => t.id), // 不硬写 `/api/graph-runs/${current_run_id}/stop`:`taskHoldsRun` 对 blocked 一律返回 // true,而 blocked 的 task 可以根本没有 run(enqueue_incomplete → URL 里就是字面量 // null,404),也可以 run 还躺在 queued(/stop 只认 running/paused/blocked,400)。 // 判据取 escapeHatchHint 那一份,它按 run 的真实状态给一条打得通的路。 hint: `先处置 ${running[0].id}:${escapeHatchHint(goal.company_id, running[0])}`, }); } // 排队中的任务要先把 run 从队里撤下再归档 —— 与 `DELETE /tasks/:taskId` 同一条理由: // `archived_at` 一盖上,`listQueue`(WHERE archived_at IS NULL)就再也看不见这条 task, // 而那条 `run_status='queued'` 的 run 永远留在库里:没人会出队它(没有 task 指向), // 也没人会停它。goal 级归档一次能盖 N 条,漏掉就是一次留 N 条幽灵,且全程不报错。 const unqueued = []; for (const t of tasks) { if (t.archived_at || t.status !== 'queued') continue; if (cancelQueuedRun(t)) { unqueued.push(t.id); continue; } // 撤不下来 = 上面那道 taskHoldsRun 查过之后、这一行之前,它被出队跑起来了(跨进程才可能)。 // **绝不能接着归档**:§13.2 ① 记着的事故就是把一条正在烧预算的 run 从界面上藏掉。 // 已经撤下的那几条如实报出来(它们本来就是本次归档的对象),不静默。 const fresh = wdb.getTask(t.id) || t; return res.status(409).json({ error: `goal ${goal.id} 下的 task ${t.id} 刚刚被出队跑起来了(当前 ${fresh.status}),归档中止`, taskId: t.id, status: fresh.status, unqueuedTasks: unqueued, hint: escapeHatchHint(goal.company_id, fresh), }); } const now = new Date().toISOString(); for (const t of tasks) { if (!t.archived_at) wdb.updateTask(t.id, { archivedAt: now }); } res.json({ success: true, goalId: goal.id, archivedTasks: tasks.filter((t) => !t.archived_at).length, goal: goalView(wdb.updateGoal(goal.id, { archivedAt: now })), }); } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * GET /api/company/:companyId/tasks — task 列表(新端点:这个路径下此前只有 POST) * query: status / goalId / includeArchived / limit * * 进度走 taskProgress 的一条 GROUP BY,不按行查(验收 10)。 * 排序沿用数据层的 created_at DESC —— **队列位次的口径(priority → seq → created_at) * 只留调度器一处**,这里再写一份就是第二个持有者,canDispatch 被前端复述了一遍还多写一个值、 * 导致公司失败一次后再也下不了任务,就是这么来的。 */ router.get('/:companyId/tasks', (req, res) => { try { const wdb = require('../services/work-graph-db'); // 同 GET /goals:公司不存在要 404,不能拿空列表冒充「这家公司没有任务」(缺陷 F9) if (!companyService.get(req.params.companyId)) { return res.status(404).json({ error: `公司 ${req.params.companyId} 不存在` }); } const limit = parseInt(req.query.limit, 10); const tasks = wdb.listCompanyTasks(req.params.companyId, { status: req.query.status || null, goalId: req.query.goalId || req.query.goal_id || null, includeArchived: req.query.includeArchived === '1' || req.query.includeArchived === 'true', limit: Number.isInteger(limit) ? limit : null, }); const progress = progressOf(tasks); const positions = queuePositions(req.params.companyId); res.json({ tasks: tasks.map((t) => taskView(t, progress.get(t.current_run_id), positions)), slots: slotsView(req.params.companyId), }); } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * POST /api/company/:companyId/tasks — 用户给公司下达一个任务 * * body: { * task | description, 必填,一句自然语言任务(= task-goal,27-NAMING §四) * goal_id?, 挂到既有 goal 下;**不传则自动建一个同描述的 goal**。 * ⚠️ 对外可见的破坏性变更:旧代码把它原样存进 * graph_runs.goal_id 且不校验,现在它必须是 goals 表里 * 本公司的真行,否则 404。外部 agent 若一直往这里传 * OKR 节点 id 之类的值,会从 201 变 404 —— 那种调用方 * 本来就在制造悬空外键(正是本批要补的窟窿),404 是对的。 * okr_node_id?, 只写进 goals 表,不进 run(§四.7 禁令一) * mode?: 'manual' | 'auto', 默认 manual;auto 前端必须先弹预估成本确认(§四.B.4) * budget?, 覆盖 DEFAULT_TASK_BUDGET,只挂 root run * report_to? 收敛后要唤醒的上级 agent(Master/CEO 代下单时传)—— * 现在会**落库**到 goals/tasks.report_to(§四.7 禁令二) * } * * **行为兼容是硬要求**:存量调用方一行不用改还能用(前端的 DispatchTaskButton 已随 GoalTaskPane * 落地删除,现在剩下的调用方是**外部 agent 的 curl** —— 那条 URL 与请求体写在 tools-base.md 里, * 等于所有 agent 的提示词,动它就是同时改所有 agent), * 响应体只是多出 goalId。为什么必须自动建 goal 而不是允许 task 悬空:goal 是验收边界的持有者, * 没有 goal 的 task 在两级树里无处可挂 —— 那正是本次要修的「界面上画不出来」。 */ router.post('/:companyId/tasks', (req, res) => { const companyId = req.params.companyId; try { const body = req.body || {}; const description = String(body.task || body.description || '').trim(); if (!description) return res.status(400).json({ error: 'task(任务描述)为必填项' }); const company = companyService.get(companyId); if (!company) return res.status(404).json({ error: `公司 ${companyId} 不存在` }); const requestedMode = normalizeMode(body.mode, body.execution_mode); if (requestedMode === null) { return res.status(400).json({ error: `mode 只能是 'manual' 或 'auto',收到: ${JSON.stringify(body.mode ?? body.execution_mode)}` }); } const mode = requestedMode || 'manual'; // 没有 org graph 就没有可执行的图(graph_kind='org' 的定义层就是它) const graph = orgGraphService.getGraph(companyId); if (!graph.nodes || graph.nodes.length === 0) { return res.status(409).json({ error: `公司 ${companyId} 还没有 org graph,无法接任务`, hint: `先调 POST /api/company/${companyId}/org-graph/replan 让 orggraph-planning 规划组织结构`, }); } const wdb = require('../services/work-graph-db'); // ⚠️ **这里以前有一道「公司已有 active root run → 409」的预检查,批 3 删掉了它。** // 这条 409 就是本批要消灭的那个东西:外部 agent 的 curl(URL 与请求体写在 tools-base.md 里, // 等于所有 agent 的提示词)撞上它只能自己重试或放弃。现在改成入队 —— 建一条 // `run_status='queued'` 的 root run,它**不占**执行位,所以入队永不失败。 const createdBy = body.created_by || body.from || 'human'; let goal; if (body.goal_id) { goal = wdb.getGoal(body.goal_id); if (!goal || goal.company_id !== companyId) { return res.status(404).json({ error: `goal ${body.goal_id} 不存在于公司 ${companyId}` }); } } else { goal = wdb.createGoal({ companyId, title: deriveTitle(description), description, priority: body.priority || 'P1', okrNodeId: body.okr_node_id || null, createdBy, reportTo: body.report_to || null, }); } let taskRow; try { taskRow = wdb.createTask({ goalId: goal.id, companyId, title: deriveTitle(description), description, executionMode: mode, priority: body.priority || goal.priority, // 落库(缺陷 F3):本次起不来 run 时(批 3 之前是 409,现在是排队)这行就是草稿,hint 让用户稍后 // dispatch —— 那时请求体早没了,不存下来 dispatch 只能用默认额度,而用户以为自己填的算数。 budgetJson: body.budget !== undefined ? body.budget : null, // 显式传 report_to 优先,其次继承 goal 的(数据层也会兜一次,这里传的是本次请求的意图) reportTo: body.report_to || null, createdBy, }); } catch (err) { const handled = sendDomainError(res, err); if (handled) return; throw err; } const launched = enqueueTaskRun({ companyId, taskRow, budget: body.budget || DEFAULT_TASK_BUDGET }); markGoalDispatched(goal); // 坑空就立刻起跑 —— 存量调用方(外部 agent 的 curl)的心智是「下达即执行」, // 队里没人时必须还是那个行为,否则等于把一条现在走得通的路改成要多打一次接口。 // 坑被占着就老老实实排队,响应里的 status/queuePosition 说清它在等什么。 const started = startNextIfSlotFree(companyId); const fresh = require('../services/work-graph-db').getTask(taskRow.id) || taskRow; res.status(201).json({ success: true, runId: launched.runId, taskId: taskRow.id, goalId: goal.id, companyId, // 从 **goal** 上回,不是从 run 上 —— run 不再带 okr_node_id(§四.7 禁令一) okrNodeId: goal.okr_node_id, executionMode: mode, // 批 3 起这里可能是 'queued':不再有 409,「没坑」的答案从错误变成状态。 // 从库里回读而不是用 launched.run —— 那是入队那一刻的快照,startNextIfSlotFree 可能已经把它推走了。 status: fresh.status, runStatus: (require('../services/work-graph-db').getRun(launched.runId) || {}).run_status, queuePosition: require('../services/task-scheduler').queuePosition(companyId, taskRow.id) || null, task: description, taskGoalArtifact: path.relative(REPO_ROOT, launched.artifact), budget: body.budget || DEFAULT_TASK_BUDGET, // 本次这条已经在上面逐字报过了(runId / status / runStatus),这里只报**别人**: // 顶上来的可能是别的 goal 早就排着的活,一次可能不止一条(批 4 的 N 个执行位)。 ...startedFields(started.filter((s) => s.task.id !== taskRow.id)), links: runLinks(launched.runId), }); } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * PATCH /api/company/:companyId/tasks/:taskId — 改 title / priority / seq / depends_on * * queued_at 非空时带 description / mode / budget 一律 409(§四.4); * depends_on 在还没跑起来的状态下可改(§六.3 定案 + §五.1.2 的「强制放行」),跑起来之后不可改。 * 两条判据都在数据层(`work-graph-db.updateTask`),这里只做入参翻译 + 改完依赖后推一次队列。 */ router.patch('/:companyId/tasks/:taskId', (req, res) => { try { const { updateTask } = require('../services/work-graph-db'); const task = requireTask(req, res); if (!task) return; const body = req.body || {}; const patch = {}; if (body.title !== undefined) patch.title = String(body.title); if (body.priority !== undefined) patch.priority = String(body.priority); // seq 可改(第二轮复核 缺陷 4):dispatch 的 409 hint 一直写着「顺序不对就改依赖或 seq」, // 而这里从不读 body.seq —— 两条自救路径只有一条通,用户照 hint 做完还是同样的 409, // 且接口回 200 假装改过了。列 TASK_PATCH_COLUMNS.seq 早就在,缺的只是这一行入参翻译。 // 必须挡非整数:seq 直接进 ORDER BY,一个 NaN / 'abc' 会让下达顺序变成谁也说不清的东西。 if (body.seq !== undefined) { const n = Number(body.seq); if (!Number.isInteger(n)) { return res.status(400).json({ error: `seq 必须是整数,收到: ${JSON.stringify(body.seq)}` }); } patch.seq = n; } if (body.depends_on !== undefined) patch.dependsOn = body.depends_on; else if (body.dependsOn !== undefined) patch.dependsOn = body.dependsOn; // 下面三个是冻结字段:照传给数据层,让它按 queued_at 决定放行还是抛 409 if (body.description !== undefined) patch.description = String(body.description); // mode 在这里就要过信任边界校验(缺陷 F7):直送 DB 会撞 tasks.execution_mode 的 CHECK 约束, // 变成 500 + 一句 SQLite 原文 —— 500 的意思是「服务端出错了」,用户会重试;正确答案是 400。 // 值必须显式区分「没传」与「传了 null」:`{"mode": null}` 走 `!== undefined` 过滤会被当没传, // 于是回 200 而什么都没改。 const patchedMode = normalizeMode(body.mode, body.execution_mode); if (patchedMode === null) { return res.status(400).json({ error: `mode 只能是 'manual' 或 'auto',收到: ${JSON.stringify(body.mode ?? body.execution_mode)}` }); } if (patchedMode) patch.executionMode = patchedMode; if (body.budget !== undefined) patch.budgetJson = JSON.stringify(body.budget); try { let updated = updateTask(task.id, patch); // 改完依赖要**立刻重算一次队列**(27-P3C §五.1.2)。 // // blocked 是每轮 tick 从依赖推导出来的**结果**,不是不可逆状态。少了这一跳, // 「强制放行」(删掉那条已经死了的上游依赖)就只是把 depends_on 写进库:任务照旧躺在 // blocked、blocked_reason 还指着一个已经不在依赖里的上游,要等 60s 兜底 sweep 才自愈—— // 界面上就是「点了没反应」。tick 是同步的(better-sqlite3),且绝不 throw。 // // **blocked_reason 不在这里单独清**:清它和 blocked → queued 是同一条条件更新 // (`requeueDerivedBlockedTask` 的 WHERE 里就有 `blocked_reason LIKE 'upstream_failed:%'`)。 // 抢在 tick 前面把 reason 清成 NULL,那条 UPDATE 就再也命中不了,任务会永远卡在 blocked // 且没有任何报错 —— 正是本轮要消灭的那类「写进去了却更糟」。 // // 回读一次再吐:不回读的话响应里带的是 tick 之前的快照,前端刷成「还是挂起」, // 而库里其实已经回队列了。 if (patch.dependsOn !== undefined) { require('../services/task-scheduler').tick(task.company_id); updated = require('../services/work-graph-db').getTask(task.id) || updated; } res.json(taskView(updated, null, queuePositions(task.company_id))); } catch (err) { const handled = sendDomainError(res, err); if (handled) return; throw err; } } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * POST /api/company/:companyId/tasks/:taskId/enqueue — 单条入队(27-P3C §六.3) * * 与 `goals/:goalId/dispatch` 的差别只是粒度:那条把 goal 下所有 draft 一起推进队列, * 这条只推一条。有了它,「在既有目标下补一条任务」才走得通 —— 在此之前前端那句 * 「+加一条任务(暂不可用)」写的理由是「POST /:cid/tasks 建完会立刻起 run, * 拿它当加行会把草稿静默变成正在跑的任务」。 * * 只接受 `draft`:已经在队里的重复入队会再建一条 run(旧的那条就成了幽灵), * 跑着的、终态的各有各的出路(stop / rerun),一律 409 并点名走哪条。 */ router.post('/:companyId/tasks/:taskId/enqueue', (req, res) => { const companyId = req.params.companyId; try { const task = requireTask(req, res); if (!task) return; if (task.status !== 'draft') { return res.status(409).json({ error: `task ${task.id} 不是草稿(当前 ${task.status}),不能入队`, status: task.status, // 三种情况三条不同的出路,且每一条都得**真的存在**:旧版对所有非 queued 状态一律 // 写 `POST /api/graph-runs/${task.current_run_id}/stop`,而 blocked 的 task 可以 // 没有 run(URL 里就是字面量 null,404)、也可以 run 还在 queued(/stop 400)。 hint: task.status === 'queued' ? `它已经在队里了:GET /api/company/${companyId}/tasks?status=queued 看位次` : require('../services/work-graph-db').TASK_TERMINAL_STATUS.includes(task.status) ? `已收尾的走重跑:POST /api/company/${companyId}/tasks/${task.id}/rerun` : escapeHatchHint(companyId, task), }); } const graph = orgGraphService.getGraph(companyId); if (!graph.nodes || graph.nodes.length === 0) { return res.status(409).json({ error: `公司 ${companyId} 还没有 org graph,无法接任务`, hint: `先调 POST /api/company/${companyId}/org-graph/replan 让 orggraph-planning 规划组织结构`, }); } enqueueTaskRun({ companyId, taskRow: task, budget: (req.body && req.body.budget) || parseBudget(task.budget_json) || DEFAULT_TASK_BUDGET, }); res.status(201).json(queueReply(companyId, task, startNextIfSlotFree(companyId))); } catch (err) { const handled = sendDomainError(res, err); if (handled) return; console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * POST /api/company/:companyId/tasks/:taskId/start — 人点「开始」,把一条排队中的 task 拉起来 * * **这是 manual 任务唯一的起跑路径**(§五.2bis)。调度器**故意**不自动出队 manual: * `startRun(manual)` 只点亮入口节点就停下等 `/next`,而那时 run 已经是 `running`、占着坑 —— * 没人点「下一步」它就把后面所有任务无限期堵死,而队列界面上看不出任何异常。 * * 少了这个端点,「dispatch 把 5 条 manual 全入队 → 第 1 条跑完 → 剩下 4 条永远躺在队里」 * 就是必然结局,且界面上和正常排队一模一样 —— 正是当初否决排队的那条理由。 * (auto 的任务不需要它:坑一空,漏斗钩子上的 `scheduleTick` 就把它们拉起来了。) */ router.post('/:companyId/tasks/:taskId/start', (req, res) => { const companyId = req.params.companyId; try { const task = requireTask(req, res); if (!task) return; if (task.status !== 'queued') { return res.status(409).json({ error: `task ${task.id} 不在队列里(当前 ${task.status}),没有可开始的东西`, status: task.status, hint: task.status === 'draft' ? `草稿要先入队:POST /api/company/${companyId}/tasks/${task.id}/enqueue` : '它已经跑起来或收尾了,刷新看看', }); } // 依赖是硬约束,绕不过去:上游没 completed 就起跑,等于把「上游产物已就绪」凭空当真 const blockers = blockersOf(task); if (blockers.length) { return res.status(409).json({ error: `task ${task.id} 的上游还没完成,不能开始`, blockedBy: blockers, hint: `先让 ${blockers.join(' / ')} 跑到 completed;确实不需要它们就改依赖:` + `PATCH /api/company/${companyId}/tasks/${task.id} {"depends_on":[...]}`, }); } // 抢坑失败(坑被占 / 被调度器抢先推走)是**正常结局**,不是故障:如实回 409 + 当前位次, // 别回 200 让界面显示「已开始」而实际什么都没跑。 const started = promoteQueuedTask(task); if (!started) { // 「位子满了」和「这一瞬间被别人推走了」是两件事,答复也不同。 // 判据与形状都取 graph-run.js 的 slotState(= 409 run-slot-busy 那一份): // 批 4 之前这里自己 `.find()` 一条 active run + 三个单数字段, // N 个执行位时只点名一条,用户停掉它之后发现还是起不来(其余的位仍占着)。 const slots = slotsOf(companyId); const busy = slots.free <= 0; const { slotBusyHint } = require('./graph-run'); return res.status(409).json({ error: busy ? `公司 ${companyId} 的执行位已满(${slots.used}/${slots.max}),这条还得排着` : `task ${task.id} 这一瞬间被别人推走了,没抢到`, // 与 graph-run.js 的 409 逐字同形(reason / active / max / holders), // 免得同一件事在两个端点上有两套字段名 ...(busy ? { reason: 'run-slot-busy', active: slots.used, max: slots.max, holders: slots.holders } : {}), queuePosition: require('../services/task-scheduler').queuePosition(companyId, task.id) || null, // 「先停掉」这句只在它此刻真停得掉时给:占位的那条可能还在 ready, // 而 /stop 的可用条件里没有 ready —— 照着做就是一个 400(判据在 slotBusyHint 里,只此一份) hint: busy ? slotBusyHint(slots) : '刷新后重试', }); } res.status(201).json(queueReply(companyId, started.task, started)); } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * POST /api/company/:companyId/tasks/:taskId/cancel — 撤出队列 / 停止(27-P3C §六.3) * * 三条分支,因为「取消」对三种状态是三件不同的事: * · `queued` → 还没开跑,run `queued → stopped`,task 由漏斗钩子收成 cancelled,没有半截产物; * · `running` → 转调既有的 `stopRun`(**软信号**:不打断手头那个节点,到节点边界才落 stopped); * · `blocked` → 「挂起,等人介入」的唯一保底出路,见下方分支注释。 * 别的状态(draft / 已终态)一律 409:对它们说「取消成功」是界面撒谎。 * * 撤下之后立刻推一次队伍:坑空了(或队头换人了)就该有人顶上,不必等 60s 兜底。 */ router.post('/:companyId/tasks/:taskId/cancel', (req, res) => { const companyId = req.params.companyId; try { const task = requireTask(req, res); if (!task) return; if (task.status === 'queued') { // 没撤下来 = 这一瞬间它被调度器抢跑了(CAS 输了)。**不许回 200**: // 用户会以为取消成功,而任务正在烧钱。如实回 409 并指出现在该走哪条路。 if (!cancelQueuedRun(task)) { const fresh = require('../services/work-graph-db').getTask(task.id) || task; return res.status(409).json({ error: `task ${task.id} 刚刚被调度器出队,已经不在队列里了(当前 ${fresh.status})`, status: fresh.status, // 不硬写 /stop:出队那一瞬间 run 可能还在 ready(/stop 的可用条件里没有 ready), // 那条 hint 会 400。escapeHatchHint 按 run 的真实状态给一条打得通的 hint: escapeHatchHint(companyId, fresh), }); } return res.json(queueReply(companyId, task, startNextIfSlotFree(companyId))); } if (task.status === 'running') { if (!task.current_run_id) { return res.status(409).json({ error: `task ${task.id} 标着执行中却没有 runId,停不了`, hint: '这是数据不一致(不是正常状态),去看服务端日志', }); } require('../services/graph-engine').stopRun(task.current_run_id); // stopRun 是软信号:run 要到节点边界才真的落 stopped,所以这里回读到的可能还是 running。 // 如实回读,不假装已经停了 —— 「点了停止就显示已停止、实际还在烧钱」正是要消灭的那类撒谎。 return res.json(queueReply(companyId, task, null)); } // blocked = 「挂起,等人介入」,**它不是终态**(TASK_TERMINAL_STATUS 里没有它)。 // // 此前它落到最下面那条 409「没有可取消的东西」、hint 指向 rerun;而 rerun 对一条 blocked // 的 task 又必然 409(`taskHoldsRun` 把 blocked 一律算成攥着 run)。两条 409 互相指着对方, // 于是一条「上游已归档、放行也放行不了」的任务在界面上没有任何出路 —— 本轮点名要消灭的 // 「不能留指向不存在行为的提示」,根子在这里没有一条真的走得通的路。取消就是那条路。 // // 三种 blocked 的 run 形态不同,出口也不同,所以判**run 的真实状态**而不是猜: if (task.status === 'blocked') { const wdb = require('../services/work-graph-db'); const run = task.current_run_id ? wdb.getRun(task.current_run_id) : null; // ① 依赖推导出来的(blocked_reason = upstream_failed:*):run 还老实躺在 queued, // 走和「取消排队」完全相同的那条路(queued → stopped,漏斗把 task 收成 cancelled)。 if (run && run.run_status === 'queued') { if (!cancelQueuedRun(task)) { const fresh = wdb.getTask(task.id) || task; return res.status(409).json({ error: `task ${task.id} 的 run 这一瞬间已经不在队列里了(当前 task=${fresh.status})`, status: fresh.status, hint: escapeHatchHint(companyId, fresh), }); } return res.json(queueReply(companyId, task, startNextIfSlotFree(companyId))); } // ② 预算耗尽那种:run 自己也是 blocked(RUN_TRANSITIONS.blocked 允许 → stopped), // 和 running 分支同一条 stopRun。 if (run) { require('../services/graph-engine').stopRun(run.id); return res.json(queueReply(companyId, task, null)); } // ③ 入队做了一半(blocked_reason = enqueue_incomplete:*):压根没有 run。 // 这是**唯一**一处绕开 transitionRun 漏斗直接写 task 状态的地方,因为漏斗的入口 // (run)根本不存在;不写的话这条任务永远卡着,而界面上没有任何按钮动得了它。 const cancelled = wdb.updateTask(task.id, { status: 'cancelled', finishedAt: new Date().toISOString(), blockedReason: null, // 已经取消了还挂着「上游未完成」,列表上会一直显示一个假原因 }); // 走 queueReply 而不是 taskView:同一个端点的三条分支必须回同一个形状, // 否则前端要按状态分支解析响应 —— 那是又一个复述服务端取值域的持有者 return res.json(queueReply(companyId, cancelled, null)); } return res.status(409).json({ error: `task ${task.id} 当前是 ${task.status},没有可取消的东西`, status: task.status, hint: task.status === 'draft' ? `草稿本来就没入队,要删就归档:DELETE /api/company/${companyId}/tasks/${task.id}` : `它已经收尾了。要再跑一次:POST /api/company/${companyId}/tasks/${task.id}/rerun`, }); } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** * POST /api/company/:companyId/tasks/:taskId/rerun — 任务级重跑(27-P3C §六.3) * * **刻意不叫 `retry`**:既有的 `POST /api/graph-runs/:id/nodes/:nodeId/retry` 是**节点级** * 重试(同一条 run 内把某个节点打回 ready)。这条是**任务级**:新起一条 run,整张图从头来。 * 两个都叫 retry 的话,实现时必然有人把新端点接到旧逻辑上。 * * **新 run,旧 run 一个字段都不动** —— 旧 run 连同它的节点行留在历史里可点开(验收 8)。 * `attempt` 不在这里 +1:它由出队时的 `claimQueuedTask` 一条 UPDATE 里自增, * 读改写在并发下会丢代次,而 attempt 是「子 run 服务的是不是本轮」的判据。 * * 前置:这条 task 手上不能还攥着 run(`taskHoldsRun`)。攥着就重跑 = 同一条 task 底下 * 两条活 run,第二条收敛时会把第一条的结论覆盖掉。 */ router.post('/:companyId/tasks/:taskId/rerun', (req, res) => { const companyId = req.params.companyId; try { const wdb = require('../services/work-graph-db'); const task = requireTask(req, res); if (!task) return; if (task.status === 'draft' || task.status === 'queued') { return res.status(409).json({ error: `task ${task.id} 还没跑过(当前 ${task.status}),没有可重跑的东西`, status: task.status, hint: task.status === 'draft' ? `要让它跑:POST /api/company/${companyId}/tasks/${task.id}/enqueue` : '它已经在队里了,等着出队就行', }); } if (taskHoldsRun(task)) { return res.status(409).json({ error: `task ${task.id} 手上还攥着 run ${task.current_run_id || '(没有 run,但状态是 ' + task.status + ')'}` + `,重跑会同时有两条活 run`, status: task.status, runId: task.current_run_id || null, // 旧版硬写 `POST /api/graph-runs/${task.current_run_id}/stop`。taskHoldsRun 对 blocked // 一律返回 true,而 blocked 的 task 可以根本没有 run(enqueue_incomplete)或 run 还在 // queued —— 两种情况那条 URL 都打不通,用户照做一次失败一次,然后无路可走。 hint: escapeHatchHint(companyId, task), }); } const graph = orgGraphService.getGraph(companyId); if (!graph.nodes || graph.nodes.length === 0) { return res.status(409).json({ error: `公司 ${companyId} 还没有 org graph,无法接任务`, hint: `先调 POST /api/company/${companyId}/org-graph/replan 让 orggraph-planning 规划组织结构`, }); } // 复用入队那一条路(建新 run + 写 task-goal.json + task 翻 queued)。 // 它顺带把上一轮留下的 started_at / finished_at / blocked_reason 清掉 —— // 不清就会在界面上显示成「排队中,已耗时 3 小时」。 enqueueTaskRun({ companyId, taskRow: task, budget: (req.body && req.body.budget) || parseBudget(task.budget_json) || DEFAULT_TASK_BUDGET, }); const fresh = wdb.getTask(task.id); res.status(201).json({ ...queueReply(companyId, fresh, startNextIfSlotFree(companyId)), // 旧 run 仍在历史里 —— 明写出来,免得调用方以为重跑会把上一次的记录冲掉 previousRunId: task.current_run_id || null, attempt: (wdb.getTask(task.id) || {}).attempt, }); } catch (err) { const handled = sendDomainError(res, err); if (handled) return; console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); /** DELETE /api/company/:companyId/tasks/:taskId — 归档一条 task(规则同 goal) */ router.delete('/:companyId/tasks/:taskId', (req, res) => { try { const { updateTask } = require('../services/work-graph-db'); const task = requireTask(req, res); if (!task) return; if (taskHoldsRun(task)) { return res.status(409).json({ error: `task ${task.id} 正在执行或挂起`, status: task.status, runId: task.current_run_id || null, // 旧版硬写 `POST /api/graph-runs/${task.current_run_id}/stop`,而 `taskHoldsRun` 对 // blocked 一律返回 true:blocked 的 task 可以没有 run(enqueue_incomplete → URL 里 // 是字面量 null,404),也可以 run 还在 queued(/stop 只认 running/paused/blocked,400)。 // 「指向不存在行为的提示」比不给提示更糟 —— 用户照做一次失败一次。判据只留 // escapeHatchHint 那一份(rerun / cancel / enqueue 早就在用它,这里是漏掉的那处)。 hint: escapeHatchHint(req.params.companyId, task), }); } // 排队中的任务先把 run 从队里撤下再归档。少了这一步,`archived_at` 一盖上, // listQueue 的 `archived_at IS NULL` 就再也看不见它,而那条 `run_status='queued'` 的 run // 永远留在库里:没人会出队它(没有 task 指向),也没人会停它 —— 一条不报错的幽灵行。 if (task.status === 'queued' && !cancelQueuedRun(task)) { // 撤不下来 = 上面那道 taskHoldsRun 查过之后、这一行之前,它被出队跑起来了(跨进程才可能)。 // **绝不能接着归档**:那正是 §13.2 ① 记着的事故 —— 把一条正在烧预算的 run 从界面上藏掉。 // taskHoldsRun 是 check-then-act,真正决胜的是 cancelQueuedRun 里那条 CAS,以它为准。 const fresh = require('../services/work-graph-db').getTask(task.id) || task; return res.status(409).json({ error: `task ${task.id} 刚刚被出队跑起来了(当前 ${fresh.status}),不能归档`, status: fresh.status, // 出队那一瞬间 run 可能还停在 ready,而 /stop 的可用条件里没有 ready —— 硬写那条 URL // 就是一个 400。escapeHatchHint 按 run 的真实状态给一条打得通的 hint: escapeHatchHint(req.params.companyId, fresh), }); } res.json({ success: true, task: taskView(updateTask(task.id, { archivedAt: new Date().toISOString() })) }); } catch (err) { console.error('[TaskEntry] Error:', err.message); res.status(500).json({ error: err.message }); } }); // POST /api/company/:companyId/deposit-template — 训练合格后沉淀为模板 router.post('/:companyId/deposit-template', (req, res) => { try { const { agentId, benchmarkScore } = req.body; if (!agentId || benchmarkScore === undefined) { return res.status(400).json({ error: 'agentId 和 benchmarkScore 为必填项' }); } const result = companyService.depositTemplate( req.params.companyId, agentId, Number(benchmarkScore) ); res.json(result); } catch (err) { res.status(400).json({ error: err.message }); } }); // GET /api/company/:companyId/agents router.get('/:companyId/agents', (req, res) => { const agents = companyService.listAgents(req.params.companyId); res.json(agents); }); // ═══════════════════════════════════════ // Company OKR (公司级 OKR / 度量树) // ═══════════════════════════════════════ const { OkrService } = require('../services/okr-service'); // 公司级 OKR 路由。路径参数统一叫 :objectiveId(原 :goalId)——第 1 层节点已正名为 Objective。 // // 27-P3C §六.1:`/:companyId/goals` 这条向后兼容别名**已退役**(核实过零调用方, // 前端与 bin/lemon.js 走的都是 /okr)。理由不是「清理」,是 `/goals` 这个词在本文件里 // 从此另有所指 —— 它是 goals 表的两级模型端点。一个 URL 两种含义,读的人和写的人 // 迟早对不上;别名活着的每一天都在把两个概念往一起粘。 const okrPath = (suffix = '') => [`/:companyId/okr${suffix}`]; // GET /api/company/:companyId/okr router.get(okrPath(), (req, res) => { const okr = new OkrService(req.params.companyId); res.json(okr.list()); }); // POST /api/company/:companyId/okr router.post(okrPath(), (req, res) => { try { const { title, description, owner } = req.body; if (!title || !owner) { return res.status(400).json({ error: 'title 和 owner 为必填项' }); } const okr = new OkrService(req.params.companyId); res.status(201).json(okr.create({ title, description, owner })); } catch (err) { res.status(500).json({ error: err.message }); } }); // GET /api/company/:companyId/okr/:objectiveId router.get(okrPath('/:objectiveId'), (req, res) => { const okr = new OkrService(req.params.companyId); const objective = okr.get(req.params.objectiveId); if (!objective) return res.status(404).json({ error: 'Objective not found' }); res.json(objective); }); // PUT /api/company/:companyId/okr/:objectiveId router.put(okrPath('/:objectiveId'), (req, res) => { const okr = new OkrService(req.params.companyId); const result = okr.updateObjective(req.params.objectiveId, req.body); if (!result) return res.status(404).json({ error: 'Objective not found' }); res.json(result); }); // POST /api/company/:companyId/okr/:objectiveId/nodes router.post(okrPath('/:objectiveId/nodes'), (req, res) => { try { const { parentNodeId, title, owner, priority } = req.body; if (!title || !owner) { return res.status(400).json({ error: 'title 和 owner 为必填项' }); } const okr = new OkrService(req.params.companyId); const result = okr.addNode(req.params.objectiveId, parentNodeId || null, { title, owner, priority, }); if (!result) return res.status(404).json({ error: 'Objective not found' }); res.status(201).json({ nodeId: result.nodeId, node: result.objective.nodes[result.nodeId] }); } catch (err) { res.status(500).json({ error: err.message }); } }); // PUT /api/company/:companyId/okr/:objectiveId/nodes/:nodeId router.put(okrPath('/:objectiveId/nodes/:nodeId'), (req, res) => { const okr = new OkrService(req.params.companyId); const result = okr.updateNode(req.params.objectiveId, req.params.nodeId, req.body); if (!result) return res.status(404).json({ error: 'Node not found' }); res.json({ nodeId: result.nodeId, node: result.node }); }); // GET /api/company/:companyId/okr/:objectiveId/progress router.get(okrPath('/:objectiveId/progress'), (req, res) => { const okr = new OkrService(req.params.companyId); const progress = okr.getProgress(req.params.objectiveId); if (!progress) return res.status(404).json({ error: 'Objective not found' }); res.json(progress); }); // ═══════════════════════════════════════ // Pipeline / SOP 流水线 // ═══════════════════════════════════════ const { PipelineService } = require('../services/pipeline-service'); // POST /api/company/:companyId/pipelines — 创建流水线 router.post('/:companyId/pipelines', (req, res) => { try { const pSvc = new PipelineService(req.params.companyId); const pipeline = pSvc.createPipeline(req.body); res.status(201).json(pipeline); } catch (err) { res.status(400).json({ error: err.message }); } }); // GET /api/company/:companyId/pipelines — 列出流水线 router.get('/:companyId/pipelines', (req, res) => { const pSvc = new PipelineService(req.params.companyId); res.json(pSvc.listPipelines()); }); // GET /api/company/:companyId/pipelines/:pipelineId — 获取流水线定义 router.get('/:companyId/pipelines/:pipelineId', (req, res) => { const pSvc = new PipelineService(req.params.companyId); const pipeline = pSvc.getPipeline(req.params.pipelineId); if (!pipeline) return res.status(404).json({ error: '流水线不存在' }); res.json(pipeline); }); // PUT /api/company/:companyId/pipelines/:pipelineId — 更新流水线 router.put('/:companyId/pipelines/:pipelineId', (req, res) => { const pSvc = new PipelineService(req.params.companyId); const pipeline = pSvc.updatePipeline(req.params.pipelineId, req.body); if (!pipeline) return res.status(404).json({ error: '流水线不存在' }); res.json(pipeline); }); // POST /api/company/:companyId/pipelines/:pipelineId/run — 启动运行 router.post('/:companyId/pipelines/:pipelineId/run', (req, res) => { try { const pSvc = new PipelineService(req.params.companyId); const run = pSvc.startRun(req.params.pipelineId, req.body); res.status(201).json(run); } catch (err) { res.status(400).json({ error: err.message }); } }); // GET /api/company/:companyId/pipelines/runs/list — 列出运行记录 router.get('/:companyId/pipelines/runs/list', (req, res) => { const pSvc = new PipelineService(req.params.companyId); res.json(pSvc.listRuns(req.query.pipelineId)); }); // GET /api/company/:companyId/pipelines/runs/:runId — 获取运行详情 router.get('/:companyId/pipelines/runs/:runId', (req, res) => { const pSvc = new PipelineService(req.params.companyId); const run = pSvc.getRun(req.params.runId); if (!run) return res.status(404).json({ error: '运行记录不存在' }); res.json(run); }); // POST /api/company/:companyId/pipelines/runs/:runId/advance — 推进步骤 router.post('/:companyId/pipelines/runs/:runId/advance', (req, res) => { try { const pSvc = new PipelineService(req.params.companyId); const { stepIndex, output, status } = req.body; if (stepIndex === undefined) return res.status(400).json({ error: 'stepIndex 为必填项' }); const run = pSvc.advanceStep(req.params.runId, Number(stepIndex), { output, status }); res.json(run); } catch (err) { res.status(400).json({ error: err.message }); } }); // GET /api/company/:companyId/pipelines/runs/:runId/current — 获取当前任务 router.get('/:companyId/pipelines/runs/:runId/current', (req, res) => { const pSvc = new PipelineService(req.params.companyId); const task = pSvc.getCurrentTask(req.params.runId); if (!task) return res.json({ message: '无就绪任务' }); res.json(task); }); // POST /api/company/:companyId/pipelines/runs/:runId/reject — COO 驳回步骤 router.post('/:companyId/pipelines/runs/:runId/reject', (req, res) => { try { const pSvc = new PipelineService(req.params.companyId); const { stepIndex, reason, reviewedBy } = req.body; if (stepIndex === undefined) return res.status(400).json({ error: 'stepIndex 为必填项' }); const run = pSvc.rejectStep(req.params.runId, Number(stepIndex), { reason: reason || '', reviewedBy: reviewedBy || 'coo', }); res.json(run); } catch (err) { res.status(400).json({ error: err.message }); } }); // Schedule (日程管理) // ═══════════════════════════════════════ // GET /api/company/:companyId/schedule — 获取日程 router.get('/:companyId/schedule', (req, res) => { res.json(scheduleService.getSchedule(req.params.companyId)); }); // POST /api/company/:companyId/schedule/slots — 添加日程 router.post('/:companyId/schedule/slots', (req, res) => { try { const slot = scheduleService.addSlot(req.params.companyId, req.body); res.json(slot); } catch (err) { res.status(400).json({ error: err.message }); } }); // PUT /api/company/:companyId/schedule/slots/:slotId — 更新日程 router.put('/:companyId/schedule/slots/:slotId', (req, res) => { const slot = scheduleService.updateSlot(req.params.companyId, req.params.slotId, req.body); if (!slot) return res.status(404).json({ error: '日程不存在' }); res.json(slot); }); // DELETE /api/company/:companyId/schedule/slots/:slotId — 删除日程 router.delete('/:companyId/schedule/slots/:slotId', (req, res) => { scheduleService.removeSlot(req.params.companyId, req.params.slotId); res.json({ success: true }); }); // ═══════════════════════════════════════ // DataSource (运营数据接入) // ═══════════════════════════════════════ const { DataSourceService } = require('../services/datasource-service'); const dataSourceService = new DataSourceService(); // GET /api/company/:companyId/data/metrics — 获取运营指标 router.get('/:companyId/data/metrics', async (req, res) => { try { const { source = 'mock', startDate, endDate } = req.query; const sd = startDate || new Date(Date.now() - 7 * 86400000).toISOString().slice(0, 10); const ed = endDate || new Date().toISOString().slice(0, 10); const data = await dataSourceService.fetchMetrics(source, { startDate: sd, endDate: ed }); res.json(data); } catch (err) { res.status(400).json({ error: err.message }); } }); // GET /api/company/:companyId/data/trending — 获取热点趋势 router.get('/:companyId/data/trending', async (req, res) => { try { const { source = 'mock', platform = 'weibo', count = '10' } = req.query; const data = await dataSourceService.fetchTrending(source, { platform, count: parseInt(count, 10) }); res.json(data); } catch (err) { res.status(400).json({ error: err.message }); } }); // GET /api/company/:companyId/data/attribution — 归因分析 router.get('/:companyId/data/attribution', async (req, res) => { try { const { source = 'mock' } = req.query; const now = new Date(); const currentEnd = now.toISOString().slice(0, 10); const currentStart = new Date(now.getTime() - 7 * 86400000).toISOString().slice(0, 10); const previousEnd = new Date(now.getTime() - 7 * 86400000).toISOString().slice(0, 10); const previousStart = new Date(now.getTime() - 14 * 86400000).toISOString().slice(0, 10); const analysis = await dataSourceService.analyzeAttribution( source, { startDate: currentStart, endDate: currentEnd }, { startDate: previousStart, endDate: previousEnd } ); res.json(analysis); } catch (err) { res.status(400).json({ error: err.message }); } }); // ═══════════════════════════════════════ // Company Files — 公司级统一文件树 // ═══════════════════════════════════════ // GET /api/company/:companyId/files — 获取公司下所有 Agent 的文件树 router.get('/:companyId/files', (req, res) => { try { const path = require('path'); const fs = require('fs'); const companyDir = path.join(COMPANIES_DIR, req.params.companyId); if (!fs.existsSync(companyDir)) { return res.json({ tree: [], totalFiles: 0 }); } // 递归扫描目录 function scanDir(dirPath, relativePath = '') { const entries = fs.readdirSync(dirPath, { withFileTypes: true }); const nodes = []; for (const entry of entries) { const fullPath = path.join(dirPath, entry.name); const relPath = relativePath ? `${relativePath}/${entry.name}` : entry.name; // 跳过隐藏文件和 node_modules if (entry.name.startsWith('.') || entry.name === 'node_modules') continue; // 跳过 conversations 目录(太大) if (entry.name === 'conversations') continue; if (entry.isDirectory()) { const children = scanDir(fullPath, relPath); nodes.push({ name: entry.name, path: relPath, type: 'directory', children, fileCount: children.reduce((sum, c) => sum + (c.type === 'file' ? 1 : (c.fileCount || 0)), 0), }); } else { const stat = fs.statSync(fullPath); nodes.push({ name: entry.name, path: relPath, type: 'file', size: stat.size, modifiedAt: stat.mtime.toISOString(), }); } } // 目录排前面, 然后按名字排序 return nodes.sort((a, b) => { if (a.type !== b.type) return a.type === 'directory' ? -1 : 1; return a.name.localeCompare(b.name); }); } const tree = scanDir(companyDir); const countFiles = (nodes) => 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/company/:companyId/files/content?path=xxx — 读取文件内容 router.get('/:companyId/files/content', (req, res) => { try { const path = require('path'); const fs = require('fs'); const filePath = req.query.path; if (!filePath) return res.status(400).json({ error: 'path 参数为必填' }); // 安全校验:防止路径穿越 const absPath = resolveInside(path.join(COMPANIES_DIR, req.params.companyId), filePath); if (!absPath) { return res.status(403).json({ error: '路径越界' }); } if (!fs.existsSync(absPath)) { return res.status(404).json({ error: '文件不存在' }); } const stat = fs.statSync(absPath); if (stat.size > 512 * 1024) { return res.json({ content: '(文件过大,跳过预览)', truncated: true, size: stat.size }); } const content = fs.readFileSync(absPath, 'utf-8'); res.json({ content, size: stat.size, path: filePath }); } catch (err) { res.status(500).json({ error: err.message }); } }); // PUT /api/company/:companyId/files/content — 保存文件内容 router.put('/:companyId/files/content', (req, res) => { try { const filePath = req.body.path; if (!filePath) return res.status(400).json({ error: 'path 参数为必填' }); const absPath = resolveInside(path.join(COMPANIES_DIR, req.params.companyId), filePath); if (!absPath) { return res.status(403).json({ error: '路径越界' }); } const dir = path.dirname(absPath); if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true }); fs.writeFileSync(absPath, req.body.content || '', 'utf-8'); console.log(`[Company] 💾 已保存文件 ${req.params.companyId}/${filePath}`); res.json({ success: true, path: filePath }); } catch (err) { res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // Company Workspace — 工作区文件管理 // ═══════════════════════════════════════ // GET /api/company/:companyId/workspace/files — 列出 workspace 文件树 router.get('/:companyId/workspace/files', (req, res) => { try { const wsDir = path.join(COMPANIES_DIR, req.params.companyId, 'workspace'); if (!fs.existsSync(wsDir)) { return res.json({ tree: [], totalFiles: 0 }); } function scanDir(dirPath, relativePath = '') { const entries = fs.readdirSync(dirPath, { withFileTypes: true }); const nodes = []; for (const entry of entries) { const fullPath = path.join(dirPath, entry.name); const relPath = relativePath ? `${relativePath}/${entry.name}` : entry.name; if (entry.name.startsWith('.') || entry.name === 'node_modules') continue; if (entry.isDirectory()) { const children = scanDir(fullPath, relPath); nodes.push({ name: entry.name, path: relPath, type: 'directory', children, fileCount: children.reduce((sum, c) => sum + (c.type === 'file' ? 1 : (c.fileCount || 0)), 0), }); } else { const stat = fs.statSync(fullPath); nodes.push({ name: entry.name, path: relPath, type: 'file', size: stat.size, createdAt: stat.birthtime.toISOString(), modifiedAt: stat.mtime.toISOString(), }); } } return nodes.sort((a, b) => { if (a.type !== b.type) return a.type === 'directory' ? -1 : 1; return a.name.localeCompare(b.name); }); } const tree = scanDir(wsDir); const countFiles = (nodes) => 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/company/:companyId/workspace/files/content?path=xxx — 读取 workspace 文件内容 router.get('/:companyId/workspace/files/content', (req, res) => { try { const filePath = String(req.query.path || ''); if (!filePath) return res.status(400).json({ error: 'path 参数为必填' }); const absPath = resolveInside(path.join(COMPANIES_DIR, req.params.companyId, 'workspace'), filePath); if (!absPath) { return res.status(403).json({ error: '路径越界' }); } if (!fs.existsSync(absPath)) { return res.status(404).json({ error: '文件不存在' }); } const stat = fs.statSync(absPath); if (stat.size > 512 * 1024) { return res.json({ content: '(文件过大,跳过预览)', truncated: true, size: stat.size }); } const content = fs.readFileSync(absPath, 'utf-8'); res.json({ content, size: stat.size, path: filePath, createdAt: stat.birthtime.toISOString(), modifiedAt: stat.mtime.toISOString(), }); } catch (err) { res.status(500).json({ error: err.message }); } }); // PUT /api/company/:companyId/workspace/files/content — 保存 workspace 文件 router.put('/:companyId/workspace/files/content', (req, res) => { try { const filePath = req.body.path; if (!filePath) return res.status(400).json({ error: 'path 参数为必填' }); const absPath = resolveInside(path.join(COMPANIES_DIR, req.params.companyId, 'workspace'), filePath); if (!absPath) { return res.status(403).json({ error: '路径越界' }); } const dir = path.dirname(absPath); if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true }); fs.writeFileSync(absPath, req.body.content || '', 'utf-8'); console.log(`[Workspace] 💾 已保存 ${req.params.companyId}/workspace/${filePath}`); res.json({ success: true, path: filePath }); } catch (err) { res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // Company Messages — 全局消息时间线 // ═══════════════════════════════════════ // GET /api/company/:companyId/messages?limit=100 — 获取公司所有 Agent 的消息时间线 router.get('/:companyId/messages', (req, res) => { try { const path = require('path'); const fs = require('fs'); const limit = Math.min(parseInt(req.query.limit) || 200, 500); const allMessages = []; // 名册查 company_agents,会话在公司态目录 system/agents/{aid}/{cid}/conversations/(28 批次 4-19) for (const row of listCompanyAgents(req.params.companyId)) { const agentId = row.role_name; const convDir = path.join(locator.companyDir(row.system_agent_id, req.params.companyId), 'conversations'); if (!fs.existsSync(convDir)) continue; const conversations = fs.readdirSync(convDir, { withFileTypes: true }).filter(d => d.isDirectory()); for (const conv of conversations) { const convId = conv.name; // 读 meta let meta = null; try { meta = JSON.parse(fs.readFileSync(path.join(convDir, convId, 'meta.json'), 'utf-8')); } catch { continue; } // 读 history let history = []; try { history = JSON.parse(fs.readFileSync(path.join(convDir, convId, 'history.json'), 'utf-8')); } catch { continue; } for (const msg of history) { allMessages.push({ agentId: `${req.params.companyId}:${agentId}`, agentName: agentId, conversationId: convId, conversationTitle: meta?.title || convId, role: msg.role, content: (msg.content || '').slice(0, 800), timestamp: msg.timestamp || meta?.createdAt, messageId: msg.messageId, taskId: convId.startsWith('task-') ? convId.replace('task-', '') : null, }); } } } // 按时间倒序 allMessages.sort((a, b) => new Date(b.timestamp).getTime() - new Date(a.timestamp).getTime()); res.json(allMessages.slice(0, limit)); } catch (err) { res.status(500).json({ error: err.message }); } }); // ═══════════════════════════════════════ // Company Stats — 统计 API (skills count, jobs status) // ═══════════════════════════════════════ // GET /api/company/:companyId/stats — 获取公司统计信息 router.get('/:companyId/stats', (req, res) => { try { const path = require('path'); const fs = require('fs'); const companyId = req.params.companyId; // 1. Agents 统计(名册查 company_agents) const agents = companyService.listAgents(companyId); const agentCount = agents.length; // 2. Skills 统计 —— Skill 挂在本体的 `.claude/skills/` 下(28 批次 4-19)。 // 注意这份计数因此是**本体级**的:两家公司雇同一个本体,各自的统计里都算它一次。 let totalSkills = 0; const agentSkills = []; for (const agent of agents) { const skillsDir = path.join(locator.bodyDir(agent.bodyId), '.claude', 'skills'); if (!fs.existsSync(skillsDir)) continue; const skillCount = fs.readdirSync(skillsDir, { withFileTypes: true }).filter(d => d.isDirectory()).length; totalSkills += skillCount; agentSkills.push({ agentId: agent.id, skillCount }); } // 3. Jobs 统计 - 从 jobs 持久化文件读取 let jobsStats = { total: 0, queued: 0, running: 0, completed: 0, failed: 0 }; const jobsFilePath = path.join(DATA_ROOT, '.data', 'jobs-openapi.json'); if (fs.existsSync(jobsFilePath)) { try { const jobsData = JSON.parse(fs.readFileSync(jobsFilePath, 'utf-8')); if (Array.isArray(jobsData)) { jobsStats.total = jobsData.length; for (const job of jobsData) { if (jobsStats[job.status] !== undefined) { jobsStats[job.status]++; } } } } catch (e) { console.warn('[Stats] 读取 Jobs 文件失败:', e.message); } } // 4. Pipelines 统计 const pSvc = new PipelineService(companyId); const pipelines = pSvc.listPipelines(); const runs = pSvc.listRuns(); const pipelinesStats = { total: pipelines.length, active: pipelines.filter(p => p.status === 'active').length, runs: { total: runs.length, running: runs.filter(r => r.status === 'running').length, completed: runs.filter(r => r.status === 'completed').length, failed: runs.filter(r => r.status === 'failed').length, } }; res.json({ companyId, timestamp: new Date().toISOString(), skills: { total: totalSkills, byAgent: agentSkills }, agents: { total: agentCount }, jobs: jobsStats, pipelines: pipelinesStats, }); } catch (err) { res.status(500).json({ error: err.message }); } }); // GET /api/company/:companyId/jobs — 获取公司相关的 Job 列表 router.get('/:companyId/jobs', (req, res) => { try { const path = require('path'); const fs = require('fs'); const jobsFilePath = path.join(DATA_ROOT, '.data', 'jobs-openapi.json'); if (!fs.existsSync(jobsFilePath)) { return res.json([]); } const jobsData = JSON.parse(fs.readFileSync(jobsFilePath, 'utf-8')); if (!Array.isArray(jobsData)) { return res.json([]); } // 过滤与公司相关的 Jobs (通过 domain 或 conversationId 包含 companyId) const limit = Math.min(parseInt(req.query.limit) || 50, 100); const companyId = req.params.companyId; const filtered = jobsData .filter(job => !job.domain || job.domain.includes(companyId)) .sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()) .slice(0, limit); res.json(filtered); } catch (err) { res.status(500).json({ error: err.message }); } }); // GET /api/company/:companyId/jobs/:jobId — 获取 Job 详情 router.get('/:companyId/jobs/:jobId', (req, res) => { try { const path = require('path'); const fs = require('fs'); const jobsFilePath = path.join(DATA_ROOT, '.data', 'jobs-openapi.json'); if (!fs.existsSync(jobsFilePath)) { return res.status(404).json({ error: 'Job 不存在' }); } const jobsData = JSON.parse(fs.readFileSync(jobsFilePath, 'utf-8')); if (!Array.isArray(jobsData)) { return res.status(404).json({ error: 'Job 不存在' }); } const job = jobsData.find(j => j.jobId === req.params.jobId); if (!job) { return res.status(404).json({ error: 'Job 不存在' }); } // 返回安全版本 (隐藏 apiKey, webhookUrl) const { apiKey, webhookUrl, ...safeJob } = job; res.json(safeJob); } catch (err) { res.status(500).json({ error: err.message }); } }); module.exports = router;