/** * subtask-planning —— 把一条 task 拆成"第一步/第二步/第三步"的步骤分解(46 号文档 Phase 1) * * 与 `task-planning`(goal → task)的关键差异:这是**建议性**产出,不落库、不创建 task 行、 * 没有重试轮次、没有写入前校验会打回 agent —— 唯一的失败模式是"没用",不是"格式错"。 * 所以本模块单轮调用、宽松解析:挑得出多少条 `steps` 就要多少条,个别字段缺失就地丢弃那一条, * 不因为一条不合法就整批作废(`system/agents/subtask-planning/agent.md` 硬约束 8 附近的 * 「你的产出是建议,不是指令」)。 * * 触发方式:`enqueueTaskRun`(`server/routes/company.js`)用 `setImmediate` 异步触发, * 照抄 `runGoalSplit` 的现成模式,不等它、不阻塞排队。**`triggerSubtaskPlanning` 是本模块 * 唯一允许抛出以外的公开入口,内部任何一步失败都只 `console.warn`,绝不 throw** —— * 33/44/47 号文档反复验证过的教训:软数据不能变成硬闸门,这里同理,招进来的这道钩子 * 更不能变成"入队本身"的新失败点。 * * 产物写共享 `artifacts/` 区、登记进 `graph_files` 索引(`kind='subtask_plan'`,已纳入 * `graph-engine.TASK_SHARED_KINDS`)——这样任意节点(org-planner 展开出的子图里的 * brd/prd/sdd 各节点、乃至后续 org-coder)的「上游输入」提示词段落都会自动带上它, * 不用为 org-planner 单独写一段 prompt-builder 注入逻辑(46 号文档「与 org-planner 的 * 衔接点」一节最初设想的软注入方案,改走这条更省事、且是引擎已验证在用的现成机制)。 * * ── Phase 2(46 号文档):接上经验库检索 + 按需 deep research ────────────────── * 第一轮调用前先查 `experience-lookup.queryExperience`(dream 沉淀的跨公司经验)拼进提示词; * 查不到经验,或 agent 自评 `confidence:"low"`,就再调 `deep-research.triggerDeepResearch` * (默认关闭,见该文件头注释),拿到调研结果就带着它做第二轮调用取代第一轮结果。 * `reasoning_source` 由服务端根据这两轮实际发生了什么算出来,不是 agent 自报的—— * agent 唯一自报的判断是 `confidence`,用来决定要不要触发调研,不直接决定 `reasoning_source`。 * * ── 48 号文档:验收契约创建时 1:1 强绑定 ──────────────────────────────────── * 每个 step 现在还要交一份 `acceptance`(benchmark 段形状)。它不是给 org-planner 参考用的 * 装饰字段——一条 step 被 `materializeDecomposedSubtasks` 物化成 `decomposed_work` 子任务时, * `acceptance` 会被 `work-graph-db.createTask` 强制要求、校验、随行落库,`enqueueTaskRun` * 起轻量图前把它写进新 run 的 `artifacts/acceptance.json`——这就是 `org-validator` 在跳过 * `org-planner` 的轻量派发链路上判卷时唯一的依据来源。 */ const path = require('path'); const fs = require('fs'); const REPO_ROOT = path.resolve(__dirname, '../..'); /** 全局唯一的一份步骤分解本体(`system/agents/subtask-planning/`) */ const PLANNING_AGENT_ID = 'subtask-planning'; /** 一条 task 最多切几步。上限是体检指标,不是配额——agent.md 讲得很清楚 */ const MAX_STEPS = 12; // ──────────────────────────────────────────────────────────────── // 注入点(仿 task-planning.setPlanTasks) // ──────────────────────────────────────────────────────────────── let planSubtasksImpl = defaultPlanSubtasks; let isInjected = false; /** 替换规划实现(传 null 回默认实现)。测试用,避免真起 CLI */ function setPlanSubtasks(fn) { planSubtasksImpl = fn || defaultPlanSubtasks; isInjected = !!fn; } /** * 更细粒度的注入点:只换掉 `defaultPlanSubtasks` 内部真正起 CLI 的那一层 * (写法仿 `dream-service.setDreamAgentCaller`),不像 `setPlanSubtasks` 那样把整段 * "查经验库 → 判断要不要调研 → 两轮调用"编排逻辑一起换掉。Phase 2 加的正是这段编排逻辑, * 留这个更细的缝才测得到它本身的分支,而不是只测"调用方传回什么就原样落盘"。 */ let agentCaller = null; /** 替换 agent 调用层(传 null 回真实现)。只给测试用 */ function setSubtaskAgentCaller(fn) { agentCaller = fn; } async function callSubtaskPlanningAgent(agentName, prompt, opts) { if (agentCaller) return agentCaller(agentName, prompt, opts); const AgentManager = require('../src/core/agent-manager'); const manager = new AgentManager({}, REPO_ROOT); return manager.callAgent(agentName, prompt, opts); } /** * 当前生效的规划实现。调用方一律走这里。 * * **踩过的坑,写在这里免得下次重犯**:`task-planning`/`workgraph-planning` 那一整套 * 「注入点」设计的前提是——调用方要触发真实实现,得先显式 `setPlanXxx(fn)`。这个前提 * 对 subtask-planning 不成立:本模块的触发点在 `enqueueTaskRun`(`server/routes/company.js` * 的 `/goals/:goalId/dispatch` 与 `/tasks/:taskId/enqueue` 两处),而**全仓库有大量单测 * 会走到这两条路径**——它们测的是排队/预算/状态流转这些跟 subtask-planning 毫无关系的东西, * 从没想过要为一个刚加进来的旁支功能显式注入桩。第一次实现时忽略了这一点,`npx vitest run * server/tests/unit` 一次性真开了 44 个 `claude` CLI 子进程挂在后台(`ps aux` 现场抓到, * 已手动 kill 干净)——正是 `docs/done/claude-cli-session-leak.md` 那份手册想防的事, * 而且是本模块自己新引入的一个源头。 * * 修法:`process.env.VITEST`(vitest 对所有测试进程都会自动置位,不是我们自己发明的判据) * 为真、且没有人显式调用过 `setPlanSubtasks` 时,直接跳过真实调用——不是「默认桩返回假数据」 * (那会让没预料到这条路径的测试断言出诡异内容),是「什么都不做」,等价于 47/44 号文档 * 反复用到的软失败路径(0 条 steps,不写文件,只留一条日志)。真想在测试里验证 subtask-planning * 的行为,显式 `setPlanSubtasks(fn)` 就行——那条路径不受这层保护影响。 */ function planSubtasks(ctx) { if (process.env.VITEST && !isInjected) { console.log('[SubtaskPlanning] VITEST 环境且未显式注入,跳过真实调用(避免测试意外起 CLI 子进程)'); return Promise.resolve({ steps: [], tokens: 0, reasoningSource: 'model_only', experienceRefs: [], researchUsed: false, }); } return planSubtasksImpl(ctx); } // ──────────────────────────────────────────────────────────────── // 默认实现 // ──────────────────────────────────────────────────────────────── /** * 调 subtask-planning 拆一次 task 的步骤。先查经验库,再按需触发一次 deep research * (Phase 2,46 号文档),最多两轮 agent 调用,没有重试。 * * @param {{companyId: string, task: object}} ctx task 是 tasks 表的整行 * @returns {Promise<{steps: object[], tokens: number, reasoningSource: string, * experienceRefs: string[], researchUsed: boolean}>} * steps 形如 `[{seq, title, why, depends_on_hint: [seq…], acceptance?}]`, * seq 恒为 1..N 连续(重新编号过)。`acceptance`(48 号文档)缺失/形状不合法的条目 * 仍会留在这里——`normalizeSteps` 不因它丢整条,深层校验留给 `materializeDecomposedSubtasks` * 建行那一刻(单步失败跳过,不作废整批)。 */ async function defaultPlanSubtasks(ctx) { const agentName = resolvePlanningAgentName(ctx.companyId); let tokens = 0; const onAcpEvent = (ev) => { if (!ev || ev.method !== 'session/result') return; const u = (ev.params && ev.params.usage) || null; if (u) tokens += (u.input || 0) + (u.output || 0); }; const callOnce = async (prompt) => { const res = await callSubtaskPlanningAgent(agentName, prompt, { onAcpEvent, metadata: { // taskId 决定 CLI 的 cwd。这里就是真的那条 task 的 id(不像 task-planning 借用 goalId // 顶替——subtask-planning 天然运行在一条具体 task 之下,不用借)。 taskId: ctx.task.id, layer: 'subtask-plan', goalId: ctx.task.goal_id, companyId: ctx.companyId, }, }); return res && res.output; }; const { queryExperience } = require('./experience-lookup'); const exp = queryExperience({ companyId: ctx.companyId, task: ctx.task }); const experienceRefs = exp.matches.map((m) => `${m.agentId}: ${m.title}`); const round1Output = await callOnce(buildSubtaskPlanningPrompt(ctx, { experience: exp.matches })); let steps = normalizeSteps(extractPlannedSteps(round1Output)); const confidence = extractConfidence(round1Output); let reasoningSource = exp.matches.length ? 'memory' : 'model_only'; let researchUsed = false; const needResearch = exp.matches.length === 0 || confidence === 'low'; if (needResearch) { const { triggerDeepResearch } = require('./deep-research'); const topic = [ctx.task.title, ctx.task.description].filter(Boolean).join(' — '); const research = await triggerDeepResearch({ companyId: ctx.companyId, task: ctx.task, topic }); if (research) { tokens += research.tokens || 0; const round2Output = await callOnce(buildSubtaskPlanningPrompt(ctx, { experience: exp.matches, research })); const round2Steps = normalizeSteps(extractPlannedSteps(round2Output)); if (round2Steps.length) { steps = round2Steps; reasoningSource = 'deep_research'; researchUsed = true; } } } return { steps, tokens, reasoningSource, experienceRefs, researchUsed }; } /** * 本体的寻址名。与 `task-planning.resolvePlanningAgentName` 同一个模式: * 按需幂等雇一次,判据查 `company_agents` 表(不是目录)。 */ function resolvePlanningAgentName(companyId) { if (!companyId) return PLANNING_AGENT_ID; const { CompanyService, listCompanyAgents } = require('./company-service'); const isHired = () => listCompanyAgents(companyId).some((r) => r.role_name === PLANNING_AGENT_ID); if (!isHired()) { try { new CompanyService().hire(companyId, PLANNING_AGENT_ID, {}); } catch (e) { console.warn(`[SubtaskPlanning] 按需雇佣 ${PLANNING_AGENT_ID} 失败(${companyId}): ${e.message}`); } } return isHired() ? `${companyId}:${PLANNING_AGENT_ID}` : PLANNING_AGENT_ID; } // ──────────────────────────────────────────────────────────────── // 提示词 // ──────────────────────────────────────────────────────────────── /** * @param {{companyId: string, task: object}} ctx * @param {{experience?: Array<{agentId:string,title:string,body:string}>, * research?: {findings:string, sources?:string[]} | null}} [extra] * `experience`:`experience-lookup.queryExperience` 的命中结果(Phase 2)。 * `research`:`deep-research.triggerDeepResearch` 的结果,只在触发过调研时传入(第二轮)。 */ function buildSubtaskPlanningPrompt(ctx, extra = {}) { const t = ctx.task || {}; const { experience = [], research = null } = extra; const lines = [ `你正在为公司 **${ctx.companyId}** 的一条 task 想"怎么做"的步骤分解。`, '', '## task 描述(要拆步骤的对象,task-planning 已经切好的一条边界)', t.description || '(调用方没给任务描述)', '', `- taskId:${t.id}(只读,不要写进你的产出)`, `- 标题:${t.title || '(无)'}`, '', ]; if (experience.length) { lines.push('## 已知相关经验(供参考,不强制采纳,来自其他公司同角色沉淀下来的经验)'); for (const e of experience) lines.push(`- [来自 ${e.agentId}] **${e.title}** —— ${e.body}`); lines.push(''); } if (research) { lines.push('## 调研参考(系统判断需要调研后跑的一次快速调研,供参考,不强制采纳)'); lines.push(research.findings); if (research.sources && research.sources.length) lines.push(`来源:${research.sources.join('、')}`); lines.push(''); } lines.push( '## 交回方式(引擎只读你回复里最后一个 ```json 代码块)', '```json', '{"steps":[{"seq":1,"title":"…","why":"…","depends_on_hint":[],' + '"acceptance":{"dimensions":[{"name":"…","weight":1,"pass_threshold":80}],"overall_pass_threshold":80}}],' + '"confidence":"high"}', '```', `- \`seq\`:1..N 连续整数、与数组顺序一致;条数上限 **${MAX_STEPS}**,一步也是合法答案。`, '- `depends_on_hint`:只能写本次产出的 `seq` 数字(1-based),没有就写 `[]`——它是提示,' + '不是硬依赖,没有任何机制会强制执行它。', '- `acceptance`:这一步的验收标准(48 号文档),形状对齐 benchmark 段:`dimensions` 至少一条,' + '每条 `{name, weight, pass_threshold?}`,`weight` 之和必须是 1;`overall_pass_threshold` 可选。' + '不能从 `why` 抄——要写清楚这一步做完该长什么样、按什么维度打分,格式不合规或空泛("基本可用"' + '这类不可度量措辞)会被直接拒收,这一步就不会被派发出去。', '- `confidence`:`"high"` 或 `"low"`——你能看到的信息(task 描述 + 上面若列出的经验/调研参考)' + '不足以让你对分解有把握时选 `"low"`;缺经验/调研参考不必然是 `"low"`,常规任务模型自身规划' + '能力足够时正常给 `"high"`。`"low"` 会让系统再帮你查一轮资料,重新问你一次。', '- 只输出这两个顶层字段(`steps`/`confidence`)。想法写在 JSON 代码块**前面**的正文里,' + '代码块之后不要再写任何字。', '', '你的产出会原样附进 org-planner 的任务提示词,当作参考。它没有义务采纳,' + '这里没有写入前校验、没有重试轮次——唯一的失败模式是列得没用,不是格式错。' + '**`acceptance` 是例外**:如果这条 task 走的是不经过 org-planner 的轻量派发链路,' + '你交的 `acceptance` 会被原样写成这一步的验收契约,`org-validator` 直接拿它判卷——' + '写得没用的后果不再只是"没人看",而是这一步真的会通不过验收(48 号文档)。' ); return lines.join('\n'); } // ──────────────────────────────────────────────────────────────── // 产出解析(宽松:单条不合法就丢那一条,不作废整批) // ──────────────────────────────────────────────────────────────── /** 从回复里取出步骤数组。取法与 task-planning.extractPlannedTasks 同思路(围栏优先、从后往前取) */ function extractPlannedSteps(output) { const text = String(output || ''); if (!text.trim()) return []; const candidates = jsonCandidates(text); for (let i = candidates.length - 1; i >= 0; i -= 1) { const doc = safeJsonParse(candidates[i]); if (doc && Array.isArray(doc.steps)) return doc.steps; if (Array.isArray(doc)) return doc; // 容错:直接回了裸数组也认 } return []; } /** 从回复里取 `confidence` 字段(`"high"`/`"low"`)。取不到就是 null——按"未知"处理,不强制触发调研 */ function extractConfidence(output) { const text = String(output || ''); if (!text.trim()) return null; const candidates = jsonCandidates(text); for (let i = candidates.length - 1; i >= 0; i -= 1) { const doc = safeJsonParse(candidates[i]); if (doc && (doc.confidence === 'high' || doc.confidence === 'low')) return doc.confidence; } return null; } function jsonCandidates(text) { const out = []; const fence = /```(?:json|jsonc)?\s*([\s\S]*?)```/g; let m; while ((m = fence.exec(text))) out.push(m[1]); out.push(...balanced(text, '[', ']'), ...balanced(text, '{', '}')); return out; } function balanced(text, open, close) { const out = []; let depth = 0; let start = -1; for (let i = 0; i < text.length; i += 1) { const c = text[i]; if (c === open) { if (depth === 0) start = i; depth += 1; } else if (c === close && depth > 0) { depth -= 1; if (depth === 0) out.push(text.slice(start, i + 1)); } } return out; } function safeJsonParse(s) { try { return JSON.parse(s); } catch { return null; } } /** * 宽松归一化:丢掉不合法的单条(没有 title 的、不是对象的),保留下来的重新编号 * (1..N 连续,不留空洞——跳过的条目不该在 seq 里留坑,下游读起来会以为漏了一条)。 * `depends_on_hint` 里指向被丢弃条目 / 越界 / 自引用的数字直接过滤掉,不报错。 */ function normalizeSteps(raw) { if (!Array.isArray(raw) || !raw.length) return []; const capped = raw.slice(0, MAX_STEPS); const out = []; capped.forEach((item, i) => { if (!item || typeof item !== 'object' || Array.isArray(item)) return; const title = String(item.title == null ? '' : item.title).trim(); if (!title) return; out.push({ _origIndex: i + 1, title, item }); }); const keptOrigSet = new Set(out.map((o) => o._origIndex)); return out.map((o, idx) => { const rawDeps = o.item.depends_on_hint !== undefined ? o.item.depends_on_hint : o.item.dependsOnHint; const deps = []; for (const d of Array.isArray(rawDeps) ? rawDeps : []) { const n = typeof d === 'number' || /^\d+$/.test(String(d).trim()) ? Number(d) : NaN; if (Number.isInteger(n) && keptOrigSet.has(n) && n !== o._origIndex && !deps.includes(n)) deps.push(n); } // 48 号文档:`acceptance` 原样透传,不在这里做形状校验——深层校验(权重之和/runner // 白名单)只有 goal-contract.validateContract 做得了,交给 materializeDecomposedSubtasks // 建行时去做(单步不合法就跳过那一步,不作废整批,同款"宽松解析"哲学,见该函数注释)。 const acceptance = o.item.acceptance && typeof o.item.acceptance === 'object' && !Array.isArray(o.item.acceptance) ? o.item.acceptance : undefined; return { seq: idx + 1, title: o.title.slice(0, 200), why: String(o.item.why == null ? '' : o.item.why).trim().slice(0, 1000), depends_on_hint: deps, ...(acceptance !== undefined ? { acceptance } : {}), }; }); } // ──────────────────────────────────────────────────────────────── // 触发入口 // ──────────────────────────────────────────────────────────────── /** * 触发一次 subtask-planning 并把结果写进这条 task 所属 run 的共享 artifacts 区 * + 登记进 `graph_files` 索引。**纯 best-effort,绝不 throw**——见文件头注释。 * * @param {{run: object, task: object}} ctx run 是 graph_runs 表整行(createRun 的返回值即可, * 只需要 company_id / root_run_id / id),task 是 tasks 表整行 */ async function triggerSubtaskPlanning({ run, task }) { try { if (!run || !task) return; // 46 号文档 Phase 3:子任务不该被再拆一层——`materializeDecomposedSubtasks` 物化出的行 // 也会经过 enqueueTaskRun(CEO 派发时),如果不挡住,会对一条已经是"第几步怎么做"粒度的 // 子任务再跑一次 subtask-planning,浪费一次 LLM 调用还可能无限递归拆下去。 // 43 号文档的验收条目子任务同理(createTask 本身已经不许它们再声明 itemized)。 if (task.parent_task_id) { console.log(`[SubtaskPlanning] task ${task.id} 本身就是子任务(parent_task_id=${task.parent_task_id}),跳过`); return; } const { steps, tokens, reasoningSource, experienceRefs, researchUsed, } = await planSubtasks({ companyId: run.company_id, task }); if (!steps.length) { console.log(`[SubtaskPlanning] task ${task.id} 没有可用的步骤分解(0 条),跳过写入`); return; } const { runRootDir } = require('./graph-engine'); const { companyRootDir } = require('./company-paths'); const artifactsDir = path.join(runRootDir(run), 'artifacts'); fs.mkdirSync(artifactsDir, { recursive: true }); const file = path.join(artifactsDir, 'subtask-plan.json'); fs.writeFileSync( file, JSON.stringify({ steps, generated_by: PLANNING_AGENT_ID, task_id: task.id, // Phase 2(46 号文档):这条分解到底是"模型自己想的"/"查到经验库"/"触发了调研", // 由服务端根据实际发生了什么算出来(见 defaultPlanSubtasks),不是 agent 自报的。 reasoning_source: reasoningSource || 'model_only', experience_refs: experienceRefs || [], research_used: !!researchUsed, }, null, 2), 'utf-8' ); const wdb = require('./work-graph-db'); wdb.recordFile({ companyId: run.company_id, taskId: task.id, relPath: path.relative(companyRootDir(run.company_id), file), kind: 'subtask_plan', runId: run.id, nodeRunId: null, }); // 46 号文档 Phase 3:把这批 steps 物化成 decomposed_work 子任务。跟上面写 artifact/登记 // 索引一样,走同一个 try/catch——物化失败不该让 subtask-plan.json 已经写好的部分失效。 const materializedIds = wdb.materializeDecomposedSubtasks({ task, steps }); console.log(`[SubtaskPlanning] task ${task.id} 写入 ${steps.length} 条步骤分解(${tokens || 0} tokens),` + `物化 ${materializedIds.length} 条待 CEO 派发的子任务`); // 46 号文档 Phase 4:拆分成功 → 子任务成为父任务唯一的执行体,父任务自己那条排队中的 // 完整 org graph run 要让位。**顺序很关键,先接管任务再取消 run,不能反过来**: // `stopRun` 会同步触发 run 收敛的回写口 `settleTaskFromRun`,那条路径会把任务写成 // `cancelled`——如果先 `stopRun` 再 `markTaskDecomposed`,`markTaskDecomposed` 的 CAS // (`WHERE status='queued'`)必然扑空,因为任务此刻已经被 `cancelled` 了(这不是小概率 // 竞态,是每次都会发生的确定性顺序,第一版实现在真实测试里当场撞上)。反过来先标 // `decomposed`,`settleTaskFromRun` 新增的 `AND status != 'decomposed'` 闸门会让 // 随后触发的 cancelled 回写扑空,任务保持 `decomposed` 不被覆盖。 // 父 run 仍有极小概率在这一步之前就已经跑完/失败(真赢了竞态)——那种情况下 // `markTaskDecomposed` 的 CAS 本来就拿不到锁(任务已是有效终态,不是 'queued'), // `stopRun` 对已终态 run 也是安全的 no-op,两者叠加不会覆盖一个已经有效终态的父任务。 if (materializedIds.length) { try { const taken = wdb.markTaskDecomposed(task.id); if (!taken) { console.warn(`[SubtaskPlanning] task ${task.id} 接管失败(已不是 queued,多半是父 run 已经赢了竞态跑完了),保留父任务原有终态`); } require('./graph-engine').stopRun(run.id); } catch (err) { console.error(`[SubtaskPlanning] task ${task.id} 取消父 run(${run.id})失败: ${err.message}`); } } } catch (err) { console.warn(`[SubtaskPlanning] task ${task && task.id} 触发失败(不影响入队):${err.message}`); } } module.exports = { PLANNING_AGENT_ID, MAX_STEPS, planSubtasks, setPlanSubtasks, setSubtaskAgentCaller, triggerSubtaskPlanning, // 下面几个导出是给测试用的。`defaultPlanSubtasks` 直接调用会绕开 `planSubtasks()` // 的 VITEST 短路——测试这条分支编排逻辑必须搭配 `setSubtaskAgentCaller` 注入, // 不要在没注入的情况下直接调用它(会真的起 CLI)。 extractPlannedSteps, extractConfidence, normalizeSteps, buildSubtaskPlanningPrompt, defaultPlanSubtasks, };