/** * 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, createWorkGraph, getWorkGraph, getRun, getRunNode, getRunNodeByDefId, // 注意:updateNodeStatus / updateRunStatus **故意不在这里 import**——状态转换一律走 // 下面的 sse.transitionNode / transitionRun,否则又会退回「改了状态但没有事件」的老路。 bumpNodeAttempt, bumpInfraFlakeCount, setRunFlags, // 批 3 复核 F1:准入被拒之后的两条出路 —— 退回队列(可自愈)或留痕(等人) RunSlotBusyError, requeueBlockedRootRun, noteTaskBlockedReason, createRun, listRunTree, listLoopRecords, insertLoopRecord, updateLoopVerdict, bumpBackwardCount, canTransitionRun, canTransitionNode, // R-19 产物索引(27-P4 §4.6):inboundContracts/findFiles 给输入侧解析, // recordFile 给产出侧记实例 —— 都由**引擎**调,不给 agent 自己写 inboundContracts, edgeContracts, findFiles, recordFile, // 27-P4 T4/T5:advanceBatch 批量拾取宽度的缺省值 DEFAULT_MAX_CONCURRENCY, } = 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 locator = require('./agent-locator'); const memoryService = require('./memory-service'); /** * 写入边界段头的**唯一定义**在 prompt-builder(28 §4.4):产出这段的是这里,据此决定 * 「坐标声明要不要再给一个交付物归宿」的是那里,两侧必须是同一个字符串常量。 * 详见 describeWriteBoundary 的注释。 */ const { GRAPH_WRITE_BOUNDARY_HEADER } = require('../src/core/prompt-builder'); const REPO_ROOT = path.resolve(__dirname, '../..'); /** * run 目录的根覆盖(R4-F1 minor 5)。**必须与 graph-run-sse.runsRootOverride 是同一个变量、 * 同一套拼法**:节点产物(nodes/{nodeRunId}/attempt-{n}.json、contract.json)和 run 的 * stream.log / timeline.ndjson 是同一个 run 目录下的两类文件,分家的后果是测试里 * 「事件写到临时目录、产物写进仓库 data/companies/」——现在的用例碰不到只是因为它们全都 * setAgentExecutor 注入了假 executor,任何一个跑到默认执行器产物路径的用例都会往仓库里写。 * * **每次调用时读,不在模块加载时快照**(27-P3C 缺陷 F8):快照的形态要求所有调用方 * 「先设 env 再 require」,而本模块被 routes/company.js、脚本、REPL 各自 require, * 谁先谁后没有任何保证。加载顺序一错,写侧落 `$GRAPH_RUNS_DIR/...`、读侧去 * `data/companies/...` 找,`describeTask` 拿到 null 后**静默退回兜底文案「执行图节点任务」**。 * 与 graph-run-sse.runsRootOverride 必须保持同一个变量、同一套拼法。 */ function runsRootOverride() { return process.env.GRAPH_RUNS_DIR || null; } /** * 子图归因节点的名字。子 run 失败时优先取这个节点的 fail_reason 透传回父层(§四.D.2.2)。 * * R-17.2 定案:`validator-reflection` → **`validator-triager`**(它做的是归因不是判定, * 与 DevLoop 已跑通的 BugTriager 对齐)。 * * **两个名字都认,不做硬切换**:改名铺在四个不同步的地方——模板目录、`_registry.json`、 * work graph 骨架 `validator.json`、以及**已经落库的 work_graph_nodes 行**。最后一处是 * 存量数据,永远改不回来。引擎只认新名字的话,改名窗口期(以及所有历史 run)里 * deriveChildOutcome 会匹配不到归因节点,静默退到「最后一个失败节点」那条兜底—— * 表现是「org 层选边突然不准了」,且没有任何报错。这正是本轮要消灭的那类静默漏接, * 所以别名集合是长期约定,不是过渡期补丁。 */ const ATTRIBUTION_NODE_NAME = 'validator-triager'; const ATTRIBUTION_NODE_ALIASES = new Set([ATTRIBUTION_NODE_NAME, 'validator-reflection']); const isAttributionNodeName = (name) => ATTRIBUTION_NODE_ALIASES.has(String(name || '')); /** 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 记账就会被架空:R-17.1 之后 subgraph 重跑虽然优先**复用**旧子 run(计数能累加), * 但复用不了时(人工停止过 / 需要重新规划)照样新建一张子 run,新行 backward_count 从 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'; /** * 30 号文档 §四:`infra_flake` 原地重跑故意不烧 attempt/backward_count("这不是节点自己的 * 错,不该烧它的重试额度"),代价是它是唯一没有熔断线的失败路径——真撞见过一次确定性 * 死循环(归因节点把"该重跑别的节点"的判词误判成自己的 infra_flake,每轮产出雷同)。 * 这条硬顶只管一件事:同一个节点连续判 infra_flake 到第几次就不再当"真的是基础设施抖动", * 改判失败走正常选边(大概率落进兜底边或直接 blocked,交给人工)。数值跟 * `LOOP_HARD_CAP_EDGE`(backward 边单边硬顶)取同一量级,没有更细的依据。 */ const INFRA_FLAKE_HARD_CAP = 3; /** * validator-triager 的归因取值域(system/agents/validator-triager/agent.md * 「归因取值域(四选一,不要发明第五个)」= §四.D.2.2 那张表)。 * * 引擎**不理解**这些取值的语义(选边只做字符串匹配),这个集合只用于一件事: * 把归因文件里自创的取值当作 null 处理(模板明说自创取值会静默落到兜底边, * 那就在入口处显式降级 + 记一条 timeline,而不是让一个假取值混进选边)。 */ const ATTRIBUTION_REASONS = new Set(['impl_bug', 'input_underspecified', 'skill_defect', INFRA_FLAKE_REASON]); /** * ── 引擎自判的 fail_reason(W12 补完)────────────────────────────── * * W12 第一轮只把失败**原因文本**送进了 timeline / `node.output.output_excerpt`(运行日志 tab), * `graph_run_nodes.fail_reason` 与 `node.output` 的 `fail_reason` 仍然恒 `null`。 * 后果是界面走查看到的那一幕:点开失败节点,**概览 tab 第一眼是「失败、原因 —」**, * 得知道去翻第三个 tab 才看得到「Agent not found: prd-planning」。 * 「引擎判定的失败」与「agent 归因产出的失败」在这一列上被区别对待,是没有道理的。 * * **为什么另起一个集合,而不是往 ATTRIBUTION_REASONS 里加**(这是本轮的取值论证): * * 1. `ATTRIBUTION_REASONS` 是 `readNodeProduct` 的**入口白名单**:它管的是 * 「agent 在 attempt-{n}.json 里**允许写**哪几个词」(模板:四选一,不要发明第五个)。 * 往里面加 `node_error`,等于告诉 agent「你也可以自称执行器炸了」—— * 放宽的是**别人**的取值域,而本轮要落的是**引擎自己**的判定。两件事不该共用一个白名单。 * 2. 反过来复用现成的四个取值更糟,逐个说: * - `infra_flake` **有引擎特判**(routeFailure:原地重跑、不烧任何计数)。 * `Agent not found` 是确定性失败,判成 infra_flake 会**原地重跑到预算耗尽**。 * - `impl_bug` / `input_underspecified` 是图上**精确 fail 边**的匹配键, * 典型图里它们指向「退回 impl 节点重写」「退回 planner 补输入」。 * 把「本体没雇进公司」判成 `impl_bug`,会让引擎**真的**退回去重写一遍代码, * 而根因一行代码都不在那儿——这是把配置故障伪装成业务失败,比 `null` 更坏。 * - `skill_defect` 语义是「技能包本身有缺陷」,同样是业务归因。 * 共同点:复用 = 借了别人的**路由语义**,误路由的代价按轮计费。 * 3. 新取值的**路由行为与今天的 `null` 完全一致**:`selectFailEdge` 先精确匹配、 * 匹配不到退兜底边;现有图骨架里没有任何边写这三个词,所以一律落兜底边—— * 与本轮之前 `fail_reason=null` 走的是同一条边。**只加信息,不改路由**。 * 将来某张图想精确路由「越界写入」,直接写 `failReason: "boundary_escape"` 就能用。 * * 三个取值与产生点(每个都有真实调用方,见各自的注释): * NODE_ERROR — advanceOne 的 executor catch(`Agent not found` 这类) * BOUNDARY_ESCAPE — applyBoundaryGuard 判越界(与 timeline 里 violation 的字面量同名,故意的) * CHILD_RUN_FAILED — deriveChildOutcome:子 run 收敛为 fail,但子图一个归因都没给出 * * ⚠️ 加取值时要同步的防线:`server/tests/unit/template-verdicts.test.js` 按源码正则取 * `ATTRIBUTION_REASONS` 校验模板与图骨架的取值。本集合**独立于**它,所以加在这里不会 * 让那道防线变红;但反过来——一旦你想在 `system/{work,org}-graph-templates/*.json` 里 * 写一条 `failReason: "boundary_escape"` 的精确边,那道防线的第 4 例会拦下来 * (它只认 `ATTRIBUTION_REASONS`),届时要把它的 `REASON_DOMAIN` 改成两个集合的并集。 */ const ENGINE_FAIL_REASONS = Object.freeze({ NODE_ERROR: 'node_error', BOUNDARY_ESCAPE: 'boundary_escape', CHILD_RUN_FAILED: 'child_run_failed', // T-mt25ecbd:produces/needs 契约以前只进提示词(软提示),agent 没写文件 / 没读文件都 // 不影响节点判 success、不影响下游点亮——三份计划文档没落盘、goal-benchmark 照样过。 // 这两个新取值让契约变成真闸门:declared 的 kind 在 graph_files 里真的查不到就判 fail, // 不发明取值域之外的字符串(否则会被 §四.D.2.2 的合法性检查当自创取值降级成 null)。 MISSING_INPUT: 'missing_required_input', PRODUCES_MISSING: 'produces_missing', // 35 号文档提案 A:本体完整性闸门只在招聘路径上查过(company-service.js 的 // 'body_not_ready' 状态),图执行侧从未查过——本体因任何原因仍是空骨架时, // 会静默拿一份没有角色指导的提示词去跑真实 CLI(真实样本见 34 号文档 §五)。 // 字符串与招聘路径的 'body_not_ready' 同名同义,不是两套判词。 BODY_NOT_READY: 'body_not_ready', }); /** 归因/结构化产出文件名(agent.md「产出格式」:runs/{rootRunId}/nodes/{nodeRunId}/attempt-{n}.json) */ const ATTEMPT_FILE_RE = /^attempt-(\d+)\.json$/; /** * ── 跨层节点地址(R-17.1)───────────────────────────────────────── * * 归因产出的是「**哪个 agent 的哪个环节**」,落到图上就是一对 (org node, 那张 work graph 里的 node)。 * R-17.1 定案「归因到哪个 work graph node 就退到哪个 node」,所以这一对必须能存、能传、能读回。 * * **形态定为 JSON 串 `{"orgNode":"org-coder","workNode":"impl-a"}`,存在 * `graph_run_nodes.fail_node_ref` 这一列上**。三个备选与否决理由: * * ① 在 org 边上加一列(`org_graph_edges.target_work_node`)——**否决**。 * 边是**定义层**,而被点名的 work graph node 是 `workgraph-planning` 在**运行时**才规划出来的, * 每次规划都是新的 `wgn_*` id。定义层根本写不出这个值;写节点名的话,一条边只能钉死一个 * 固定目标,而归因是每一轮都可能不同的运行时结论。 * ② 让 `fail_reason` 携带(`impl_bug@impl-a`)——**否决**。`fail_reason` 是 §四.D.2.2 的 * 选边匹配键,掺进地址后所有精确 fail 边一律失配,全部落兜底边;而且它有封闭取值域 * (ATTRIBUTION_REASONS),复合串会被合法性检查当成自创取值直接降级成 null。 * 这就是「归因产出被静默丢掉」的第 13 个变体。 * ③ 放 loop record 上——**只放一半**。loop record 只在命中 backward 边时才有行, * 而地址在**归因发生时**(子 run 内部)就产生了,比回退早好几步;借它当唯一载体等于 * 重演 R4-F1 那个「判词只在 backward 路径落库」的缺陷。所以: * 归因的**主张**存节点行(`fail_node_ref`,与 `fail_reason` 同一次条件更新落库), * 回退的**动作**存账本(`target_node_ref`,记解析后的 childRunId/nodeRunId,供控制台展示粒度)。 * * `orgNode` 可缺省(归因只说得出「哪个环节」时):缺省时以选中的那条 backward 边的目标为准。 * 两者都给且对不上时**不做定点重置**(边才是图上真实存在的路由),退化成整张图重跑并记一条 timeline。 */ function normalizeNodeRef(value) { let v = value; if (typeof v === 'string') { const s = v.trim(); if (!s) return null; v = safeJsonParse(s); } if (!v || typeof v !== 'object' || Array.isArray(v)) return null; const workNode = typeof v.workNode === 'string' ? v.workNode.trim() : ''; if (!workNode) return null; // 没点到具体 node 的地址等于没有地址 const orgNode = typeof v.orgNode === 'string' && v.orgNode.trim() ? v.orgNode.trim() : null; return JSON.stringify({ orgNode, workNode }); } /** JSON 串 → {orgNode, workNode};坏值一律 null(读不出地址 = 没有地址,不抛) */ function parseNodeRef(json) { const norm = normalizeNodeRef(json); return norm ? JSON.parse(norm) : null; } /** 判「这次失败了」的 verdict(agent.md 产出格式那三个取值里的两个);no_prior_failure 是放行 */ /** 任务级共享产物的 kind(不受拓扑相邻性限制,任何下游都可能要读,R-19.3 定案①) */ const TASK_SHARED_KINDS = ['acceptance', 'contract']; /** 按文件名兜底认 kind 的白名单(出边没声明契约时用;认不出来一律 'artifact') */ const KNOWN_FILE_KINDS = ['brd', 'prd', 'sdd', 'acceptance', 'contract']; const FAILING_VERDICTS = new Set(['attributed', 'inconclusive']); /** * 「必须留下判词」的节点(27-NAMING §3.1/§3.3/§3.5 的 validator 命名轴): * 归因节点 validator-triager(含改名前的 validator-reflection、以及 org-validator-triager * 这种带父节点全名的写法)与一切 `{...}-validator` 节点。这些节点的**判定结果只存在于 * 产出文件里**(verdict / fail_reason),调用本身正常退出说明不了任何事。文件缺失时按 * 「无法判定」结算,不能静默判 success,否则 fail 边 / 回退 / infra_flake 原地重跑在生产里 * 永远触发不了(R3-F1 缺陷 3)。 */ const ATTRIBUTION_REQUIRED_RE = /(^|[-_])(triager|reflection|validator)$/i; function requiresAttributionProduct(ctx) { const name = String((ctx && ctx.nodeName) || ''); return isAttributionNodeName(name) || ATTRIBUTION_REQUIRED_RE.test(name); } const now = () => new Date().toISOString(); // ──────────────────────────────────────────────────────────────── // 注入点:两个外部依赖都可替换(测试注入假实现,避免真起 Claude CLI) // ──────────────────────────────────────────────────────────────── /** * workgraph-planning 注入点(§四.D.0)。 * * 默认实现是**真的调 workgraph-planning 规划子图**(defaultPlanWorkGraph,见下)。 * 这个注入点保留只为一件事:测试注入假图,避免真起 Claude CLI。 * * R5 之前这里是一句 throw。而 org graph 的节点**全部**是 subgraph(work-graph-db * readDefGraph 里写死 nodeKind:'subgraph'),所以生产环境跑 org run 的第一个节点就撞上它, * 被 advanceOne 的 catch 兜成一个没有 fail_reason 的 fail 走兜底边——「一句自然语言 → * 一张图 → 跑完」这条线在生产里从来没通过。 * * @type {(ctx: NodeContext) => Promise<{workGraphId: string, tokens?: number}>} */ let planWorkGraph = defaultPlanWorkGraph; /** 替换 workgraph-planning 实现(传 null 回默认实现;测试注入假图) */ function setPlanWorkGraph(fn) { planWorkGraph = fn || defaultPlanWorkGraph; } /** * 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] * @property {string|null} [remediationNote] 「该怎么修」,由 settleNode 挂到下游节点 * @property {number} [tokens] 本次调用的 token 用量,进 §四.A.3 的 total_tokens 维度 */ /** * agent_call:调 AgentManager 跑 system_agent_id。 * * 成功/失败只看调用本身是否抛异常——**引擎不发明 fail_reason**。 * 按 §四.D.2.2 的定案,归因是一次显式的 agent 调用(validator-triager 干的活), * 不是失败节点顺手吐一个值;执行器只在结果里带上 agent 自己给的 failReason(若有)。 * 没有 failReason 时选边会退到兜底边,不会莫名其妙死路(§三.7 第五项校验保证兜底边存在)。 */ async function defaultAgentExecutor(ctx) { const AgentManager = require('../src/core/agent-manager'); // 空 config 即可:getAgentConfig 有基于文件系统的动态回退(本体恒在 system/agents/{aid}), // 与 agent-chat.js 的用法一致。 const manager = new AgentManager({}, REPO_ROOT); const task = ctx.remediationNote ? `${ctx.task}\n\n[上一轮回退的修复建议]\n${ctx.remediationNote}` : ctx.task; // total_tokens 这一维的**产生端**(R4-F1 major 2)。checkBudget/spendBudget 两头早就通了, // 缺的一直是「谁来数」——callAgent 的返回值只有 {agent, output, status},没有 usage。 // 用量走的是事件轨:ClaudeCliProtocol 在 type=result 时 emit // acp_event { method:'session/result', params:{ usage:{input,output}, cost_usd, ... } } // 而 callAgent → _runViaProtocol → runPromptInSession 会把 options.onAcpEvent 原样透传下去 // (agent-manager.js:104 / :325),所以这里不需要改 agent-manager 就能取到真实计数。 // 同一次调用里 _runViaProtocol 可能 failover 重试一轮,两轮的 usage 都真花了钱,累加是对的。 // // ponytail: 只有 claude-cli 协议上报 usage;AcpProtocol 不 emit session/result, // 这类 agent 的 tokens 恒为 0(**这一维在它们身上会低计**,不是负数也不是崩溃, // 预算仍由 total_agent_calls / wallclock_ms 两维兜底)。 // 升级路径:在 AcpProtocol 收尾处补一条同形状的 session/result 事件,本函数不用改。 let tokens = 0; const onAcpEvent = (ev) => { if (!ev) return; if (ev.method === 'session/result') { const u = (ev.params && ev.params.usage) || null; if (u) tokens += (u.input || 0) + (u.output || 0); return; } // mock 模式下 agent 的「产出文件」只能靠这条事件落盘 —— ClaudeCliProtocol._sendMockPrompt // 把 fixture.files 原样 emit 出来,真实 CLI 则自己写盘、不发这条。 // DevLoop 的 loop-controller 一直接着它,图执行这侧没接,于是 mock 下 // readNodeProduct 永远找不到 attempt-{n}.json —— validator 系节点因此恒 blocked, // 整条链在 mock 下走不完。补上这一环,mock 才真正能驱动 P3。 if (ev.method === 'mock/files' && Array.isArray(ev.params && ev.params.files)) { const base = safeNodeArtifactDir(ctx); if (!base) return; for (const f of ev.params.files) { if (!f || typeof f.path !== 'string') continue; // fixture 不知道自己这轮是第几次 attempt,用占位符让它跨轮复用同一份。 const rel = f.path.replace(/\{attempt\}/g, String(ctx.attempt)); const abs = path.resolve(base, rel); // 越界写入直接丢弃:fixture 虽是仓库里的文件,产物目录之外的路径也不该由它决定。 if (abs !== base && !abs.startsWith(base + path.sep)) { timelineNote(ctx, 'node.output', `mock 产物路径越界,已丢弃: ${f.path}`, { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, degraded: 'mock_file_escapes_artifact_dir', }); continue; } try { fs.mkdirSync(path.dirname(abs), { recursive: true }); fs.writeFileSync(abs, String(f.content == null ? '' : f.content), 'utf8'); } catch (e) { timelineNote(ctx, 'node.output', `mock 产物落盘失败 ${rel}: ${e.message}`, { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, degraded: 'mock_file_write_failed', }); } } } }; let res = null; let err = null; try { res = await manager.callAgent(resolveAgentName(ctx.companyId, ctx.systemAgentId), task, { onAcpEvent, metadata: { taskId: ctx.runId, layer: 'graph-node', nodeRunId: ctx.nodeRunId, // 产物目录是**提示词**里那份约定的机器可读副本(提示词由 describeArtifactContract 拼, // 见 advanceOne);metadata 只进日志与 hook,agent 读不到,所以两边都要有。 rootRunId: ctx.rootRunId, attempt: ctx.attempt, artifactDir: safeNodeArtifactDir(ctx), // 项目根同理:提示词那份给 agent 读,这份给日志/hook 用(W2) projectRoot: ctx.projectRoot || null, }, }); } catch (e) { err = e; } // 归因/产出是**文件**,不是返回值:agent 崩了也可能已经把 attempt-{n}.json 写出来了, // 所以成功失败两条路径都读一次(§四.D.2.2 + validator-triager/agent.md 的产出约定)。 const produced = readNodeProduct(ctx); if (err) { return { status: 'fail', output: err.message, failReason: produced.failReason, remediationNote: produced.remediationNote, failNodeRef: produced.failNodeRef, outputRef: produced.outputRef, tokens, // 炸掉之前烧掉的 token 一样要扣(预算「只增不减」,§四.A.3) }; } // 判词节点(validator / triager)没留下产出文件 = **无法判定**,不是通过(R3-F1 缺陷 3b)。 // 按 §四.D.2.2 的 inconclusive 语义结算:fail + fail_reason=null → 走那条兜底边 // (§三.7 第五项校验保证它存在;没有兜底边就是死路 blocked,等人工)。 // 静默判 success 的后果不是「失败但走兜底边」,而是「validator 判的失败根本不存在」。 if (produced.missing && requiresAttributionProduct(ctx)) { timelineNote(ctx, 'node.triage', `判词节点没有可用的 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, failNodeRef: null, outputRef: produced.outputRef, tokens, }; } 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, // 跨层地址只在真判失败时才有意义(R-17.1):success 路径显式写 null, // 不留上一轮的地址在节点行上(同 fail_reason 的处理) failNodeRef: produced.failingVerdict ? produced.failNodeRef : null, // 首轮进场(verdict=no_prior_failure)也可能留了话给下一轮:成功路径带回来之后由 // settleNode → activateSuccessTargets → markReady 写到被点亮的下游节点那一行 // (R4-F1 major 1:这句注释以前是个假承诺,下游实际收到的一直是 null)。 remediationNote: produced.remediationNote, outputRef: produced.outputRef, tokens, }; } /** * 读节点这一 attempt 的结构化产出,解析出三值透传需要的两个值 + output_ref。 * * 约定路径(system/agents/validator-triager/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, failNodeRef: 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 } = resolveTriageFile(dir, ctx.attempt); if (!file) { if (stale) { // 目录按 nodeRunId 分配、跨 attempt 复用,编号比本轮小的那份是**上一轮**的判词。 // 拿它当本轮结论会让节点被旧判词反复判失败直到撞硬顶(R3-F1 缺陷 4)——宁可当没有。 timelineNote(ctx, 'node.triage', `只找到上一轮的归因文件 ${path.basename(stale)}(< attempt-${ctx.attempt}),不复用`, { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, file: stale, degraded: 'stale_attribution', }); } return { ...empty, outputRef }; } const doc = readJsonOrNull(file); if (!doc || typeof doc !== 'object') { timelineNote(ctx, 'node.triage', `归因文件解析失败,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.triage', `归因取值非法(${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); // R-17.1 跨层地址:归因文件里的 target_node(+ 可选 target_org_node)。 // 写不出合法地址(没写 / 写成对象 / 只写了 org 那一半)一律按「没有地址」处理 —— // 回退退化成整张 work graph 重跑,与返工前同语义,不会因为地址坏了就卡住。 const failNodeRef = normalizeNodeRef({ orgNode: doc.target_org_node, workNode: doc.target_node, }); if (doc.target_node != null && !failNodeRef) { timelineNote(ctx, 'node.triage', `跨层地址不合法(target_node=${JSON.stringify(doc.target_node)}),按无地址处理`, { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, file, degraded: 'invalid_target_node', }); } if (failingVerdict || remediationNote) { timelineNote(ctx, 'node.triage', `归因 verdict=${verdict || '(缺省)'} fail_reason=${failReason || 'null'}` + (failNodeRef ? ` → ${failNodeRef}` : ''), { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, file, verdict, fail_reason: failReason, remediation_note: remediationNote, fail_node_ref: failNodeRef, failing: failingVerdict, }); } return { failReason, remediationNote, failNodeRef, 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 resolveTriageFile(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 = ATTEMPT_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 }; } /** * 事件 data 里携带的文本上限(W12)。 * * 为什么是「截断进事件」而不是「加一列 output 落库」: * 1. outcome.output 是 agent 的**完整输出**,长度无上限;graph_run_nodes 每次 * getRunState 都是 `SELECT *` 全表读回去(控制台每次轮询都拉一遍所有节点), * 把整份输出塞进节点行 = 每次轮询把所有历史输出重传一遍。 * 2. 完整产物已经有地址了 —— output_ref 指向 nodes/{nodeRunId}/ 那个目录, * 再存一份是同一份数据的第二个真相源。 * 3. fail_reason 不能借用:它是**选边的匹配键**(精确 fail 边靠它匹配), * 掺进自由文本会让所有精确 fail 边失配(§四.D.2.2「引擎不发明 fail_reason」)。 * 所以留痕轨道选 timeline.ndjson + stream.log(append-only、按 run 分目录、天然有上限), * 节点行只保留定长的结构化列。 */ const EVENT_TEXT_MAX = 600; /** 事件 data 里的自由文本一律截断,避免 stream.log / timeline.ndjson 被单条巨型输出撑爆 */ function clipText(s, n = EVENT_TEXT_MAX) { if (s === null || s === undefined || s === '') return null; const str = typeof s === 'string' ? s : String(s); return str.length <= n ? str : `${str.slice(0, n)}…(已截断,完整输出见 output_ref)`; } /** 只记一条审计(同时也发 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:aid`,不存在给裸 `aid`。28 号设计把公司实体清空之后(批次 4)这个判据恒为 false, * 于是**地址里的公司这一半会随目录一起消失** —— 而公司恰恰是运行态分区的键 * (memory / conversations 按 (agentId, companyId) 归属)。所以地址先立住: * 恒为 `cid:aid`,`aid` 恒指向 `system/agents/` 下的本体,「私有目录还在不在」 * 降级成 agent-locator.resolveAgentDir 里那条过渡期分支。 */ function resolveAgentName(companyId, systemAgentId) { return locator.formatAddress(companyId, systemAgentId); } // ──────────────────────────────────────────────────────────────── // workgraph-planning 的默认实现(§四.D.0 的第一步,R5 接上的断链 1) // ──────────────────────────────────────────────────────────────── /** 全局唯一的一份 work graph 规划本体(R-15.2:不按 org node 复制副本) */ const PLANNING_AGENT_ID = 'workgraph-planning'; /** * 规划的最多轮次。第 2 轮起把上一轮的校验错误原样贴回去让它自己修 * (workgraph-planning/agent.md Step 6「返回 400 怎么办:读懂 errors,自己修,重试」, * 与 org-graph POST 端点返回 `errors[{rule, detail}]` 是同一套约定,见 company.js:1128)。 * 模板给 agent 的自查上限是 5 轮,引擎这一侧收紧到 3:每一轮都是一次真实 LLM 调用, * 而且失败还有外层的回退/熔断/预算三道兜底,不必在这一层烧完。 */ const PLAN_MAX_ROUNDS = 3; /** * 调 workgraph-planning 规划一张子图并落库(§四.D.0 的 `节点 → running → 调 workgraph-planning`)。 * * 与 defaultAgentExecutor 共用同一套调用方式(AgentManager + onAcpEvent 累加 usage)—— * **规划调用也是真花钱的一次 agent 调用**,token 必须计入同一个 root run 预算池, * 否则「一层规划 + 一层执行」里有一半的花费是隐形的。 * * 落库前必过 work-graph-db 的五项校验(createWorkGraph 内部调 validateWorkGraph); * 不过就把 errors 原样回给 agent 再来一轮(模板 Step 6 定义的那次交互)。 * * @returns {Promise<{workGraphId: string, tokens: number, rounds: number}>} */ async function defaultPlanWorkGraph(ctx) { const AgentManager = require('../src/core/agent-manager'); const manager = new AgentManager({}, REPO_ROOT); 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); }; let errors = null; let lastOutput = ''; for (let round = 1; round <= PLAN_MAX_ROUNDS; round++) { const prompt = buildPlanningPrompt(ctx, { round, errors, lastOutput }); timelineNote(ctx, 'plan.call', `调 ${PLANNING_AGENT_ID} 规划子图(第 ${round}/${PLAN_MAX_ROUNDS} 轮)`, { node_run_id: ctx.nodeRunId, node_name: ctx.nodeName, agent: agentName, round, retry_errors: errors, }); let res; try { res = await manager.callAgent(agentName, prompt, { onAcpEvent, metadata: { taskId: ctx.runId, layer: 'graph-plan', nodeRunId: ctx.nodeRunId, rootRunId: ctx.rootRunId, attempt: ctx.attempt, projectRoot: ctx.projectRoot || null, }, }); } catch (e) { // 调用本身炸了(本体寻址失败 / CLI 起不来)不是「图画错了」,重贴 errors 再来一轮没有意义。 // 直接上抛,但**把已经烧掉的 token 挂上去**,否则前几轮的花费在这条路径上就蒸发了。 e.tokens = (e.tokens || 0) + tokens; throw e; } lastOutput = (res && res.output) || ''; const picked = extractPlannedGraph(lastOutput); if (!picked) { errors = [{ rule: 'output', detail: '回复里既没有已落库的 workGraphId(wg_...),也没有可解析的图 JSON' + '(内联 ```json 代码块或一个 .json 文件的绝对路径,形状 {"nodes":[...],"edges":[...]})', }]; continue; } // ① agent 自己已经把图落库了(模板 Step 6 的 POST 成功路径)→ 认它的 id if (picked.workGraphId) { if (getWorkGraph(picked.workGraphId)) { timelineNote(ctx, 'plan.done', `子图规划完成(agent 已落库): ${picked.workGraphId}`, { node_run_id: ctx.nodeRunId, work_graph_id: picked.workGraphId, round, tokens, }); return { workGraphId: picked.workGraphId, tokens, rounds: round }; } errors = [{ rule: 'reference', detail: `work graph ${picked.workGraphId} 在库里不存在,请把图 JSON 直接交回来` }]; continue; } // ② 更常见的一条:端点还没上线(模板里的 404 降级路径),agent 把图交回引擎落库 const created = createWorkGraphSafely(ctx, picked.draft); if (created.ok) { timelineNote(ctx, 'plan.done', `子图规划完成并落库: ${created.id}(${picked.source})`, { node_run_id: ctx.nodeRunId, work_graph_id: created.id, round, tokens, nodes: (picked.draft.nodes || []).length, edges: (picked.draft.edges || []).length, }); return { workGraphId: created.id, tokens, rounds: round }; } errors = created.errors; timelineNote(ctx, 'plan.rejected', `子图未过写入前校验,回给 ${PLANNING_AGENT_ID} 自行修正(第 ${round} 轮)`, { node_run_id: ctx.nodeRunId, round, errors, }); } // 耗尽轮次:抛给 advanceOne 的 catch,节点按 fail 结算走兜底边(引擎不发明 fail_reason)。 // 抛之前把 tokens 挂在错误上,让已经烧掉的规划调用照样扣预算(「只增不减」,§四.A.3)。 const err = new Error( `${PLANNING_AGENT_ID} 连续 ${PLAN_MAX_ROUNDS} 轮未产出合法子图(节点 ${ctx.nodeName}):` + JSON.stringify(errors) ); err.tokens = tokens; throw err; } /** * 规划本体的寻址名。 * * workgraph-planning 是基础设施角色,不出现在 org graph 里,没有任何招聘流程会把它雇进公司—— * 生产里第一次规划必然撞 `Agent not found`。(28 批次 1 把 system/templates/ 并进 * system/agents/ 之后,「本体不在 AgentManager 认得的两条路径上」这半条理由消失了, * 但「没人雇它」这半条还在,所以下面的按需雇佣照旧。) * * 解法与 company.js 对 orggraph-planning 的做法一致(replan 端点先 hire 再触发): * 用到时按需雇一次,hire 本身幂等(已存在返回 already_exists)。P1 引用模式下不拷贝 * agent.md,只建目录 + 注入 company-context.md,这也正是模板说的「靠记忆分区承载差异」。 */ function resolvePlanningAgentName(companyId) { if (!companyId) return PLANNING_AGENT_ID; // 「雇没雇过」查 company_agents(28 批次 4-19)。原来判的是公司私有目录存不存在, // 那个目录清空之后判据恒为 false:每次都重新 hire 一遍(幂等,无害), // 但最后那个三元表达式也恒为 false → 返回裸名,公司限定地址整个丢掉。 const { CompanyService, listCompanyAgents } = require('./company-service'); const isHired = () => listCompanyAgents(companyId).some((r) => r.role_name === PLANNING_AGENT_ID); if (!isHired()) { try { const hired = new CompanyService().hire(companyId, PLANNING_AGENT_ID, {}); // hire() 会顺手把被雇的角色 upsert 进 org_graph_nodes,只对 config.type='meta-bootstrap' // 的基础设施角色开了豁免;workgraph-planning 的 type 是 'meta-planning',**不在豁免里**。 // 不撤掉这一行的话,规划本体会作为一个孤立 active 节点混进它正在为之规划的那张 org graph // (agent.md 约束八:「你自己不入图」;P2 §4.1 对 orggraph-planning 是同一条规矩), // 下一次建 run 时它就会变成一个谁也点不亮的 subgraph 节点。 // 撤在这里而不是改 company-service:那不是本轮能动的文件;这条 guard 的正解是把 // company-service 的判据从 'meta-bootstrap' 放宽成 'meta-*',届时这段可以删掉。 if (hired && hired.status === 'hired_from_template') { require('./org-graph-service').removeNode(companyId, PLANNING_AGENT_ID); } } catch (e) { // 公司记录不存在 / 模板缺失:不在这里报错,退回裸名让 AgentManager 给出它自己的诊断 console.warn(`[graph-engine] 按需雇佣 ${PLANNING_AGENT_ID} 失败(${companyId}): ${e.message}`); } } return isHired() ? `${companyId}:${PLANNING_AGENT_ID}` : PLANNING_AGENT_ID; } /** * §四.D.0 那张清单里 workgraph-planning 的四项输入,一项都不能少 * (模板 Step 1「四项输入,一项都不能跳过不读」): * ① 该 org node 的角色定义 ② 任务描述 ③ 可读的 artifacts/ ④ 上一轮 remediation_note * 外加引擎侧的三样东西:图要交回哪里、宿主 org node 的 id(模板 body 里的 ownerOrgNodeId)、 * 上一轮的校验错误(重试轮才有)。 */ function buildPlanningPrompt(ctx, { round = 1, errors = null, lastOutput = '' } = {}) { const lines = [ `你正在为 org node **${ctx.nodeName}** 规划它这一次任务的 work graph。`, '', '## ① 宿主 org node 的角色定义', `- 节点名(org graph 内):${ctx.nodeName}`, `- 引用的本体(systemAgentId):${ctx.systemAgentId || '(未绑定)'}`, `- ownerOrgNodeId(提交图时要带上):${ctx.defNodeId}`, `- 公司:${ctx.companyId}`, describeRoleDefinition(ctx), '', '## ② 任务描述(task-goal)', ctx.task || '(调用方没给任务描述)', '', '## ③ 可读的 artifacts/(root run 共享产物区,只读)', describeArtifactsDir(ctx.run), // R-19 org 层契约的读回方。返回 null(没有任何契约行)时**整段连标题一起消失**, // 靠下面那个 filter 滤掉 —— 编号用 ③bis 而不是插进 ①②③④,那四项是模板 Step 1 // 「四项输入一项都不能跳」的原文清单,重编号会让模板与提示词对不上。 describeOrgContracts(ctx), '', '## ④ 上一轮 remediation_note', ctx.remediationNote || '无,首次规划。', '', // W2:规划本体也是一次真 CLI 调用,也会写盘(模板 Step 6 的降级路径就叫它把图写成 // workspace/workgraph-.json)。不给它位置,它同样会写到最近的仓库里去。 // // 段头必须是 prompt-builder 那个常量,**不能手写一个同义的中文标题**(原来写的是 // `## 写入边界(硬约束)`):PromptBuilder.buildRuntimeCoordinates 靠 // taskCarriesGraphWriteBoundary(task) 判断本次调用是 graph 链路还是 agent-chat 链路, // 判据就是这个字面量在不在任务描述里。手写的标题让判据落空 → 坐标声明会再补一行 // 「你的交付物必须写到 {companyRoot}/workspace/」,而下面这行又说「禁止写 REPO_ROOT // 下的任何路径」—— companyRoot 正在 REPO_ROOT 下。同一段提示词里两条互相排斥的指令, // 规划本体听哪条都跑得完、都不报错(28 §4.4 要挡的正是这种静默失败)。 GRAPH_WRITE_BOUNDARY_HEADER, ctx.projectRoot ? `本次任务的项目根:${ctx.projectRoot}\n` + '你若要落任何中间文件(图 JSON、草稿),写在这个项目根里面,用绝对路径。' : '本次任务没有分配到项目根(分配失败,见 timeline):不要在磁盘上留任何文件,把图直接贴在回复里。', `禁止写 ${REPO_ROOT} 下的任何路径(那是引擎源码仓库);引擎会在调用前后扫宿主仓库的 git 状态,` + '发现越界会把本节点判失败。你规划出来的每个节点将来也会收到同一条边界约束。', '', '## 落库口径(本次调用的硬约束)', `- goalId=${ctx.run.goal_id || 'null'} / taskId=${ctx.run.task_id || 'null'} / okrNodeId=${ctx.run.okr_node_id || 'null'},原样带上,别自己编。`, '- 图必须过五项校验:无环 / 单一入口 / 全可达 / 回退边显式标 backward /' + ' **每个有 fail 出边的节点都要有一条不带 failReason 的兜底边**。', `- 节点引用的 systemAgentId 必须是**已存在的本体**;不存在的本体会被引擎以 rule=reference 打回。`, '', '## 交回方式(引擎读你回复的最后一段)', // 措辞必须与 workgraph-planning/agent.md 的 NO-SUCH-ENDPOINT 一致:说「尚未上线」会被读成 // 「将来会上线」,agent 就还会去试一次、吃一个 404、白烧一轮(前五次真跑每轮都在烧)。 // 这个端点从来没有设计过,落库始终是引擎的活。 '`POST /api/company/{公司ID}/work-graphs` **不存在、也不会上线** —— 落库是引擎的活:', '**把最终图直接贴在回复里的一个 ```json 代码块中**(或写成一个 .json 文件后给出它的绝对路径),形状:', '```json', // 示例的边必须带 produces:这段是 agent 唯一见过的「一条边长什么样」, // 示例里没有声明,它照抄出来的就是一张不传数据的图 —— 引擎自己的示例在教反面教材。 '{"nodes":[{"nodeName":"...","systemAgentId":"...","nodeKind":"agent_call"}],' + '"edges":[{"from":"...","to":"...","condition":"success","produces":["brd"]}]}', '```', // 刻意**不**再提「已落库时回一行 workGraphId」:没有任何受支持的落库途径, // 所以这句话只会诱发它编一个 wg_ id;而 extractPlannedGraph 优先匹配那个形状、 // 查库查不到就整轮作废(模板 agent.md:345 把这条列成硬禁令,两边对立时 agent 得赌)。 // 引擎侧仍保留认真实 wg_ id 的分支,只是不再邀请它写。 ]; if (round > 1) { lines.push( '', `## ⚠️ 上一轮没通过(第 ${round} 轮)`, '引擎对你上一轮的产出跑了写入前校验,结果如下。**读懂 rule,改图,再交一次**;', '不要原样重发,也不要绕过校验:', '```json', JSON.stringify(errors || [], null, 2), '```', ); if (lastOutput) lines.push(`(上一轮回复摘要:${String(lastOutput).slice(-800)})`); } // 只滤掉 null/undefined(拼不出内容的可选段),`''` 是**故意的空行**,不能一起滤掉 return lines.filter((l) => l != null).join('\n'); } /** 宿主 org node 的角色定义:优先公司私有目录(过渡期,批次 4 后消失),其次共享本体,再次本体目录直读 */ function describeRoleDefinition(ctx) { const candidates = []; if (ctx.companyId && ctx.systemAgentId) { candidates.push(path.join(locator.legacyCompanyAgentDir(ctx.companyId, ctx.systemAgentId), 'agent.md')); } if (ctx.systemAgentId) { try { const shared = require('./graph-service').lookupSharedAgentMd(ctx.systemAgentId); if (shared) candidates.push(shared); } catch { /* 查不到共享本体不影响规划,下面还有本体目录兜底 */ } // 兜底候选原来指 `system/templates/`,而 28 批次 1 已把该目录整体并进 `system/agents/` —— // 目录没了,这条候选就永远 existsSync=false,等于悄悄少了一层兜底(本体在磁盘上、 // 只是没入库时正好落进这个缝)。改指本体目录,恢复它本来的作用。 candidates.push(path.join(locator.bodyDir(ctx.systemAgentId), 'agent.md')); } for (const f of candidates) { try { if (!fs.existsSync(f)) continue; // 只给开头一段:规划要的是「这个节点负责什么职能」,不是它的完整工作流。 // 但截断必须**可见**——原来悄悄切掉后半段,agent.md 写得长一点(常见于职能说明 // 排在文件后半的本体),「这个节点负责什么职能」这句关键描述就可能根本没进提示词, // 而 workgraph-planning 拿不到信号、只会按看到的那半截自行推断,不会知道自己漏看了。 // 加一句提示 + 完整路径,给它一个自己去读全文的退路,不是把上限拉大(2000 仍是故意的)。 const full = fs.readFileSync(f, 'utf8'); const excerpt = full.slice(0, 2000); const truncNote = full.length > 2000 ? `\n(⚠️ 原文共 ${full.length} 字,已截断到前 2000 字——如果上面没读到「这个节点负责什么职能」,说明它在后半段,去读完整文件 ${f})` : ''; return `- 角色定义摘录(${path.relative(REPO_ROOT, f)}):\n${excerpt}${truncNote}`; } catch { /* 读不动就换下一个候选 */ } } return '- 角色定义:未找到该本体的 agent.md,请按节点名推断职能,并在规划记录里写明这一项缺失。'; } /** 任务级共享产物区的清单 + task-goal.json 正文(§四.A.5:跨 org node 只读) */ function describeArtifactsDir(run) { let dir; try { dir = path.join(runRootDir(run), 'artifacts'); } catch { return '- 无(run 缺 company_id/root_run_id,拼不出产物目录)'; } const out = [`- 目录:${dir}`]; let names = []; try { names = fs.readdirSync(dir); } catch { return out.concat('- (目录还不存在,说明上游 org node 还没留下任何产物)').join('\n'); } out.push(names.length ? `- 文件:${names.join(', ')}` : '- (空目录)'); const goal = readJsonOrNull(path.join(dir, 'task-goal.json')); if (goal && goal.task) out.push(`- task-goal.json 里的原始任务:${goal.task}`); return out.join('\n'); } /** 本节点**出边**上声明为 produces 的 kind(去重)。role 为 NULL 的历史行按 produces 读, * 与 org-graph-service.readGraph 回读契约时同一个口径。backward 出边不算:引擎解析输入 * 只看 forward 入边,写在回退边上的声明本就不生效(validateGraph 也拦了)。 */ function outboundProducedKinds(layer, edges, defNodeId) { const out = []; for (const e of edges || []) { if (String(e.from) !== String(defNodeId)) continue; if ((e.edgeKind || 'forward') === 'backward') continue; for (const c of edgeContracts(layer, e.id)) { if ((c.role || 'produces') !== 'needs') out.push(c.kind); } } return [...new Set(out)]; } /** * R-19 org 层:把 org 图上的数据契约递给 **workgraph-planning**(这就是 org 层契约的读回方)。 * * 为什么不能靠 describeInputs:它第一行就是 `ctx.nodeKind === 'subgraph'` 时返回空串, * 而 readDefGraph 把 org 图的节点**全部**写成 nodeKind='subgraph'(work-graph-db.js * 「org graph 的节点全部是 subgraph executor」)。两条合起来 = org 层的边契约在输入侧 * 一个读回方都没有。先补模板去教 orggraph-planning 声明,产出的就是「声明了没人读」。 * * 语义也不该复用 describeInputs 的措辞。org 节点自己不干活,它派给规划本体的是**约束**: * 入边声明的 kind → 你规划的子图的**入口节点**要吃这些(引擎已经把绝对路径查好) * 出边声明的 kind → 你规划的子图**必须有节点产出**这些,否则下游 org node 拿不到输入 * 这才是两层递归里 org 层约束 work 层的方式。 * * 拼不出来(没有契约行 / def 读不出来)时返回 **null**,整段连标题一起省略: * 留一个空标题会被 agent 读成「引擎查过了,确实没有」,而真相是「引擎没查」。 * * ponytail: 这里自己再读一次 def(defIndex)而不是让 advanceOne 把 def 塞进 ctx —— * 代价是每一轮规划多一次定义层查询(org 图是个位数节点,可忽略),换来的是 ctx 形状不变: * ctx 会原样传给 setAgentExecutor 注入的外部实现,往里塞一个含 Map 的字段是更大的面。 * 规划轮数一多(PLAN_MAX_ROUNDS=3)也就三次;真要省,把 def 提到 executeSubgraph 里传参。 */ function describeOrgContracts(ctx) { if (!ctx || !ctx.run) return null; let inboundLines = []; let produced = []; try { const layer = ctx.run.graph_kind === 'org' ? 'org' : 'work'; const edges = defIndex(ctx.run).edges || []; inboundLines = resolveInboundLines(ctx, inboundContracts(layer, edges, ctx.defNodeId)); produced = outboundProducedKinds(layer, edges, ctx.defNodeId); } catch { return null; // 索引/定义层读不出来不该拖垮规划调用 } if (!inboundLines.length && !produced.length) return null; const lines = ['', `## ③bis ${ctx.nodeName} 在上一层图上的数据契约(引擎已解析,不要自己猜路径)`]; if (inboundLines.length) { lines.push( '', '### 上游节点交给你的输入', ...inboundLines, '标 ⚠️ 的表示上一层图上声明了这份输入、但产物索引里还没有实例:先确认是不是上游还没跑完,', '不要自己去别的节点私有目录里翻。', '**这些路径要落到你规划的子图的入口节点手上** —— 引擎不会自动把它们拼进那个节点的提示词,', '你要写进入口节点的任务描述里。', ); } if (produced.length) { lines.push( '', '### 你这一层必须产出的东西(硬约束)', `上一层图上你的出边声明了这些 kind,下游节点就是按它们去产物索引里找输入的:${produced.join(' / ')}`, '**你规划的子图里必须有节点真的产出上面每一个 kind**,一个都不能少。', '少一个的后果不是现在报错,而是下游节点跑到那一步说「我拿不到输入」—— 那时整个子图的预算已经烧掉了。', '落地方式(两条都要做到位才算产出):', ` ① 在产出它的那条 work 边上写 \`"produces":["${produced[0]}"]\`(引擎按出边声明给文件定 kind);`, // 出口节点没有出边 → 它的 indexNodeOutputs 拿不到任何声明,只能按文件名认 kind。 // 这条不写清楚的话,org 层最后一个 kind 恰好就是最容易掉的那个。 ' ② 出口节点没有出边、写不了 produces 时,把**文件名**直接起成 kind 本身' + `(引擎只看去掉扩展名的那一段,认得出的是 ${KNOWN_FILE_KINDS.join(' / ')})。`, ); } return lines.join('\n'); } /** * 从 agent 的回复里取出图。三种形态按优先级: * ① `workGraphId: wg_xxx`(它自己 POST 成功了) * ② 内联 ```json 代码块 / 裸 JSON 对象,含 nodes * ③ 一个 .json 文件路径(模板 404 降级路径写的 workspace/workgraph-.json) * @returns {{workGraphId?: string, draft?: object, source: string}|null} */ function extractPlannedGraph(output) { const text = String(output || ''); if (!text.trim()) return null; // 严格按 work-graph-db.newId 的形状匹配(`wg_` + 12 位十六进制),不要写成 \w+ —— // 那样 agent 起的 `wg_impl` 这类**节点名**会被当成图 id,白白多烧一轮调用。 const wg = /\bwg_[0-9a-f]{12}\b/.exec(text); if (wg) return { workGraphId: wg[0], source: 'workGraphId' }; for (const block of jsonCandidates(text)) { const doc = safeJsonParse(block); if (doc && Array.isArray(doc.nodes)) return { draft: doc, source: 'inline-json' }; } // 文件路径:取最后一个提到的 .json(agent 通常在末尾汇报产物路径) const files = text.match(/(?:[A-Za-z]:)?[\w./-]+\.json/g) || []; for (const f of files.reverse()) { const abs = path.isAbsolute(f) ? f : path.join(REPO_ROOT, f); const doc = readJsonOrNull(abs); if (doc && Array.isArray(doc.nodes)) return { draft: doc, source: abs }; } return null; } /** ```json 围栏块优先,其次从第一个 `{` 起做括号配平取一个对象 */ function jsonCandidates(text) { const out = []; const fence = /```(?:json|jsonc)?\s*([\s\S]*?)```/g; let m; while ((m = fence.exec(text))) out.push(m[1]); let depth = 0; let start = -1; for (let i = 0; i < text.length; i++) { const c = text[i]; if (c === '{') { if (depth === 0) start = i; depth++; } else if (c === '}' && depth > 0) { depth--; if (depth === 0) out.push(text.slice(start, i + 1)); } } return out; } function safeJsonParse(s) { try { return JSON.parse(s); } catch { return null; } } /** * 落库。校验不过返回 errors 给上层贴回给 agent; * FK 之类的**抛错**同样翻译成一条 errors,不让「引用了不存在的本体」变成整个 run 的崩溃 * (模板 Step 3 情形 3 明说 agent 可能引用尚不存在的本体)。 */ function createWorkGraphSafely(ctx, draft) { try { const created = createWorkGraph({ companyId: ctx.companyId, ownerOrgNodeId: Number.isNaN(Number(ctx.defNodeId)) ? null : Number(ctx.defNodeId), plannedBy: PLANNING_AGENT_ID, goalId: ctx.run.goal_id, taskId: ctx.run.task_id, okrNodeId: ctx.run.okr_node_id, nodes: draft.nodes, edges: draft.edges || [], // 31-work-graph-fail-fallback-coverage 决策 A:这是 workgraph-planning 真实提交的图 // 落库前的唯一校验点,也是它 Step 7 那张 errors 表实际读到的来源——只在这里收紧。 strictFailFallback: true, }); return created.ok ? created : { ok: false, errors: created.errors }; } catch (e) { return { ok: false, errors: [{ rule: /FOREIGN KEY/i.test(e.message) ? 'reference' : 'schema', detail: `落库失败:${e.message}` + (/FOREIGN KEY/i.test(e.message) ? '(多半是某个 systemAgentId 不是已注册的本体,换一个清单里真实存在的)' : ''), }], }; } } /** * subgraph:起一张子图并等它收敛(§四.D.0)。 * * 「等」不是在这里 await 几小时——手动模式下 advanceOne 由一次 HTTP 请求驱动,不可能挂住。 * 落地方式是:本函数只负责规划图 + 建子 run + 回填 child_run_id,然后返回 status='pending' * 表示「父节点继续停在 running」;子 run 收敛时由 settleRun → propagateToParent 回写父节点。 * 两条路径最终都进同一个 settleNode,且靠 running→终态 的条件更新保证只生效一次。 */ async function executeSubgraph(ctx) { // ── 幂等判据(R-17.1 重新设计)───────────────────────────────────────── // // 判据是「这张子 run 服务的是**本轮**(本节点的这一次 attempt)吗」, // 落在 graph_run_nodes.child_run_attempt 这一列上。 // // 为什么必须换掉旧判据:旧判据是「旧子 run 还活着吗」——活着就等它,终态就**另起一张**。 // 它的正确性正好架在「每次回退都新建子 run」这条上(P3 第二轮修死锁的修法)。 // R-17.1 要求复用旧子 run,那条修法被拆掉,「上一轮留下的终态」和「这一轮刚重置过的子 run」 // 在 run_status 上长得一模一样,只看终态就会重演那个永久死锁(守卫读到终态子 run 返回 // pending,父节点永远停在 running,没有任何人再回写它)。代次是唯一能把两者分开的信息。 // // 三条分支,**没有一条会在子 run 已终态时返回 pending**(死锁那一类在结构上不再存在): // servedAttempt === attempt 且子 run 还活着 → 它正在为本轮干活,等它回写(pending) // servedAttempt === attempt 且子 run 已终态 → 回写在某一步丢了(进程重启 / 事件竞态), // 就地按它的结果结算,**绝不返回 pending** // servedAttempt !== attempt(含旧行的 NULL)→ 新一轮:复用定义与子 run(reopenChildRun), // 复用不了才落回「重新规划一张」 // // infra_flake 原地重跑不加 attempt(§四.D.2.1 末段),所以它走不到「新一轮」那条—— // routeFailure 在那条路径上显式清掉 child_run_id + child_run_attempt,回到「没有子 run」分支。 const existing = getRunNode(ctx.nodeRunId); const prior = existing && existing.child_run_id ? getRun(existing.child_run_id) : null; if (prior) { const served = existing.child_run_attempt == null ? null : Number(existing.child_run_attempt); if (served === ctx.attempt) { if (!isChildSettled(prior)) return { status: 'pending', childRunId: prior.id }; // 本轮的子 run 已经交待完了却又被拾取:唯一安全的动作是就地结算,不是继续等。 timelineNote(ctx, 'node.subgraph', `子 run ${prior.id} 已为本轮(attempt ${ctx.attempt})收敛到 ${prior.run_status},就地结算而不是继续等回写`, { node_run_id: ctx.nodeRunId, child_run_id: prior.id, attempt: ctx.attempt, child_run_status: prior.run_status, degraded: 'orphan_settled_child', }); return deriveChildOutcome(prior); } // 新一轮:R-17.1「复用既有 work graph 定义,只重置被点名的 node」 if (reopenChildRun(ctx, 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})`); // 规划本身就是一次真实 agent 调用,它烧的 token 必须回到 advanceOne 手里扣同一个预算池。 // 不透传的话 total_tokens 这一维在 subgraph 节点上恒为 0 —— 「产生了但没传上去」, // 正是本轮在盯的那类漏接(注入假规划的测试不返回 tokens,这里按 0 处理)。 const tokens = (planned && planned.tokens) || 0; 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, }); // child_run_id 与 child_run_attempt **必须同一句写**:分两句写会留下「有子 run、代次未知」 // 的中间态,守卫读在那一刻就会把本轮的子 run 当成上一轮的,重开一遍。 getDb() .prepare(`UPDATE graph_run_nodes SET child_run_id = ?, child_run_attempt = ?, updated_at = ? WHERE id = ?`) .run(child.id, ctx.attempt, now(), ctx.nodeRunId); markRunReady(child.id); startRunSync(child.id); return { status: 'pending', childRunId: child.id, tokens }; } /** * 子 run 是否已经「交待完了」——终态或 blocked 都算:settleRun 在这两种情形下都会 * propagateToParent 回写父节点,父节点不该再等它。 */ function isChildSettled(childRun) { return TERMINAL_RUN_STATUS.has(childRun.run_status) || childRun.run_status === 'blocked'; } /** * 复用旧子 run 跑新一轮(R-17.1「复用既有 work graph 定义」)。 * * 两种入场姿势,靠**子 run 里有没有已经被点亮的节点**区分,不另存指令: * - 有 ready/running 节点 → 上游已经做过**定点重置**(handleBackwardEdge 把被点名的那个 * node 打回 ready 了),本函数只把 run 拉回 running,其余节点原样不动 * —— 这就是「只重置被点名的 node」;contract_hash unchanged 的下游因此一步都不重跑。 * - 没有 → 这一轮是**整张图作废**(上游 org node 重做导致输入全变,resetForwardDownstream * 打回来的),全量重置 + 重新点亮入口,但仍然复用同一张 work graph 定义和同一个 child_run_id, * 不再调 workgraph-planning 重画一张(返工前每次回退都要重画,那是一次真实 LLM 调用)。 * * **重置在 backward 那一刻就做完,run 状态却留到这里才拉起**:子 run 一变 running, * advanceTree 就会开始推它,而那时父节点可能还没轮到重新点亮 —— 子图会抢在上游重做完之前先跑。 * 所以「改节点」和「开跑」必须分在两个时刻。 * * @returns {boolean} 复用成功;false = 复用不了(人工停止过 / 转换被拒),调用方回退到重新规划 */ function reopenChildRun(ctx, child) { // stopped 是**人工主动终止**的结果,自动流程不许把它复活(与 resumeDescendants 不碰 blocked 同理) if (child.run_status === 'stopped') return false; const db = getDb(); const rows = db.prepare(`SELECT id, status FROM graph_run_nodes WHERE run_id = ?`).all(child.id); const targeted = rows.some((r) => r.status === 'ready' || r.status === 'running'); if (!targeted) resetRunNodes(child.id); // 已收敛的(completed/failed)先退回 ready ——「图还在、等启动」,这是 §四.A.4 给终态开的 // 唯一出口;blocked / paused 本来就有直达 running 的转换,不必绕这一步。 const fresh = getRun(child.id); const summary = `子 run 复用(父节点 ${ctx.nodeName} 第 ${ctx.attempt + 1} 轮,` + `${targeted ? '定点重置:只重跑被点名的节点' : '整张图作废重跑'})`; if (fresh.run_status !== 'running' && !canTransitionRun(fresh.run_status, 'running')) { if (!canTransitionRun(fresh.run_status, 'ready')) return false; if (!sse.transitionRun(child.id, fresh.run_status, 'ready', {}, { summary: `${fresh.run_status} → ready:${summary}`, reused_by: ctx.nodeRunId, attempt: ctx.attempt, scope: targeted ? 'node' : 'graph', })) { return false; } } // 走与「启动」完全同一条路径拉起(顺带清掉上一轮可能残留的 pause/stop 软信号 —— 不清的话 // 复用后第一次 advanceOne 就被自己挡回去) startRunSync(child.id); // 全量作废后必须重新点亮入口:startRunSync 只有从 ready 起跑时才 seed, // 而 blocked → running 这条路上它不 seed,漏了这一句整张图会全是 pending 直接假收敛 if (!targeted) seedEntryNodes(child.id); db.prepare(`UPDATE graph_run_nodes SET child_run_attempt = ?, updated_at = ? WHERE id = ?`) .run(ctx.attempt, now(), ctx.nodeRunId); sse.emitEvent(ctx.runId, 'node.subgraph', `${ctx.nodeName}: 复用子 run ${child.id}(${targeted ? '只重跑被点名的节点' : '整张图重跑'},不重新规划)`, { node_run_id: ctx.nodeRunId, child_run_id: child.id, attempt: ctx.attempt, scope: targeted ? 'node' : 'graph', replanned: false, }, { ref: ctx.nodeRunId }); return true; } /** * 把一张 run 里所有节点打回 pending(整图作废)。 * * 与 resetForwardDownstream 共用同一条 UPDATE 语义(条件更新 + 受影响行数 = 是否我改的), * 区别只是范围:那个按 forward 可达裁剪,这个是整张图。 * **child_run_id 不清**:嵌套 subgraph 节点靠 child_run_attempt 判代次,清了反而丢掉可复用的定义。 */ function resetRunNodes(runId) { const db = getDb(); const reset = []; for (const rn of db.prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ?`).all(runId)) { if (rn.status === 'pending' || rn.status === 'running') continue; const changed = sse.emitTransition( runId, { event: 'node.status', ref: rn.id, summary: `node ${rn.def_node_id}: ${rn.status} → pending(子 run 被复用,整张图作废重跑)`, data: { node_run_id: rn.id, from: rn.status, to: 'pending', attempt: rn.attempt, reset_by: 'reopen' }, }, () => db .prepare(` UPDATE graph_run_nodes SET status = 'pending', fail_reason = NULL, fail_node_ref = NULL, remediation_note = 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) { bumpNodeAttempt(rn.id); // 新的一轮:attempt 不走,产物文件名不变,归因会读到上一轮那份 reset.push(rn.id); } } return reset; } // ──────────────────────────────────────────────────────────────── // 核心原语: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 {string[]} reset 沿 backward 边被打回 pending 的下游 node_run_id * @property {object|null} crossLayer R-17.1 跨层定点重置的落点 {childRunId, nodeRunId, * workNodeName, orgNodeName, ref};null = 整张图重跑 * @property {boolean} infraRerun 本次是不是 infra_flake 原地重跑(不烧任何计数) * @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 }; } // 27-P4 T5:账户欠费门,与 advanceBatch 共用同一个 checkBilling/suspendRunTree // (advanceBatch 的 maxConcurrency<=1 分支整个委托给这个函数,这道门只写一份)。 const billing = await checkBilling(run); if (!billing.ok) { suspendRunTree(run, { reason: 'billing-arrears' }); return { ...result(runId, (getRun(runId) || {}).run_status, 'billing-arrears'), billing }; } 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) }; } return runClaimedNode(runId, run, def, defNode, nodeName, picked, opts); } /** * 已经拾取(`ready → running` 条件更新已经成功)的节点,从建 ctx 到执行到结算的完整一段。 * 27-P4 T6 从 `advanceOne` 里原样抽出来,**行为逐字不变**——只是把原来闭包捕获的局部变量 * 变成参数,好让 `advanceBatch` 的并行路径也能复用同一段逻辑(§4.3 决策:并发下的 join * 正确性不依赖新增锁/事务,靠的正是这段收尾——`settleNode` 全程同步——的既有写法, * 所以这里一个字都不能改成异步中途掺一次 DB 之外的 await)。 * * @param {string} runId * @param {object} run getRun(runId) 的快照(调用方已经读过一次,不在这里重读) * @param {object} def defIndex(run)(调用方已经算过一次,批量拾取时给多个节点共用) * @param {object} defNode def.nodes.get(picked.def_node_id) || {} * @param {string} nodeName * @param {object} picked graph_run_nodes 行(此刻已经是 running) * @param {object} [opts] * @returns {Promise} */ async function runClaimedNode(runId, run, def, defNode, nodeName, picked, opts = {}) { 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(picked.id), executionMode: run.execution_mode, }; // W2:产出位置必须在**发起调用之前**就存在,且两种节点都要有——subgraph 节点起的是 // workgraph-planning,那同样是一次真 CLI 调用,同样会写盘。 const projectRoot = ensureProjectRoot(ctx); ctx.projectRoot = projectRoot ? projectRoot.root_path : null; // W9:失败上下文(哪个节点 / 第几 attempt / 失败摘要)进提示词。 // 两种节点都拼:agent_call 的 ctx.task 直接就是提示词正文;subgraph 的 ctx.task 被 // buildPlanningPrompt 原样放进「## ② 任务描述」,规划本体重规划时同样需要知道上一轮死在哪。 // 没有前序失败时返回空串,不塞空段落(见 describeFailureContext)。 ctx.task += describeFailureContext(ctx); // 产物目录必须**进提示词**:读的一侧(readNodeProduct)按 §四.A.5 的路径去读, // 写的一侧不告诉 agent 往哪写、文件名怎么起,这份文件在生产里就根本不会被写出来, // 于是 fail 边 / 回退 / infra_flake 全都触发不了(R3-F1 缺陷 3a)。 // 项目根同理,且比产物目录更早断:不写进提示词,agent 就会写到它能看见的最近的仓库。 if (ctx.nodeKind !== 'subgraph') ctx.task += describeArtifactContract(ctx); // R-19:把**上游产物的真实路径**递到节点手上。 // // 这是第四次真跑那个断点的正解。当时 prd 节点的归因原话是「不是上游没产出, // 是它的产出**结构上到不了我手上**」—— 边只承载控制流,上游的 brd.md 躺在它自己的 // 私有目录里,下游的提示词里没有任何可读路径,只能靠硬拼,拼不出来。 // // 刻意**不是**「注入一句『索引在某个文件里,自己去检索』」:那仍然依赖 agent 自觉去查、 // 自觉拼对路径,而真跑已经证明它拼不出来。引擎直接把路径列出来,不给它猜的机会。 ctx.task += describeInputs(ctx, def); // 35 号文档提案 A:本体完整性闸门。此前 `missingBodyDefinition` 只在招聘/建公司 // 路径上被查过(graph-service.js / graph-seed-service.js / company-service.js), // 图执行的核心路径(这里)从未查过——本体因任何原因(装配没跑完、手工改坏、 // 迁移遗留……)仍是空骨架时,引擎会原样把它当成正常本体,派发一次真实 CLI 调用, // 静默产出一份「像是完成了、其实没有任何角色指导」的结果(真实样本见 34 号文档 §五: // `unit-tester` 全程顶着 `_base` 骨架跑完了一整条真实任务)。 // // 只查 `agent_call` 节点:subgraph 节点派发的是 workgraph-planning 去规划子图, // 它的 `systemAgentId` 不是「即将被真实调用的本体」,不是这里要拦的场景。 // // 且只在**真会起 CLI 的执行器**下查(与 `boundaryGuardApplicable` 同一个判据): // 测试注入的假 executor 是进程内函数,压根不会读任何 agent.md,用的 `systemAgentId` // 也大多是 `w-prd`/`impl` 这类不对应仓库里任何真实目录的夹具名——扫真实文件系统 // 对这些用例既没意义,又会把全仓库几百个注入假 executor 的用例一次性判成 // body_not_ready(本改动第一版就是这么把测试套件炸穿的,改用这条判据后收窄回来)。 const isRealDispatch = ctx.nodeKind !== 'subgraph' && !!ctx.systemAgentId && agentExecutor === defaultAgentExecutor; const missingBody = isRealDispatch ? locator.missingBodyDefinition(ctx.systemAgentId) : []; let outcome; if (missingBody.length) { // 没有真的发起调用:不扣预算(「一次调用一定发生了」这条记账理由在这里不成立 // ——这次压根没有调用发生),不起边界守卫(没有子进程会去动宿主仓库)。 outcome = { status: 'fail', failReason: ENGINE_FAIL_REASONS.BODY_NOT_READY, output: `本体 ${ctx.systemAgentId} 仍缺角色定义(${missingBody.join(', ')}),本节点未派发真实调用`, }; timelineNote(ctx, 'node.boundary', `本体 ${ctx.systemAgentId} 尚未装配完成(缺 ${missingBody.join(', ')}),节点直接判 fail,不派发真实调用`, { node_run_id: ctx.nodeRunId, node_name: nodeName, system_agent_id: ctx.systemAgentId, missing: missingBody, degraded: 'body_not_ready', }); } else { // 抢到节点即记账:一次调用一定发生了(哪怕它随后炸掉),预算「只增不减」(§四.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 }, ); // W3:调用前后各拍一张宿主仓库快照。能拦什么拦不了什么见 applyBoundaryGuard 的注释。 const guardSnapshot = beginBoundaryGuard(ctx); try { outcome = ctx.nodeKind === 'subgraph' ? await executeSubgraph(ctx) : await agentExecutor(ctx); } catch (err) { // executor 自己炸了(规划失败 / 配置缺失等)算节点失败。 // W12 补完:fail_reason 落 `node_error`(不再是 null)。选边行为与 null 完全一致 // ——没有任何图骨架写这条边,`selectFailEdge` 精确匹配不到就退兜底边—— // 但概览 tab 从「失败、原因 —」变成「失败、node_error」,操作者第一眼就知道 // 这是**引擎/配置**层面炸的,不是 agent 判的业务失败。取值论证见 ENGINE_FAIL_REASONS。 // err.tokens:规划连着跑了几轮才放弃时,那几轮的 token 是真花了的,照扣不误(§四.A.3「只增不减」)。 outcome = { status: 'fail', output: err.message, failReason: ENGINE_FAIL_REASONS.NODE_ERROR, tokens: err.tokens || 0, }; // W12:这条路径以前是**净损失** —— err.message 写进 outcome.output 之后没有任何一环消费它: // 不落库(graph_run_nodes 没有 output 列,且不该有,见 EVENT_TEXT_MAX 的注释)、 // 不进 SSE、也不写 timeline,fail_reason 还是 null。操作者看到的是「节点 fail、无任何原因」。 // 真跑第一个 work 节点时 AgentManager 抛的「Agent not found」正是从这里出去的。 // 现在原因先进审计轨(timeline.ndjson,永不改写)+ 实时轨(stream.log / SSE), // 排障至少有的可查;错误类型与调用目标一并记,好判断是「本体没雇进公司」还是别的。 timelineNote(ctx, 'node.error', `节点执行器抛异常(${(err && err.name) || 'Error'}):${err && err.message}`, { node_run_id: ctx.nodeRunId, node_name: nodeName, attempt: ctx.attempt, node_kind: ctx.nodeKind, system_agent_id: ctx.systemAgentId || null, error: clipText(err && err.message ? err.message : String(err)), error_type: (err && err.name) || 'Error', // 栈顶几行足够定位抛点(整份栈进事件没意义,也会把 stream.log 撑大) error_stack: clipText(err && err.stack ? String(err.stack).split('\n').slice(0, 6).join('\n') : null), }); } // 炸掉的那条路径同样要查:agent 写完仓库再崩,痕迹一样在。 outcome = applyBoundaryGuard(ctx, guardSnapshot, outcome); } // executor 报了 token 用量就一并扣。产生端见 defaultAgentExecutor 的 onAcpEvent // (claude-cli 协议的 session/result.usage);不上报 usage 的协议这一维恒为 0, // 届时仍由 total_agent_calls(上面那行)与 wallclock_ms(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, // W12:outcome.output 有三个产生方(defaultAgentExecutor 的 res.output、 // applyBoundaryGuard 的越界 banner、上面 catch 的 err.message),以前**一个消费方都没有**。 // 这里是它唯一的出口:定长摘要进事件(前端 NodeRunLog 已经在按 node_run_id 消费 // node.output 这条事件),完整输出仍然只在 output_ref 指的产物目录里。 output_excerpt: clipText(outcome && outcome.output), }, { 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, }; } const settled = settleNode(runId, picked.id, outcome); // 36 号文档 §三.3:图执行的验收反馈此前从不进 agent 自己的 memory.md(只有人类对话路径 // agent-chat.js 会写),Dream 天天扫全是空转。这里补上唯一的桥接点——只在真派发、且这个 // systemAgentId 真的是已雇佣的本体(bodyDir 存在)时才记,避免测试夹具/假 executor 用例 // 把假地址写进真实 system/agents/ 目录(isRealDispatch 与 missingBody 同一条判据, // bodyDir 存在性检查是多一层防线)。 if (isRealDispatch && settled && settled.node && fs.existsSync(locator.bodyDir(ctx.systemAgentId))) { recordNodeOutcomeMemory(ctx, settled.node); } return settled; } /** * 把节点这次运行的结算结果记一行进它自己的 memory.md——36 号文档提案的桥接点。 * 只记摘要(成功/失败+原因),不是提炼;提炼是 Dream 的活,这里只负责「让原料存在」。 */ function recordNodeOutcomeMemory(ctx, nodeSnapshot) { try { const addr = resolveAgentName(ctx.companyId, ctx.systemAgentId); const task = `${ctx.nodeName}(第 ${ctx.attempt + 1} 次运行)`; const result = nodeSnapshot.status === 'success' ? 'success' : `fail${nodeSnapshot.failReason ? `: ${nodeSnapshot.failReason}` : ''}`; memoryService.appendToMemoryMd(addr, task, result, { stats: { toolCallCount: 1 } }); } catch (err) { // 记忆写入是旁路信号,失败不该反过来影响图执行本身 console.error('[graph-engine] 节点结果写入 memory.md 失败:', err.message); } } /** * subgraph 节点(org 层)结算时,把「这次规划出的图最终跑得怎么样」记进 * `workgraph-planning` 自己的 memory.md——36 号文档追加发现:它的图形状裁剪判据 * 现在只有"允许裁剪"没有"具体判据"(真实撞见:单页静态网站任务原样照抄 * planner.json 全部 5 节点,一个都没砍),而它自己完全没有可积累的历史——这是补上 * 这个原料通道的桥接点,跟 recordNodeOutcomeMemory 记 agent_call 节点是同一个精神。 * * 不能直接复用 recordNodeOutcomeMemory:subgraph 节点的 `ctx.systemAgentId` 是**宿主 * org node 自己的身份**(如 "org-coder"),不是真正执行规划的那个 agent——规划本体 * 全局唯一,恒为 `PLANNING_AGENT_ID`(R-15.2)。 * * 只在真实规划(`planWorkGraph === defaultPlanWorkGraph`)时记:测试夹具几乎总会注入 * 假 planWorkGraph(defaultPlanWorkGraph 会真的起 CLI 调 workgraph-planning),这道闸门 * 挡住的正是"把假地址写进真实 system/agents/workgraph-planning/ 目录"这类污染 * (与 runClaimedNode 里 agentExecutor === defaultAgentExecutor 同一个判据)。 */ function recordWorkgraphPlanningMemory(companyId, nodeName, status, failReason) { if (planWorkGraph !== defaultPlanWorkGraph) return; try { const addr = resolveAgentName(companyId, PLANNING_AGENT_ID); const task = `为 ${nodeName} 规划的 work graph`; const result = status === 'success' ? 'success' : `fail${failReason ? `: ${failReason}` : ''}`; memoryService.appendToMemoryMd(addr, task, result, { stats: { toolCallCount: 1 } }); } catch (err) { console.error('[graph-engine] workgraph-planning 结果写入 memory.md 失败:', err.message); } } /** * 批量推进(27-P4 T4/T6):一次拾取至多 `maxConcurrency` 个 ready 节点,并发执行。 * * `maxConcurrency <= 1` 时**直接委托给 `advanceOne`**,不走下面的批量路径——零行为差异, * `advanceOne` 现有的调用方/测试不用改一个字(见 27-P4 §4.1 决策:不是「advanceOne 包 advanceBatch」, * 是反过来,这样 advanceOne 的行为永远只由它自己的测试背书,不会被批量路径的改动间接影响)。 * * `maxConcurrency > 1` 时: * 1. 先做一遍与 `advanceOne` 完全相同的 run 级检查(stop/pause/running/预算)—— * 这段刻意重复一份而不是抽共享函数,因为 `advanceOne` 是被大量测试钉住的稳定代码, * 抽共享会让这次改动的 diff 面盖到它头上,得不偿失。 * 2. `SELECT ... LIMIT maxConcurrency` 取候选,逐个走 `sse.transitionNode('ready'→'running')` * —— 单行条件更新本身就是并发闸门,不需要包裹事务(§4.1 决策的完整论证)。 * 3. 拾到的节点各自 `runClaimedNode`,`Promise.allSettled` 并发执行——一个节点的 * executor 抛出去不会波及同批其它节点(`runClaimedNode` 内部已经把 agent 调用异常 * 转成 fail outcome,这里的 catch 是防它自己有 bug 的最后一道网)。 * * @param {string} runId * @param {object} [opts] * @param {number} [opts.maxConcurrency] 不传则读 `run.max_concurrency`,再不然用 DEFAULT_MAX_CONCURRENCY * @param {boolean} [opts.ignorePause] * @param {string} [opts.task] * @returns {Promise<{ok: boolean, runId: string, runStatus: string|null, reason: string|null, * results: AdvanceResult[], activated: Array}>} */ async function advanceBatch(runId, opts = {}) { const run0 = getRun(runId); const maxConcurrency = opts.maxConcurrency || (run0 && run0.max_concurrency) || DEFAULT_MAX_CONCURRENCY; if (maxConcurrency <= 1) { const res = await advanceOne(runId, opts); return { ...res, results: res.node ? [res] : [] }; } const run = run0; if (!run) return { ok: false, runId, runStatus: null, reason: 'run-not-found', results: [], activated: [] }; if (run.stop_requested) { return { ok: false, runId, runStatus: run.run_status, reason: 'stop-requested', results: [], activated: [] }; } if (run.pause_requested && !opts.ignorePause) { landPause(runId); return { ok: false, runId, runStatus: (getRun(runId) || {}).run_status, reason: 'pause-requested', results: [], activated: [], }; } if (run.run_status !== 'running') { return { ok: false, runId, runStatus: run.run_status, reason: 'run-not-running', results: [], activated: [] }; } // 候选先查、budget/billing 后查——次序必须跟 advanceOne 一致(其注释原话:「拾到了 ready // 节点(=真要花钱了)才查,避免空转时把已跑完的 run 误挂起」)。审 T5 时发现这里以前是反的 // (先查 budget 再查候选):一个已经没有 ready 节点、只是恰好也把预算花完了的收敛 run, // 会被误判成「预算耗尽而挂起」而不是正确收敛到 completed。顺手连着billing gate一起接进来, // 两处判定共用同一个「先确认真有工作要做」的前提。 const def = defIndex(run); const candidates = getDb() .prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? AND status = 'ready' ORDER BY created_at, id LIMIT ?`) .all(runId, maxConcurrency); if (!candidates.length) { // ok:false 跟 advanceOne 的 result() 惯例对齐——「这一次调用没有推进任何节点」, // 不是「出错了」。runAuto 的循环靠 !res.ok 判断要不要停,这里如果写成 true, // 一个已经收敛的 run 会被 runAuto 一路空转到 maxSteps 才停(§4.3 决策审这段代码时发现, // 上线前修,不留到接 runAuto 那一步才炸)。 return { ok: false, runId, runStatus: settleRun(runId), reason: 'no-ready-node', results: [], activated: [] }; } const gate = checkBudget(run); if (!gate.ok) { suspendRunTree(run); return { ok: false, runId, runStatus: (getRun(runId) || {}).run_status, reason: 'budget-exhausted', results: [], activated: [], budget: gate, }; } // 27-P4 T5:账户欠费门。billingGateFn 未注入时 checkBilling 恒返回 ok:true(桩子默认「未欠费」, // 真实计费接口接通之前不影响任何行为)。放在 checkBudget 之后同一个理由:真要花钱才查。 const billing = await checkBilling(run); if (!billing.ok) { suspendRunTree(run, { reason: 'billing-arrears' }); return { ok: false, runId, runStatus: (getRun(runId) || {}).run_status, reason: 'billing-arrears', results: [], activated: [], billing, }; } const claimed = []; for (const picked of candidates) { const defNode = def.nodes.get(picked.def_node_id) || {}; const nodeName = defNode.name || picked.def_node_id; // 逐行条件更新——两个并发 advanceBatch 即使 SELECT 到同一批候选,这里也只会有 // 一边拿到 changes=1(§4.1 决策),拿不到的直接跳过,不算错误。 if ( sse.transitionNode(runId, picked.id, 'ready', 'running', { startedAt: now() }, { node_name: nodeName, node_kind: defNode.nodeKind || 'agent_call', attempt: picked.attempt, }) ) { claimed.push({ picked, defNode, nodeName }); } } if (!claimed.length) { // 候选全被别的 advanceBatch/advanceOne 抢走:不是出错,但同样是「这一次没推进任何东西」, // ok:false 理由同上——taken-by-other 在 advanceOne 里也是 ok:false(见其 result() 用法)。 return { ok: false, runId, runStatus: run.run_status, reason: 'taken-by-other', results: [], activated: [] }; } const settled = await Promise.allSettled( claimed.map(({ picked, defNode, nodeName }) => runClaimedNode(runId, run, def, defNode, nodeName, picked, opts)) ); const results = settled.map((s, i) => { if (s.status === 'fulfilled') return s.value; // 防的是 runClaimedNode 自己的 bug(它内部的 try/catch 已经把 agent 调用异常转成 // fail outcome,正常情况下不会走到这里)。节点此刻卡在 running——不代它结算, // 留给人工「重试」或下一次真跑复现;这里只保证同批其它节点不受影响。 const { picked } = claimed[i]; return { ...result(runId, run.run_status, 'node-executor-rejected'), node: nodeView(runId, getRunNode(picked.id)), error: s.reason && s.reason.message, }; }); return { ok: true, runId, runStatus: settleRun(runId), reason: null, results, activated: results.flatMap((r) => r.activated || []), }; } /** * 节点执行结束后的落地:写终态 → 评估出边 → 收敛判定。 * 也是子 run 收敛时回写父节点走的同一条路(propagateToParent)。 * * @param {object} outcome {status, failReason?, remediationNote?, outputRef?, agentRunId?, childRunId?} */ function settleNode(runId, nodeRunId, outcome = {}) { let to = outcome.status === 'success' ? 'success' : 'fail'; let failReason = outcome.failReason === undefined ? null : outcome.failReason; // T-mt25ecbd:executor 报 success 不等于契约兑现了 —— agent 可能压根没写声明过的文件 // (brd/prd/sdd 三个 planner 节点复现过)。产出闸门放在这里、卡在 transitionNode 落库 // 之前,这样一旦判 fail 就走下面 to==='fail' 的 routeFailure 分支,不会先点亮下游 // 再回头发现产物不存在。 if (to === 'success') { const run0 = getRun(runId); const nodeRun0 = getRunNode(nodeRunId); const def0 = run0 ? defIndex(run0) : null; // subgraph 节点(org 层)没有自己的产物目录——它的「产出」是子 work run 里的节点登记的 // (org-layer-contracts.test.js ⑤ 的口径:output_ref 恒为 null)。这两道闸门只对 // agent_call 节点生效,subgraph 节点的契约兑现与否由子 run 自己的节点去过这两道闸门。 const isSubgraphNode = !!(def0 && def0.nodes.get(nodeRun0 && nodeRun0.def_node_id) && def0.nodes.get(nodeRun0.def_node_id).nodeKind === 'subgraph'); if (run0 && nodeRun0 && def0 && !isSubgraphNode) { const missingProduces = unsatisfiedRequiredProduces(run0, def0, nodeRun0); if (missingProduces.length) { to = 'fail'; failReason = ENGINE_FAIL_REASONS.PRODUCES_MISSING; timelineNote({ runId, nodeRunId }, 'node.output', `节点声明会产出 ${missingProduces.join(', ')},但产物目录里没找到对应文件,按 fail 结算`, { node_run_id: nodeRunId, missing_kinds: missingProduces, degraded: 'produces_missing' }); } else { // T-mt25ecbd 症状②:goal-benchmark 的入边 needs:[brd,prd,sdd] 以前只用来拼 // describeInputs 那句 ⚠️ 提示词——agent 完全可能对它视而不见、照样判自己成功 // (复现原样:三份文档一份没写出来,benchmark 仍产出合法 acceptance)。 // 这里在真正判 success 之前把同一份 needs 契约当真:声明了 required 却在 // graph_files 里连一份实例都查不到,就不认这次 success,按 fail 结算。 const missingInputs = requiredInputsMissing( { run: run0, companyId: run0.company_id, defNodeId: nodeRun0.def_node_id, nodeKind: 'agent_call' }, def0 ); if (missingInputs.length) { to = 'fail'; failReason = ENGINE_FAIL_REASONS.MISSING_INPUT; timelineNote({ runId, nodeRunId }, 'node.output', `节点声明必需的上游产物 ${missingInputs.join(', ')} 在索引里查不到实例,按 fail 结算`, { node_run_id: nodeRunId, missing_kinds: missingInputs, degraded: 'missing_required_input' }); } } } } const changed = sse.transitionNode(runId, nodeRunId, 'running', to, { failReason, // 跨层地址与 fail_reason 同一次条件更新落库(R-17.1)。恒写显式值:success 或本轮没归因 // 就写 null,不给「读到上一轮地址」留窗口。normalize 让执行器可以直接给对象或 JSON 串。 failNodeRef: normalizeNodeRef(outcome.failNodeRef), 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, crossLayer: null, // reset(backward 边打回 pending 的下游)与 infraRerun(原地重跑)在 handleBackwardEdge / // routeFailure 里赋值,先在这里声明,免得返回对象里出现「有时有、有时没有」的字段(minor 4) reset: [], infraRerun: false, }; // 36 号文档:subgraph 节点结算 = 「这次规划出的图最终跑得怎么样」这个信号第一次齐全的时刻 // (outcome 是 deriveChildOutcome 派生的,代表子 work graph run 的真实收敛结果)。 const settledDefNode = def.nodes.get(nodeRun.def_node_id); if (settledDefNode && settledDefNode.nodeKind === 'subgraph') { recordWorkgraphPlanningMemory(run.company_id, settledDefNode.name || nodeRun.def_node_id, to, failReason); } if (to === 'success') { // R-19 产出侧:把这次落盘的文件登记进产物索引,下游节点靠它拿到真实路径。 // 只在 success 时记 —— 失败节点的半成品不该被下游当输入读。 indexNodeOutputs(run, def, nodeRun); // 人工「重试」留下的契约快照在这里判:契约变了才级联下游(§四.C 末段) out.contractDiff = applyRetryCascade(runId, def, nodeRun); activateSuccessTargets(runId, def, nodeRun.def_node_id, out, outcome.remediationNote || null); } 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, // minor 4:这两个之前算出来了却没进返回值 —— 重试级联的重置靠 contractDiff.reset 上报了, // backward 边的重置和 infra_flake 原地重跑却只有 SSE 看得到,REST/advance 的调用方拿不到。 reset: out.reset, infraRerun: out.infraRerun, contractDiff: out.contractDiff, // R-17.1:本次回退到底是「定点重置了子图里的某个 node」还是「整张图重跑」—— // 算出来了就必须上报,否则控制台只能看见「退了一次」,看不见退到了哪一层(同 minor 4 的教训) crossLayer: out.crossLayer, childRunId: nodeRun.child_run_id || null, }; } // ──────────────────────────────────────────────────────────────── // §四.D.1 主流程:success 后的出边评估 // ──────────────────────────────────────────────────────────────── /** * 遍历 condition='success' 的出边,目标节点若**所有**入边条件都满足则置 ready。 * 「全部满足」= 该目标所有 forward + success 的入边,其源节点都已 success(汇合点语义)。 * fail 边和 backward 边不参与这道闸——它们是异常流程,各自单独把目标点亮。 */ function activateSuccessTargets(runId, def, defNodeId, out, remediationNote = null) { for (const e of def.edges) { if (e.from !== defNodeId || e.condition !== 'success') continue; if (!allIncomingSatisfied(runId, def, e.to)) continue; // 首轮进场(verdict=no_prior_failure)也可能留了话给下一轮,成功路径同样带过去 —— // defaultAgentExecutor 那句「成功路径同样带回」以前是个假承诺,现在它是真的。 const activated = markReady(runId, def, e.to, remediationNote); 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。 * remediationNote 跟着这一次条件更新落到收件人那一行(见上方「判词的载体」一节)。 */ function markReady(runId, def, defNodeId, remediationNote = null) { 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', notePatch(remediationNote), { 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; // 30 号文档 §四:这是唯一没有熔断线的失败路径(attempt/backward_count 都故意不烧), // 真撞过一次确定性死循环——连续 N 次判 infra_flake 就不再当"基础设施抖动",改判失败, // 交给人工(不硬猜该走哪条 fail 边:infra_flake 这个取值域通常没人给它连边,直接 // blockNode 比"碰巧选中一条不相关的边"更可预期)。 const flakeCount = bumpInfraFlakeCount(nodeRun.id); if (flakeCount > INFRA_FLAKE_HARD_CAP) { sse.emitEvent(runId, 'loop.terminated', `熔断:节点 ${name} 连续 ${flakeCount} 次被判 infra_flake,不像基础设施抖动,改为人工介入`, { node_run_id: nodeRun.id, node_name: name, infra_flake_count: flakeCount, cap: INFRA_FLAKE_HARD_CAP, verdict: 'force_terminated' }, { ref: nodeRun.id }); blockNode(runId, nodeRun.id, remediationNote || `连续 ${flakeCount} 次被判 infra_flake(超过硬顶 ${INFRA_FLAKE_HARD_CAP}),已停止原地重跑`); return; } // 原地重跑 = 一轮新的执行,但它**不加 attempt**(§四.D.2.1 末段),所以 R-17.1 的代次判据 // 在这条路径上分辨不出新旧轮 —— 因此 child_run_id 与 child_run_attempt 必须一起清空, // 让 executeSubgraph 走「没有子 run」那条分支重新规划。少清 child_run_attempt 那一半, // 守卫会读到「代次 == 本轮 attempt」+ 终态子 run,虽然不再死锁(就地结算)但会把 // 上一轮的失败结果原样再判一遍,基础设施抖动就永远重跑不出来。 if ( sse.transitionNode(runId, nodeRun.id, 'fail', 'ready', { childRunId: null, childRunAttempt: null, failReason: null, failNodeRef: null, ...notePatch(remediationNote) }, { node_name: name, attempt: nodeRun.attempt, summary: `node ${name}: infra_flake 原地重跑(不烧计数,第 ${flakeCount}/${INFRA_FLAKE_HARD_CAP} 次)` }) ) { 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,等人工介入(区别于人工主动「停止」)。 // 判词的收件人是**它自己**:人工「重试」把它打回 ready 之后(retryNode 不动这一列), // 下一次 advanceOne 就会从它自己那一行把判词读进提示词。 if (!edge) return blockNode(runId, nodeRun.id, remediationNote); 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, remediationNote); return; } // forward 的 fail 边 = 普通失败分支(如「失败转人工审核节点」),直接点亮目标。 // 判词跟着边走:escalate / 人工审核节点正是最需要「该怎么修」这段文本的那个节点。 const activated = markReady(runId, def, edge.to, remediationNote); 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; // R-17.1 跨层寻址:先**只解析不动手**(纯查询),因为解析结果要写进第 ① 步的备案里。 // 解析不出来(没地址 / 图上没这个节点 / org node 对不上)返回 null,第 ③ 步就退化成 // 「整张 work graph 重跑」——与返工前同语义,账本上 target_node_ref 为空即是这一情形的证据。 const cross = resolveCrossLayerTarget(runId, def, nodeRun, edge, target); // ⓪ 整棵树已经越顶:不再登记新记录,直接收敛。 // 子层每重跑一轮就是一张新子 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, remediationNote); return true; } // ① 备案:两个计数一起加,谁都不许跳过 const rec = insertLoopRecord({ runId, edgeId: edge.id, fromNodeRunId: nodeRun.id, toNodeRunId: target.id, verdict: 'registered', remediationNote, targetNodeRef: cross ? cross.ref : null, }); 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}` + (cross ? `(定点重置 ${cross.workNodeName})` : '(整张图重跑)'), { 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, target_node_ref: cross ? cross.ref : null, }, { 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, remediationNote); return true; } // ③ 回退。先把目标的 forward 下游打回 pending,再(跨层)定点重置,最后点亮目标本身。 out.reset = resetForwardDownstream(runId, def, edge.to); // 定点重置必须**先于**点亮父节点:顺序反了而重置又失败,父节点就会带着一张终态子 run // 被点亮 —— 那正是第二轮那个死锁的形状。先重置、再点亮,重置不成也只是退化成整张图重跑。 out.crossLayer = cross ? applyCrossLayerReset(runId, cross, remediationNote) : null; 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 = 新一轮。**child_run_id 不再清空**(R-17.1「复用旧子 run」): // 换代靠 attempt+1 让 child_run_attempt 对不上,executeSubgraph 据此复用同一张子 run // 而不是重新规划一张。清空反而会丢掉「只重置被点名的 node」的全部前提。 if ( sse.transitionNode(runId, target.id, target.status, 'ready', { failReason: null, failNodeRef: null, ...notePatch(remediationNote) }, { node_name: name, attempt: target.attempt + 1, backward_edge_id: edge.id, cross_layer: out.crossLayer, summary: `node ${name}: 沿回退边 ${edge.id} 重新点亮(attempt #${target.attempt + 2}` + `${out.crossLayer ? `,定点重置 ${out.crossLayer.workNodeName}` : ''})` }) ) { bumpNodeAttempt(target.id); out.activated.push({ id: target.id, defNodeId: edge.to, name }); } } return true; } /** * 解析跨层地址 → 「该退到哪个子 run 的哪一行」(**纯查询,不改任何状态**)。 * * 输入的地址来自失败节点行的 `fail_node_ref`(三值透传的第四值,见 normalizeNodeRef 那段)。 * 四道关,任何一道不过都返回 null = 退化成整张 work graph 重跑(不是报错,也不是死路): * ① 有地址吗 ── 归因没点名具体 node(老图 / 老 agent)时就是没有 * ② 回退目标是 subgraph 节点吗 ── 目标本身就是 agent_call 时,它自己就是最细粒度,没有「跨层」可言 * ③ 目标手上有可复用的子 run 吗 ── 没起过 / 被人工停止过 → 只能整张图重来 * ④ 地址与边指向同一个 org node、且那张子 run 里真有这个节点名吗 * —— org node 对不上时**以边为准**(边是图上真实存在的路由,地址只是归因的主张), * 不做定点重置并记一条 timeline;对不上还照做,等于让一句 LLM 输出改写图的拓扑。 * * @returns {{childRunId, nodeRunId, workNodeName, orgNodeName, ref}|null} */ function resolveCrossLayerTarget(runId, def, nodeRun, edge, target) { const ref = parseNodeRef(nodeRun && nodeRun.fail_node_ref); if (!ref) return null; const d = def.nodes.get(edge.to) || {}; const orgNodeName = d.name || edge.to; if ((d.nodeKind || 'agent_call') !== 'subgraph') { timelineNote({ runId, nodeRunId: target.id }, 'node.triage', `归因点名了 ${ref.workNode},但回退目标 ${orgNodeName} 不是 subgraph 节点,忽略跨层地址`, { node_run_id: target.id, target_node: ref.workNode, degraded: 'target_not_subgraph', }); return null; } if (ref.orgNode && ref.orgNode !== orgNodeName) { // 主张与路由打架:边赢。静默照做的话,一次归因就能把回退落到图上另一条路径上去。 timelineNote({ runId, nodeRunId: target.id }, 'node.triage', `归因声称的 org node(${ref.orgNode})与回退边的目标(${orgNodeName})不一致,忽略跨层地址、整张图重跑`, { node_run_id: target.id, claimed_org_node: ref.orgNode, edge_target: orgNodeName, degraded: 'org_node_mismatch', }); return null; } const childRun = target.child_run_id ? getRun(target.child_run_id) : null; if (!childRun || childRun.run_status === 'stopped') return null; const childDef = defIndex(childRun); const hit = [...childDef.nodes.values()].find((n) => n.name === ref.workNode); if (!hit) { timelineNote({ runId, nodeRunId: target.id }, 'node.triage', `子 run ${childRun.id} 的图上没有名为 ${ref.workNode} 的节点,忽略跨层地址、整张图重跑`, { node_run_id: target.id, child_run_id: childRun.id, target_node: ref.workNode, degraded: 'work_node_not_found', }); return null; } const row = getRunNodeByDefId(childRun.id, hit.defNodeId); if (!row) return null; return { childRunId: childRun.id, nodeRunId: row.id, workNodeName: hit.name, orgNodeName, ref: JSON.stringify({ orgNode: orgNodeName, workNode: hit.name, childRunId: childRun.id, nodeRunId: row.id, }), }; } /** * 定点重置:把子 run 里被点名的那一行打回 ready,**其余节点一动不动**。 * * 直接复用人工「重试」那条路(retryNode)而不是另写一套 —— R-17.1 的「下游按 contract_hash * 四分类级联」正是 retryNode → snapshotContract → applyRetryCascade 这条既有链: * 快照现在拍,等这个节点重跑成功时 applyRetryCascade 才比 hash,unchanged 的下游一步都不重跑。 * 另写一套的话这条判据在跨层路径上就是死的(「实现在、零调用方」的第五次)。 * * `reopenRun: false` 是本路径唯一的差别:子 run 的 run_status 留给 reopenChildRun 在 * 父节点真正被拾取时才拉起来,免得子图抢在上游重做完之前先跑(见 reopenChildRun 的说明)。 */ function applyCrossLayerReset(runId, cross, remediationNote = null) { const res = retryNode(cross.childRunId, cross.nodeRunId, { reopenRun: false, remediationNote, }); if (!res.ok) { // 目标节点正在跑 / 已被别人重置:不重复动手,本轮退化成整张图重跑(reopenChildRun 会全量重置) timelineNote({ runId, nodeRunId: cross.nodeRunId }, 'node.triage', `定点重置未生效(${res.reason}),本轮退化成整张 work graph 重跑`, { child_run_id: cross.childRunId, node_run_id: cross.nodeRunId, target_node: cross.workNodeName, degraded: 'targeted_reset_skipped', }); return null; } sse.emitEvent(runId, 'loop.cross_layer', `跨层定点重置:${cross.orgNodeName} / ${cross.workNodeName}(复用子 run ${cross.childRunId},只重跑这一个节点)`, { child_run_id: cross.childRunId, node_run_id: cross.nodeRunId, org_node: cross.orgNodeName, work_node: cross.workNodeName, contract_snapshot: res.contractSnapshot ? res.contractSnapshot.present : null, }, { ref: cross.nodeRunId }); return { ...cross, ok: true }; } /** * 外层熔断计数的**共池读**:整棵 run 树的总回退次数,记在 root run 那一行。 * * 为什么不按当前 run 读:R-17.1 之后 executeSubgraph 重跑优先**复用**旧子 run(reopenChildRun), * 但复用不了就仍会新建一张(backward_count 从 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; // remediation_note 清掉:这个节点下一轮会被重新点亮(届时由 markReady 重新写判词), // 留着上一轮那条会让它照着作废了的建议干活。fail_node_ref 同理。 // **child_run_id 不清**(R-17.1 改):subgraph 节点下一轮复用同一张子 run 的定义, // 换代靠下面那句 bumpNodeAttempt 让 child_run_attempt 对不上;旧行为(清空 → 重新规划) // 每退一次就多烧一次 workgraph-planning 的真实 LLM 调用,且丢掉了「只重置被点名的 node」的前提。 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, fail_node_ref = NULL, remediation_note = 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; } /** * fail → blocked。remediationNote 一并写在**这个节点自己**那一行:blocked 的唯一出路是人工 * 「重试」,重试后重跑的还是它,所以判词的收件人就是它自己(R4-F1 major 1 的第二个入口)。 * 死路、回退目标失踪、撞熔断硬顶三条路径一视同仁,不做特例。 */ function blockNode(runId, nodeRunId, remediationNote = null) { sse.transitionNode(runId, nodeRunId, 'fail', 'blocked', notePatch(remediationNote), { summary: 'fail → blocked:没有可走的 fail 边或已撞熔断硬顶,等人工介入', }); } /** * ── 判词(remediation_note)的载体 ────────────────────────────────── R4-F1 major 1 * * 修之前:readNodeProduct 产生它、settleNode 传递它,但**唯一的落库点是 * handleBackwardEdge 里的 insertLoopRecord**。于是走 forward fail 边(escalate / 人工审核 * 节点)、死路 blocked、infra_flake 原地重跑这三条路径时,判词只进 timeline.ndjson —— * 那是纯审计轨,没有任何代码读回它。「产生 OK → 传递 OK → 落库 MISSING → 读回 MISSING」, * 与前面已修的四处同型。最需要这段文本的恰恰是 escalate 节点:它的提示词里一个字都没有。 * * 载体定在 **graph_run_nodes.remediation_note**(work-graph-db 已备好列 + NODE_PATCH_COLUMNS * 白名单 + getRunState 读回),语义是「留给**这个节点**下一轮的修复建议」,所以: * - 判词**写在收件人那一行**,不写在判它失败的那一行; * - 写入时机 = 收件人被点亮的那次状态转换,作为 patch 跟着 updateNodeStatus 的**同一次 * 条件更新**落库,不另开 setter —— 「状态改了但判词没落」这条岔路根本不存在; * - 每一次点亮都显式写值(没有判词就写 null),所以不会留下上一轮的陈旧判词 * (R3-F1 缺陷 4 那个「被旧判词反复判失败」的坑,在这条链上没有第二个入口)。 * * 不用 graph_run_loop_records 当载体:那是熔断账本,只在命中 backward 边时才有行; * 借它存判词等于给「回退发生了」造假记录。它仍然照旧记 backward 那一份(跨 run 传播、 * 控制台循环徽标要用),两者不冲突:账本记「退过几次」,节点列记「下一轮该看什么」。 */ /** 收件人被点亮时把判词一起写上;note 恒为显式值(null = 清掉上一轮的) */ function notePatch(remediationNote) { return { remediationNote: remediationNote || null }; } // ──────────────────────────────────────────────────────────────── // 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) return; if (parentNode.status === 'running') { // remediationNote 不是 graph_run_nodes 的列,随 outcome 一起交给父层的选边/回退用 settleNode(child.parent_run_id, child.parent_node_run_id, deriveChildOutcome(child)); return; } reviveParentFromChild(child, parentNode); } /** * 迟到的好消息:父节点已经因这个子 run 的**上一次**失败结算成 fail/blocked, * 而子 run 后来被人从下一层救回来跑到了 completed。 * * 原来这里的幂等守卫是 `parentNode.status !== 'running' → return`,判据用错了维度: * 它想防的是「同一个结果被结算两次」,实际拦掉的是「子 run 换了个结果」。 * 界面走查实测的后果是**任务永久卡死** —— 子 run 已 completed、父节点钉在 blocked、 * root run 跟着钉住,org 层再也推不下去;唯一人工出路是对 org 节点点「重试」, * 而那会把整张子图重置到下一个 attempt,**刚跑完的成果全丢**。 * * subgraph 节点的状态本来就是**从子 run 派生**的,子 run 换了终态就该重新派生一次。 * * 只处理「转好」(fail/blocked → success),不处理「转坏」:子 run 从 completed 掉回 * blocked 只可能是上层回退把它重新拉起又失败了,那条路径本来就会重走 running → 正常结算。 * * `blocked`/`fail` 的合法出路只有 `ready`(NODE_TRANSITIONS),所以这里走 * **blocked → ready → running → success 三步合法转换**,而不是放宽转换表 —— * 转换表是给所有调用方的安全网,不该为一条恢复路径把 `blocked → success` 对全局开放。 * 三条 node.status 事件也让「这个节点是被救回来的」在 timeline 上看得见。 */ function reviveParentFromChild(child, parentNode) { if (parentNode.status !== 'fail' && parentNode.status !== 'blocked') return; if (parentNode.child_run_id !== child.id) return; // 不是它在等的那个子 run if (child.run_status !== 'completed') return; // 只认「转好」 const runId = child.parent_run_id; const nodeRunId = child.parent_node_run_id; const nodeName = (defIndex(getRun(runId)).nodes.get(parentNode.def_node_id) || {}).name || nodeRunId; const why = `子 run ${child.id} 已收敛为 completed,父节点从 ${parentNode.status} 重新派生`; // 父 run 多半也跟着钉在 blocked(唯一的 blocked 节点就是它)。settleRun 在转换非法时是 // **静默跳过**的,而 RUN_TRANSITIONS 里没有 blocked → completed,所以不先把 run 拉回 // running 的话:节点复活成 success 了,run 还留在 blocked,org 层照样推不下去 —— // 这一条是补测试时才暴露出来的,光复活节点并不够。 const parentRun = getRun(runId); if (parentRun && parentRun.run_status === 'blocked') { // 坑满在这条路上**不许静默**(批 3 复核 F1 场景 C):本函数是被 settleRun → propagateToParent // 顺带调起来的,而那一层挂在救援方那条 run 的 `runAuto(...).catch` 下面 —— 抛出去只会被 // 吞成一行 console,父节点与父 run 双双钉死在 blocked,控制台上什么都看不出来。 let outcome; try { outcome = admitOrRequeue(runId, 'blocked', 'running', { summary: `run ${runId}: blocked → running:${why}`, revived_from_child: child.id, }); } catch (err) { if (!(err instanceof RunSlotBusyError)) throw err; // 连队列都退不回去(这条 run 没有 task / task 已不是 blocked)。三处留痕,缺一处就是 // 「救不回来」这件事在界面上不可见:timeline+SSE 事件、进程日志、以及 task 行上那句 // 人在任务列表里直接能读到的原因。父节点保持 blocked 原样 —— 不留半截状态。 const note = `revive_slot_busy:坑被 ${err.heldBy.join(', ')} 占着,子 run ${child.id} 已完成但父 run 拉不回 running`; // 复用 `node.error` 而不是造一个新事件名:控制台的 EventSource 是**逐个具名订阅**的 // (node-signals.RUN_EVENT_KINDS),发一个没人 addEventListener 的名字等于把这条痕迹 // 发进真空 —— 那就又成了「注释承诺了可见性、实际看不见」。挂 ref=nodeRunId, // 于是它落在人真正会去看的那个卡住的节点的事件流里。 sse.emitEvent(runId, 'node.error', `救援受阻:${note}`, { error_type: 'RunSlotBusyError', error: err.message, held_by: err.heldBy, child_run_id: child.id, node_run_id: nodeRunId, company_id: err.companyId, }, { ref: nodeRunId }); if (parentRun.task_id) noteTaskBlockedReason(parentRun.task_id, runId, note); console.error(`[graph-engine] reviveParentFromChild 拿不到执行坑 run=${runId}: ${err.message}`); return; } // requeued:父 run 已连同 task 退回队列,节点照常复活 —— 调度器把它出队时链路接着往下走。 if (outcome === 'requeued') { console.warn(`[graph-engine] 父 run ${runId} 坑满,已退回队列继续救援(子 run ${child.id})`); } } // 每一步都是条件更新(WHERE status = from),任一步被别人抢走就整体放弃,不留半截状态 if (!sse.transitionNode(runId, nodeRunId, parentNode.status, 'ready', {}, { nodeName, summary: `${parentNode.status} → ready:${why}`, revived_from_child: child.id, })) return; if (!sse.transitionNode(runId, nodeRunId, 'ready', 'running', {}, { nodeName, summary: `ready → running:${why}`, revived_from_child: child.id, })) return; settleNode(runId, nodeRunId, deriveChildOutcome(child)); } /** 从子 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-triager 节点的产出(改名前叫 validator-reflection,两个名字都认, // 见 ATTRIBUTION_NODE_ALIASES);没有这个节点时退到最近一个失败/阻塞节点。 // R-17.1 已定案(原 R-16「归因保持 org node 粒度封装、不跨层寻址」的假设**作废**): // 父层除了「子图整体失败 + 根因」,还拿走同一行的 fail_node_ref,org 层回退可以直接 // 定点重置子图里的那一个 node(handleBackwardEdge → targetedReset)。 const attribution = rows.find((r) => { const d = def.nodes.get(r.def_node_id); return d && isAttributionNodeName(d.name) && r.fail_reason; }); const lastBad = rows.find((r) => (r.status === 'fail' || r.status === 'blocked') && r.fail_reason); // W12 补完:子 run 收敛为 fail,但子图里**一个归因都没有**(全是死路 blocked、 // 或失败节点的 fail_reason 全是 null)时,以前父节点拿到的是 null —— org 层控制台上 // 一个 subgraph 节点显示「失败、原因 —」,而失败明明发生在子图里。 // 现在署名 `child_run_failed`:它至少告诉操作者「原因要下钻到子 run 里看」。 // 注意顺序:真有归因(含子图里落的 node_error / boundary_escape)时一律优先透传, // 这个值只是**兜底**,不会盖掉任何更具体的根因。 const failReason = status === 'fail' ? (attribution || lastBad || {}).fail_reason || ENGINE_FAIL_REASONS.CHILD_RUN_FAILED : null; // R-17.1:跨层地址与 fail_reason 从**同一行**取,一起透传上去。分开取会出现 // 「根因来自归因节点、地址来自另一个失败节点」的错配,父层据此定点重置的就是错的 node。 const failNodeRef = status === 'fail' ? (attribution || lastBad || {}).fail_node_ref || null : null; // 判词优先取子图里回退备案上的那条(它是 validator 明确写给某个节点的), // 子 run 走死路 / forward fail 边收场时**一条 loop record 都不会有**,此时退到归因节点 // (或最后一个失败节点)自己那一行的 remediation_note —— 少了这个回退,子图整体挂掉但 // 没退过任何一次的那类失败,父层永远拿不到「该怎么修」(R4-F1 major 1 的跨 run 面)。 const records = listLoopRecords(childRun.id); const remediationNote = (records.find((r) => r.remediation_note) || {}).remediation_note || (attribution || lastBad || {}).remediation_note || null; return { status, failReason, remediationNote, failNodeRef, 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; // 预算字段坏了不该把执行卡死,按不限额处理并留给控制台暴露 } } /** * 墙钟已用时长 = **已结算的活跃时长** + 当前这一段还没结算的。 * * ── 为什么不是「now - 第一次花钱的时刻」 ────────────────────────────────── * 原来就是那么算的,它和 blocked 的语义直接对撞:预算耗尽 / 节点判罚都会把 run 打到 * `blocked`,而 blocked 的定义就是**挂起等人工**。人隔夜再来处置时,墙钟把这段等待 * 一并算进了「已用」,于是一起跑就报 `wallclock_ms` 耗尽 → 再 blocked → 再等人。 * **一条过夜的 run 从此永远醒不过来**,而报出来的原因是「跑太久」,实际是「等人等太久」。 * * 实例:`run_9168bd24f46c`(e2e-demo 做官网)2026-08-18 15:48 起跑,预算 2h, * 只花了 7/100 次调用、69,961/5,000,000 tokens,08-18 16:19 判 blocked 之后就再没能起来。 * * ponytail: 只认 `running` 一个态。manual 模式下两次点击之间 run 也停在 running, * 那段人类思考时间仍会计入 —— 上限已知,比原来跨天累计小几个数量级。 * 要更准就得按 agent 调用的起止配对计时,那是另一套账。 */ function wallclockElapsedMs(spent = {}) { const settled = spent.active_ms || 0; if (!spent.started_at) return settled; return settled + Math.max(0, Date.now() - Date.parse(spent.started_at)); } /** * 墙钟的起止结算。挂在 `graph-run-sse.afterRunTransition` 这个**单一漏斗**上, * 而不是去枚举「哪些地方会让 run 停下来」—— 那种枚举漏一条就是静默少算。 * * 进 running:开一段新的计时。离开 running:把这一段并进 active_ms 并断开锚点。 * * **存量行**(`active_ms` 缺失):旧锚点里混着等待时间,真实活跃时长已经复原不出来了。 * 这里把它按 0 起算,等于把过去那段一笔勾销 —— 有意为之:另外两维(调用次数 / tokens) * 是精确的、且没受影响,真正兜住花钱的是那两维;墙钟只是跑飞保护。 */ function settleWallclock(runId, to) { const run = getRun(runId); if (!run) return null; // 预算行永远在 root(readBudget 自动上溯),但**活动发生在树里任何一层**: // 人工重试一个 work 层节点时,动的是子 run(blocked → running),root 一直是 blocked。 // 只认 root 的转换 = 子 run 跑整场、墙钟一次都没开始计,而且旧锚点还留着 —— // 实测就是这么漏的:那条卡了两天的 run 重试后子 run 起来了,root 的 started_at // 仍是 08-18,第一次 checkBudget 立刻判耗尽。 const held = readBudget(run); if (!held || held.budget.wallclock_ms == null) return null; const b = held.budget; if (!b.spent) b.spent = {}; const s = b.spent; const legacy = s.active_ms == null; // 存量行:旧锚点混着等待时间,复原不出来 if (to === 'running') { if (legacy) { s.active_ms = 0; s.started_at = now(); // 旧锚点一笔勾销,从这次起重新计 } else if (!s.started_at) { s.started_at = now(); } else { return null; // 树里已经有别的 run 在跑,这一段的计时早开着了,别重锚 } } else { // 这一层停了不等于树停了 —— 还有兄弟 / 父 run 在跑就不结算,否则计时会被提前掐断。 // 转换此刻已落库,所以直接查库就能拿到最新状态,不用把 runId 排除在外。 if (treeHasRunningRun(held.rootRunId)) return null; if (s.started_at) { s.active_ms = (s.active_ms || 0) + Math.max(0, Date.now() - Date.parse(s.started_at)); } s.active_ms = s.active_ms || 0; s.started_at = null; } getDb() .prepare(`UPDATE graph_runs SET budget_json = ?, updated_at = ? WHERE id = ?`) .run(JSON.stringify(b), now(), held.rootRunId); return s; } /** 这棵 run 树里还有没有 running 的 run(墙钟该不该继续走) */ function treeHasRunningRun(rootRunId) { return !!getDb() .prepare(`SELECT 1 FROM graph_runs WHERE root_run_id = ? AND run_status = 'running' LIMIT 1`) .get(rootRunId); } /** * 余额检查。任何一层发起 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 && wallclockElapsedMs(spent) >= b.wallclock_ms) { exhausted.push('wallclock_ms'); } return { ok: exhausted.length === 0, exhausted, rootRunId: held.rootRunId, budget: b }; } /** * 27-P4 T5:账户欠费门的可插拔挂载点。 * * 与 `checkBudget` 不是一回事——`checkBudget` 判的是**任务级**的 agent-call/token 预算池 * (R-15 那套「只增不减的共池」),这里判的是**账户级**的欠费状态,两个维度互不代替 * (27-P4 §4.4 决策记录了这条区分,别合并成一个判据)。 * * 真实的计费接口还没确认(依赖计费系统那边给出可查的欠费状态),先按 `setAgentExecutor` / * `setPlanWorkGraph` 同款的可注入模式搭桩子:不注入时恒返回「未欠费」,`checkBudget`/ * `checkBilling` 两道门因此在真实接口接通前对现有行为零影响。 */ let billingGateFn = null; /** * @param {null | (companyId: string) => boolean | Promise | * {inArrears: boolean} | Promise<{inArrears: boolean}>} fn * 传 null 恢复成「恒未欠费」的默认桩子(测试 afterEach 用)。 */ function setBillingGate(fn) { billingGateFn = fn; } /** * @param {object} run * @returns {Promise<{ok: boolean, inArrears: boolean}>} */ async function checkBilling(run) { if (!billingGateFn) return { ok: true, inArrears: false }; const raw = await billingGateFn(run.company_id); const inArrears = typeof raw === 'boolean' ? raw : !!(raw && raw.inArrears); return { ok: !inArrears, inArrears }; } /** * 27-P4 T5:欠费清零后的自动恢复。 * * 今天从 `blocked` 恢复靠人手点「启动」(`startRunSync` 里 `blocked → running` 那跳)—— * 欠费清零不会有人去点,这正是「7×24 无人值守缺看门狗」那个更大缺口的一个具体实例 * (27-P4 §4.4 决策块)。这个函数补的就是那个自动触发器:扫一遍所有因为欠费被挂起的 * root run,对每一条重新查一次欠费状态,清了就代它点「启动」。 * * **谁来调它、多久调一次,不是这个函数的职责**——那是计费系统那边事件/轮询接进来之后 * 才能定的调度问题(T5 设计决策里写过),这里只保证「调了就正确」。桩子还没换成真实 * 计费接口时,`checkBilling` 恒 ok,这个函数扫到的挂起 run 集合恒为空,调用零风险。 * * @param {object} [opts] * @param {string} [opts.companyId] 只扫这一家公司;不传扫全部 * @returns {Promise<{scanned: number, resumed: string[], stillBlocked: string[]}>} */ async function resumeBillingClearedRuns(opts = {}) { const rows = getDb() .prepare( opts.companyId ? `SELECT * FROM graph_runs WHERE run_status = 'blocked' AND block_reason = 'billing-arrears' AND parent_run_id IS NULL AND company_id = ?` : `SELECT * FROM graph_runs WHERE run_status = 'blocked' AND block_reason = 'billing-arrears' AND parent_run_id IS NULL` ) .all(...(opts.companyId ? [opts.companyId] : [])); const resumed = []; const stillBlocked = []; for (const row of rows) { const billing = await checkBilling(row); if (!billing.ok) { stillBlocked.push(row.id); continue; } // startRun 的「启动」路径本身就是 blocked→running 的准入 + 三层级联恢复(§4.4 决策 // 复用的正是这条路),不用在这里另写一套恢复逻辑。 startRun(row.id, { mode: row.execution_mode }); resumed.push(row.id); } return { scanned: rows.length, resumed, stillBlocked }; } /** * 墙钟预算的兜底扫描——「7×24 无人值守缺看门狗」(27-P4 §4.4)的又一个具体实例, * 这次挡的是 `wallclock_ms` 那一维而不是欠费。 * * `checkBudget` 只在 `advanceOne`/`advanceBatch` 拾到 ready 节点时才查(§四.A.3 的 * 「拾到了才查」,即将真花钱时才查,不在空转时误判)——这对**正在被推进**的 run 没问题, * 出问题的是**没人再推进它**的 run:manual 模式没人点「下一步」(`POST /graph-runs/:id/next`), * 或 auto 模式的驱动循环(`runAuto`)是纯内存 promise 链,进程一重启就消失、 * 没有任何东西在启动时把它接回去。两条路径的共同症状是 `run_status='running'` * 却再也没有任何调用会触发 checkBudget——wallclock_ms 早就超了,界面上却和正常在跑 * 一模一样,直到有人手工发现。 * * 实例:`T-msx6toej`/`run_56ea2c598153`,manual 模式,2026-08-17 12:06 起跑, * `wallclock_ms` 预算 2 小时,卡到 2026-08-23 才被发现——中途没有任何一次 checkBudget。 * * 只扫 root run:`budget_json` 恒挂在 root(`readBudget` 的口径,子 run 永远是 null), * 扫子 run 拿到的预算池和扫 root 是同一份,多扫一遍纯浪费。 * * 复用 `checkBudget` + `suspendRunTree`——跟 `advanceOne` 判耗尽时走的是同一条路径, * 不发明第二种「超时」形态。 * * @param {object} [opts] * @param {string} [opts.companyId] 只扫这一家公司;不传扫全部 * @returns {{scanned: number, suspended: string[]}} */ function sweepWallclockBudgets(opts = {}) { const rows = getDb() .prepare( opts.companyId ? `SELECT * FROM graph_runs WHERE run_status = 'running' AND parent_run_id IS NULL AND budget_json LIKE '%wallclock_ms%' AND company_id = ?` : `SELECT * FROM graph_runs WHERE run_status = 'running' AND parent_run_id IS NULL AND budget_json LIKE '%wallclock_ms%'` ) .all(...(opts.companyId ? [opts.companyId] : [])); const suspended = []; for (const run of rows) { try { const gate = checkBudget(run); if (!gate.ok && gate.exhausted.includes('wallclock_ms')) { suspendRunTree(run); suspended.push(run.id); } } catch (err) { console.error(`[graph-engine] sweepWallclockBudgets run=${run.id} 异常: ${err.message}`); } } return { scanned: rows.length, suspended }; } const WALLCLOCK_SWEEP_INTERVAL_MS = 60 * 1000; let wallclockSweepTimer = null; /** 起兜底定时器。`unref()`——不该拖着进程不退出;幂等——重复调只留一个(同 task-scheduler 的形状)。 */ function startWallclockWatchdog(intervalMs = WALLCLOCK_SWEEP_INTERVAL_MS) { if (wallclockSweepTimer) return wallclockSweepTimer; wallclockSweepTimer = setInterval(() => { try { sweepWallclockBudgets(); } catch (err) { console.error(`[graph-engine] 墙钟看门狗异常: ${err.message}`); } }, intervalMs); if (typeof wallclockSweepTimer.unref === 'function') wallclockSweepTimer.unref(); return wallclockSweepTimer; } function stopWallclockWatchdog() { if (wallclockSweepTimer) clearInterval(wallclockSweepTimer); wallclockSweepTimer = null; } /** * 扣减(「只增不减」,§四.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; // 墙钟不在这里锚 —— 它按「run 处于 running 的时长」累计,起止都归 settleWallclock 管。 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; } /** * 运维操作:给一条 blocked 的 run 追加预算上限,为「启动」铺路。 * * 「只增不减」(§四.A.3 的既有约定)在这里延伸成:只吃正数增量、只加到已经配了额度的 * 那一维上(`b[key] == null` = 那一维本来就不限额,追加没有意义,忽略而不是凭空开一维新限额)。 * * 顺手结清卡死的墙钟锚点——不做这一步的话,`startRunSync` 一「启动」就会走 * `settleWallclock(runId,'running')`:`spent.started_at` 非空会被判成「树里已经有别的 run * 在跑,别重锚」(见该函数注释),于是沿用的还是 blocked 之前那个旧锚点,预算刚加完 * 下一次 `checkBudget` 立刻用「挂起等待的时长」把新加的额度吃光,run 秒进秒出。 * 这里手动做一次 `settleWallclock` 本该在「离开 running」那一刻做的事:把锚点到现在的 * 时长并进 `active_ms`、断开锚点,让「启动」重新开一段干净的计时。 */ function extendBudget(runId, patch = {}) { const run = getRun(runId); if (!run) throw new Error(`graph_runs 不存在: ${runId}`); const held = readBudget(run); if (!held) throw new Error(`run ${runId} 没有挂 budget_json,没有预算可加`); const b = held.budget; if (!b.spent) b.spent = {}; const before = { ...b, spent: { ...b.spent } }; const keys = ['total_agent_calls', 'total_tokens', 'wallclock_ms']; const applied = {}; for (const key of keys) { if (patch[key] == null) continue; if (typeof patch[key] !== 'number' || !(patch[key] > 0)) { throw new Error(`${key} 增量必须是正数,收到: ${patch[key]}`); } if (b[key] == null) continue; // 该维度本来就不限额,追加没有意义 b[key] += patch[key]; applied[key] = patch[key]; } if (Object.keys(applied).length === 0) { throw new Error('没有任何一维预算被追加(增量缺失,或目标维度本来就不限额)'); } if (b.spent.started_at) { b.spent.active_ms = (b.spent.active_ms || 0) + Math.max(0, Date.now() - Date.parse(b.spent.started_at)); b.spent.started_at = null; } getDb() .prepare(`UPDATE graph_runs SET budget_json = ?, updated_at = ? WHERE id = ?`) .run(JSON.stringify(b), now(), held.rootRunId); sse.emitEvent(runId, 'budget.extended', '运维追加预算并结清墙钟锚点', { root_run_id: held.rootRunId, applied, before, after: b, }, { ref: held.rootRunId }); return { ok: true, rootRunId: held.rootRunId, applied, budget: b }; } /** * 资源类原因(预算耗尽 / 账户欠费)→ **整棵 run 树挂起等人工**,不是失败(§四.A.3 / 27-P4 §4.4): * root run → blocked,树上所有非终态 run 一律置 pause_requested=1。 * 同样只做一次 listRunTree 索引查询。 * * @param {object} run * @param {object} [opts] * @param {string} [opts.reason='budget-exhausted'] 写进 `graph_runs.block_reason`(机器可读, * 欠费自动恢复的扫描靠它过滤),默认值保留旧调用方(checkBudget 那条路径)的行为不变。 * @param {string} [opts.summary] 人读的事件摘要,不传则按 reason 给个默认句子。 */ function suspendRunTree(run, opts = {}) { const reason = opts.reason || 'budget-exhausted'; const summary = opts.summary || (reason === 'billing-arrears' ? '账户欠费:整棵 run 树挂起等人工(不是失败)' : '预算耗尽:整棵 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', { blockReason: reason }, { summary, reason, }); } return (getRun(run.id) || {}).run_status; } // ──────────────────────────────────────────────────────────────── // §四.C 五个控制操作(手动 / 自动共用上面的 advanceOne) // ──────────────────────────────────────────────────────────────── /** * 「进坑;进不去就回队列」—— **一切恢复类 run 转换的唯一入口**(27-P3C 批 3 复核 F1)。 * * 背景:`updateRunStatus` 的准入守卫判的是「从活跃集外进入活跃集」,而三条恢复路径的 from * 恰好全在活跃集外 —— 人加完预算点启动(`blocked → running`)、子 run 救回父 run(同)、 * 对已收敛 run 里的节点点重试(`completed → ready` / `blocked → running`)。批 3 之后 * 「坑一空就被队里下一条填上」是每次都会发生的事,于是这三条在有队列的公司里几乎必然撞坑满。 * * 为什么不给恢复开一条「优先通道」(抢占正在跑的那条 / 绕过守卫): * · 绕过守卫 = 直接撞 `idx_graph_runs_one_active_root`,把一次可解释的拒绝换成一句 * SQLite 原文的 500,比现在更糟; * · 「让调度器刚出队那条 `queued→ready` 的 run 退回队列」听着更好,实际抓不住: * `tickInner` 5c-5e 是**同一次同步调用**里出队 + 起跑的,run 停在 ready 的窗口只有几微秒, * 人点下按钮时坑上躺着的一定是 `running`。要抢占就得中断一条正在烧 token 的 run —— 那是 * 批 4 「max>1 + 优先级抢占」的活,不是批 3 能可靠做对的事。 * * 所以选的是**另一半**:拿不到坑的不是「失败」,是「该重新排队」。把这条 root run 连同它的 * task 一起退回 queued(`requeueBlockedRootRun`,两行同事务),调度器就会在坑空时按 * priority→seq→created_at 把它接回来。「T1 永远回不来」这个后果因此消失:它重新变成队里的一员, * 而不是躺在一个非终态上等一次不可能赢的赛跑。 * * 退不回去的(run 没有 task / task 已不是 blocked)照抛 `RunSlotBusyError` —— 那是干净的拒绝: * 抛出来时一个字都还没写,路由层翻 409 即可(错误里带着 companyId / heldBy)。 * * @returns {'admitted'|'requeued'|'lost'} lost = 条件更新匹配 0 行(别人抢先推走了这条 run) * @throws {RunSlotBusyError} 坑满且这条 run 连队列都回不去 */ function admitOrRequeue(runId, from, to, extra = {}) { try { return sse.transitionRun(runId, from, to, {}, extra) ? 'admitted' : 'lost'; } catch (err) { if (!(err instanceof RunSlotBusyError)) throw err; if (!requeueBlockedRootRun(runId)) throw err; // 状态改了就必须有事件:requeueBlockedRootRun 要把 run 行和 task 行包进同一个事务, // 包不进 sse.transitionRun(它的 apply 只能改一张表),所以事件在这里补。 // 用 `run.status` 这个事件名不是凑合:控制台正是靠它去重拉一次全量快照。 sse.emitEvent(runId, 'run.status', `执行坑被 ${err.heldBy.join(', ')} 占着:${from} → queued,退回队列等调度器按位次接回`, { from, to: 'queued', intended: to, held_by: err.heldBy, requeued: true, }); return 'requeued'; } } /** * planning → ready(图规划完成 + 通过写入前五项校验之后调,校验在 work-graph-db)。 * * ⚠️ **它只认 planning**:`from` 写死成 'planning',对别的状态(最典型的是排队中的 `queued` run) * 是一次 `WHERE run_status='planning'` 匹配 0 行的空转。所以**出队不能复用它**, * 出队原语是 `updateRunStatus(runId,'queued','ready')`(27-P3C §五.1)。 * * 收尾那次回读是必需的:以前这里无条件 `return {ok:true, runStatus:'ready'}`, * 于是「没推进」和「推进了」返回的东西一模一样 —— 调用方(launchTaskRun / 调度器) * 拿着一个假的 ready 继续往下走,表现为 task 标了 running、库里却没有活的 run,且全程无报错。 */ 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' }); const actual = (getRun(runId) || {}).run_status; if (actual !== 'ready') { throw new Error(`markRunReady 未生效 run=${runId}:期望 ready,实际 ${actual}(它只认 planning,${actual === 'queued' ? '排队中的 run 请走 updateRunStatus(queued→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 时复用。 * * **恢复与暂停必须成镜像**(R5 接上的断链 2)。pauseRun 是沿 root_run_id 向下三层级联的 * (§四.C 定案),而这里以前只清自己那一层:点「暂停」再点「启动」之后,子 run / 孙 run * 的 pause_requested 仍然是 1、run_status 仍然是 paused,advanceTree 只捡 running 的 run, * 于是整棵树一步也推不动——控制台表现为「点了启动没反应」,验收 4 的后半从来不成立。 * 级联恢复只做两件事:清全树软信号 + 把**被级联暂停的那些 paused 子 run** 拉回 running。 */ function startRunSync(runId, { mode } = {}) { const run = getRun(runId); if (!run) throw new Error(`graph_runs 不存在: ${runId}`); // 排队中的 run 必须先出队(queued → ready)再启动,不能反过来(27-P3C §九.1)。 // 单独拦一句而不是让它去撞 RUN_TRANSITIONS:那样报出来的是一句 InvalidTransitionError, // 看不出「你把出队那一步跳过了」这个真正的原因。 if (run.run_status === 'queued') { throw new Error(`run ${runId} 还在排队中:启动前必须先出队 updateRunStatus(runId,'queued','ready'),见 27-P3C §五.1`); } const fresh = run.run_status === 'ready'; if (run.run_status !== 'running') { // 走 admitOrRequeue 而不是裸 transitionRun:`blocked → running`(人加完预算点「启动」) // 是准入转换,坑可能刚被队里下一条占走。那时把这条 run 退回队列,而不是抛一个 // 让它永远躺在 blocked 的错误 —— blocked 不是终态,躺着的后果是 goal 永不收敛。 // `ready/paused → running` 是活跃集内移动,走这里与以前逐字等价(不可能坑满)。 if (admitOrRequeue(runId, run.run_status, 'running', { summary: `启动:${run.run_status} → running${mode ? `(${mode} 模式)` : ''}`, execution_mode: mode || run.execution_mode, }) === 'requeued') { // 早返回:后面那次「回读必须是 running」的断言对退回队列的 run 不成立, // 而软信号 / 级联恢复留给真正出队那一次做(走的是同一条 startRunSync)。 return { ok: true, runId, runStatus: 'queued', requeued: true, seeded: [], resumed: [] }; } } // 「启动」清掉软信号,否则 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); } // 27-P4 T5:block_reason 只在 suspendRunTree 挂起那一刻写,「启动」(不管从 blocked/paused/ // ready 起)一律清掉——不清的话,一次欠费挂起恢复之后再因为别的原因二次挂起,界面还会显示 // 上一次的旧理由。这条路径只有「启动」会碰它,不存在并发写冲突,用无条件 UPDATE 而不是走 // sse.transitionRun 的 patch(那边的 from 已经在上面的 admitOrRequeue 里转掉了,这里补一刀更简单)。 getDb().prepare(`UPDATE graph_runs SET block_reason = NULL WHERE id = ?`).run(runId); const resumed = resumeDescendants(run, mode); const seeded = fresh ? seedEntryNodes(runId) : []; // 同 markRunReady:以前这里无条件报 running,一次没抢到的启动和一次成功的启动长得一模一样。 // 回读一次,真没起来就抛 —— 调用方(调度器)据此把 task 收回,而不是留下一条「标着 running // 却没有活 run」的任务。回读而不是看 transitionRun 的返回值:run 本来就是 running 时 // 这里根本不转换,那也是合法的「已经起来了」。 const actual = (getRun(runId) || {}).run_status; if (actual !== 'running') { throw new Error(`startRun 未生效 run=${runId}:期望 running,实际 ${actual}`); } return { ok: true, runId, runStatus: 'running', seeded, resumed }; } /** * 级联恢复(pauseRun 的镜像)。 * * 只对**非终态**的后代动手(与 pauseRun 的过滤条件逐字一致): * - 一律清 pause_requested / stop_requested,否则子层在节点边界又会自己落回 paused; * - run_status='paused' 的拉回 running —— 它们的 paused 就是上一次级联暂停留下的。 * **blocked 的子 run 不碰**:blocked 的语义是「等人工介入」(熔断 / 死路 / 预算耗尽), * 它没有 ready 节点,拉成 running 只会在下一次 settleRun 时原地又判回 blocked, * 白白制造一串状态抖动事件。解除 blocked 的正当入口是对具体节点点「重试」(retryNode * 里有 blocked → running 那一跳),不是父层点一下启动就把所有人的阻塞一起抹掉。 * * @returns {string[]} 被拉回 running 的后代 run id */ function resumeDescendants(run, mode) { const resumed = []; for (const r of descendantRuns(run)) { if (TERMINAL_RUN_STATUS.has(r.run_status)) continue; setRunFlags(r.id, { pauseRequested: false, stopRequested: false }); if (r.run_status !== 'paused') continue; if (sse.transitionRun(r.id, 'paused', 'running', {}, { summary: `级联恢复:paused → running(随父 run ${run.id} 一起启动)`, execution_mode: mode || r.execution_mode, cascaded_from: run.id, })) { resumed.push(r.id); } } return resumed; } /** * 「启动」(对外)。auto 模式立刻挂一个自我调度的循环上去;manual 模式只标 running, * 停在原地等「下一步」。返回里的 driver 是 auto 循环的 promise(路由可以不管它,测试可以 await)。 */ function startRun(runId, opts = {}) { const res = startRunSync(runId, opts); // 退回队列的不挂 auto 驱动:那条 run 现在是 queued,驱动循环只会空转一遍, // 真正的起跑发生在调度器把它出队之后(那时会再走一次 startRun)。 if (res.requeued) return res; 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}`); // 与 pauseRun 同一条过滤:**终态 run 不碰**(R5 顺带修的第三处不对称)。 // 原来对整棵树无差别置 stop_requested,会给早已 completed 的子 run 也盖上停止标志; // 那条 run 之后不可能再被推进,标志位纯属噪声,但它会让「这棵树是怎么停下来的」 // 在控制台和 timeline 里读起来像是「所有层都被停了」。 const targets = [run, ...descendantRuns(run)].filter((r) => !TERMINAL_RUN_STATUS.has(r.run_status)); // 软信号先在整棵子树上一次置齐,再逐层处理:否则处理深层的时候浅层还能拾取新节点 for (const r of targets) setRunFlags(r.id, { stopRequested: true }); // **自底向上**(targets 里父一定排在子前面,倒序即最深优先)。 // // 倒序 + 每层落 stopped 后补一次 propagateToParent,修的是「终态 run 上留着永远 running // 的节点」(R6-F3 minor):子 run 被 stopRun 直接转 stopped,不经过 settleRun, // 父节点那个 subgraph 节点就永远停在 running —— run 已是终态、节点却还在跑, // 控制台读起来像没停下来。回写走的是子 run 正常收敛用的**同一条** propagateToParent, // 不另造一条路径;它自带幂等(父节点已不是 running 就直接返回)。 // 倒序保证父层在**收到回写之后**才轮到自己 skip 剩余节点 → stopped, // 于是回写顺带点亮的下游(fail 边 / 兜底边的目标)同样会被 skip,不留悬空 ready。 // 只停子树时父 run 不在 targets 里,它就此拿到「子 run 停了」这个失败结果,按图上的 // fail 边自行处置 —— 这正是「停止一个子图」在父层应有的语义,而不是父层永远干等。 for (const r of [...targets].reverse()) { 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, }); } propagateToParent(r.id); } 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; } /** * 人工「重试」的 run 侧:把这条 run 重新拉进可推进的状态。**由 retryNode 在动节点之前调**。 * * `reopenRun=false` 是跨层定点重置专用:run 状态留给 reopenChildRun 在父节点被拾取时才拉起, * 免得子图抢在上游 org node 重做完之前先跑起来。 * * `completed` 也要能重新拉起(R-17.1 给 RUN_TRANSITIONS 开的 `completed → ready`)—— * 不然对一张已收敛的 run 里的节点点「重试」,节点是 ready 了、run 还是 completed, * 而 advanceTree 只捡 running 的 run,这个节点永远不会被推(重试按钮点了没反应)。 * * 坑满时的两种结局都在 admitOrRequeue 里: * · run 是 blocked(预算耗尽那一类)→ 连 task 一起退回队列,返回 requeued。节点随后照常 * 置 ready,等调度器把这条 run 出队时它就是第一个被推的 —— 重试没白点。 * · run 是 completed → 退不回队列(task 早已是终态,把终态任务塞回队列是另一码事, * 属批 4 的 rerun 语义),`RunSlotBusyError` 原样抛给路由翻 409。此时**一个字都没写**。 * * @returns {void} 没有「失败」这个出口:要么推进了,要么被别人抢先推走了(都可以继续动节点), * 要么抛 RunSlotBusyError(整条重试作废,节点一个字都没动)。 */ function reopenRunForRetry(runId, reopenRun) { if (!reopenRun) return; const run = getRun(runId); if (run.run_status === 'completed') { admitOrRequeue(runId, 'completed', 'ready', { summary: '人工重试重开已收敛的 run:completed → ready' }); } const cur = (getRun(runId) || {}).run_status; if (cur === 'blocked' || cur === 'ready') { admitOrRequeue(runId, cur, 'running', { summary: `人工重试重开链路:${cur} → running` }); } } /** * 「重试」:节点 status ∈ {fail, blocked} → ready,attempt + 1。 * * **默认不级联重置下游**(§四.C 末段):级联与否不由重试这个动作决定,而由**产出契约是否 * 真的变了**决定。判定发生在重跑成功之后(applyRetryCascade),这里只把当前契约拍个快照。 */ function retryNode(runId, nodeRunId, { reopenRun = true, remediationNote = null } = {}) { const nodeRun = getRunNode(nodeRunId); if (!nodeRun || nodeRun.run_id !== runId) throw new Error(`节点不属于该 run: ${nodeRunId}`); if (!canTransitionNode(nodeRun.status, 'ready')) { // ready/running/pending 的节点本来就已经排上队了,再重置一次没有意义(跨层定点重置会撞到这里) return { ok: false, reason: 'not-retryable', nodeStatus: nodeRun.status }; } // ① **run 的转换必须排在节点转换之前**(批 3 复核 F1 场景 B)。 // 以前的顺序是「节点 → ready(提交)→ attempt+1(提交)→ 才动 run」,而动 run 那一步 // 是准入转换(`completed → ready` / `blocked → running`,from 都在活跃集外),坑满时它抛。 // 抛的时候前两步**已经落库**:节点 ready、attempt 已 +1、run 还是 completed,而 // advanceTree 只捡 running 的 run —— 这个节点从此再也不会被推,且没有任何一条路径能修。 // 先做 run:转不动就整段放弃,一个字都没写,人重试一次仍是干净的初态。 reopenRunForRetry(runId, reopenRun); // 人工重试同样是「新的一轮」,靠 **attempt+1** 换代(R-17.1):subgraph 节点的 // child_run_id 保留不清,executeSubgraph 比 child_run_attempt 与 attempt 就知道这是新一轮, // 于是复用同一张 work graph 定义重跑,而不是再花一次 LLM 调用重画一张。 // 返工前这里清 child_run_id,是因为当时的守卫只看「子 run 是否终态」,不清就死锁。 // 判词**只在调用方明确给了新的**时才覆盖,不给就保留节点行上原有的那条。 // 无条件写 notePatch(null) 会把 blockNode 留在这一行的判词抹掉,而「人工重试 blocked 节点」 // 恰恰要靠它把「该怎么修」带进下一轮的提示词(§八.14 那条四环全通的链路就断在这里)。 if ( !sse.transitionNode(runId, nodeRunId, nodeRun.status, 'ready', { failReason: null, failNodeRef: null, ...(remediationNote ? notePatch(remediationNote) : {}) }, { attempt: nodeRun.attempt + 1, summary: `${reopenRun ? '人工「重试」' : '跨层定点重置'}:${nodeRun.status} → ready(attempt #${nodeRun.attempt + 2})`, }) ) { return { ok: false, reason: 'taken-by-other', nodeStatus: (getRunNode(nodeRunId) || {}).status }; } bumpNodeAttempt(nodeRunId); const snapshot = snapshotContract(getRun(runId), nodeRunId); 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}/ * GRAPH_RUNS_DIR 覆盖时以它为 {rootRunId} 的父目录 —— 与 graph-run-sse.runDir 完全同源, * 保证 run 的产物、stream.log、timeline.ndjson 永远落在同一棵目录树下(R4-F1 minor 5)。 */ function nodeArtifactDir(run, nodeRunId) { return path.join(runRootDir(run), 'nodes', nodeRunId); } /** run 目录(§四.A.5),与 graph-run-sse.runDir 同源;产物 / stream.log / timeline 共用它 */ function runRootDir(run) { const runsRoot = runsRootOverride(); return runsRoot ? path.join(runsRoot, run.root_run_id) : path.join(REPO_ROOT, 'data', 'companies', run.company_id, 'runs', run.root_run_id); } // ──────────────────────────────────────────────────────────────── // W2 · 每个 root run 一个项目根(27-P3B §四 W2) // ──────────────────────────────────────────────────────────────── /** * **根因不是「缺沙箱」而是「缺产出位置」**:节点收到「实现某个功能」的任务,却没有 * 「这次任务的代码写哪」,于是写到它能看见的最近的仓库(真 CLI 实测已往宿主仓库写过 * server/routes/echo.js 等三处)。所以 W2 补的是位置,W3 才是边界。 * * 位置的分配**复用 DevLoop 的 path-resolver.allocateProjectRoot(sandbox 模式)**, * 不新造一套:`data/companies/{cid}/projects/{pid}`,与 DevLoop 项目同一棵目录树。 * * ── 存哪:`runs/{rootRunId}/artifacts/project-root.json`,**不加 schema 列** ── * 三条理由: * 1. graph_runs 加列要动 work-graph-db.js(本轮不归本文件),且这条信息**天然按 * root_run_id 唯一**——而 runRootDir 本来就是按 root_run_id 拼的,子 run 拼出来是 * 同一个目录,天然共享同一份记录,不需要每行都存一份再对齐。 * 2. 它是「这次任务的工作区在哪」,与 artifacts/task-goal.json 是同一类 run 级事实, * 放在一起读回的路径已经通了(describeArtifactsDir / describeTask 都在读这个目录)。 * 3. 产生 → 传递 → 落库 → 读回 四环都在本文件里闭合: * ensureProjectRoot 分配(产生)→ ctx.projectRoot 进提示词 + callAgent metadata(传递) * → project-root.json(落库)→ 下一个节点/子 run 再进来时 readProjectRootRecord 读回。 */ const PROJECT_ROOT_FILE = 'project-root.json'; const PROJECT_ROOT_SCHEMA = 'graph-run-project-root/v1'; function projectRootFile(run) { return path.join(runRootDir(run), 'artifacts', PROJECT_ROOT_FILE); } function readProjectRootRecord(run) { let rec = null; try { rec = readJsonOrNull(projectRootFile(run)); } catch { return null; // run 缺 company_id/root_run_id:拼不出路径 = 没有记录 } return rec && typeof rec.root_path === 'string' && rec.root_path ? rec : null; } /** * 真正在盘上开一个目录。 * * GRAPH_RUNS_DIR 覆盖时项目根跟着走(`{RUNS_ROOT}/{rootRunId}/project`),理由与本文件 * 顶部 RUNS_ROOT 那段注释完全一样:run 的产物被重定向到临时目录,项目根却还落在仓库 * `data/companies/` 的话,测试每跑一次就往仓库里堆一个空项目目录。两处必须同源。 */ function allocateRunProjectRoot(run) { const runsRoot = runsRootOverride(); if (runsRoot) { const root_path = path.join(runsRoot, run.root_run_id, 'project'); fs.mkdirSync(root_path, { recursive: true }); return { pid: `p-run-${run.root_run_id}`, root_path, mode: 'runs-dir-override' }; } return require('./devloop/path-resolver').allocateProjectRoot({ company_id: run.company_id, mode: 'sandbox', }); } /** * 取本 root run 的项目根,没有就分配一个并落库。**幂等**:先读回、再分配。 * * @returns {{pid: string, root_path: string, mode: string}|null} null = 这次拿不到写入位置 * (提示词会如实说明,而不是编一个假路径塞给 agent) */ function ensureProjectRoot(ctx) { const run = ctx && ctx.run; if (!run) return null; const existing = readProjectRootRecord(run); if (existing) { // 目录被人删掉过就照原路径重建:记录里的路径是提示词/产物的唯一真源,不能换。 try { fs.mkdirSync(existing.root_path, { recursive: true }); } catch { /* 建不出来下面照样返回,agent 会自己报错 */ } return existing; } let alloc; try { alloc = allocateRunProjectRoot(run); } catch (e) { timelineNote(ctx, 'node.boundary', `项目根分配失败(${e.message}),本节点没有产出位置`, { node_run_id: ctx.nodeRunId, degraded: 'project_root_alloc_failed', }); return null; } // W3 的第一道(也是唯一一道**调用前就能强制**的)守卫:复用 path-resolver 的黑名单。 // sandbox 模式自己算出来的路径不会命中,但 GRAPH_RUNS_DIR 指到 /etc 这类配置错误会。 try { if (require('./devloop/path-resolver').isSensitivePath(alloc.root_path)) { timelineNote(ctx, 'node.boundary', `项目根落在敏感路径黑名单里,拒绝使用:${alloc.root_path}`, { node_run_id: ctx.nodeRunId, root_path: alloc.root_path, degraded: 'sensitive_project_root', }); return null; } } catch { /* 黑名单查不了不该把 run 弄挂,继续用分配到的路径 */ } const rec = { schema: PROJECT_ROOT_SCHEMA, pid: alloc.pid, root_path: alloc.root_path, mode: alloc.mode, root_run_id: run.root_run_id, company_id: run.company_id, allocated_by: ctx.nodeRunId || null, allocated_at: now(), }; const file = projectRootFile(run); try { fs.mkdirSync(path.dirname(file), { recursive: true }); // 'wx' = 先到先得。并发下第二个节点会 EEXIST,读回第一个人写的那份(下面 catch)。 fs.writeFileSync(file, JSON.stringify(rec, null, 2), { flag: 'wx' }); } catch { const raced = readProjectRootRecord(run); if (raced) { // ponytail: 输给别人的那次 allocateProjectRoot 会剩一个空目录(sandbox 模式下是 // data/companies/{cid}/projects/{pid}/,里面什么都没有)。不清理:删目录的代价错误 // 方向更贵。升级路径是把分配挪到 createRun 那一刻(届时只有一个调用方)。 return raced; } // 落不了库但目录是真的:仍然给 agent 一个写入位置,但要吼出来——下一个节点会**另分配一个**, // 跨节点的产物会散在两个根里。 timelineNote(ctx, 'node.boundary', `项目根记录写入失败(${file}),后续节点可能分配到不同的项目根`, { node_run_id: ctx.nodeRunId, root_path: rec.root_path, degraded: 'project_root_not_persisted', }); return rec; } timelineNote(ctx, 'node.boundary', `本次任务的项目根:${rec.root_path}(${rec.mode})`, { node_run_id: ctx.nodeRunId, root_path: rec.root_path, pid: rec.pid, mode: rec.mode, }); return rec; } // ──────────────────────────────────────────────────────────────── // W3 · 产出边界守卫(27-P3B §四 W3) // ──────────────────────────────────────────────────────────────── /** * **能拦住什么、拦不住什么,先说清楚**(27-P3B §四 W3 的状态栏同文): * * 真 CLI 是子进程直接写盘、还带着 `--dangerously-skip-permissions`,引擎**不可能**在写发生的 * 那一刻拦住它。所以这里落的是「事后取证 + 结算拒收」,不是沙箱: * * 能做到:① 调用前后各扫一次**宿主仓库**的 `git status --porcelain`,差集就是这次调用在 * 仓库里留下的痕迹;有痕迹就把节点判失败(success 也降级成 fail)并写 timeline。 * 这正好覆盖已经实际发生过的那次事故(写 server/routes/echo.js 并改 server/index.js)。 * ② 项目根本身过 isSensitivePath 黑名单(调用前,真强制)。 * 做不到:① 宿主仓库**之外**的写入(`~/Documents/别的项目`、`/tmp`)完全看不见; * ② 仓库内被 .gitignore 忽略的路径(含 `data/companies/`)看不见; * ③ 同一个文件在调用前就已经是 modified、调用中又被改一次 → porcelain 行不变,看不见; * ④ 越界的写**已经发生**,守卫只拒收结果,**绝不自动回滚**——回滚等于替开发者 * 丢弃未提交的改动,代价方向比误判更贵; * ⑤ 别人(开发者本人、并行的另一个 agent)在同一时刻改仓库会被算到本节点头上(误判)。 * 真正的隔离要么改 AgentManager 让 CLI 的 cwd/权限落在项目根(`agent.workDir` 现在写死是 * agent 自己的目录,不归本文件),要么上容器/文件系统级沙箱。两者都不在本轮。 * * 守卫只在**真会起 CLI 的执行器**下开(默认执行器 / 默认规划器):测试注入的假 executor * 是进程内函数,扫仓库既没意义又要给 719 个用例每人加两次 git 调用。 */ function boundaryGuardApplicable(ctx) { return ctx.nodeKind === 'subgraph' ? planWorkGraph === defaultPlanWorkGraph : agentExecutor === defaultAgentExecutor; } function hostRepoStatus() { try { return require('child_process').execFileSync('git', ['status', '--porcelain'], { cwd: REPO_ROOT, encoding: 'utf8', timeout: 20_000, maxBuffer: 8 * 1024 * 1024, }); } catch { return null; // 不是 git 仓库 / 没有 git / 超时:守卫不可用,如实记一条,不影响执行 } } /** * 两次 `git status --porcelain` 的差集 = 本次调用期间新出现的仓库改动。 * 纯函数,唯一调用方是 applyBoundaryGuard(导出只为让它能被单测直接验)。 */ function diffRepoStatus(before, after) { const seen = new Set(String(before == null ? '' : before).split('\n').map((l) => l.trim()).filter(Boolean)); return String(after == null ? '' : after) .split('\n').map((l) => l.trim()).filter(Boolean) .filter((l) => !seen.has(l)); } /** 调用前拍一张宿主仓库快照;守卫不适用/不可用时返回 null */ function beginBoundaryGuard(ctx) { if (!boundaryGuardApplicable(ctx)) return null; const snap = hostRepoStatus(); if (snap === null) { timelineNote(ctx, 'node.boundary', '边界守卫不可用(宿主仓库 git status 跑不了),本次调用不做越界检测', { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, degraded: 'boundary_guard_unavailable', }); } return snap; } /** * 调用后比对。发现越界 → 记 timeline + **把结果降级成 fail**(引擎不发明 fail_reason, * 越界走的是那条兜底边)。subgraph 的 pending 不降级:它的子 run 还在跑,就地判失败会 * 留下一张没人回写的孤儿子 run,只留痕。 */ /** * `system/agents/` 这棵树 —— 28 号改造之后它是**agent 注册表**,不再只是源码。 * * 本体的落点从 `data/companies/{cid}/agents/`(整个目录在 .gitignore 里、天然不进 * git status)搬到了被 git 跟踪的 `system/agents/{id}/`,于是「注册表」和「仓库源码」 * 住进了同一棵树。而运行期**合法地**写这棵树的有三方: * ① `hireFromBaseMetaAgent` 从 `_base` 复制骨架 → 新建整个 `system/agents/{newId}/`; * ② agent-creator 往里写 agent.md / soul.md / SKILL.md; * ③ evolver 迭代已有本体。 * 守卫**本来就无法把这些改动归属到当前节点** —— 同期可能有别的招聘在跑, * 整仓 git status 的差集里根本没有「谁写的」这一维。 * * 旧判据是「深度 1 的五个固定文件名」(`todo.md` / `plan.md` / … ,旧名 * `AGENT_RUNTIME_STATE_RE`),那是按**旧落点**裁剪的:agent 只会在自己既有的本体目录里 * 改几个运行态文件。新建本体在 porcelain 里是 `?? system/agents/{newId}/` **一整条目录项**, * 五个文件名的名单接不住 → 落进 escapes → 触发招聘的那个节点被降级成 fail + * `violation:'boundary_escape'`,正是 28 号设计 §九 风险表第一行点名的症状: * agent 明明成功却 fail,且归因指向错误的节点。 * * 所以判据放宽到整棵注册表树:归进 unattributable = **保留证据、放弃判罚**, * 与该桶已有的语义一致(timeline 照记,不参与降级)。判罚留给真正的越界: * 写进 server/ web/ packages/ 这些纯源码目录。 * ponytail: 上限没变 —— 同一次调用里 agent 若在注册表树**之外**写了源码,仍算当前节点头上。 */ const AGENT_REGISTRY_TREE_RE = /^system\/agents\//; /** * 「这个 worktree 还有别人在写」—— 判罚前必须先回答的问题。 * * 守卫的前提是「本次调用窗口内新出现的仓库改动 = 这个节点写的」。多会话共用同一个 * worktree 时这个前提**是假的**:`git status` 的差集里没有「谁写的」这一维, * 一个并行的开发会话改 `server/services/x.js`,会被原样记到当时正在跑的那个节点头上。 * 实测已经发生过:27-P3C 批 3 的开发改动把一条 run 判成了 `boundary_escape` 卡住 * (§15.5 第 1 条)。 * * **可检测的信号**:如果节点**开跑那一刻**宿主仓库就已经有源码目录的未提交改动, * 那说明有人正在这个 worktree 上开发 —— 此后窗口内新增的源码改动,是他的可能性 * 至少不低于是 agent 的。这不是猜,是快照里读得出来的事实。 * * 这种情况下**降级为只记证据、不判罚**(与 `unattributable` 桶同一语义), * 而不是关掉守卫:证据照留,人一眼能看出「这次没判罚,因为当时仓库是脏的」。 * * `BOUNDARY_GUARD` 可覆盖:`strict` = 无论仓库脏不脏都判罚(CI / 干净环境用); * `record` = 一律只记不判。缺省是上面那条自动判断。 */ const JUDGED_SOURCE_RE = /^(server|web|packages|scripts)\//; function sharedWorktreeReason(before) { const mode = process.env.BOUNDARY_GUARD; if (mode === 'strict') return null; if (mode === 'record') return 'BOUNDARY_GUARD=record:本次一律只记证据、不判罚'; const dirty = String(before == null ? '' : before) .split('\n').map((l) => l.trim()).filter(Boolean) .map((l) => l.replace(/^\S{1,2}\s+/, '').trim().replace(/^"|"$/g, '')) .filter((p) => JUDGED_SOURCE_RE.test(p)); if (!dirty.length) return null; return `本节点开跑时宿主仓库已有 ${dirty.length} 处源码未提交改动(如 ${dirty[0]}),` + '说明有别的会话正在这个 worktree 上开发;窗口内新增的源码改动无法归属到本节点,只记证据不判罚'; } function splitUnattributable(entries) { const escapes = []; const unattributable = []; for (const e of entries) { // diffRepoStatus 已经把前导空格去掉了,所以 porcelain 的两字符状态位在这里长度不定: // 修改类是 "M path"(1 字符),未跟踪类是 "?? path"(2 字符),新建本体正是后者。 // 按「首个空白之前」切,不要按固定偏移切 —— 固定 slice(3) 会把 "M system/..." 切成 // "stem/...",恰好对**修改类**失效,而 agent-creator 改自己 todo.md 正是修改类。 // 末尾的去引号同样是必需的:路径含空格 / 中文时 git 会整条加引号("system/agents/新 人/"), // 不剥掉就成了 `"system/...` 开头、前缀匹配不上。 const p = String(e).replace(/^\S{1,2}\s+/, '').trim().replace(/^"|"$/g, ''); (AGENT_REGISTRY_TREE_RE.test(p) ? unattributable : escapes).push(e); } return { escapes, unattributable }; } /** * @param {string|null} [after] 调用后的仓库快照。生产里由 hostRepoStatus() 现拍 * (advanceOne 只传前三个参数);**留这个可选参数是为了给「越界 → 判 fail」这条分支一道 * 不起真 CLI 的测试缝**——第一批终验记录点名了它:守卫只在 `agentExecutor === * defaultAgentExecutor` 时才开,而单测一律注入假 executor,于是这条唯一有实际后果的分支 * 以前只有真 CLI 才走得到(= 实际零覆盖)。注入 after 后判罚逻辑可以逐条验, * 而生产路径一个字符都没变。 */ function applyBoundaryGuard(ctx, before, outcome, injectedAfter = undefined) { // 这个早返回必须在拍快照**之前**:守卫不适用时 before 是 null,此时一次 git 子进程都不该起。 // (所以不能写成 `after = hostRepoStatus()` 的默认参数——默认参数在函数体之前就求值了。) if (before === null || before === undefined) return outcome; const after = injectedAfter === undefined ? hostRepoStatus() : injectedAfter; if (after === null) return outcome; let { escapes, unattributable } = splitUnattributable(diffRepoStatus(before, after)); // 多会话共用 worktree 时,源码改动无法归属到本节点 —— 全部降级为「只记证据」 const sharedReason = escapes.length ? sharedWorktreeReason(before) : null; if (sharedReason) { timelineNote(ctx, 'node.boundary', `本次调用窗口内有 ${escapes.length} 处仓库改动,但无法归属到本节点,不计入越界`, { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, violation: null, entries: escapes.slice(0, 20), reason: sharedReason, note: '守卫的前提是「窗口内新增的改动 = 本节点写的」,而 git status 的差集里没有' + '「谁写的」这一维。要在干净 worktree 上强制判罚,设 BOUNDARY_GUARD=strict。', }); unattributable = unattributable.concat(escapes); escapes = []; } if (unattributable.length) { timelineNote(ctx, 'node.boundary', `本次调用窗口内有 ${unattributable.length} 处 agent 注册表(system/agents/)改动,无法归属到本节点,不计入越界`, { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, violation: null, unattributable: unattributable.slice(0, 20), note: '注册表在运行期本就会被合法写入(招聘新建本体 / agent-creator 补 agent.md / ' + 'evolver 迭代 / agent 改自己的 todo.md),且同期可能有别的招聘在跑,无法归属;' + '这里只留证据不判罚,判罚留给 server/ web/ packages/ 这类纯源码目录', }); } if (escapes.length === 0) return outcome; const shown = escapes.slice(0, 20); timelineNote(ctx, 'node.boundary', `越界写入:本次调用在宿主仓库留下 ${escapes.length} 处改动(应写在项目根 ${ctx.projectRoot || '(未分配)'} 内)`, { node_run_id: ctx.nodeRunId, attempt: ctx.attempt, violation: 'boundary_escape', project_root: ctx.projectRoot || null, repo_root: REPO_ROOT, count: escapes.length, entries: shown, note: '引擎不会自动回滚这些改动(见 W3 的能力边界),请人工核对 git status', }); if (outcome && outcome.status === 'pending') return outcome; // W12:banner 里不再重复改动清单 —— 清单已经在上面那条 node.boundary 事件的 entries 里 // (证据不丢),而 outcome.output 现在会被截成 output_excerpt 进 node.output 事件, // 塞 20 行清单只会把 agent 真正的输出挤出摘要窗口。这里只留「为什么被判 fail」这一句。 const banner = `[边界守卫] 本次调用在宿主仓库(${REPO_ROOT})留下 ${escapes.length} 处改动,` + '按越界判失败(改动清单见同一节点的 node.boundary 事件 entries)。'; return { ...(outcome || {}), status: 'fail', // W12 补完:以前这里恒 `|| null`,理由写的是「引擎不发明 fail_reason」。 // 那条理由针对的是**归因语义**(不许引擎替 validator 说「这是 impl_bug」), // 不是「引擎不许说自己为什么判失败」——结果就是概览 tab 上一个越界被判失败的节点 // 显示「原因 —」,而真正的原因(越界清单)躺在第三个 tab 里。 // 现在落 `boundary_escape`:agent 自己给了归因就尊重它(越界只是附加事实), // 没给才由引擎署名。取值不在任何图骨架的精确边上 → 仍走兜底边,路由不变。 failReason: (outcome && outcome.failReason) || ENGINE_FAIL_REASONS.BOUNDARY_ESCAPE, output: `${banner}\n\n${(outcome && outcome.output) || ''}`, }; } function readJsonOrNull(file) { try { return JSON.parse(fs.readFileSync(file, 'utf8')); } catch { return null; } } /** 节点产出契约(写的一侧由 describeArtifactContract 进提示词,读的一侧就是这里) */ const CONTRACT_FILE = 'contract.json'; /** 人工重试时拍下的上一轮契约快照(含「当时压根没有契约」这一情形,见 snapshotContract) */ const CONTRACT_SNAPSHOT_FILE = 'contract.prev.json'; const CONTRACT_SNAPSHOT_SCHEMA = 'contract-snapshot/v1'; /** * 人工重试前拍一张契约快照,**无论当前有没有 contract.json 都要拍**。 * * 旧实现是「文件不在就 return null 不拍」,配上 applyRetryCascade「没快照就什么都不做」, * 等于把三件不同的事塌缩成同一个「不级联」: * ① 这个节点本来就不产契约;② 重试前有契约、重试后没了;③ 重试前没有、重试后产出了。 * ②③ 是**契约真的变了**,级联是对的,却被静默吞掉。所以快照恒落盘,并显式记下 * `present` 这一位,让「有→无 / 无→有」两种变化能被 diff 看见。 * * @returns {{file: string, present: boolean, hash: string|null}} */ function snapshotContract(run, nodeRunId) { const dir = nodeArtifactDir(run, nodeRunId); const cur = path.join(dir, CONTRACT_FILE); const present = fs.existsSync(cur); const contract = present ? readJsonOrNull(cur) : null; const hash = present ? hashExport(contract) : null; const file = path.join(dir, CONTRACT_SNAPSHOT_FILE); try { fs.mkdirSync(dir, { recursive: true }); fs.writeFileSync( file, JSON.stringify({ schema: CONTRACT_SNAPSHOT_SCHEMA, present, hash, contract, capturedAt: now() }), 'utf8', ); } catch (e) { // 磁盘写不进去不该把「重试」这个操作本身弄失败:退化成「没快照」= 不级联(同下面的降级理由) console.warn(`[graph-engine] 契约快照写入失败 node=${nodeRunId}: ${e.message}`); return { file: null, present, hash }; } return { file, present, hash }; } /** * 重跑成功后判级联:有快照(说明这个节点被人工重试过)才判,没快照什么都不做。 * 四分类(added / changed / removed / unchanged,同 loops-l3.md 的增量算法): * **unchanged 保持原状不重跑**;只要出现任一 added/changed/removed 就把 forward 下游打回 * pending,等本节点的 success 出边重新点亮它们。 * * ── 读不到 contract.json 时的降级:**视为契约未变、不级联**(显式选择,不是遗漏)── * 理由三条: * 1. 代价不对称。级联的后果是把下游已完成的产物全部作废重跑,烧的是同一个 root run * 预算池里的真钱;不级联的后果是下游拿着旧产物,人工在控制台对单个下游点一次「重试」 * 就能补救。拿「无证据」当「变了」,等于把 §四.C 明确否决的「重试默认级联下游」 * 变成事实上的默认行为。 * 2. 绝大多数节点本来就不产契约(模板里 contract.json 是可选项),把它们全判成 dirty * 会让这条判据退化成一个恒真的开关,等于判据不存在。 * 3. 真正的变化并没有落进这个降级里:快照恒落盘且带 present 位,所以「有→无」「无→有」 * 都会被 diffContracts 算成 removed / added 而级联。落到「不级联」的只剩 * **重试前后都没有契约**这一种,那正是「这个节点不产契约」的准确刻画。 * * 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_SNAPSHOT_FILE); if (!fs.existsSync(prevFile)) return null; const snap = readJsonOrNull(prevFile); // 快照信封(本版)与裸契约(旧版残留文件)都要认,否则升级那一刻的在途重试会把 // 信封本身当契约 diff,判出一堆假的 changed。 const enveloped = snap && snap.schema === CONTRACT_SNAPSHOT_SCHEMA; const prev = enveloped ? snap.contract : snap; const prevPresent = enveloped ? Boolean(snap.present) : snap !== null; const curFile = path.join(dir, CONTRACT_FILE); const nextPresent = fs.existsSync(curFile); const next = nextPresent ? readJsonOrNull(curFile) : null; try { fs.unlinkSync(prevFile); // 快照一次性,判完即弃 } catch { /* 已经被别人删掉:判据照常出结果 */ } const diff = diffContracts(prev, next); // evidence 让控制台/调用方分得清「判过且没变」和「压根没有契约可判」—— // 两者都不级联,但只有后者是降级,前者是真结论。 diff.evidence = prevPresent || nextPresent ? 'compared' : 'no-contract'; diff.prevHash = enveloped ? snap.hash : (prevPresent ? hashExport(prev) : null); diff.nextHash = nextPresent ? hashExport(next) : null; diff.reset = diff.dirty ? resetForwardDownstream(runId, def, nodeRun.def_node_id) : []; sse.emitEvent(runId, 'node.contract', diff.evidence === 'no-contract' ? `${nodeRun.def_node_id}: 重试前后都没有 ${CONTRACT_FILE},无判据 → 按契约未变处理,不级联下游` : `${nodeRun.def_node_id}: 契约 ${diff.dirty ? '变了' : '未变'}(+${diff.added.length}/~${diff.changed.length}/-${diff.removed.length})` + ` → ${diff.dirty ? `级联重置 ${diff.reset.length} 个下游` : '不级联'}`, { node_run_id: nodeRun.id, evidence: diff.evidence, dirty: diff.dirty, added: diff.added, changed: diff.changed, removed: diff.removed, unchanged: diff.unchanged, prev_hash: diff.prevHash, next_hash: diff.nextHash, reset: diff.reset, }, { ref: nodeRun.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 } }); } // ──────────────────────────────────────────────────────────────── // 自动模式 = 不停调 advanceBatch 的循环(不是另一套实现) // ──────────────────────────────────────────────────────────────── /** * 在一棵 run **子树**里推进一步:优先推最深的 active 子 run(父节点正卡在它上面)。 * 单张 run 时退化成 advanceBatch(runId)(27-P4 T4/T6 接入:以前这里调的是 advanceOne, * 一次只推一个节点;现在同一个「最深 active run」内部能一次并发跑多个 ready 节点)。 * * **形参是子树根,不要求是 root run**(R6-F3 问题 2)。控制台「进入子图」后那一层的五个 * 控制按钮作用在子 run 上(§六 验收 15),对子 run 点自动启动时 runAuto 调的就是 * advanceTree(子 runId)。旧实现直接 `listRunTree(传入值)` —— 那是按 `root_run_id = ?` 查的, * 传子 run 拿到**零行**,循环第 0 步就 break:HTTP 回 `ok:true / auto:true`, * 而那个 run 里的 ready 节点纹丝不动(返回 ok 却什么都不做,是最坏的形态)。 * * ponytail: 仍然只推**一个**「最深 active run」,同一 tick 里的**兄弟**子 run(比如两个并行 * subgraph 节点各自起的子 run)不会同时被推——它们靠 runAuto 的外层循环一轮一轮轮流服务, * 不是真并发。真并发只发生在**同一个 run 内部**(advanceBatch 的 fan-out)。上限是「兄弟子 run * 之间」这一维的吞吐,不影响正确性(不会有子 run 被饿死,只是不同时跑);升级路径是把这个 * for 循环也从「找到第一个 ok 就 return」改成对多个 active run 并发调 advanceBatch。 */ async function advanceTree(runId, opts = {}) { const self = getRun(runId); if (!self) return result(runId, null, 'run-not-found'); // 仍然只做索引查询:按 root_run_id 拿全树,再在内存里裁出「自己 + 后代」这棵子树。 // 保留 listRunTree 的 created_at 顺序 —— 后建的一定更深,倒序即「最深优先」。 const scope = new Set([self.id, ...descendantRuns(self).map((r) => r.id)]); const tree = listRunTree(self.root_run_id) .filter((r) => scope.has(r.id) && r.run_status === 'running'); // 后建的 run 一定更深(子 run 总是在父节点 running 之后建的),倒序即「最深优先」 let informative = null; for (const r of tree.reverse()) { const res = await advanceBatch(r.id, opts); if (res.ok) return res; // minor 3:ok:false 时**不能**把所有原因都塌缩成 'no-ready-node'。子层返回 // 'budget-exhausted'(还带着 budget.exhausted 明细)而树层报「没活可干」, // 自动模式的调用方和 REST 层就分不清「预算耗尽等人工加额度」和「真的跑完了」。 // 最深的那条有信息量的原因胜出;全是 no-ready-node 才落回原来的兜底。 if (!informative && res.reason && res.reason !== 'no-ready-node') informative = res; } return { ...result(runId, (getRun(runId) || {}).run_status, informative ? informative.reason : 'no-ready-node'), // budget 只有预算闸门那条路径才带,透上来给控制台顶栏用 ...(informative && informative.budget ? { budget: informative.budget } : {}), // 原因来自子树上的哪一层(runId 保持子树根,调用方拿它继续操作这棵子树) ...(informative ? { sourceRunId: informative.runId } : {}), }; } /** * 自动驱动:只要 pause_requested / stop_requested 都是假就在上一次 resolve 后立刻发起下一次。 * 与 advanceTree 同款语义 —— runId 是**子树根**,对子 run 起自动就只驱动它这棵子树。 * maxSteps 只是防跑飞的安全阀,真正的终止条件是预算(§四.A.3)与熔断(§四.D.2.1)。 * 27-P4 T4/T6 接入后,「同一个 run 内部」已经能并发;跨兄弟子 run 的并发仍是 advanceTree * 那条 ponytail(一次只推最深的一个),不在本函数里。 */ async function runAuto(runId, opts = {}) { const maxSteps = opts.maxSteps || 1000; let steps = 0; for (; steps < maxSteps; steps++) { const run = getRun(runId); 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(runId, opts); if (!res.ok) break; } return { ok: true, runId, steps, runStatus: (getRun(runId) || {}).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, reset: [], infraRerun: false, contractDiff: null, crossLayer: 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, }; } /** * 公司根 / task 产物目录:唯一实现在 `company-paths.js`(R-19.3)。 * 这里曾有一份本地实现、goal-contract.js 里有一份镜像,控制台路由是第三个调用方 —— * 三处都必须算出同一个字符串,靠注释约束「保持逐字同构」在第三份出现时就不成立了。 */ const { companyRootDir, taskDirRel } = require('./company-paths'); /** * R-19 产出侧:把节点这次落盘的文件登记进 graph_files。 * * **由引擎调,不给 agent 自己写** —— 原设计(§4.6 v0.1)也是这么定的,理由是 * 「否则索引必然和实际文件漂移」。本轮反复验证过这条:凡「靠 agent 自觉」的环节 * 几乎全部落空过(goal-contract 零调用、setPlanWorkGraph 零调用、三值透传只接了读那一半)。 * * kind 怎么定:先看这个节点的**出边**上声明了什么(规划时就写下了「我会产一份 brd」), * 再按文件名兜底。声明优先,因为那是图作者的意图;文件名只是巧合。 */ /** * T-mt25ecbd 产出闸门:这个节点的出边声明了**必需**(required)produces kind, * 而它的产物目录里**连一个候选文件都没有**——判「压根没执行到位」,按 fail 结算。 * * 刻意只判「空目录」,不判「declared kind 认领不上具体哪个文件」:后者是 * indexNodeOutputs 的分类问题(ambiguous 时退成 kind=artifact,下游拿 ⚠️ 提示词), * 分类不准不代表节点没干活——把它也当失败拦下来会误杀`⑦ 有歧义时不猜`那类合法场景 * (agent 确实写了东西,只是文件名让引擎猜不出哪份对应哪个 kind)。 * 本轮要拦的是 brd/prd/sdd 三个节点复现的那种更粗的情形:声明了要产出、 * 目录里什么候选文件都没有——那不是分类模糊,是根本没写。 * * 目录读不了/没声明时一律放行(同 indexNodeOutputs 的兜底:不该让基础设施问题 * 或没声明契约的节点被这道新闸门误杀)。 * * @returns {string[]} 一个都没写出来时返回声明的 kind 列表;有任何候选文件 = 放行([]) */ function unsatisfiedRequiredProduces(run, def, nodeRun) { if (!run || !nodeRun) return []; let declared = []; try { const layer = run.graph_kind === 'org' ? 'org' : 'work'; const outbound = ((def && def.edges) || []).filter( (e) => String(e.from) === String(nodeRun.def_node_id) && (e.edgeKind || 'forward') !== 'backward' ); for (const e of outbound) { for (const c of edgeContracts(layer, e.id)) { if ((c.role || 'produces') === 'produces' && c.required) declared.push(c.kind); } } declared = [...new Set(declared)]; } catch { return []; // 契约查不到不该拖垮结算(同 indexNodeOutputs 的兜底) } if (!declared.length) return []; let dir; let files = []; try { dir = nodeArtifactDir(run, nodeRun.id); if (!fs.existsSync(dir) || !fs.statSync(dir).isDirectory()) return declared; files = fs.readdirSync(dir).filter((f) => !f.startsWith('.')); } catch { return []; // 目录读不了不该拖垮结算 } const candidates = files.filter( (f) => !/^attempt-\d+\.json$/.test(f) && f !== CONTRACT_SNAPSHOT_FILE ); return candidates.length ? [] : declared; } function indexNodeOutputs(run, def, nodeRun) { if (!run || !nodeRun || !nodeRun.output_ref) return; let dir = nodeRun.output_ref; let files = []; try { if (!fs.existsSync(dir) || !fs.statSync(dir).isDirectory()) return; files = fs.readdirSync(dir).filter((f) => !f.startsWith('.')); } catch { return; // 目录读不了不该拖垮结算 } if (!files.length) return; // 出边声明的 kind:优先用它给文件定 kind。 // // **只收 role='produces'**:出边上的 needs 说的是「下游要吃这个 kind,而产出方不是紧邻上游」 // (中继声明,写在消费方入边上 —— 那与本节点的出边是同一条边),它不是本节点的产出。 // 混进来会坏两件事: // ① 认领作废:p-prd 的出边写 produces[prd]+needs[brd] → 未认领 kind 数 2 ≠ 1 → // 下面那条「一对一无歧义」的兜底不成立,文件退成 artifact,下游查不到实例。 // ② 更坏的错标:文件名里带 brd(如 brd_notes.md)会命中中继的 needs[brd], // 把它错标成 brd —— 错的 kind 让下游读到一份不是它要的文件,比 artifact 查不到更坏。 // role 列是随 needs/produces 一起落库的(work-graph-db 的 graph_edge_contracts), // 这里正是它存在的理由;旧库迁移行回填成 produces,所以缺省当 produces 处理。 let declared = []; try { const layer = run.graph_kind === 'org' ? 'org' : 'work'; const outbound = ((def && def.edges) || []).filter( (e) => String(e.from) === String(nodeRun.def_node_id) && (e.edgeKind || 'forward') !== 'backward' ); for (const e of outbound) { for (const c of edgeContracts(layer, e.id)) { if ((c.role || 'produces') !== 'needs') declared.push(c.kind); } } declared = [...new Set(declared)]; } catch { /* 契约查不到就只靠文件名 */ } const companyRoot = companyRootDir(run.company_id); // ── 第一遍:按文件名认领 kind // 引擎自己的簿记文件不是节点交付物,不进索引: // attempt-*.json 归因文件(readNodeProduct 读它) // contract.prev.json retryNode 为算 contract_hash 留的上一轮快照(CONTRACT_SNAPSHOT_FILE) // 漏掉后者的话,每次人工重试都给 graph_files 添一行 kind=artifact 的噪声。 const candidates = files.filter( (f) => !/^attempt-\d+\.json$/.test(f) && f !== CONTRACT_SNAPSHOT_FILE ); const assigned = candidates.map((f) => { const base = f.replace(/\.[^.]+$/, '').toLowerCase(); // 声明优先:出边说会产 brd,且文件名里就有 brd → 定 brd const kind = declared.find((k) => base === k || base.includes(k)) || (KNOWN_FILE_KINDS.includes(base) ? base : null); return { file: f, kind }; }); // ── 第二遍:文件名认不出来时,靠**声明**认领 // // 第五次真跑里这一步成立纯属侥幸:brd 节点恰好把文件命名成 `brd.md`。同一个本体 // 换写 `业务需求文档.md` 就会落成 kind=artifact,下游入边声明的 brd 查不到实例, // 提示词退化成「⚠️ 索引里查不到实例」—— 刚关掉的那个断点换个形式重现。 // agent 给文件起中文名 / 加前缀是完全正常的行为,不该由它决定索引对不对。 // // 只在**一对一无歧义**时认领:还差一个 kind 没人产、正好还剩一个认不出的文件。 // ponytail: 多个未认领 kind 配多个未认领文件时不猜(猜错比留 artifact 更坏 —— // 错的 kind 会让下游读到一份不是它要的文件,而 artifact 只是查不到)。 // 升级路径:让 agent 在 attempt-*.json 里自报 {kind: 文件名} 的映射,产出侧不再靠推断。 const unmatchedKinds = declared.filter((k) => !assigned.some((a) => a.kind === k)); const unclaimed = assigned.filter((a) => a.kind === null); if (unmatchedKinds.length === 1 && unclaimed.length === 1) { unclaimed[0].kind = unmatchedKinds[0]; } for (const { file: f, kind } of assigned) { const abs = path.join(dir, f); try { recordFile({ companyId: run.company_id, taskId: run.task_id, relPath: path.relative(companyRoot, abs), kind: kind || 'artifact', runId: run.id, nodeRunId: nodeRun.id, }); } catch (e) { timelineNote({ run, runId: run.id, nodeRunId: nodeRun.id, attempt: nodeRun.attempt }, 'node.output', `产物索引写入失败 ${f}: ${e.message}`, { degraded: 'file_index_write_failed' }); } } } /** * 入边声明的 kind → 产物索引里的绝对路径,一个 kind 一行。 * * 抽出来是为了让**执行侧**(describeInputs,agent_call 节点)与**规划侧** * (describeOrgContracts,subgraph 节点交给 workgraph-planning 的那份提示词) * 走同一段代码:三种口径(查得到给绝对路径 / required 查不到给 ⚠️ / * 可选查不到整行省略)就不可能在两处漂移 —— 复制一遍再各自演化正是本轮在防的那类断链。 */ /** * T-mt25ecbd 输入闸门:入边声明的**必需**(required)kind,有几个在产物索引里真的 * 一份实例都查不到——与 describeInputs 给 agent 看的 ⚠️ 提示同一份数据源 * (inboundContracts + findFiles),区别是这里把它变成执行前的硬拦截,不再只是 * 「提示词里有一行警告,agent 爱信不信」。 * * 只看 required 且缺实例的:可选(required:false)输入本来就允许没有,不该拦。 * 索引读不出来(层查询异常)时放行——同 describeInputs 的兜底,基础设施问题不该 * 误伤节点执行。 * * @returns {string[]} 缺实例的必需 kind 列表;空数组 = 放行 */ function requiredInputsMissing(ctx, def) { if (!ctx || !ctx.run || ctx.nodeKind === 'subgraph') return []; let contracts = []; try { const layer = ctx.run.graph_kind === 'org' ? 'org' : 'work'; contracts = inboundContracts(layer, (def && def.edges) || [], ctx.defNodeId); } catch { return []; } const requiredKinds = [...new Set(contracts.filter((c) => c.required).map((c) => c.kind))]; if (!requiredKinds.length) return []; let found; try { found = findFiles({ companyId: ctx.companyId, taskId: ctx.run.task_id, kinds: requiredKinds }); } catch { return []; } const haveKinds = new Set(found.map((f) => f.kind)); return requiredKinds.filter((k) => !haveKinds.has(k)); } function resolveInboundLines(ctx, contracts) { const declared = [...new Set(contracts.map((c) => c.kind))]; if (!declared.length) return []; const found = findFiles({ companyId: ctx.companyId, taskId: ctx.run.task_id, kinds: declared }); const byKind = new Map(); for (const f of found) byKind.set(f.kind, f); const root = companyRootDir(ctx.companyId); const lines = []; for (const kind of declared) { const hit = byKind.get(kind); const required = contracts.some((c) => c.kind === kind && c.required); if (hit) lines.push(`- ${kind}:${path.join(root, hit.rel_path)}`); else if (required) lines.push(`- ${kind}:⚠️ 上游声明会产出,但索引里查不到实例`); } return lines; } /** * 「这个节点需要什么输入、在哪」—— 两类数据流都查(R-19.3 定案①): * ① 沿边传递:入边上声明的 kind(受拓扑相邻性限制) * ② 任务级共享:本 task 下的 acceptance / contract 这类,**不受相邻性限制** * —— R-9.3 的「planner 生成、validator 执行」是隔着一个 coder 的非相邻依赖, * 纯边绑定表达不了,所以必须有这一类。 * * 声明了却查不到实例时**明说缺哪个 kind**,不要静默省略:那正是引擎能在跑之前 * 发现「你要吃 brd,但上游没人产 brd」的地方(R-19.4)。 */ function describeInputs(ctx, def) { if (!ctx || !ctx.run || ctx.nodeKind === 'subgraph') return ''; let contracts = []; let shared = []; try { const layer = ctx.run.graph_kind === 'org' ? 'org' : 'work'; contracts = inboundContracts(layer, (def && def.edges) || [], ctx.defNodeId); shared = findFiles({ companyId: ctx.companyId, taskId: ctx.run.task_id, kinds: TASK_SHARED_KINDS, }); } catch { return ''; // 索引读不出来不该拖垮节点执行 } const declared = [...new Set(contracts.map((c) => c.kind))]; const lines = resolveInboundLines(ctx, contracts); // 任务级共享按 kind 去重,只列最新的那一份(findFiles 是 created_at 升序,所以后写的覆盖前写的 // —— 与上面沿边那一半的 byKind.set 同一个口径)。 // // 不去重的后果在第五次真跑里已经出现:brd 与 prd 两个节点各写了一份 contract.json, // 都是 kind=contract,于是往后每个节点的提示词里「contract(任务级共享)」逐个累加。 // 节点一多就是噪声,而且 agent 无从判断该信哪一份 —— 给它两个同名同类的路径, // 等于把一个我们自己都没定的选择推给它。 // ponytail: 「最新的赢」是约定而非事实推导。根因是节点私有目录里的文件也会按文件名 // 落成 task-shared 的 kind;真正的修法是任务级产物只认 tasks/{taskId}/ 下的那份 // (R-19.3 的产物区),但那会改掉「节点私有目录写 acceptance 也算共享」这条现有语义, // 归属 R-9.3 非相邻依赖那条线一起定。 const sharedByKind = new Map(); for (const f of shared) { if (declared.includes(f.kind)) continue; // 已在上面列过 sharedByKind.set(f.kind, f); } for (const [kind, f] of sharedByKind) { lines.push(`- ${kind}(任务级共享):${path.join(companyRootDir(ctx.companyId), f.rel_path)}`); } if (!lines.length) return ''; return [ '', '', '[上游输入 —— 引擎已按图上声明的数据契约解析好,直接读这些路径]', ...lines, '标 ⚠️ 的表示图上声明了这份输入、但索引里没有实例:先确认是不是上游没产出,', '不要自己去别的节点私有目录里翻(跨节点私有目录不可读,见写入边界)。', ].join('\n'); } /** * 拼进提示词的产物目录约定(§四.A.5 + validator-triager/agent.md「产出格式」)。 * * 路径与 readNodeProduct 读的**是同一个**(nodeArtifactDir),文件名给绝对路径, * 不留 0 基 / 1 基的解释空间——模板里 `attempt-{n}.json` 的 n 没定死基数, * 这里直接把本轮该写的文件名算好告诉它。 * * **contract.json 同理**(R5 接上的断链 3):§四.C 末段定死了「重试要不要级联下游, * 由产出契约是否真的变了决定」,判据是 contract_hash。但在此之前提示词里一个字都没提 * contract.json,于是 snapshotContract 恒返回 null、整条判据是死路径。写的一侧不被告知 * 往哪写,读的一侧就永远读不到——与产物目录当初那个缺口同型。 */ function describeArtifactContract(ctx) { const dir = safeNodeArtifactDir(ctx); if (!dir) return ''; // run 缺 company_id/root_run_id:拼不出目录就不编一个假的塞给 agent return [ ...describeWriteBoundary(ctx, dir), '', '', '[节点产物目录]', `你的私有目录(可读可写):${dir}`, `本轮的结构化产出/归因写到:${path.join(dir, `attempt-${ctx.attempt}.json`)}`, '格式见 system/agents/validator-triager/agent.md「产出格式」' + '(schema=validator-triager/v1,改名前叫 validator-reflection/v1,' + '引擎只读 verdict / fail_reason / remediation_note 三个字段、不校验 schema 串,两种写法都能解析;' + 'verdict ∈ attributed|inconclusive|no_prior_failure,fail_reason ∈ ' + `${[...ATTRIBUTION_REASONS].join('|')} 或 null)。`, // R-17.1 的**产生端**。四环里的第一环只能在这里接:模板 agent.md 不归本轮改, // 而所有 agent_call 节点的任务描述都会拼上这一段,所以提示词这一侧是通的。 // 不写这段的话,target_node 永远没人产出,跨层寻址整条链是死路径(§八.15 同型)。 '判失败时若能精确到「是哪个环节」,再多写两个字段,引擎会据此**只重跑那个节点**而不是整张图重来(R-17.1):', ' `"target_node"`:被点名的 work graph **节点名**(不是本体名,是图上那个节点的 nodeName);', ' `"target_org_node"`:可选,它属于哪个 org node。给了就必须与回退边的目标一致,' + '对不上引擎会忽略整个地址、退回整张图重跑。写不出来就别写,留空不是错误。', '', '[产出契约 contract.json]', `你若产出了**下游节点要依赖的东西**(接口、数据格式、文件路径、配置键…),写到:${path.join(dir, CONTRACT_FILE)}`, '形状:{"exports": {"<键>": <任意可 JSON 序列化的值>}},一个键 = 一件下游依赖的东西。', '这份文件是「人工重试后要不要把下游全部作废重跑」的**唯一判据**(§四.C 末段):' + '引擎按 exports 逐键算 hash,只要有 added/changed/removed 就级联重置下游,全 unchanged 就不动下游。', '所以**每一轮都要把它完整重写一遍**(不是增量补丁);确实没有对外契约的节点可以不写这个文件。', '', `任务级共享产物区(只读):${path.join(runRootDir(ctx.run), 'artifacts')}`, '不要读写其它节点的私有目录(§四.A.5 的隔离规则)。', '判词类节点(validator / triager):**没有 attempt-*.json = 无法判定**,引擎会按失败结算走兜底边。', ].join('\n'); } /** * 写入边界进提示词(W2 的「告诉它写哪」+ W3 的「不许写别处」)。 * * 这是**唯一在真 CLI 跑起来之前就能生效的一半**:引擎拦不住子进程写盘(见 applyBoundaryGuard), * 提示词是「必要但不充分」的那一半,事后扫描是另一半。两半都写在同一个文件里,改一处 * 不会忘了另一处。 * * 段头用 prompt-builder 导出的常量:**graph 节点链路的交付物归宿只有这一份**。 * PromptBuilder 的运行时坐标声明按这段在不在场分叉(taskCarriesGraphWriteBoundary), * 在场就不再写「交付物写公司 workspace」那一行 —— 那条对 graph 节点是错的: * workspace 既不被 indexNodeOutputs 索引,又在 REPO_ROOT 下、会被边界守卫判越界。 * 所以这个段头改字面量等于悄悄让分叉失效,必须动常量。 * * @param {string} artifactDir 节点私有目录(调用方已算好,避免重复拼) * @returns {string[]} 拼进任务描述的行;调用方负责 join */ function describeWriteBoundary(ctx, artifactDir) { const root = (ctx && ctx.projectRoot) || null; const lines = ['', '', GRAPH_WRITE_BOUNDARY_HEADER]; if (root) { lines.push( `本次任务的项目根:${root}`, '**一切代码 / 测试 / 配置 / 交付物都写在这个项目根里面**(它是个空目录,结构你自己定)。', `你只被允许写两个位置:① 上面这个项目根及其子目录;② 你的节点私有目录 ${artifactDir}(结构化产出/归因/契约)。`, ); } else { lines.push( '本次任务**没有分配到项目根**(分配失败,见 timeline)。', `在拿到项目根之前,你只被允许写你的节点私有目录 ${artifactDir},不要在别处创建任何文件。`, ); } lines.push( `禁止写这两处之外的任何路径,**尤其是 ${REPO_ROOT}**(这是引擎自己的源码仓库,不是你的工作区)。`, '写文件一律用上面给的**绝对路径**拼接:你的进程 cwd 不是项目根,用相对路径会写到别人家里。', '引擎会在你这次调用前后各扫一次宿主仓库的 git 状态;发现越界改动会把本节点判失败并留痕' + '(引擎不会替你回滚,越界等于这一轮白跑)。', ); return lines; } /** 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 §四)。 * * **先读回 artifacts/task-goal.json**(R5 发现的第 10 处同型漏接): * 用户下达的那句自然语言由 company.js 的 writeTaskGoalArtifact 落盘到 * `runs/{rootRunId}/artifacts/task-goal.json`,它自己的注释写着「手动模式下每次 /next * 由控制台发起,需要能把原始 task-goal 读回来」——但全仓库**没有任何代码读过这个文件**。 * 于是「产生 → 传递 → 落库 → 读回」在最后一环断掉,后果不是少个字段而是: * - manual 模式(`POST /:id/next` 不带 body.task,控制台也不会带) * - 「暂停后点启动恢复自动」(`POST /:id/start` 根本没有 task 参数) * 这两条路径上,agent 收到的任务描述退化成 `执行图节点任务(goal=..., task=T-xxx)` * ——生产里 agent 压根不知道要干什么,却照样烧预算跑一遍。 * opts.task 仍然优先(调用方显式给了就听调用方的),这里只补上「没人给时去读那份落盘的」。 */ function describeTask(run) { const ids = [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); const suffix = ids.length ? `(${ids.join(', ')})` : ''; let goal = null; try { goal = readJsonOrNull(path.join(runRootDir(run), 'artifacts', 'task-goal.json')); } catch { goal = null; // run 缺列拼不出目录:退回下面的兜底文案 } const task = goal && typeof goal.task === 'string' ? goal.task.trim() : ''; return task ? `${task}\n\n[任务标识]${suffix}` : `执行图节点任务${suffix}`; } // ──────────────────────────────────────────────────────────────── // W9 · 失败上下文进提示词(27-P3B §四 W9) // ──────────────────────────────────────────────────────────────── /** * `validator-triager/agent.md` 与 `tools.md` 都写着:「本次调用的提示词 = **失败上下文** * (哪个节点、第几 attempt、失败摘要)+ task-goal + 上一轮 remediation_note」。 * 前两项一直是真的(`describeTask` / `latestRemediationNote`),**「失败上下文」这一项 * 全仓库没有任何代码提供** —— 于是 agent.md 给 `target_node` 列的三个取名来源里, * 第 (a) 个「引擎告诉你的那个失败节点名」在生产里永远是空的,归因只能靠猜。 * * ── 数据源为什么是 timeline 而不是 graph_run_nodes ────────────────── * 直觉上应该 `SELECT * FROM graph_run_nodes WHERE status IN ('fail','blocked')`, * 但那样在**最需要它的那条路径上恰好取不到值**: * benchmark-validator 失败 → backward 边退回 validator-triager → * `handleBackwardEdge` 调 `resetForwardDownstream`,而 benchmark-validator 正是 * validator-triager 的 forward 下游 → 它被重置成 `pending`, * **`fail_reason` / `remediation_note` 一并被 SET NULL**(见该函数)。 * 等 validator-triager 真正跑起来时,库里已经没有任何一行记着「刚才谁失败了」。 * 而 `timeline.ndjson` 是 append-only、永不改写已有行的审计轨(§三.3), * 失败那一刻的 `node.output`(带 W12 那一轮加的 `output_excerpt`)原样躺在里面。 * 所以这里读的是审计轨 —— 这也顺带给 W12 的 `output_excerpt` 加了第二个消费方: * 它不再只是给人看的,引擎自己会把它读回来喂给下一个节点。 * * ponytail: 每个节点起跑前读一次整份 timeline.ndjson(切尾 2000 行)。 * 上限 = O(文件大小),而文件大小被任务级预算(十几到几十次调用)钳住,眼下无所谓。 * 真要拆的话,升级路径是给 sse 加一个「按 ref 倒读最近 N 条」的接口,而不是加一列缓存。 */ const FAILURE_CONTEXT_SCAN_LINES = 2000; function readRunTimeline(runId) { try { const p = sse.timelinePath(runId); if (!fs.existsSync(p)) return []; return fs.readFileSync(p, 'utf8') .split('\n').filter(Boolean) .slice(-FAILURE_CONTEXT_SCAN_LINES) .map(safeJsonParse) .filter((e) => e && e.payload); } catch { return []; // 审计轨读不出来不该把节点跑挂,退化成「没有失败上下文」 } } /** * 同一个 node_run_id 上、不晚于 idx 的那些事件里凑齐四样东西。 * 一条事件凑不齐是常态:`node.status`(settleNode 那次)带 `failReason` 但**不带 node_name** * (它的 extra 只有 attempt/remediation_note),`agent.call` / `node.output` 才带名字。 * 键名两种写法都认:`node.output` 的 data 是手写的 snake_case,`node.status` 的 data 由 * `transitionNode` 把 patch 原样摊进去,是 camelCase。 */ function collectFailureDetail(entries, idx, nodeRunId) { const d = { nodeName: null, attempt: null, failReason: null, excerpt: null, from: null }; for (let i = idx; i >= 0; i--) { const e = entries[i]; const p = e.payload || {}; if (p.node_run_id !== nodeRunId) continue; if (d.nodeName == null && p.node_name) d.nodeName = p.node_name; if (d.attempt == null && Number.isInteger(p.attempt)) d.attempt = p.attempt; if (d.failReason == null) d.failReason = p.fail_reason || p.failReason || null; if (d.excerpt == null) { const text = e.kind === 'node.error' ? p.error : e.kind === 'node.output' ? p.output_excerpt : e.kind === 'node.boundary' && p.violation ? e.summary : e.kind === 'node.triage' ? e.summary : null; if (text) { d.excerpt = text; d.from = e.kind; } } } return d; } /** 本 run 里最近一次节点失败;没有就 null(首轮进场的绝大多数节点都是 null) */ function findLastFailure(ctx) { const entries = readRunTimeline(ctx.runId).filter((e) => e.run_id === ctx.runId); for (let i = entries.length - 1; i >= 0; i--) { const e = entries[i]; const p = e.payload; if (!p.node_run_id) continue; // 两种判据都要:`node.output` 覆盖 agent_call 节点(带摘要), // `node.status → fail/blocked` 覆盖 subgraph 回写、死路 blocked、熔断这几条没有 node.output 的路径。 const failed = (e.kind === 'node.output' && p.status === 'fail') || (e.kind === 'node.status' && (p.to === 'fail' || p.to === 'blocked')); if (!failed) continue; return { nodeRunId: p.node_run_id, ...collectFailureDetail(entries, i, p.node_run_id) }; } return null; } /** node_run_id → 图上的节点名(timeline 里没带名字时的兜底) */ function nodeNameOf(ctx, nodeRunId) { try { const rn = getRunNode(nodeRunId); if (!rn) return null; const d = defIndex(ctx.run).nodes.get(rn.def_node_id); return (d && d.name) || rn.def_node_id; } catch { return null; } } /** * 拼进任务描述的「失败上下文」段。**没有前序失败就返回空串**(不塞空段落白烧上下文)—— * 首轮进场的节点本来就该看到「无」,而模板已经教了 `verdict: "no_prior_failure"` 这条路。 */ function describeFailureContext(ctx) { const hit = findLastFailure(ctx); if (!hit) return ''; const name = hit.nodeName || nodeNameOf(ctx, hit.nodeRunId) || hit.nodeRunId; const attempt = Number.isInteger(hit.attempt) ? hit.attempt : 0; const isSelf = hit.nodeRunId === ctx.nodeRunId; return [ '', '', '[前序失败上下文 —— 引擎提供,本 run 里最近的一次节点失败]', `失败节点:**${name}**${isSelf ? '(就是你自己的上一轮)' : ''}(node_run_id=${hit.nodeRunId})`, `第几次尝试:attempt-${attempt}(人类计数第 ${attempt + 1} 次)`, `当时落库的 fail_reason:${hit.failReason || 'null(没有归因,走的是兜底边)'}`, `失败摘要${hit.from ? `(来自 ${hit.from})` : ''}:${clipText(hit.excerpt) || '(那一轮没有留下任何输出摘要)'}`, '你若是归因节点(validator-triager):上面这个**节点名**就是 agent.md 里 `target_node` ' + '三个取名来源中的第 (a) 个(引擎点名的失败节点),可直接用;' + '判断它不是真正的根因时再走 (b)/(c)。', '(本段只在确实有前序失败时出现;没有这一段 = 本 run 至今没有节点失败过。)', ].join('\n'); } /** * 「应该如何修复」,作为下一轮的上下文补充(§四.D.2.1 第 3 步)。 * * 判词就在**这个节点自己**那一行(graph_run_nodes.remediation_note),由点亮它的那次 * 状态转换写入,见上方「判词的载体」一节。这样天然满足两条以前靠约定维持的要求: * - **不串台**:验收 9b 那种多条回退边并存的图里,给 a 的判词只可能写在 a 那一行, * 不可能被 b/c 读到(以前按 to_node_run_id 过滤 loop record 是在补这个); * - **不吃陈货**:每次点亮都显式写值(无判词写 null),不存在「读到上一轮那条」的窗口 * —— loop record 那条路上 created_at 只有毫秒精度、同毫秒的多条记录排序不定, * 用它取「最新一条」本身就不可靠。 */ function latestRemediationNote(nodeRunId) { const rn = nodeRunId ? getRunNode(nodeRunId) : null; return (rn && rn.remediation_note) || null; } module.exports = { // 核心原语 advanceOne, advanceBatch, runClaimedNode, advanceTree, runAuto, settleNode, settleRun, // 五个控制操作 startRun, nextStep, pauseRun, stopRun, retryNode, descendantRuns, landPause, // 任务级总预算(§四.A.3) checkBudget, spendBudget, extendBudget, settleWallclock, wallclockElapsedMs, suspendRunTree, // 27-P4 T5:账户欠费门(可插拔桩子)+ 自动恢复 checkBilling, setBillingGate, resumeBillingClearedRuns, // 27-P4 §4.4:墙钟预算看门狗(manual 模式无人点「下一步」/ auto 驱动循环随进程重启丢失时的兜底) sweepWallclockBudgets, startWallclockWatchdog, stopWallclockWatchdog, // 重试级联判据(§四.C 末段) diffContracts, applyRetryCascade, snapshotContract, // run 生命周期辅助 markRunReady, seedEntryNodes, // executor 与注入点 executeSubgraph, defaultAgentExecutor, // 导出只为让真实链路 smoke 能直接问一句「守卫在这条路径上开着没有」 // (graph-real-chain-smoke.test.js ③)。生产里它只被 advanceOne 调。 boundaryGuardApplicable, defaultPlanWorkGraph, setAgentExecutor, setPlanWorkGraph, extractPlannedGraph, // W2/W3 项目根与边界守卫(27-P3B)。 // ensureProjectRoot 的真实调用方是 advanceOne,diffRepoStatus 的是 applyBoundaryGuard; // 导出只是让单测能直接验这两条判据,不是新开一圈公开面。 ensureProjectRoot, // readProjectRootRecord:控制台「产物」面板要知道这个任务的真实交付物写在哪—— // graph_files 只索引每个节点自己私有目录下的产出,project-root.json 指向的才是 // 跨节点共享的项目根(真实代码落盘的地方,见本文件 W2 那段大注释)。 readProjectRootRecord, diffRepoStatus, // 真实调用方是 advanceOne(每次节点执行前后各一次)。导出 + 可注入的 after 快照, // 是为了让「越界 → success 降级成 fail + fail_reason=boundary_escape」这条分支 // 不必起真 CLI 也能验(第一批终验记录点名的那处零覆盖)。 applyBoundaryGuard, // 归属判定:agent 运行态文件不算越界(真跑冒烟的断点就在这,见函数上方注释) splitUnattributable, sharedWorktreeReason, // 36 号文档:图执行结果 → memory.md 桥接。真实调用方是 runClaimedNode(结算之后); // 导出只为让"success 那一行怎么格式化"不必靠真起 CLI 触发 isRealDispatch 才能验 // (fail 分支可以借道 body_not_ready 走端到端,success 分支目前只有这条直调路径)。 recordNodeOutcomeMemory, // 36 号文档追加:subgraph 节点(workgraph-planning 规划出的图)结算 → memory.md 桥接。 // 真实调用方是 settleNode;导出只为不必真起 CLI 触发 planWorkGraph===defaultPlanWorkGraph // 就能验证格式化与闸门逻辑(同 recordNodeOutcomeMemory 的导出理由)。 recordWorkgraphPlanningMemory, // 选边 / 熔断 / 透传 selectFailEdge, handleBackwardEdge, resetForwardDownstream, deriveChildOutcome, // R-17.1 跨层寻址的四个函数(normalizeNodeRef / parseNodeRef / resolveCrossLayerTarget / // reopenChildRun)**故意不导出**:它们只有本模块内的调用方,导出等于凭空多一圈公开面。 // 要验它们的行为就通过 advanceOne / handleBackwardEdge 这些真入口验(用例就是这么写的)。 ATTRIBUTION_NODE_NAME, INFRA_FLAKE_REASON, // W12 补完:引擎自判失败的三个取值。导出是给防线用例断言「概览 tab 上会看到哪几个词」, // 以及给将来想写精确 fail 边的图规划者一个单一真相源(别再各自抄字符串)。 ENGINE_FAIL_REASONS, // W9:失败上下文段的构造与它的数据源。describeFailureContext 的真实调用方是 advanceOne, // 导出只是让单测能不起真 CLI 直接验这一段文本(同 ensureProjectRoot / diffRepoStatus 的口径)。 describeFailureContext, // 「本体装配静默坑」修复:截断可见性只有一条判据(截断了就必须带⚠️提示), // 导出给单测直接验,同 describeFailureContext 的口径。 describeRoleDefinition, LOOP_HARD_CAP_EDGE, LOOP_HARD_CAP_RUN, };