graph-engine.js 259 KB

Edit Raw Blame History

/**
 * 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,
  // 39 号文档「确定性通路」：回退熔断收敛时顺手开一条 draft task，见 maybeProposeReplanTask
  createTask,
  getGoal,
} = 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<AgentResult>}
 */
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-<taskId>.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-<taskId>.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<AdvanceResult>}
 *
 * @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<AdvanceResult>}
 */
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);
    // 39 号文档「确定性通路」：熔断收敛之后顺手把这个卡点回写成 goal/task 板上一条 draft
    // task，不等 ceo 心跳才发现。best-effort，绝不能反过来影响上面已经做完的熔断收敛。
    maybeProposeReplanTask(runId, rootRunId, nodeRun, def, 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;
}

/**
 * 39 号文档「确定性通路」：run 树级/单边回退熔断（force_terminated）收敛时，顺手在同一个
 * goal 下开一条 draft task，把卡住的诊断信息交回 goal/task 板，而不是只留一个 blocked
 * 节点等人翻控制台才发现。
 *
 * 只在根 run 触发一次（`runId === rootRunId`）：嵌套子图自己的回退熔断会先在子层
 * blockNode，结论再沿子 run 边界冒泡成父层（org）节点的失败——冒泡到根时同一次「卡住」
 * 只应该开一条 task，不是每层各开一条。
 *
 * 落库时机是 `draft`，不自动 enqueue（与 §四.9 task-planning 同一条哲学：这是一次
 * 「猜」，不该自动花钱）；`report_to` 继承原 goal 的 `report_to`（§四.2 已拍板）。
 *
 * 纯 best-effort：任何一步查不到/写不进都吞掉，不能让「顺手开个 task」反过来打断
 * 熔断收敛本身——熔断收敛是安全关键路径，这个不是。
 */
function maybeProposeReplanTask(runId, rootRunId, nodeRun, def, remediationNote) {
  if (runId !== rootRunId) return;
  try {
    const db = getDb();
    const task = db.prepare(`SELECT * FROM tasks WHERE current_run_id = ?`).get(rootRunId);
    if (!task) return; // 这条 run 树当前不是任何 task 的 current_run_id（例如已被重跑顶替），不开
    const goal = getGoal(task.goal_id);
    if (!goal) return;

    const d = def.nodes.get(nodeRun.def_node_id) || {};
    const nodeName = d.name || nodeRun.def_node_id;
    const reason = remediationNote || `节点 ${nodeName} 反复回退撞了熔断硬顶，未能收敛`;

    createTask({
      goalId: goal.id,
      companyId: goal.company_id,
      title: `[待重新规划] ${(task.title || task.description || goal.title || '').slice(0, 160)}`,
      description: `任务 ${task.id}（${task.title || task.description || ''}）执行时在节点 `
        + `${nodeName} 反复回退未能收敛，已停止自动重试。\n\n诊断：${reason}\n\n`
        + `此任务由系统自动生成（草稿），需要人工或 ceo 审阅后决定：调整原任务描述重新下达，还是关闭。`,
      executionMode: 'manual',
      priority: goal.priority || 'P1',
      reportTo: goal.report_to || null,
      createdBy: 'system',
    });
  } catch (err) {
    // best-effort：吞掉。上一步的熔断收敛（blockNode）已经做完，这里出错不影响主路径。
    console.error('[graph-engine] maybeProposeReplanTask 失败（不影响熔断收敛）：', err.message);
  }
}

/**
 * 解析跨层地址 → 「该退到哪个子 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<boolean> |
 *   {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;
}

/**
 * 39 号文档「公司级目标代码目录」：读 `company.json.target_dir`/`target_mode`。
 *
 * 走 `companyRootDir` 而不是 `company-service.get`——理由与 `work-graph-db.maxActiveRootRuns`
 * 头部注释一致（那边的判据是这个仓库唯一持有者，这里只是同一份配置的第二个读者，
 * 字段名必须逐字对齐，不能另起 camelCase）：`companyRootDir` 认 GRAPH_RUNS_DIR 测试覆盖，
 * `company-service.get` 不认，用后者会导致设了 GRAPH_RUNS_DIR 的单测读到真仓库的
 * `data/companies/`，撞上同名公司就读到真配置。
 *
 * 读不到 / 格式不对一律当没配（回落 sandbox），不抛错——目标目录是可选加成，
 * 不能因为它读取失败就让整个项目根分配跟着炸。
 */
function readCompanyTargetConfig(companyId) {
  if (!companyId) return null;
  try {
    const fp = path.join(companyRootDir(companyId), 'company.json');
    if (!fs.existsSync(fp)) return null;
    const cfg = JSON.parse(fs.readFileSync(fp, 'utf-8'));
    if (!cfg.target_dir || typeof cfg.target_dir !== 'string') return null;
    const target_mode = cfg.target_mode === 'linked-new' ? 'linked-new' : 'linked-existing';
    return { target_dir: cfg.target_dir, target_mode };
  } catch {
    return null;
  }
}

/**
 * 真正在盘上开一个目录。
 *
 * GRAPH_RUNS_DIR 覆盖时项目根跟着走（`{RUNS_ROOT}/{rootRunId}/project`），理由与本文件
 * 顶部 RUNS_ROOT 那段注释完全一样：run 的产物被重定向到临时目录，项目根却还落在仓库
 * `data/companies/` 的话，测试每跑一次就往仓库里堆一个空项目目录。两处必须同源。
 *
 * 配了 `company.json.target_dir` 时，该公司**所有** run 解析到同一个固定 root_path，
 * 不再走 sandbox 那套「每 run 一个新 pid」——39 号文档 P5：公司级目标代码目录。
 * 校验失败（比如 git 树不干净）直接抛出，不静默降级回 sandbox：配错了必须看得见，
 * 不能让 agent 写到一个跟用户预期不一样的空目录里却毫无察觉。
 */
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' };
  }
  const targetCfg = readCompanyTargetConfig(run.company_id);
  if (targetCfg) {
    const root_path = require('./devloop/path-resolver').resolveFixedTarget({
      target_path: targetCfg.target_dir, mode: targetCfg.target_mode,
    });
    return { pid: 'company-target', root_path, mode: targetCfg.target_mode };
  }
  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,
};