/** * Graph Engine — P3 图执行引擎(T2) * * 设计见 docs/todo/system-design/27-p3-p4-work-graph-console.md * §四.A.4 状态机 §四.A.6 幂等(条件更新 + 受影响行数) * §四.C 五个控制操作 + 「手动/自动共用同一个 advanceOne」 * §四.D.0 两种 node executor(agent_call / subgraph) * §四.D.1 主流程(success → 出边 → 下游 ready → 收敛 → 回写父节点) * §四.D.2 异常流程骨架(fail → 选边:精确匹配 → 兜底边 → 死路 blocked) * §四.D.2.1 双层回退熔断(单边 loop_count + run 级 backward_count,T7) * §四.A.3 任务级总预算挂 root run,扣完整棵 run 树挂起(T8) * §四.C 暂停/停止三层级联 + 重试的 contract_hash 判据(T8) * * 本文件只管「跑」,不管「存」——所有落库走 work-graph-db.js,不在这里写一句 SQL DDL。 * * 唯一推进原语是 advanceOne(runId):拾取一个 ready 节点 → running → 执行 → 评估出边 → * 把满足条件的下游置 ready → 返回。**它不循环往下跑**。 * 手动模式 = 路由 handler 每收到一次点击调一次; * 自动模式 = runAuto() 在 pause/stop 标志位为假时不停调它。 * 只有这一套推进逻辑,自动模式不是另一套实现(§四.C 末段)。 * * 仍然不做、留给后续任务的: * - running 节点的进程级硬取消(§3.6,调用链由 T9 核实)→ 当前是「等它跑完但不再调度后续」的退化路径 */ const path = require('path'); const fs = require('fs'); const { getDb, readDefGraph, getRun, getRunNode, getRunNodeByDefId, // 注意:updateNodeStatus / updateRunStatus **故意不在这里 import**——状态转换一律走 // 下面的 sse.transitionNode / transitionRun,否则又会退回「改了状态但没有事件」的老路。 bumpNodeAttempt, setRunFlags, createRun, listRunTree, listLoopRecords, insertLoopRecord, updateLoopVerdict, bumpBackwardCount, canTransitionRun, canTransitionNode, } = require('./work-graph-db'); /** * 状态转换的唯一出口(§三.3 末段)。**本文件不再直调 work-graph-db 的裸 * updateNodeStatus / updateRunStatus**:所有转换都走 sse.transitionNode / transitionRun, * 让「改状态 + 写 timeline + 发 SSE」在同一次同步调用里一起发生——DevLoop 那个 * 「审计轨与实时轨不桥接」的缺口就是这么来的,这里不重复它。 * 非状态机类事件(回退备案 / 预算扣减 / agent 调用起止 / 归因产出)走 sse.emitEvent。 */ const sse = require('./graph-run-sse'); const REPO_ROOT = path.resolve(__dirname, '../..'); /** * 子图归因节点的名字。子 run 失败时优先取这个节点的 fail_reason 透传回父层(§四.D.2.2)。 * // 假设(R-16待定案): 归因节点名保留 validator-reflection,不改 validator-triager。 */ const ATTRIBUTION_NODE_NAME = 'validator-reflection'; /** run 的终态,进了就不再被引擎改写 */ const TERMINAL_RUN_STATUS = new Set(['completed', 'failed', 'stopped']); /** * 双层回退硬顶(§四.D.2.1 第 2 步)。两个都是**纯整数**,判定处不调用任何 LLM—— * 这是熔断不被 agent 的自由发挥绕过的唯一保证。 * LOOP_HARD_CAP_EDGE:单边计数,防同一条路径上的死循环; * LOOP_HARD_CAP_RUN :run 级总回退计数,防「换个借口继续退」(R-16 发现的 3×N 架空漏洞)。 * 与 §四.A.3 的任务级总预算是两回事:预算管花了多少钱,这两个管绕了多少圈。 * * ⚠️ 外层计数按 **root_run_id 共池**(见 treeBackwardCount / handleBackwardEdge)。 * 按当前 run 记账时,subgraph 节点每重跑一轮就新建一张子 run,子层计数每轮从 0 重新开始, * 外层硬顶被彻底架空——这正是 27-E2E §7.1「每一级都重置下一级的计数器,整个循环就没有 * 终止条件」的第二个变体(第一个是多条边各自计数把硬顶架空成 3×N)。 */ const LOOP_HARD_CAP_EDGE = 3; const LOOP_HARD_CAP_RUN = 5; /** * 引擎唯一硬编码的 fail_reason 取值(§四.D.2.1 末段把它定成了引擎行为,不是图语义): * 进程超时 / 被杀这类基础设施抖动,原地重跑,不烧任何计数。 * 其余取值(impl_bug / input_underspecified / skill_defect …)引擎一律只做字符串匹配选边, * 语义由图的规划者与 validator 约定;`skill_defect` 的「挂起等人工」由规划者把边连到人工节点 * 或兜底边表达,引擎不特判(L4 在 §一 是非目标)。 */ const INFRA_FLAKE_REASON = 'infra_flake'; /** * validator-reflection 的归因取值域(system/templates/validator-reflection/agent.md * 「归因取值域(四选一,不要发明第五个)」= §四.D.2.2 那张表)。 * * 引擎**不理解**这些取值的语义(选边只做字符串匹配),这个集合只用于一件事: * 把归因文件里自创的取值当作 null 处理(模板明说自创取值会静默落到兜底边, * 那就在入口处显式降级 + 记一条 timeline,而不是让一个假取值混进选边)。 */ const ATTRIBUTION_REASONS = new Set(['impl_bug', 'input_underspecified', 'skill_defect', INFRA_FLAKE_REASON]); /** 归因/结构化产出文件名(agent.md「产出格式」:runs/{rootRunId}/nodes/{nodeRunId}/attempt-{n}.json) */ const REFLECTION_FILE_RE = /^attempt-(\d+)\.json$/; /** 判「这次失败了」的 verdict(agent.md 产出格式那三个取值里的两个);no_prior_failure 是放行 */ const FAILING_VERDICTS = new Set(['attributed', 'inconclusive']); /** * 「必须留下判词」的节点(27-NAMING §3.1/§3.3/§3.5 的 validator 命名轴): * 归因节点 validator-reflection(含 org-validator-reflection 这种带父节点全名的写法)与 * 一切 `{...}-validator` 节点。这些节点的**判定结果只存在于产出文件里**(verdict / fail_reason), * 调用本身正常退出说明不了任何事。文件缺失时按「无法判定」结算,不能静默判 success, * 否则 fail 边 / 回退 / infra_flake 原地重跑在生产里永远触发不了(R3-F1 缺陷 3)。 */ const ATTRIBUTION_REQUIRED_RE = /(^|[-_])(reflection|validator)$/i; function requiresAttributionProduct(ctx) { const name = String((ctx && ctx.nodeName) || ''); return name === ATTRIBUTION_NODE_NAME || ATTRIBUTION_REQUIRED_RE.test(name); } const now = () => new Date().toISOString(); // ──────────────────────────────────────────────────────────────── // 注入点:两个外部依赖都可替换(测试注入假实现,避免真起 Claude CLI) // ──────────────────────────────────────────────────────────────── /** * workgraph-planning 注入点(§四.D.0)。 * * P3 阶段这一步还没接上(属阶段 D),默认实现直接抛错。接上之后替换成: * 调 workgraph-planning 本体 → 产出图 → 过 §三.7 五项校验 → createWorkGraph()。 * * @type {(ctx: NodeContext) => Promise<{workGraphId: string}>} */ let planWorkGraph = async (ctx) => { throw new Error( `workgraph-planning 未接入(阶段 D):节点 ${ctx.nodeName} 是 subgraph,` + `需要先 setPlanWorkGraph(fn) 注入图规划实现,或在建 run 前预先备好 work graph` ); }; /** 替换 workgraph-planning 实现(阶段 D 接真实现;测试注入假图) */ function setPlanWorkGraph(fn) { planWorkGraph = fn; } /** * agent_call 执行器注入点。默认实现调 AgentManager(见 defaultAgentExecutor)。 * @type {(ctx: NodeContext) => Promise} */ let agentExecutor = defaultAgentExecutor; /** 替换 agent_call 执行器(测试注入假 executor,避免真起 Claude CLI) */ function setAgentExecutor(fn) { agentExecutor = fn || defaultAgentExecutor; } // ──────────────────────────────────────────────────────────────── // 两种 node executor(§四.D.0) // ──────────────────────────────────────────────────────────────── /** * @typedef {object} NodeContext * @property {string} runId @property {object} run * @property {string} rootRunId @property {string} companyId * @property {string} nodeRunId @property {string} defNodeId * @property {string} nodeName @property {string} systemAgentId @property {string} nodeKind * @property {number} attempt * @property {string} task 任务描述(调用方透传,缺省从 run 的 goal/task id 拼) * @property {string|null} remediationNote 上一轮回退带过来的「应该如何修复」 * @property {string} executionMode * * @typedef {object} AgentResult * @property {'success'|'fail'} status * @property {string} [output] @property {string|null} [failReason] * @property {string} [outputRef] @property {string} [agentRunId] */ /** * agent_call:调 AgentManager 跑 system_agent_id。 * * 成功/失败只看调用本身是否抛异常——**引擎不发明 fail_reason**。 * 按 §四.D.2.2 的定案,归因是一次显式的 agent 调用(validator-reflection 干的活), * 不是失败节点顺手吐一个值;执行器只在结果里带上 agent 自己给的 failReason(若有)。 * 没有 failReason 时选边会退到兜底边,不会莫名其妙死路(§三.7 第五项校验保证兜底边存在)。 */ async function defaultAgentExecutor(ctx) { const AgentManager = require('../src/core/agent-manager'); // 空 config 即可:getAgentConfig 有基于文件系统的动态回退(system/agents/ 与 // data/companies/{cid}/agents/ 两条路径),与 agent-chat.js 的用法一致。 const manager = new AgentManager({}, REPO_ROOT); const task = ctx.remediationNote ? `${ctx.task}\n\n[上一轮回退的修复建议]\n${ctx.remediationNote}` : ctx.task; let res = null; let err = null; try { res = await manager.callAgent(resolveAgentName(ctx.companyId, ctx.systemAgentId), task, { metadata: { taskId: ctx.runId, layer: 'graph-node', nodeRunId: ctx.nodeRunId, // 产物目录是**提示词**里那份约定的机器可读副本(提示词由 describeArtifactContract 拼, // 见 advanceOne);metadata 只进日志与 hook,agent 读不到,所以两边都要有。 rootRunId: ctx.rootRunId, attempt: ctx.attempt, artifactDir: safeNodeArtifactDir(ctx), }, }); } catch (e) { err = e; } // 归因/产出是**文件**,不是返回值:agent 崩了也可能已经把 attempt-{n}.json 写出来了, // 所以成功失败两条路径都读一次(§四.D.2.2 + validator-reflection/agent.md 的产出约定)。 const produced = readNodeProduct(ctx); if (err) { return { status: 'fail', output: err.message, failReason: produced.failReason, remediationNote: produced.remediationNote, outputRef: produced.outputRef, }; } // 判词节点(validator / reflection)没留下产出文件 = **无法判定**,不是通过(R3-F1 缺陷 3b)。 // 按 §四.D.2.2 的 inconclusive 语义结算:fail + fail_reason=null → 走那条兜底边 // (§三.7 第五项校验保证它存在;没有兜底边就是死路 blocked,等人工)。 // 静默判 success 的后果不是「失败但走兜底边」,而是「validator 判的失败根本不存在」。 if (produced.missing && requiresAttributionProduct(ctx)) { timelineNote(ctx, 'node.reflection', `判词节点没有可用的 attempt-${ctx.attempt}.json(缺失/不可解析/只剩上一轮的),` + '按「无法判定」结算(fail + fail_reason=null → 兜底边)', { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, degraded: 'missing_product', expected_file: path.join(safeNodeArtifactDir(ctx) || '', `attempt-${ctx.attempt}.json`), }); return { status: 'fail', output: res && res.output, agentRunId: (res && res.runId) || null, failReason: null, remediationNote: null, outputRef: produced.outputRef, }; } return { // 调用没抛不等于节点 success:validator 类节点判「这次失败了」是**写在产出文件里**的 // (verdict=attributed / inconclusive),CLI 本身当然是正常退出的。产出说失败就按失败结算, // 否则 fail 边、回退边、§四.D.2.1 的 infra_flake 原地重跑在生产里永远触发不了 // ——引擎仍然没有「发明」fail_reason,它只是照抄归因节点写下的判词(§四.D.2.2)。 status: produced.failingVerdict ? 'fail' : 'success', output: res && res.output, // RunRegistry 的 runId 目前没有从 callAgent 透出来(调用链由 T9 核实/补齐), // 执行器返回了就写,没有就留空。 agentRunId: (res && res.runId) || null, failReason: produced.failingVerdict ? produced.failReason : null, // 首轮进场(verdict=no_prior_failure)也可能留了话给下一轮,成功路径同样带回 remediationNote: produced.remediationNote, outputRef: produced.outputRef, }; } /** * 读节点这一 attempt 的结构化产出,解析出三值透传需要的两个值 + output_ref。 * * 约定路径(system/templates/validator-reflection/agent.md「产出格式」): * data/companies/{cid}/runs/{rootRunId}/nodes/{nodeRunId}/attempt-{n}.json * attempt 编号在模板里没写死是 0 基还是 1 基,所以先按 ctx.attempt 与 ctx.attempt+1 找, * 都没有就退到目录里编号最大的那个 attempt-*.json。 * * **一律不抛**(§四.D.2.2:读不到就是没归因,不是节点执行失败): * 文件缺失 → 返回三个 null 且 `missing=true`(绝大多数节点本来就不产归因文件,不刷 timeline; * 是不是判词节点、缺了算不算失败,由调用方按 requiresAttributionProduct 决定); * 文件存在但解析失败 / 取值非法 → 降级为 null **并记一条 timeline**,让控制台看得见归因白做了。 * * @returns {{failReason: string|null, remediationNote: string|null, outputRef: string|null, * failingVerdict: boolean, missing: boolean}} */ function readNodeProduct(ctx) { const empty = { failReason: null, remediationNote: null, outputRef: null, failingVerdict: false, missing: true, }; if (!ctx || !ctx.run) return empty; let dir; try { dir = nodeArtifactDir(ctx.run, ctx.nodeRunId); if (!fs.existsSync(dir)) return empty; } catch { return empty; // 路径都拼不出来(run 缺列)就是没有产物 } // 节点私有目录本身就是 output_ref 指向的「被测产物」(goal-contract.js 顶部注释) const outputRef = dir; const { file, stale } = resolveReflectionFile(dir, ctx.attempt); if (!file) { if (stale) { // 目录按 nodeRunId 分配、跨 attempt 复用,编号比本轮小的那份是**上一轮**的判词。 // 拿它当本轮结论会让节点被旧判词反复判失败直到撞硬顶(R3-F1 缺陷 4)——宁可当没有。 timelineNote(ctx, 'node.reflection', `只找到上一轮的归因文件 ${path.basename(stale)}(< attempt-${ctx.attempt}),不复用`, { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, file: stale, degraded: 'stale_reflection', }); } return { ...empty, outputRef }; } const doc = readJsonOrNull(file); if (!doc || typeof doc !== 'object') { timelineNote(ctx, 'node.reflection', `归因文件解析失败,fail_reason/remediation_note 降级为 null: ${path.basename(file)}`, { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, file, degraded: 'unparsable', }); return { ...empty, outputRef }; } const raw = doc.fail_reason == null ? null : String(doc.fail_reason); let failReason = raw; if (raw !== null && !ATTRIBUTION_REASONS.has(raw)) { // 模板:「四个值以外一律不要输出」。自创取值不报错,当 null 处理(=走兜底边)并留痕。 failReason = null; timelineNote(ctx, 'node.reflection', `归因取值非法(${raw}),按 null 处理并走兜底边`, { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, file, degraded: 'invalid_fail_reason', fail_reason: raw, }); } const remediationNote = typeof doc.remediation_note === 'string' && doc.remediation_note.trim() ? doc.remediation_note : null; // verdict 决定节点判不判失败(模板「产出格式」:attributed | inconclusive | no_prior_failure)。 // inconclusive 也算失败——模板明写「让引擎走那条 fail_reason IS NULL 的兜底边」。 // 没写 verdict 但给了 fail_reason 的,按失败处理(有根因就是有失败)。 const verdict = doc.verdict == null ? null : String(doc.verdict); const failingVerdict = verdict === null ? raw !== null : FAILING_VERDICTS.has(verdict); if (failingVerdict || remediationNote) { timelineNote(ctx, 'node.reflection', `归因 verdict=${verdict || '(缺省)'} fail_reason=${failReason || 'null'}`, { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, file, verdict, fail_reason: failReason, remediation_note: remediationNote, failing: failingVerdict, }); } return { failReason, remediationNote, outputRef, failingVerdict, missing: false }; } /** * attempt-{n}.json 的定位:先精确(n 与 n+1 两种基数),再退到编号最大的那份。 * * 兜底**只认编号 >= 本轮 attempt 的文件**:节点私有目录按 nodeRunId 分配、跨 attempt 复用, * 无条件取「编号最大的」会读到上一轮的归因文件,节点被旧判词反复判失败直到撞硬顶 * (R3-F1 缺陷 4)。编号更小的那份原样回报给调用方去记一条 degraded timeline。 * * @returns {{file: string|null, stale: string|null}} */ function resolveReflectionFile(dir, attempt) { for (const n of [attempt, attempt + 1]) { const p = path.join(dir, `attempt-${n}.json`); if (fs.existsSync(p)) return { file: p, stale: null }; } let best = null; let bestN = -1; let names = []; try { names = fs.readdirSync(dir); } catch { return { file: null, stale: null }; // 目录刚被删/不可读:当作没有归因产出 } for (const name of names) { const m = REFLECTION_FILE_RE.exec(name); if (m && Number(m[1]) > bestN) { bestN = Number(m[1]); best = path.join(dir, name); } } return bestN >= attempt ? { file: best, stale: null } : { file: null, stale: best }; } /** 只记一条审计(同时也发 SSE,走 emitEvent 保证两轨不分家);失败绝不影响执行 */ function timelineNote(ctx, event, summary, data) { try { sse.emitEvent(ctx.runId, event, summary, data, { ref: ctx.nodeRunId }); } catch { // 审计写不进去(磁盘满 / 目录只读)不该把节点跑挂 } } /** * agent 本体名 → AgentManager 能寻址的名字。 * 公司雇过这个本体时目录在 data/companies/{cid}/agents/{id},寻址名带 `cid:` 前缀 * (与 agent-chat.js getAgentDir 的约定一致);否则用裸本体名走 system/agents/。 */ function resolveAgentName(companyId, systemAgentId) { if (!companyId) return systemAgentId; const scoped = path.join(REPO_ROOT, 'data', 'companies', companyId, 'agents', systemAgentId); return fs.existsSync(scoped) ? `${companyId}:${systemAgentId}` : systemAgentId; } /** * subgraph:起一张子图并等它收敛(§四.D.0)。 * * 「等」不是在这里 await 几小时——手动模式下 advanceOne 由一次 HTTP 请求驱动,不可能挂住。 * 落地方式是:本函数只负责规划图 + 建子 run + 回填 child_run_id,然后返回 status='pending' * 表示「父节点继续停在 running」;子 run 收敛时由 settleRun → propagateToParent 回写父节点。 * 两条路径最终都进同一个 settleNode,且靠 running→终态 的条件更新保证只生效一次。 */ async function executeSubgraph(ctx) { // 幂等守卫的判据是「**本轮**是不是已经起过子 run」,不是「child_run_id 非空」。 // // 「child_run_id 非空即跳过」曾经是个永久死锁(P3 首轮缺陷 1):节点第二次变 ready 时 // 拿到的是上一轮**已终态**的旧子 run,直接返回 pending,父节点永远停在 running。 // §四.D.3 定案是「该 org node 整张 work graph 重跑(起新子 run,attempt+1)」—— // 旧子 run 留作历史(控制台按 parent_node_run_id 仍查得到),本轮另起一张。 // // 判据选「旧子 run 是否还活着」而不是「attempt 变没变」:infra_flake 原地重跑 // (§四.D.2.1 末段)明确**不加 attempt**,光比 attempt 会把它误判成同一轮。 // 活着 = run_status 还没落到终态/blocked,即父节点确实还在等它回写; // 同一轮内的并发进入(进程重启重放)看到的一定是活的子 run,仍然只起一张图。 const existing = getRunNode(ctx.nodeRunId); const prior = existing && existing.child_run_id ? getRun(existing.child_run_id) : null; if (prior && !isChildSettled(prior)) { return { status: 'pending', childRunId: prior.id }; } const planned = await planWorkGraph(ctx); const workGraphId = planned && (planned.workGraphId || planned.id); if (!workGraphId) throw new Error(`planWorkGraph 未返回 workGraphId(节点 ${ctx.nodeName})`); const child = createRun({ graphKind: 'work', graphId: workGraphId, companyId: ctx.companyId, parentRunId: ctx.runId, parentNodeRunId: ctx.nodeRunId, goalId: ctx.run.goal_id, taskId: ctx.run.task_id, okrNodeId: ctx.run.okr_node_id, // 子 run 默认继承父 run 的执行模式(§四.D.0);预算只挂 root run,这里不传 executionMode: ctx.executionMode, }); getDb() .prepare(`UPDATE graph_run_nodes SET child_run_id = ?, updated_at = ? WHERE id = ?`) .run(child.id, now(), ctx.nodeRunId); markRunReady(child.id); startRunSync(child.id); return { status: 'pending', childRunId: child.id }; } /** * 子 run 是否已经「交待完了」——终态或 blocked 都算:settleRun 在这两种情形下都会 * propagateToParent 回写父节点,父节点不该再等它。 */ function isChildSettled(childRun) { return TERMINAL_RUN_STATUS.has(childRun.run_status) || childRun.run_status === 'blocked'; } // ──────────────────────────────────────────────────────────────── // 核心原语:advanceOne(§四.C / §四.D.1) // ──────────────────────────────────────────────────────────────── /** * 推进一步:拾取**一个** ready 节点 → running → 按 node_kind 分派执行 → 评估出边 → * 把满足条件的下游置 ready → 返回本次结果。不继续往下跑。 * * 幂等(§四.A.6):ready→running 是条件更新,受影响行数 = 0 即「已被别人拿走」, * 返回当前状态而不是报错。 * * @param {string} runId * @param {object} [opts] * @param {boolean} [opts.ignorePause] 「下一步」用:auto 已 paused 时也允许单步推一下 * @param {string} [opts.task] 传给 executor 的任务描述 * @returns {Promise} * * @typedef {object} AdvanceResult * @property {boolean} ok 本次是否真的推进了一个节点 * @property {string} runId * @property {string} runStatus 推进后的 run 状态 * @property {string|null} reason 未推进的原因码 / 特殊情形码,见下 * @property {object|null} node {id, defNodeId, name, kind, status, attempt, * failReason, outputRef, agentRunId, childRunId} * @property {Array} activated 本次被置 ready 的下游 [{id, defNodeId, name}] * @property {string|null} backwardEdgeId 命中的 backward 边 id * @property {object|null} loop 本次回退的熔断账:{edgeId, recordId, loopCount, * backwardCount, forceTerminated, cap:'edge'|'run'|null} * @property {object|null} contractDiff 人工重试后的契约四分类 + 被打回 pending 的下游 * @property {string|null} childRunId subgraph 节点起的子 run * * reason 取值:run-not-found | run-not-running | stop-requested | pause-requested | * no-ready-node | taken-by-other | awaiting-child-run | budget-exhausted | * null(正常推进完成) */ async function advanceOne(runId, opts = {}) { const run = getRun(runId); if (!run) return result(runId, null, 'run-not-found'); // 软信号在**节点边界**生效:不打断已经在跑的节点,只是不再拾取新的(§四.C) if (run.stop_requested) return result(runId, run.run_status, 'stop-requested'); if (run.pause_requested && !opts.ignorePause) { // 到了节点边界,「暂停」这时候才真正落到 run_status 上 landPause(runId); return result(runId, (getRun(runId) || {}).run_status, 'pause-requested'); } if (run.run_status !== 'running') return result(runId, run.run_status, 'run-not-running'); const picked = getDb() .prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? AND status = 'ready' ORDER BY created_at, id LIMIT 1`) .get(runId); if (!picked) { // 没有 ready 节点:可能是全跑完了,也可能全都卡在子 run 上,交给收敛判定 return result(runId, settleRun(runId), 'no-ready-node'); } // §四.A.3:任何一层发起 agent 调用前先按 root_run_id 查同一个预算池的余额。 // 拾到了 ready 节点(=真要花钱了)才查,避免空转时把已跑完的 run 误挂起。 const gate = checkBudget(run); if (!gate.ok) { // 扣完不是失败,是**整棵 run 树挂起等人工**:root run → blocked,向下级联 pause_requested=1 suspendRunTree(run); return { ...result(runId, (getRun(runId) || {}).run_status, 'budget-exhausted'), budget: gate }; } const def = defIndex(run); const defNode = def.nodes.get(picked.def_node_id) || {}; const nodeName = defNode.name || picked.def_node_id; // ready → running 的条件更新就是并发闸门(§四.A.6) if ( !sse.transitionNode(runId, picked.id, 'ready', 'running', { startedAt: now() }, { node_name: nodeName, node_kind: defNode.nodeKind || 'agent_call', attempt: picked.attempt, }) ) { const cur = getRunNode(picked.id); return { ...result(runId, run.run_status, 'taken-by-other'), node: nodeView(runId, cur) }; } const ctx = { runId, run, rootRunId: run.root_run_id, companyId: run.company_id, nodeRunId: picked.id, defNodeId: picked.def_node_id, nodeName, systemAgentId: defNode.systemAgentId || null, nodeKind: defNode.nodeKind || 'agent_call', attempt: picked.attempt, task: opts.task || describeTask(run), remediationNote: latestRemediationNote(runId, picked.id), executionMode: run.execution_mode, }; // 产物目录必须**进提示词**:读的一侧(readNodeProduct)按 §四.A.5 的路径去读, // 写的一侧不告诉 agent 往哪写、文件名怎么起,这份文件在生产里就根本不会被写出来, // 于是 fail 边 / 回退 / infra_flake 全都触发不了(R3-F1 缺陷 3a)。 if (ctx.nodeKind !== 'subgraph') ctx.task += describeArtifactContract(ctx); // 抢到节点即记账:一次调用一定发生了(哪怕它随后炸掉),预算「只增不减」(§四.A.3)。 // subgraph 节点这一次花的是 workgraph-planning 的调用,同样按一次计。 spendBudget(run, { agentCalls: 1 }); sse.emitEvent( runId, 'agent.call', `${ctx.nodeKind === 'subgraph' ? '规划子图' : '调用 agent'} ${ctx.systemAgentId || nodeName}(attempt #${picked.attempt + 1})`, { node_run_id: picked.id, node_name: nodeName, node_kind: ctx.nodeKind, system_agent_id: ctx.systemAgentId, attempt: picked.attempt, remediation_note: ctx.remediationNote, }, { ref: picked.id }, ); let outcome; try { outcome = ctx.nodeKind === 'subgraph' ? await executeSubgraph(ctx) : await agentExecutor(ctx); } catch (err) { // executor 自己炸了(规划失败 / 配置缺失等)算节点失败,不带 failReason,走兜底边 outcome = { status: 'fail', output: err.message, failReason: null }; } // executor 报了 token 用量就一并扣(没报就只扣调用次数,wallclock 由 spent.started_at 兜) if (outcome && outcome.tokens) spendBudget(run, { tokens: outcome.tokens }); sse.emitEvent( runId, 'node.output', `${nodeName} → ${(outcome && outcome.status) || 'unknown'}${outcome && outcome.failReason ? `(${outcome.failReason})` : ''}`, { node_run_id: picked.id, node_name: nodeName, attempt: picked.attempt, status: outcome && outcome.status, fail_reason: (outcome && outcome.failReason) || null, output_ref: (outcome && outcome.outputRef) || null, child_run_id: (outcome && outcome.childRunId) || null, }, { ref: picked.id }, ); if (outcome && outcome.status === 'pending') { // subgraph:父节点继续停在 running,等子 run 收敛回写 return { ...result(runId, run.run_status, 'awaiting-child-run'), ok: true, node: nodeView(runId, getRunNode(picked.id)), childRunId: outcome.childRunId || null, }; } return settleNode(runId, picked.id, outcome); } /** * 节点执行结束后的落地:写终态 → 评估出边 → 收敛判定。 * 也是子 run 收敛时回写父节点走的同一条路(propagateToParent)。 * * @param {object} outcome {status, failReason?, remediationNote?, outputRef?, agentRunId?, childRunId?} */ function settleNode(runId, nodeRunId, outcome = {}) { const to = outcome.status === 'success' ? 'success' : 'fail'; const changed = sse.transitionNode(runId, nodeRunId, 'running', to, { failReason: outcome.failReason === undefined ? null : outcome.failReason, outputRef: outcome.outputRef, agentRunId: outcome.agentRunId, childRunId: outcome.childRunId, completedAt: now(), }, { attempt: (getRunNode(nodeRunId) || {}).attempt, remediation_note: outcome.remediationNote || null }); // 0 行 = 这个节点已经被别人结算过了(§四.A.6),原样返回当前状态 if (!changed) { return { ...result(runId, (getRun(runId) || {}).run_status, 'taken-by-other'), node: nodeView(runId, getRunNode(nodeRunId)), }; } const nodeRun = getRunNode(nodeRunId); const run = getRun(runId); const def = defIndex(run); const out = { activated: [], backwardEdgeId: null, loop: null, contractDiff: null }; if (to === 'success') { // 人工「重试」留下的契约快照在这里判:契约变了才级联下游(§四.C 末段) out.contractDiff = applyRetryCascade(runId, def, nodeRun); activateSuccessTargets(runId, def, nodeRun.def_node_id, out); } else { routeFailure(runId, def, nodeRun, out, outcome.remediationNote || null); } return { ok: true, runId, runStatus: settleRun(runId), reason: null, node: nodeView(runId, getRunNode(nodeRunId)), activated: out.activated, backwardEdgeId: out.backwardEdgeId, loop: out.loop, contractDiff: out.contractDiff, childRunId: nodeRun.child_run_id || null, }; } // ──────────────────────────────────────────────────────────────── // §四.D.1 主流程:success 后的出边评估 // ──────────────────────────────────────────────────────────────── /** * 遍历 condition='success' 的出边,目标节点若**所有**入边条件都满足则置 ready。 * 「全部满足」= 该目标所有 forward + success 的入边,其源节点都已 success(汇合点语义)。 * fail 边和 backward 边不参与这道闸——它们是异常流程,各自单独把目标点亮。 */ function activateSuccessTargets(runId, def, defNodeId, out) { for (const e of def.edges) { if (e.from !== defNodeId || e.condition !== 'success') continue; if (!allIncomingSatisfied(runId, def, e.to)) continue; const activated = markReady(runId, def, e.to); if (activated) out.activated.push(activated); } } function allIncomingSatisfied(runId, def, toDefNodeId) { return def.edges .filter((e) => e.to === toDefNodeId && e.condition === 'success' && e.edgeKind !== 'backward') .every((e) => { const src = getRunNodeByDefId(runId, e.from); return src && src.status === 'success'; }); } /** pending → ready;已经不是 pending(被别人点亮或已跑过)就跳过,返回 null */ function markReady(runId, def, defNodeId) { const rn = getRunNodeByDefId(runId, defNodeId); if (!rn || rn.status !== 'pending') return null; const d = def.nodes.get(defNodeId) || {}; const name = d.name || defNodeId; if (!sse.transitionNode(runId, rn.id, 'pending', 'ready', {}, { node_name: name, attempt: rn.attempt })) { return null; } return { id: rn.id, defNodeId, name }; } // ──────────────────────────────────────────────────────────────── // §四.D.2 异常流程骨架:fail 后选边 // ──────────────────────────────────────────────────────────────── /** * 选边规则(§四.D.2.2,引擎只做字符串匹配,不理解取值语义): * 先找 fail_reason 精确匹配的边 → 无匹配退到 fail_reason IS NULL 的兜底边 → 再无则死路。 * 取值域不在引擎里硬编码,由图的规划者与对应 validator 约定。 */ function selectFailEdge(def, defNodeId, failReason) { const candidates = def.edges.filter((e) => e.from === defNodeId && e.condition === 'fail'); if (candidates.length === 0) return null; if (failReason) { const exact = candidates.find((e) => e.failReason === failReason); if (exact) return exact; } return candidates.find((e) => e.failReason === null || e.failReason === undefined || e.failReason === '') || null; } function routeFailure(runId, def, nodeRun, out, remediationNote = null) { // infra_flake 特例(§四.D.2.1 末段):基础设施抖动直接**原地重跑**,不选边、不写 // loop record、不加 backward_count、不扣该节点的 retry 预算(attempt 也不加一)。 // 终止性由 §四.A.3 的任务级总预算兜底——每次重跑仍算一次 agent 调用,池子空了整棵树挂起。 if (nodeRun.fail_reason === INFRA_FLAKE_REASON) { const d = def.nodes.get(nodeRun.def_node_id) || {}; const name = d.name || nodeRun.def_node_id; // 原地重跑 = 一轮新的执行:subgraph 节点必须**清空 child_run_id**,否则下一次 // executeSubgraph 会拿着上一轮的旧子 run 直接返回 pending(缺陷 1 的另一条入口)。 if ( sse.transitionNode(runId, nodeRun.id, 'fail', 'ready', { childRunId: null, failReason: null }, { node_name: name, attempt: nodeRun.attempt, summary: `node ${name}: infra_flake 原地重跑(不烧计数)` }) ) { out.activated.push({ id: nodeRun.id, defNodeId: nodeRun.def_node_id, name }); out.infraRerun = true; } return; } const edge = selectFailEdge(def, nodeRun.def_node_id, nodeRun.fail_reason); // 死路:没有可走的 fail 边 → 节点 blocked,等人工介入(区别于人工主动「停止」) if (!edge) return blockNode(runId, nodeRun.id); if (edge.edgeKind === 'backward') { out.backwardEdgeId = edge.id; const handled = handleBackwardEdge(runId, def, nodeRun, edge, out, remediationNote); // 目标节点不在本 run(定义层被改过)等异常情形:源节点 blocked,不静默留在 fail 态 if (!handled) blockNode(runId, nodeRun.id); return; } // forward 的 fail 边 = 普通失败分支(如「失败转人工审核节点」),直接点亮目标 const activated = markReady(runId, def, edge.to); if (activated) out.activated.push(activated); } /** * 沿一条 backward 边回退 —— §四.D.2.1 的三步,**无条件按顺序执行**: * * 0. 越顶即收敛:整棵 run 树的 backward_count 已经越过 run 级硬顶时,**连备案都不再做**, * 直接 blockNode。少了这一条,撞顶之后每多退一次就多一条 force_terminated 记录, * 而回退目标仍会被重新点亮(第 3 步),熔断只是「每轮记一笔」而不是「停下来」。 * 1. 备案:insertLoopRecord(verdict='registered'),loop_count = 该 (run_id, edge_id) 已有 * 记录数 + 1(由 work-graph-db 算);同时 backward_count += 1。 * 「无条件」指两个计数在任何回退上都写,先记账再判,不能先判后记。 * 2. 双层硬顶:loop_count > 3(单边)**或** backward_count > 5(run 树级)→ 改判 * force_terminated,源节点 blocked,链路收敛等人工。 * **这一步是纯整数比较,不调用任何 LLM**——单边计数防同一路径死循环,外层计数防 * 「换个借口继续退」(R-16:多条回退边各自计数会把硬顶架空成 3×N)。 * 3. 回退:把 to_node 置 ready、attempt+1,remediation_note 落在 loop record 里, * 由 latestRemediationNote 在下一轮取给回退目标当上下文补充。 * * @returns {boolean} 是否已经处理掉这次失败 */ function handleBackwardEdge(runId, def, nodeRun, edge, out, remediationNote = null) { const target = getRunNodeByDefId(runId, edge.to); if (!target) return false; // 边指向的节点不在本 run(定义层被换过)→ 交给调用方 blockNode const rootRunId = (getRun(runId) || {}).root_run_id || runId; // ⓪ 整棵树已经越顶:不再登记新记录,直接收敛。 // 子层每重跑一轮就是一张新子 run,若只按「本 run」判,越顶后的每一轮都会重新开一条 // 记录并把目标重新点亮(父层因此烧满 4 轮),熔断形同虚设。 const already = treeBackwardCount(rootRunId); if (already > LOOP_HARD_CAP_RUN) { out.loop = { edgeId: edge.id, recordId: null, loopCount: null, backwardCount: already, forceTerminated: true, cap: 'run', }; sse.emitEvent(runId, 'loop.terminated', `熔断:run 树级硬顶已撞(${already}/${LOOP_HARD_CAP_RUN}),不再备案,直接收敛等人工`, { edge_id: edge.id, backward_count: already, root_run_id: rootRunId, verdict: 'force_terminated', cap: 'run', registered: false, from_node_run_id: nodeRun.id, to_node_run_id: target.id, }, { ref: edge.id }); blockNode(runId, nodeRun.id); return true; } // ① 备案:两个计数一起加,谁都不许跳过 const rec = insertLoopRecord({ runId, edgeId: edge.id, fromNodeRunId: nodeRun.id, toNodeRunId: target.id, verdict: 'registered', remediationNote, }); const backwardCount = bumpTreeBackwardCount(runId, rootRunId); sse.emitEvent(runId, 'loop.registered', `回退备案 loop_count=${rec.loopCount}/${LOOP_HARD_CAP_EDGE} · backward_count=${backwardCount}/${LOOP_HARD_CAP_RUN}`, { record_id: rec.id, edge_id: edge.id, loop_count: rec.loopCount, backward_count: backwardCount, from_node_run_id: nodeRun.id, to_node_run_id: target.id, verdict: 'registered', remediation_note: remediationNote, }, { ref: edge.id }); out.loop = { edgeId: edge.id, recordId: rec.id, loopCount: rec.loopCount, backwardCount, forceTerminated: false, cap: null, }; // ② 双层硬顶:纯整数比较,零 LLM 参与(backwardCount 是整棵 run 树的共池计数) if (rec.loopCount > LOOP_HARD_CAP_EDGE || backwardCount > LOOP_HARD_CAP_RUN) { updateLoopVerdict(rec.id, 'force_terminated'); out.loop.forceTerminated = true; out.loop.cap = rec.loopCount > LOOP_HARD_CAP_EDGE ? 'edge' : 'run'; sse.emitEvent(runId, 'loop.terminated', `熔断:${out.loop.cap === 'edge' ? '单边' : 'run 级'}硬顶已撞,链路收敛等人工`, { record_id: rec.id, edge_id: edge.id, loop_count: rec.loopCount, backward_count: backwardCount, verdict: 'force_terminated', cap: out.loop.cap, }, { ref: edge.id }); blockNode(runId, nodeRun.id); return true; } // ③ 回退。先把目标的 forward 下游打回 pending,再点亮目标本身。 out.reset = resetForwardDownstream(runId, def, edge.to); if (target.status !== 'ready' && target.status !== 'running' && canTransitionNode(target.status, 'ready')) { const d = def.nodes.get(edge.to) || {}; const name = d.name || edge.to; // 回退目标重新变 ready = 新一轮:subgraph 节点的 child_run_id 必须清空, // 否则 executeSubgraph 会复用上一轮的终态子 run 并永远返回 pending(缺陷 1 主入口)。 if ( sse.transitionNode(runId, target.id, target.status, 'ready', { childRunId: null, failReason: null }, { node_name: name, attempt: target.attempt + 1, backward_edge_id: edge.id, summary: `node ${name}: 沿回退边 ${edge.id} 重新点亮(attempt #${target.attempt + 2})` }) ) { bumpNodeAttempt(target.id); out.activated.push({ id: target.id, defNodeId: edge.to, name }); } } return true; } /** * 外层熔断计数的**共池读**:整棵 run 树的总回退次数,记在 root run 那一行。 * * 为什么不按当前 run 读:subgraph 节点每重跑一轮都新建一张子 run(executeSubgraph), * 子层的计数会每轮从 0 重新开始 —— 27-E2E §7.1「每一级都重置下一级的计数器,整个循环就 * 没有终止条件」。预算(§四.A.3)早就是按 root_run_id 共池的,这里对齐同一个口径。 */ function treeBackwardCount(rootRunId) { const root = getRun(rootRunId); return root ? root.backward_count || 0 : 0; } /** * 共池写:root run 那一行是**全树**总数(熔断判据),子 run 自己那一行仍记本层次数 * (控制台按 run 展示用)。root 与当前 run 是同一行时只加一次。 * work-graph-db.bumpBackwardCount 不归本文件改,这里直接按 root_run_id 寻址它。 * @returns {number} 加完之后的**全树**总数 */ function bumpTreeBackwardCount(runId, rootRunId) { if (runId !== rootRunId) bumpBackwardCount(runId); return bumpBackwardCount(rootRunId); } /** * 把某节点 forward 可达的全部下游打回 pending。 * * 回退必须做这件事:下游节点此刻多半已经是 success/fail 终态,而 markReady 只认 pending, * 不重置的话第二轮永远点不亮,run 会假收敛成 completed。 * 这与「重试默认不级联下游」不矛盾——那条讲的是**人工重试**,判据是 contract_hash * (见 applyRetryCascade);自动回退的语义本来就是「上游要重做,下游全部作废」。 * * ponytail: NODE_TRANSITIONS 里没有 *→pending 这一条(work-graph-db 不归本文件改), * 所以这里走一次带 status 条件的直接 UPDATE,仍然是「受影响行数 = 是否我改的」的幂等语义。 * 升级路径:等 work-graph-db 在转换表里补上 success/fail/blocked/skipped → pending。 */ function resetForwardDownstream(runId, def, fromDefNodeId) { const db = getDb(); const reset = []; for (const defNodeId of forwardReachable(def, fromDefNodeId)) { const rn = getRunNodeByDefId(runId, defNodeId); // running 的不动(它正在跑,结算时自己会落地);pending 的本来就是 pending if (!rn || rn.status === 'pending' || rn.status === 'running') continue; const d = def.nodes.get(defNodeId) || {}; const name = d.name || defNodeId; // child_run_id 一起清:这个节点下一轮会被重新点亮,subgraph 节点必须起**新**子 run, // 不能复用上一轮那张已终态的图(缺陷 1 的第二条入口)。旧子 run 行留着不动, // 控制台按 parent_node_run_id 仍查得到历史子图。 const changed = sse.emitTransition( runId, { event: 'node.status', ref: rn.id, summary: `node ${name}: ${rn.status} → pending(上游回退,本轮产出作废)`, data: { node_run_id: rn.id, node_name: name, from: rn.status, to: 'pending', attempt: rn.attempt, reset_by: fromDefNodeId, }, }, () => db .prepare(` UPDATE graph_run_nodes SET status = 'pending', fail_reason = NULL, child_run_id = NULL, started_at = NULL, completed_at = NULL, updated_at = ? WHERE id = ? AND status IN ('ready','success','fail','blocked','skipped') `) .run(now(), rn.id).changes, ); if (changed) { // 被作废的节点下一次执行是**新的一轮**,attempt 必须往前走。 // 不加的话它的 attempt 永远停在上一轮:产物文件名 attempt-{n}.json 不变, // readNodeProduct 每轮都精确命中**上一轮**那份归因文件,节点被旧判词反复判失败 // 直到撞硬顶(R3-F1 缺陷 4 的主入口;兜底取最大编号只是它的第二条入口)。 // 回退目标自己的 attempt 由 handleBackwardEdge 加,两边不重叠(forwardReachable 不含起点)。 bumpNodeAttempt(rn.id); reset.push(rn.id); } } return reset; } /** 从某节点出发、沿非 backward 边可达的下游集合(不含自己)。图小,直接 BFS */ function forwardReachable(def, fromDefNodeId) { const seen = new Set(); const queue = [fromDefNodeId]; while (queue.length) { const cur = queue.shift(); for (const e of def.edges) { if (e.from !== cur || e.edgeKind === 'backward') continue; if (e.to === fromDefNodeId || seen.has(e.to)) continue; seen.add(e.to); queue.push(e.to); } } return seen; } function blockNode(runId, nodeRunId) { sse.transitionNode(runId, nodeRunId, 'fail', 'blocked', {}, { summary: 'fail → blocked:没有可走的 fail 边或已撞熔断硬顶,等人工介入', }); } // ──────────────────────────────────────────────────────────────── // run 收敛(§四.D.1 末两行) // ──────────────────────────────────────────────────────────────── /** * 收敛判定:所有节点到达终态且无 pending/ready 悬空 → completed;有 blocked → blocked。 * 收敛后若本 run 有 parent_node_run_id,回写父节点(三值透传见 deriveChildOutcome)。 * @returns {string} 判定后的 run_status */ function settleRun(runId) { const run = getRun(runId); if (!run || TERMINAL_RUN_STATUS.has(run.run_status)) return run ? run.run_status : null; const counts = {}; for (const r of getDb() .prepare(`SELECT status, COUNT(*) AS n FROM graph_run_nodes WHERE run_id = ? GROUP BY status`) .all(runId)) { counts[r.status] = r.n; } // 还有活跃节点(含停在 running 等子 run 的 subgraph 节点)→ 没收敛 if (counts.ready || counts.running) return run.run_status; let target; if (counts.blocked) { // 需要人工介入:保留 pending 节点原样,人工「重试」blocked 节点后链路还能继续 target = 'blocked'; } else { // 没有 blocked 也没有活跃节点 → 剩下的 pending 再也不会被点亮(点亮只发生在节点结算时), // 批量 skipped,避免「所有可达节点都完成了但 run 永远不 completed」 for (const r of getDb() .prepare(`SELECT id FROM graph_run_nodes WHERE run_id = ? AND status = 'pending'`) .all(runId)) { sse.transitionNode(runId, r.id, 'pending', 'skipped', {}, { summary: 'pending → skipped:本 run 已收敛,这条分支再也不会被点亮', }); } target = 'completed'; } if (target !== run.run_status && canTransitionRun(run.run_status, target)) { sse.transitionRun(runId, run.run_status, target, {}, { summary: `run ${runId}: ${run.run_status} → ${target}(收敛判定)`, counts, }); } const settled = (getRun(runId) || {}).run_status; if (TERMINAL_RUN_STATUS.has(settled) || settled === 'blocked') propagateToParent(runId); return settled; } /** * 子 run 收敛 → 回写父节点。**必须回写三个值**(§四.D.0 的 ⚠️,R-16 硬要求): * success/fail ← 子 run 的 run_status * fail_reason ← 子图归因节点的产出,父层 §四.D.2.2 靠它选边 * remediation_note ← 同上,随回退传给目标节点,成为它下一次规划的输入 * 少了后两个,org 层选边就无据可依,只能走兜底边。 */ function propagateToParent(childRunId) { const child = getRun(childRunId); if (!child || !child.parent_run_id || !child.parent_node_run_id) return; const parentNode = getRunNode(child.parent_node_run_id); if (!parentNode || parentNode.status !== 'running') return; // 已经结算过 → 幂等返回 const outcome = deriveChildOutcome(child); // remediationNote 不是 graph_run_nodes 的列,随 outcome 一起交给父层的选边/回退用 settleNode(child.parent_run_id, child.parent_node_run_id, outcome); } /** 从子 run 提取三值透传的内容 */ function deriveChildOutcome(childRun) { const status = childRun.run_status === 'completed' ? 'success' : 'fail'; const def = defIndex(childRun); const rows = getDb() .prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? ORDER BY updated_at DESC, id DESC`) .all(childRun.id); // 归因优先取 validator-reflection 节点的产出;没有这个节点时退到最近一个失败/阻塞节点 // // 假设(R-16待定案): 归因保持 org node 粒度封装 —— 父层只拿到「子图整体失败 + 根因」, // org 层回退只退到 org node,不跨层寻址子图里的具体节点。 const attribution = rows.find((r) => { const d = def.nodes.get(r.def_node_id); return d && d.name === ATTRIBUTION_NODE_NAME && r.fail_reason; }); const lastBad = rows.find((r) => (r.status === 'fail' || r.status === 'blocked') && r.fail_reason); const failReason = status === 'fail' ? (attribution || lastBad || {}).fail_reason || null : null; const records = listLoopRecords(childRun.id); const remediationNote = (records.find((r) => r.remediation_note) || {}).remediation_note || null; return { status, failReason, remediationNote, childRunId: childRun.id }; } // ──────────────────────────────────────────────────────────────── // §四.A.3 任务级总预算:只挂 root run,全树按 root_run_id 扣同一个池 // ──────────────────────────────────────────────────────────────── /** * budget_json 的形状(§四.A.3): * { total_agent_calls, total_tokens, wallclock_ms, spent: {agent_calls, tokens, started_at} } * 三个上限都可选,缺了就是这一维不限额;整个 budget_json 为空 = 完全不限额。 * 子 run 的 budget_json 恒为 null(createRun 保证),所以永远上溯到 root_run_id 那一行读写。 */ function readBudget(run) { // 必须重新读一次 root 行:调用方手里的 run 往往是 advanceOne 开头取的快照, // 拿它的 budget_json 做读-改-写会把同一次推进里前一笔扣减覆盖掉。 const root = getRun(run.root_run_id); if (!root || !root.budget_json) return null; try { return { rootRunId: root.id, budget: JSON.parse(root.budget_json) }; } catch { return null; // 预算字段坏了不该把执行卡死,按不限额处理并留给控制台暴露 } } /** * 余额检查。任何一层发起 agent 调用前都要过这一关(advanceOne 拾取节点之后立刻查)。 * @returns {{ok: boolean, exhausted: string[], rootRunId?: string, budget?: object}} */ function checkBudget(run) { const held = readBudget(run); if (!held) return { ok: true, exhausted: [], budget: null }; const b = held.budget; const spent = b.spent || {}; const exhausted = []; if (b.total_agent_calls != null && (spent.agent_calls || 0) >= b.total_agent_calls) { exhausted.push('total_agent_calls'); } if (b.total_tokens != null && (spent.tokens || 0) >= b.total_tokens) exhausted.push('total_tokens'); if ( b.wallclock_ms != null && spent.started_at && Date.now() - Date.parse(spent.started_at) >= b.wallclock_ms ) { exhausted.push('wallclock_ms'); } return { ok: exhausted.length === 0, exhausted, rootRunId: held.rootRunId, budget: b }; } /** * 扣减(「只增不减」,§四.A.3)。始终写 root run 那一行。 * ponytail: 读-改-写不是原子的,当前单进程串行驱动(runAuto 一次只推一个节点)下够用; * 升级路径是把自增下沉成一条 `UPDATE ... SET budget_json = json_set(...)` 的 SQL。 */ function spendBudget(run, { agentCalls = 0, tokens = 0 } = {}) { const held = readBudget(run); if (!held) return null; const b = held.budget; if (!b.spent) b.spent = {}; b.spent.agent_calls = (b.spent.agent_calls || 0) + agentCalls; b.spent.tokens = (b.spent.tokens || 0) + tokens; if (!b.spent.started_at) b.spent.started_at = now(); // 墙钟从第一次真花钱开始算 getDb() .prepare(`UPDATE graph_runs SET budget_json = ?, updated_at = ? WHERE id = ?`) .run(JSON.stringify(b), now(), held.rootRunId); // 预算是控制台顶栏那三个进度条的数据源,扣一次播一次(前端监听 budget.spent) sse.emitEvent(run.id, 'budget.spent', `预算扣减 +${agentCalls} 次调用 / +${tokens} tokens`, { root_run_id: held.rootRunId, delta: { agent_calls: agentCalls, tokens }, spent: b.spent, total_agent_calls: b.total_agent_calls ?? null, total_tokens: b.total_tokens ?? null, }, { ref: held.rootRunId }); return b; } /** * 预算耗尽 → **整棵 run 树挂起等人工**,不是失败(§四.A.3): * root run → blocked,树上所有非终态 run 一律置 pause_requested=1。 * 同样只做一次 listRunTree 索引查询。 */ function suspendRunTree(run) { const rootRunId = run.root_run_id; const tree = listRunTree(rootRunId); for (const r of tree) { if (TERMINAL_RUN_STATUS.has(r.run_status)) continue; setRunFlags(r.id, { pauseRequested: true }); } const root = getRun(rootRunId); if (root && canTransitionRun(root.run_status, 'blocked')) { sse.transitionRun(rootRunId, root.run_status, 'blocked', {}, { summary: '预算耗尽:整棵 run 树挂起等人工(不是失败)', reason: 'budget-exhausted', }); } return (getRun(run.id) || {}).run_status; } // ──────────────────────────────────────────────────────────────── // §四.C 五个控制操作(手动 / 自动共用上面的 advanceOne) // ──────────────────────────────────────────────────────────────── /** planning → ready(图规划完成 + 通过写入前五项校验之后调,校验在 work-graph-db) */ function markRunReady(runId) { const run = getRun(runId); if (!run) throw new Error(`graph_runs 不存在: ${runId}`); if (run.run_status === 'ready') return { ok: true, runStatus: 'ready' }; sse.transitionRun(runId, 'planning', 'ready', {}, { summary: '图规划完成并通过写入前校验:planning → ready' }); return { ok: true, runStatus: 'ready' }; } /** * 入口节点 = 没有任何 forward 入边的节点(不分 success / fail),run 起跑时点亮。 * * **fail 入边同样算入边**:只有 fail 入边的节点(人工审核 / escalate 那一类)是失败分支的 * 落点,它要等失败真的发生、routeFailure 选中那条边时才被 markReady 点亮。按 condition 过滤 * 会让它在 run 一起跑就被点亮执行 —— 在任何失败发生之前先真调一次 LLM 并计入预算 * (spendBudget 在 advanceOne 抢到节点时就扣),随后又被 resetForwardDownstream 打回 pending。 * * backward 入边仍要排除:回退目标(如 coder)除了回退边可能没有别的入边, * 不排除它就永远起不来。 */ function seedEntryNodes(runId) { const run = getRun(runId); const def = defIndex(run); const hasIncoming = new Set(def.edges.filter((e) => e.edgeKind !== 'backward').map((e) => e.to)); const seeded = []; for (const [defNodeId] of def.nodes) { if (hasIncoming.has(defNodeId)) continue; const activated = markReady(runId, def, defNodeId); if (activated) seeded.push(activated); } return seeded; } /** * 「启动」:run_status ∈ {ready, paused, blocked} → running。 * 从 ready 起跑时顺带点亮入口节点;恢复(paused/blocked)时不动已有节点状态。 * 同步部分单独抽出来,供 executeSubgraph 起子 run 时复用。 */ function startRunSync(runId, { mode } = {}) { const run = getRun(runId); if (!run) throw new Error(`graph_runs 不存在: ${runId}`); const fresh = run.run_status === 'ready'; if (run.run_status !== 'running') { sse.transitionRun(runId, run.run_status, 'running', {}, { summary: `启动:${run.run_status} → running${mode ? `(${mode} 模式)` : ''}`, execution_mode: mode || run.execution_mode, }); } // 「启动」清掉软信号,否则 advanceOne 一进来就被上一次的暂停挡住 setRunFlags(runId, { pauseRequested: false, stopRequested: false }); if (mode) { getDb().prepare(`UPDATE graph_runs SET execution_mode = ?, updated_at = ? WHERE id = ?`) .run(mode, now(), runId); } const seeded = fresh ? seedEntryNodes(runId) : []; return { ok: true, runId, runStatus: 'running', seeded }; } /** * 「启动」(对外)。auto 模式立刻挂一个自我调度的循环上去;manual 模式只标 running, * 停在原地等「下一步」。返回里的 driver 是 auto 循环的 promise(路由可以不管它,测试可以 await)。 */ function startRun(runId, opts = {}) { const res = startRunSync(runId, opts); const mode = opts.mode || (getRun(runId) || {}).execution_mode; if (mode !== 'auto') return res; // 自动模式不是另一套推进逻辑,就是一个不停调 advanceOne 的循环(§四.C 末段) const driver = runAuto(runId, opts).catch((err) => { console.error(`[graph-engine] auto 驱动异常 run=${runId}: ${err.message}`); return { ok: false, error: err.message }; }); return { ...res, driver }; } /** * 「下一步」:拾取一个 ready 节点执行完就停,不继续往下跑。 * 已 paused 时也允许单步(§四.C 表格「可用条件」那一列)——单步临时把 run 拉回 running, * 步完只要 pause_requested 还在就自动落回 paused。 * **「下一步」只作用于当前层级,不级联**(§四.C 定案:暂停是全局意图,单步是局部操作)。 */ function nextStep(runId, opts = {}) { const run = getRun(runId); if (!run) throw new Error(`graph_runs 不存在: ${runId}`); if (run.run_status === 'paused') { sse.transitionRun(runId, 'paused', 'running', {}, { summary: '「下一步」:paused → running(单步)' }); } const cur = getRun(runId); if (cur.run_status !== 'running') { return Promise.resolve(result(runId, cur.run_status, 'run-not-running')); } return advanceOne(runId, { ...opts, ignorePause: true }).then((res) => { landPause(runId); return { ...res, runStatus: (getRun(runId) || {}).run_status }; }); } /** * 「暂停」:软信号,不打断正在 running 的节点,下一个节点边界生效。 * **沿 parent_run_id 向下级联到所有 active 子 run**(§四.C 三层级联定案)—— * 层级是三层(root → org node → 嵌套子图),不级联就会出现「点了暂停但子图还在偷偷跑」。 */ function pauseRun(runId) { const run = getRun(runId); if (!run) throw new Error(`graph_runs 不存在: ${runId}`); const targets = [run, ...descendantRuns(run)].filter((r) => !TERMINAL_RUN_STATUS.has(r.run_status)); for (const r of targets) { setRunFlags(r.id, { pauseRequested: true }); landPause(r.id); // 该 run 手头没有 running 节点就立刻落 paused,有就等它到节点边界 } return { ok: true, runId, pauseRequested: 1, runStatus: (getRun(runId) || {}).run_status, cascaded: targets.map((r) => r.id).filter((id) => id !== runId), }; } /** * running → paused 的落库时机 = **节点边界**(§四.C:暂停不打断当前节点)。 * 手头还有 running 节点就先不落,等 advanceOne 下一次被 pause_requested 挡住时再落。 */ function landPause(runId) { const run = getRun(runId); if (!run || !run.pause_requested || run.run_status !== 'running') return null; const busy = getDb() .prepare(`SELECT 1 FROM graph_run_nodes WHERE run_id = ? AND status = 'running' LIMIT 1`) .get(runId); if (busy) return null; sse.transitionRun(runId, 'running', 'paused', {}, { summary: '到达节点边界,暂停落库:running → paused' }); return 'paused'; } /** * 「停止」:人工主动终止,**同样向下级联**。每一层:置 stop_requested → 仍 pending/ready * 的节点批量 skipped → run → stopped。 * * 正在 running 的节点:§3.6 的进程级硬取消调用链由 T9 核实,当前是文档给的降级路径—— * 「等它跑完、记录结果、但不再调度后续节点」(stop_requested 让 advanceOne 不再拾取新节点)。 */ function stopRun(runId) { const run = getRun(runId); if (!run) throw new Error(`graph_runs 不存在: ${runId}`); const targets = [run, ...descendantRuns(run)]; for (const r of targets) { setRunFlags(r.id, { stopRequested: true }); for (const n of getDb() .prepare(`SELECT id, status FROM graph_run_nodes WHERE run_id = ? AND status IN ('pending','ready')`) .all(r.id)) { sse.transitionNode(r.id, n.id, n.status, 'skipped', {}, { summary: '人工「停止」:未开跑的节点批量 skipped' }); } const fresh = getRun(r.id); if (fresh && canTransitionRun(fresh.run_status, 'stopped')) { sse.transitionRun(r.id, fresh.run_status, 'stopped', {}, { summary: `人工「停止」:${fresh.run_status} → stopped`, cascaded_from: runId, }); } } return { ok: true, runId, runStatus: (getRun(runId) || {}).run_status, cascaded: targets.map((r) => r.id).filter((id) => id !== runId), }; } /** * 一个 run 的全部后代(不含自己)。 * `root_run_id` 冗余列的用处就在这里:**一次索引查询**把整棵树拉出来,再在内存里沿 * parent_run_id 走一遍取子树,不做递归 SQL(§四.C)。 */ function descendantRuns(run) { const all = listRunTree(run.root_run_id); const byParent = new Map(); for (const r of all) { if (!byParent.has(r.parent_run_id)) byParent.set(r.parent_run_id, []); byParent.get(r.parent_run_id).push(r); } const out = []; const stack = [run.id]; while (stack.length) { for (const child of byParent.get(stack.pop()) || []) { out.push(child); stack.push(child.id); } } return out; } /** * 「重试」:节点 status ∈ {fail, blocked} → ready,attempt + 1。 * * **默认不级联重置下游**(§四.C 末段):级联与否不由重试这个动作决定,而由**产出契约是否 * 真的变了**决定。判定发生在重跑成功之后(applyRetryCascade),这里只把当前契约拍个快照。 */ function retryNode(runId, nodeRunId) { const nodeRun = getRunNode(nodeRunId); if (!nodeRun || nodeRun.run_id !== runId) throw new Error(`节点不属于该 run: ${nodeRunId}`); // 人工重试同样是「新的一轮」:subgraph 节点清 child_run_id,重跑时起新子 run // (§四.D.3;不清的话 executeSubgraph 会拿旧的终态子 run 返回 pending → 永久卡 running) if ( !sse.transitionNode(runId, nodeRunId, nodeRun.status, 'ready', { childRunId: null }, { attempt: nodeRun.attempt + 1, summary: `人工「重试」:${nodeRun.status} → ready(attempt #${nodeRun.attempt + 2})`, }) ) { return { ok: false, reason: 'taken-by-other', nodeStatus: (getRunNode(nodeRunId) || {}).status }; } bumpNodeAttempt(nodeRunId); const run = getRun(runId); const snapshot = snapshotContract(run, nodeRunId); if (run.run_status === 'blocked') { sse.transitionRun(runId, 'blocked', 'running', {}, { summary: '人工重试解除阻塞:blocked → running' }); } return { ok: true, runId, nodeRunId, nodeStatus: 'ready', contractSnapshot: snapshot, runStatus: (getRun(runId) || {}).run_status, }; } // ──────────────────────────────────────────────────────────────── // §四.C 末段:重试后级联下游的判据 = contract_hash 变没变 // ──────────────────────────────────────────────────────────────── /** 节点私有产物目录(§四.A.5):data/companies/{cid}/runs/{rootRunId}/nodes/{nodeRunId}/ */ function nodeArtifactDir(run, nodeRunId) { return path.join(REPO_ROOT, 'data', 'companies', run.company_id, 'runs', run.root_run_id, 'nodes', nodeRunId); } function readJsonOrNull(file) { try { return JSON.parse(fs.readFileSync(file, 'utf8')); } catch { return null; } } /** 重试前把节点当前的 contract.json 拍成 contract.prev.json;没有契约就不拍(=不会级联) */ function snapshotContract(run, nodeRunId) { const dir = nodeArtifactDir(run, nodeRunId); const cur = path.join(dir, 'contract.json'); if (!fs.existsSync(cur)) return null; const prev = path.join(dir, 'contract.prev.json'); fs.copyFileSync(cur, prev); return prev; } /** * 重跑成功后判级联:有快照(说明这个节点被人工重试过)才判,没快照什么都不做。 * 四分类(added / changed / removed / unchanged,同 loops-l3.md 的增量算法): * **unchanged 保持原状不重跑**;只要出现任一 added/changed/removed 就把 forward 下游打回 * pending,等本节点的 success 出边重新点亮它们。 * * ponytail: 引擎不知道「哪个 export 被哪个下游节点消费」(那是图规划者的知识),所以四分类 * 算出来之后退化成一个整体的 dirty 位。升级路径是让 workgraph-planning 在边上声明消费的 * export key,届时按 key 精确挑下游即可,diffContracts 的输出形状不用改。 */ function applyRetryCascade(runId, def, nodeRun) { const run = getRun(runId); const dir = nodeArtifactDir(run, nodeRun.id); const prevFile = path.join(dir, 'contract.prev.json'); if (!fs.existsSync(prevFile)) return null; const prev = readJsonOrNull(prevFile); const next = readJsonOrNull(path.join(dir, 'contract.json')); fs.unlinkSync(prevFile); // 快照一次性,判完即弃 const diff = diffContracts(prev, next); diff.reset = diff.dirty ? resetForwardDownstream(runId, def, nodeRun.def_node_id) : []; return diff; } /** * 按 contract_hash 把两份契约的 exports 分成四类。 * 单个 export 的 hash 复用 devloop/doc-store.js:382 的 hashContract() * (对 exports 做递归 key 排序的规范化序列化后 sha256 取前 16 位),不另造一套。 */ function diffContracts(prev, next) { const a = exportsOf(prev); const b = exportsOf(next); const res = { added: [], changed: [], removed: [], unchanged: [] }; for (const key of new Set([...Object.keys(a), ...Object.keys(b)])) { if (!(key in a)) res.added.push(key); else if (!(key in b)) res.removed.push(key); else if (hashExport(a[key]) === hashExport(b[key])) res.unchanged.push(key); else res.changed.push(key); } res.dirty = res.added.length + res.changed.length + res.removed.length > 0; return res; } function exportsOf(contract) { if (!contract || typeof contract !== 'object') return {}; const e = contract.exports || contract; return e && typeof e === 'object' && !Array.isArray(e) ? e : { _: e }; } function hashExport(value) { // 懒 require:doc-store 会拉起 path-resolver 等一串 DevLoop 依赖,不该在本模块加载时就付这个钱 const { hashContract } = require('./devloop/doc-store'); return hashContract({ exports: { value } }); } // ──────────────────────────────────────────────────────────────── // 自动模式 = 不停调 advanceOne 的循环(不是另一套实现) // ──────────────────────────────────────────────────────────────── /** * 在一棵 run 树里推进一步:优先推最深的 active 子 run(父节点正卡在它上面)。 * 单张图时退化成 advanceOne(rootRunId)。 */ async function advanceTree(rootRunId, opts = {}) { const tree = listRunTree(rootRunId).filter((r) => r.run_status === 'running'); // 后建的 run 一定更深(子 run 总是在父节点 running 之后建的),倒序即「最深优先」 for (const r of tree.reverse()) { const res = await advanceOne(r.id, opts); if (res.ok) return res; } return result(rootRunId, (getRun(rootRunId) || {}).run_status, 'no-ready-node'); } /** * 自动驱动:只要 pause_requested / stop_requested 都是假就在上一次 resolve 后立刻发起下一次。 * maxSteps 只是防跑飞的安全阀,真正的终止条件是预算(§四.A.3)与熔断(§四.D.2.1)。 * ponytail: 串行单步循环,一次只跑一个节点,并行分支不会并发执行;上限是吞吐, * 升级路径是按 ready 集合并发 advanceOne(节点状态机已经是幂等条件更新,能直接支撑)。 */ async function runAuto(rootRunId, opts = {}) { const maxSteps = opts.maxSteps || 1000; let steps = 0; for (; steps < maxSteps; steps++) { const run = getRun(rootRunId); if (!run || run.pause_requested || run.stop_requested) break; if (TERMINAL_RUN_STATUS.has(run.run_status) || run.run_status === 'blocked') break; const res = await advanceTree(rootRunId, opts); if (!res.ok) break; } return { ok: true, runId: rootRunId, steps, runStatus: (getRun(rootRunId) || {}).run_status }; } // ──────────────────────────────────────────────────────────────── // 内部小工具 // ──────────────────────────────────────────────────────────────── /** 定义层索引:nodes 用 Map(defNodeId → node),edges 原样(图规模小,不做二次索引) */ function defIndex(run) { const def = readDefGraph(run.graph_kind, run.graph_id); return { nodes: new Map(def.nodes.map((n) => [n.defNodeId, n])), edges: def.edges }; } function result(runId, runStatus, reason) { return { ok: false, runId, runStatus: runStatus || null, reason, node: null, activated: [], backwardEdgeId: null, loop: null, contractDiff: null, childRunId: null, }; } function nodeView(runId, r) { if (!r) return null; const run = getRun(runId); const d = run ? defIndex(run).nodes.get(r.def_node_id) : null; return { id: r.id, defNodeId: r.def_node_id, name: d ? d.name : null, kind: d ? d.nodeKind : null, status: r.status, attempt: r.attempt, failReason: r.fail_reason, outputRef: r.output_ref, agentRunId: r.agent_run_id, childRunId: r.child_run_id, }; } /** * 拼进提示词的产物目录约定(§四.A.5 + validator-reflection/agent.md「产出格式」)。 * * 路径与 readNodeProduct 读的**是同一个**(nodeArtifactDir),文件名给绝对路径, * 不留 0 基 / 1 基的解释空间——模板里 `attempt-{n}.json` 的 n 没定死基数, * 这里直接把本轮该写的文件名算好告诉它。 */ function describeArtifactContract(ctx) { const dir = safeNodeArtifactDir(ctx); if (!dir) return ''; // run 缺 company_id/root_run_id:拼不出目录就不编一个假的塞给 agent return [ '', '', '[节点产物目录]', `你的私有目录(可读可写):${dir}`, `本轮的结构化产出/归因写到:${path.join(dir, `attempt-${ctx.attempt}.json`)}`, '格式见 system/templates/validator-reflection/agent.md「产出格式」' + '(schema=validator-reflection/v1;引擎只读 verdict / fail_reason / remediation_note 三个字段,' + 'verdict ∈ attributed|inconclusive|no_prior_failure,fail_reason ∈ ' + `${[...ATTRIBUTION_REASONS].join('|')} 或 null)。`, `任务级共享产物区(只读):${path.join(REPO_ROOT, 'data', 'companies', ctx.companyId || '', 'runs', ctx.rootRunId || '', 'artifacts')}`, '不要读写其它节点的私有目录(§四.A.5 的隔离规则)。', '判词类节点(validator / reflection):**没有这份文件 = 无法判定**,引擎会按失败结算走兜底边。', ].join('\n'); } /** nodeArtifactDir 的不抛版本(run 缺列时返回 null) */ function safeNodeArtifactDir(ctx) { try { return ctx && ctx.run ? nodeArtifactDir(ctx.run, ctx.nodeRunId) : null; } catch { return null; } } /** 没有显式任务描述时给 executor 的兜底文案(goal/task/okr 三者基数不同,见 27-NAMING §四) */ function describeTask(run) { const parts = [run.goal_id && `goal=${run.goal_id}`, run.task_id && `task=${run.task_id}`, run.okr_node_id && `okr=${run.okr_node_id}`].filter(Boolean); return parts.length ? `执行图节点任务(${parts.join(', ')})` : '执行图节点任务'; } /** * 「应该如何修复」,作为下一轮的上下文补充(§四.D.2.1 第 3 步)。 * * **必须按 to_node_run_id 过滤**:§四.D.2.1 第 3 步说的是把判词带给**回退目标**。 * 同一个 run 里多条回退边并存时(验收 9b 那种图),不过滤会把给 a 的修复建议原样塞进 * b/c 的提示词 —— 判词串台比没有判词更糟,下一轮那个 agent 会照着别人的活干。 */ function latestRemediationNote(runId, toNodeRunId) { const rec = listLoopRecords(runId) .find((r) => r.remediation_note && (!toNodeRunId || r.to_node_run_id === toNodeRunId)); return rec ? rec.remediation_note : null; } module.exports = { // 核心原语 advanceOne, advanceTree, runAuto, settleNode, settleRun, // 五个控制操作 startRun, nextStep, pauseRun, stopRun, retryNode, descendantRuns, landPause, // 任务级总预算(§四.A.3) checkBudget, spendBudget, suspendRunTree, // 重试级联判据(§四.C 末段) diffContracts, applyRetryCascade, snapshotContract, // run 生命周期辅助 markRunReady, seedEntryNodes, // executor 与注入点 executeSubgraph, defaultAgentExecutor, setAgentExecutor, setPlanWorkGraph, // 选边 / 熔断 / 透传 selectFailEdge, handleBackwardEdge, resetForwardDownstream, deriveChildOutcome, ATTRIBUTION_NODE_NAME, INFRA_FLAKE_REASON, LOOP_HARD_CAP_EDGE, LOOP_HARD_CAP_RUN, };