宜博 / lemonboss

/**
 * Graph Engine — P3 图执行引擎（T2）
 *
 * 设计见 docs/todo/system-design/27-p3-p4-work-graph-console.md
 *   §四.A.4 状态机  §四.A.6 幂等（条件更新 + 受影响行数）
 *   §四.C   五个控制操作 + 「手动/自动共用同一个 advanceOne」
 *   §四.D.0 两种 node executor（agent_call / subgraph）
 *   §四.D.1 主流程（success → 出边 → 下游 ready → 收敛 → 回写父节点）
 *   §四.D.2 异常流程骨架（fail → 选边：精确匹配 → 兜底边 → 死路 blocked）
 *   §四.D.2.1 双层回退熔断（单边 loop_count + run 级 backward_count，T7）
 *   §四.A.3 任务级总预算挂 root run，扣完整棵 run 树挂起（T8）
 *   §四.C   暂停/停止三层级联 + 重试的 contract_hash 判据（T8）
 *
 * 本文件只管「跑」，不管「存」——所有落库走 work-graph-db.js，不在这里写一句 SQL DDL。
 *
 * 唯一推进原语是 advanceOne(runId)：拾取一个 ready 节点 → running → 执行 → 评估出边 →
 * 把满足条件的下游置 ready → 返回。**它不循环往下跑**。
 *   手动模式 = 路由 handler 每收到一次点击调一次；
 *   自动模式 = runAuto() 在 pause/stop 标志位为假时不停调它。
 * 只有这一套推进逻辑，自动模式不是另一套实现（§四.C 末段）。
 *
 * 仍然不做、留给后续任务的：
 *   - running 节点的进程级硬取消（§3.6，调用链由 T9 核实）→ 当前是「等它跑完但不再调度后续」的退化路径
 */

const path = require('path');
const fs = require('fs');

const {
  getDb,
  readDefGraph,
  getRun,
  getRunNode,
  getRunNodeByDefId,
  // 注意：updateNodeStatus / updateRunStatus **故意不在这里 import**——状态转换一律走
  // 下面的 sse.transitionNode / transitionRun，否则又会退回「改了状态但没有事件」的老路。
  bumpNodeAttempt,
  setRunFlags,
  createRun,
  listRunTree,
  listLoopRecords,
  insertLoopRecord,
  updateLoopVerdict,
  bumpBackwardCount,
  canTransitionRun,
  canTransitionNode,
} = require('./work-graph-db');

/**
 * 状态转换的唯一出口（§三.3 末段）。**本文件不再直调 work-graph-db 的裸
 * updateNodeStatus / updateRunStatus**：所有转换都走 sse.transitionNode / transitionRun，
 * 让「改状态 + 写 timeline + 发 SSE」在同一次同步调用里一起发生——DevLoop 那个
 * 「审计轨与实时轨不桥接」的缺口就是这么来的，这里不重复它。
 * 非状态机类事件（回退备案 / 预算扣减 / agent 调用起止 / 归因产出）走 sse.emitEvent。
 */
const sse = require('./graph-run-sse');

const REPO_ROOT = path.resolve(__dirname, '../..');

/**
 * 子图归因节点的名字。子 run 失败时优先取这个节点的 fail_reason 透传回父层（§四.D.2.2）。
 * // 假设(R-16待定案): 归因节点名保留 validator-reflection，不改 validator-triager。
 */
const ATTRIBUTION_NODE_NAME = 'validator-reflection';

/** run 的终态，进了就不再被引擎改写 */
const TERMINAL_RUN_STATUS = new Set(['completed', 'failed', 'stopped']);

/**
 * 双层回退硬顶（§四.D.2.1 第 2 步）。两个都是**纯整数**，判定处不调用任何 LLM——
 * 这是熔断不被 agent 的自由发挥绕过的唯一保证。
 *   LOOP_HARD_CAP_EDGE：单边计数，防同一条路径上的死循环；
 *   LOOP_HARD_CAP_RUN ：run 级总回退计数，防「换个借口继续退」（R-16 发现的 3×N 架空漏洞）。
 * 与 §四.A.3 的任务级总预算是两回事：预算管花了多少钱，这两个管绕了多少圈。
 *
 * ⚠️ 外层计数按 **root_run_id 共池**（见 treeBackwardCount / handleBackwardEdge）。
 * 按当前 run 记账时，subgraph 节点每重跑一轮就新建一张子 run，子层计数每轮从 0 重新开始，
 * 外层硬顶被彻底架空——这正是 27-E2E §7.1「每一级都重置下一级的计数器，整个循环就没有
 * 终止条件」的第二个变体（第一个是多条边各自计数把硬顶架空成 3×N）。
 */
const LOOP_HARD_CAP_EDGE = 3;
const LOOP_HARD_CAP_RUN = 5;

/**
 * 引擎唯一硬编码的 fail_reason 取值（§四.D.2.1 末段把它定成了引擎行为，不是图语义）：
 * 进程超时 / 被杀这类基础设施抖动，原地重跑，不烧任何计数。
 * 其余取值（impl_bug / input_underspecified / skill_defect …）引擎一律只做字符串匹配选边，
 * 语义由图的规划者与 validator 约定；`skill_defect` 的「挂起等人工」由规划者把边连到人工节点
 * 或兜底边表达，引擎不特判（L4 在 §一 是非目标）。
 */
const INFRA_FLAKE_REASON = 'infra_flake';

/**
 * validator-reflection 的归因取值域（system/templates/validator-reflection/agent.md
 * 「归因取值域（四选一，不要发明第五个）」= §四.D.2.2 那张表）。
 *
 * 引擎**不理解**这些取值的语义（选边只做字符串匹配），这个集合只用于一件事：
 * 把归因文件里自创的取值当作 null 处理（模板明说自创取值会静默落到兜底边，
 * 那就在入口处显式降级 + 记一条 timeline，而不是让一个假取值混进选边）。
 */
const ATTRIBUTION_REASONS = new Set(['impl_bug', 'input_underspecified', 'skill_defect', INFRA_FLAKE_REASON]);

/** 归因/结构化产出文件名（agent.md「产出格式」：runs/{rootRunId}/nodes/{nodeRunId}/attempt-{n}.json） */
const REFLECTION_FILE_RE = /^attempt-(\d+)\.json$/;

/** 判「这次失败了」的 verdict（agent.md 产出格式那三个取值里的两个）；no_prior_failure 是放行 */
const FAILING_VERDICTS = new Set(['attributed', 'inconclusive']);

/**
 * 「必须留下判词」的节点（27-NAMING §3.1/§3.3/§3.5 的 validator 命名轴）：
 * 归因节点 validator-reflection（含 org-validator-reflection 这种带父节点全名的写法）与
 * 一切 `{...}-validator` 节点。这些节点的**判定结果只存在于产出文件里**（verdict / fail_reason），
 * 调用本身正常退出说明不了任何事。文件缺失时按「无法判定」结算，不能静默判 success，
 * 否则 fail 边 / 回退 / infra_flake 原地重跑在生产里永远触发不了（R3-F1 缺陷 3）。
 */
const ATTRIBUTION_REQUIRED_RE = /(^|[-_])(reflection|validator)$/i;

function requiresAttributionProduct(ctx) {
  const name = String((ctx && ctx.nodeName) || '');
  return name === ATTRIBUTION_NODE_NAME || ATTRIBUTION_REQUIRED_RE.test(name);
}

const now = () => new Date().toISOString();

// ────────────────────────────────────────────────────────────────
// 注入点：两个外部依赖都可替换（测试注入假实现，避免真起 Claude CLI）
// ────────────────────────────────────────────────────────────────

/**
 * workgraph-planning 注入点（§四.D.0）。
 *
 * P3 阶段这一步还没接上（属阶段 D），默认实现直接抛错。接上之后替换成：
 * 调 workgraph-planning 本体 → 产出图 → 过 §三.7 五项校验 → createWorkGraph()。
 *
 * @type {(ctx: NodeContext) => Promise<{workGraphId: string}>}
 */
let planWorkGraph = async (ctx) => {
  throw new Error(
    `workgraph-planning 未接入（阶段 D）：节点 ${ctx.nodeName} 是 subgraph，` +
    `需要先 setPlanWorkGraph(fn) 注入图规划实现，或在建 run 前预先备好 work graph`
  );
};

/** 替换 workgraph-planning 实现（阶段 D 接真实现；测试注入假图） */
function setPlanWorkGraph(fn) {
  planWorkGraph = fn;
}

/**
 * agent_call 执行器注入点。默认实现调 AgentManager（见 defaultAgentExecutor）。
 * @type {(ctx: NodeContext) => Promise<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]
 */

/**
 * agent_call：调 AgentManager 跑 system_agent_id。
 *
 * 成功/失败只看调用本身是否抛异常——**引擎不发明 fail_reason**。
 * 按 §四.D.2.2 的定案，归因是一次显式的 agent 调用（validator-reflection 干的活），
 * 不是失败节点顺手吐一个值；执行器只在结果里带上 agent 自己给的 failReason（若有）。
 * 没有 failReason 时选边会退到兜底边，不会莫名其妙死路（§三.7 第五项校验保证兜底边存在）。
 */
async function defaultAgentExecutor(ctx) {
  const AgentManager = require('../src/core/agent-manager');
  // 空 config 即可：getAgentConfig 有基于文件系统的动态回退（system/agents/ 与
  // data/companies/{cid}/agents/ 两条路径），与 agent-chat.js 的用法一致。
  const manager = new AgentManager({}, REPO_ROOT);

  const task = ctx.remediationNote
    ? `${ctx.task}\n\n[上一轮回退的修复建议]\n${ctx.remediationNote}`
    : ctx.task;

  let res = null;
  let err = null;
  try {
    res = await manager.callAgent(resolveAgentName(ctx.companyId, ctx.systemAgentId), task, {
      metadata: {
        taskId: ctx.runId, layer: 'graph-node', nodeRunId: ctx.nodeRunId,
        // 产物目录是**提示词**里那份约定的机器可读副本（提示词由 describeArtifactContract 拼，
        // 见 advanceOne）；metadata 只进日志与 hook，agent 读不到，所以两边都要有。
        rootRunId: ctx.rootRunId, attempt: ctx.attempt, artifactDir: safeNodeArtifactDir(ctx),
      },
    });
  } catch (e) {
    err = e;
  }

  // 归因/产出是**文件**，不是返回值：agent 崩了也可能已经把 attempt-{n}.json 写出来了，
  // 所以成功失败两条路径都读一次（§四.D.2.2 + validator-reflection/agent.md 的产出约定）。
  const produced = readNodeProduct(ctx);

  if (err) {
    return {
      status: 'fail',
      output: err.message,
      failReason: produced.failReason,
      remediationNote: produced.remediationNote,
      outputRef: produced.outputRef,
    };
  }
  // 判词节点（validator / reflection）没留下产出文件 = **无法判定**，不是通过（R3-F1 缺陷 3b）。
  // 按 §四.D.2.2 的 inconclusive 语义结算：fail + fail_reason=null → 走那条兜底边
  // （§三.7 第五项校验保证它存在；没有兜底边就是死路 blocked，等人工）。
  // 静默判 success 的后果不是「失败但走兜底边」，而是「validator 判的失败根本不存在」。
  if (produced.missing && requiresAttributionProduct(ctx)) {
    timelineNote(ctx, 'node.reflection',
      `判词节点没有可用的 attempt-${ctx.attempt}.json（缺失/不可解析/只剩上一轮的），` +
      '按「无法判定」结算（fail + fail_reason=null → 兜底边）', {
        node_run_id: ctx.nodeRunId, attempt: ctx.attempt, degraded: 'missing_product',
        expected_file: path.join(safeNodeArtifactDir(ctx) || '', `attempt-${ctx.attempt}.json`),
      });
    return {
      status: 'fail',
      output: res && res.output,
      agentRunId: (res && res.runId) || null,
      failReason: null,
      remediationNote: null,
      outputRef: produced.outputRef,
    };
  }

  return {
    // 调用没抛不等于节点 success：validator 类节点判「这次失败了」是**写在产出文件里**的
    // （verdict=attributed / inconclusive），CLI 本身当然是正常退出的。产出说失败就按失败结算，
    // 否则 fail 边、回退边、§四.D.2.1 的 infra_flake 原地重跑在生产里永远触发不了
    // ——引擎仍然没有「发明」fail_reason，它只是照抄归因节点写下的判词（§四.D.2.2）。
    status: produced.failingVerdict ? 'fail' : 'success',
    output: res && res.output,
    // RunRegistry 的 runId 目前没有从 callAgent 透出来（调用链由 T9 核实/补齐），
    // 执行器返回了就写，没有就留空。
    agentRunId: (res && res.runId) || null,
    failReason: produced.failingVerdict ? produced.failReason : null,
    // 首轮进场（verdict=no_prior_failure）也可能留了话给下一轮，成功路径同样带回
    remediationNote: produced.remediationNote,
    outputRef: produced.outputRef,
  };
}

/**
 * 读节点这一 attempt 的结构化产出，解析出三值透传需要的两个值 + output_ref。
 *
 * 约定路径（system/templates/validator-reflection/agent.md「产出格式」）：
 *   data/companies/{cid}/runs/{rootRunId}/nodes/{nodeRunId}/attempt-{n}.json
 * attempt 编号在模板里没写死是 0 基还是 1 基，所以先按 ctx.attempt 与 ctx.attempt+1 找，
 * 都没有就退到目录里编号最大的那个 attempt-*.json。
 *
 * **一律不抛**（§四.D.2.2：读不到就是没归因，不是节点执行失败）：
 * 文件缺失 → 返回三个 null 且 `missing=true`（绝大多数节点本来就不产归因文件，不刷 timeline；
 *            是不是判词节点、缺了算不算失败，由调用方按 requiresAttributionProduct 决定）；
 * 文件存在但解析失败 / 取值非法 → 降级为 null **并记一条 timeline**，让控制台看得见归因白做了。
 *
 * @returns {{failReason: string|null, remediationNote: string|null, outputRef: string|null,
 *            failingVerdict: boolean, missing: boolean}}
 */
function readNodeProduct(ctx) {
  const empty = {
    failReason: null, remediationNote: null, outputRef: null, failingVerdict: false, missing: true,
  };
  if (!ctx || !ctx.run) return empty;
  let dir;
  try {
    dir = nodeArtifactDir(ctx.run, ctx.nodeRunId);
    if (!fs.existsSync(dir)) return empty;
  } catch {
    return empty; // 路径都拼不出来（run 缺列）就是没有产物
  }

  // 节点私有目录本身就是 output_ref 指向的「被测产物」（goal-contract.js 顶部注释）
  const outputRef = dir;

  const { file, stale } = resolveReflectionFile(dir, ctx.attempt);
  if (!file) {
    if (stale) {
      // 目录按 nodeRunId 分配、跨 attempt 复用，编号比本轮小的那份是**上一轮**的判词。
      // 拿它当本轮结论会让节点被旧判词反复判失败直到撞硬顶（R3-F1 缺陷 4）——宁可当没有。
      timelineNote(ctx, 'node.reflection',
        `只找到上一轮的归因文件 ${path.basename(stale)}（< attempt-${ctx.attempt}），不复用`, {
          node_run_id: ctx.nodeRunId, attempt: ctx.attempt, file: stale, degraded: 'stale_reflection',
        });
    }
    return { ...empty, outputRef };
  }

  const doc = readJsonOrNull(file);
  if (!doc || typeof doc !== 'object') {
    timelineNote(ctx, 'node.reflection', `归因文件解析失败，fail_reason/remediation_note 降级为 null: ${path.basename(file)}`, {
      node_run_id: ctx.nodeRunId, attempt: ctx.attempt, file, degraded: 'unparsable',
    });
    return { ...empty, outputRef };
  }

  const raw = doc.fail_reason == null ? null : String(doc.fail_reason);
  let failReason = raw;
  if (raw !== null && !ATTRIBUTION_REASONS.has(raw)) {
    // 模板：「四个值以外一律不要输出」。自创取值不报错，当 null 处理（=走兜底边）并留痕。
    failReason = null;
    timelineNote(ctx, 'node.reflection', `归因取值非法（${raw}），按 null 处理并走兜底边`, {
      node_run_id: ctx.nodeRunId, attempt: ctx.attempt, file, degraded: 'invalid_fail_reason', fail_reason: raw,
    });
  }
  const remediationNote =
    typeof doc.remediation_note === 'string' && doc.remediation_note.trim() ? doc.remediation_note : null;

  // verdict 决定节点判不判失败（模板「产出格式」：attributed | inconclusive | no_prior_failure）。
  // inconclusive 也算失败——模板明写「让引擎走那条 fail_reason IS NULL 的兜底边」。
  // 没写 verdict 但给了 fail_reason 的，按失败处理（有根因就是有失败）。
  const verdict = doc.verdict == null ? null : String(doc.verdict);
  const failingVerdict = verdict === null ? raw !== null : FAILING_VERDICTS.has(verdict);

  if (failingVerdict || remediationNote) {
    timelineNote(ctx, 'node.reflection', `归因 verdict=${verdict || '(缺省)'} fail_reason=${failReason || 'null'}`, {
      node_run_id: ctx.nodeRunId, attempt: ctx.attempt, file,
      verdict, fail_reason: failReason, remediation_note: remediationNote, failing: failingVerdict,
    });
  }
  return { failReason, remediationNote, outputRef, failingVerdict, missing: false };
}

/**
 * attempt-{n}.json 的定位：先精确（n 与 n+1 两种基数），再退到编号最大的那份。
 *
 * 兜底**只认编号 >= 本轮 attempt 的文件**：节点私有目录按 nodeRunId 分配、跨 attempt 复用，
 * 无条件取「编号最大的」会读到上一轮的归因文件，节点被旧判词反复判失败直到撞硬顶
 * （R3-F1 缺陷 4）。编号更小的那份原样回报给调用方去记一条 degraded timeline。
 *
 * @returns {{file: string|null, stale: string|null}}
 */
function resolveReflectionFile(dir, attempt) {
  for (const n of [attempt, attempt + 1]) {
    const p = path.join(dir, `attempt-${n}.json`);
    if (fs.existsSync(p)) return { file: p, stale: null };
  }
  let best = null;
  let bestN = -1;
  let names = [];
  try {
    names = fs.readdirSync(dir);
  } catch {
    return { file: null, stale: null }; // 目录刚被删/不可读：当作没有归因产出
  }
  for (const name of names) {
    const m = REFLECTION_FILE_RE.exec(name);
    if (m && Number(m[1]) > bestN) {
      bestN = Number(m[1]);
      best = path.join(dir, name);
    }
  }
  return bestN >= attempt ? { file: best, stale: null } : { file: null, stale: best };
}

/** 只记一条审计（同时也发 SSE，走 emitEvent 保证两轨不分家）；失败绝不影响执行 */
function timelineNote(ctx, event, summary, data) {
  try {
    sse.emitEvent(ctx.runId, event, summary, data, { ref: ctx.nodeRunId });
  } catch {
    // 审计写不进去（磁盘满 / 目录只读）不该把节点跑挂
  }
}

/**
 * agent 本体名 → AgentManager 能寻址的名字。
 * 公司雇过这个本体时目录在 data/companies/{cid}/agents/{id}，寻址名带 `cid:` 前缀
 * （与 agent-chat.js getAgentDir 的约定一致）；否则用裸本体名走 system/agents/。
 */
function resolveAgentName(companyId, systemAgentId) {
  if (!companyId) return systemAgentId;
  const scoped = path.join(REPO_ROOT, 'data', 'companies', companyId, 'agents', systemAgentId);
  return fs.existsSync(scoped) ? `${companyId}:${systemAgentId}` : systemAgentId;
}

/**
 * subgraph：起一张子图并等它收敛（§四.D.0）。
 *
 * 「等」不是在这里 await 几小时——手动模式下 advanceOne 由一次 HTTP 请求驱动，不可能挂住。
 * 落地方式是：本函数只负责规划图 + 建子 run + 回填 child_run_id，然后返回 status='pending'
 * 表示「父节点继续停在 running」；子 run 收敛时由 settleRun → propagateToParent 回写父节点。
 * 两条路径最终都进同一个 settleNode，且靠 running→终态 的条件更新保证只生效一次。
 */
async function executeSubgraph(ctx) {
  // 幂等守卫的判据是「**本轮**是不是已经起过子 run」，不是「child_run_id 非空」。
  //
  // 「child_run_id 非空即跳过」曾经是个永久死锁（P3 首轮缺陷 1）：节点第二次变 ready 时
  // 拿到的是上一轮**已终态**的旧子 run，直接返回 pending，父节点永远停在 running。
  // §四.D.3 定案是「该 org node 整张 work graph 重跑（起新子 run，attempt+1）」——
  // 旧子 run 留作历史（控制台按 parent_node_run_id 仍查得到），本轮另起一张。
  //
  // 判据选「旧子 run 是否还活着」而不是「attempt 变没变」：infra_flake 原地重跑
  // （§四.D.2.1 末段）明确**不加 attempt**，光比 attempt 会把它误判成同一轮。
  // 活着 = run_status 还没落到终态/blocked，即父节点确实还在等它回写；
  // 同一轮内的并发进入（进程重启重放）看到的一定是活的子 run，仍然只起一张图。
  const existing = getRunNode(ctx.nodeRunId);
  const prior = existing && existing.child_run_id ? getRun(existing.child_run_id) : null;
  if (prior && !isChildSettled(prior)) {
    return { status: 'pending', childRunId: prior.id };
  }

  const planned = await planWorkGraph(ctx);
  const workGraphId = planned && (planned.workGraphId || planned.id);
  if (!workGraphId) throw new Error(`planWorkGraph 未返回 workGraphId（节点 ${ctx.nodeName}）`);

  const child = createRun({
    graphKind: 'work',
    graphId: workGraphId,
    companyId: ctx.companyId,
    parentRunId: ctx.runId,
    parentNodeRunId: ctx.nodeRunId,
    goalId: ctx.run.goal_id,
    taskId: ctx.run.task_id,
    okrNodeId: ctx.run.okr_node_id,
    // 子 run 默认继承父 run 的执行模式（§四.D.0）；预算只挂 root run，这里不传
    executionMode: ctx.executionMode,
  });

  getDb()
    .prepare(`UPDATE graph_run_nodes SET child_run_id = ?, updated_at = ? WHERE id = ?`)
    .run(child.id, now(), ctx.nodeRunId);

  markRunReady(child.id);
  startRunSync(child.id);

  return { status: 'pending', childRunId: child.id };
}

/**
 * 子 run 是否已经「交待完了」——终态或 blocked 都算：settleRun 在这两种情形下都会
 * propagateToParent 回写父节点，父节点不该再等它。
 */
function isChildSettled(childRun) {
  return TERMINAL_RUN_STATUS.has(childRun.run_status) || childRun.run_status === 'blocked';
}

// ────────────────────────────────────────────────────────────────
// 核心原语：advanceOne（§四.C / §四.D.1）
// ────────────────────────────────────────────────────────────────

/**
 * 推进一步：拾取**一个** ready 节点 → running → 按 node_kind 分派执行 → 评估出边 →
 * 把满足条件的下游置 ready → 返回本次结果。不继续往下跑。
 *
 * 幂等（§四.A.6）：ready→running 是条件更新，受影响行数 = 0 即「已被别人拿走」，
 * 返回当前状态而不是报错。
 *
 * @param {string} runId
 * @param {object} [opts]
 * @param {boolean} [opts.ignorePause] 「下一步」用：auto 已 paused 时也允许单步推一下
 * @param {string}  [opts.task] 传给 executor 的任务描述
 * @returns {Promise<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 {object|null} contractDiff   人工重试后的契约四分类 + 被打回 pending 的下游
 * @property {string|null} childRunId     subgraph 节点起的子 run
 *
 * reason 取值：run-not-found | run-not-running | stop-requested | pause-requested |
 *              no-ready-node | taken-by-other | awaiting-child-run | budget-exhausted |
 *              null（正常推进完成）
 */
async function advanceOne(runId, opts = {}) {
  const run = getRun(runId);
  if (!run) return result(runId, null, 'run-not-found');

  // 软信号在**节点边界**生效：不打断已经在跑的节点，只是不再拾取新的（§四.C）
  if (run.stop_requested) return result(runId, run.run_status, 'stop-requested');
  if (run.pause_requested && !opts.ignorePause) {
    // 到了节点边界，「暂停」这时候才真正落到 run_status 上
    landPause(runId);
    return result(runId, (getRun(runId) || {}).run_status, 'pause-requested');
  }
  if (run.run_status !== 'running') return result(runId, run.run_status, 'run-not-running');

  const picked = getDb()
    .prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? AND status = 'ready' ORDER BY created_at, id LIMIT 1`)
    .get(runId);
  if (!picked) {
    // 没有 ready 节点：可能是全跑完了，也可能全都卡在子 run 上，交给收敛判定
    return result(runId, settleRun(runId), 'no-ready-node');
  }

  // §四.A.3：任何一层发起 agent 调用前先按 root_run_id 查同一个预算池的余额。
  // 拾到了 ready 节点（=真要花钱了）才查，避免空转时把已跑完的 run 误挂起。
  const gate = checkBudget(run);
  if (!gate.ok) {
    // 扣完不是失败，是**整棵 run 树挂起等人工**：root run → blocked，向下级联 pause_requested=1
    suspendRunTree(run);
    return { ...result(runId, (getRun(runId) || {}).run_status, 'budget-exhausted'), budget: gate };
  }

  const def = defIndex(run);
  const defNode = def.nodes.get(picked.def_node_id) || {};
  const nodeName = defNode.name || picked.def_node_id;

  // ready → running 的条件更新就是并发闸门（§四.A.6）
  if (
    !sse.transitionNode(runId, picked.id, 'ready', 'running', { startedAt: now() }, {
      node_name: nodeName, node_kind: defNode.nodeKind || 'agent_call', attempt: picked.attempt,
    })
  ) {
    const cur = getRunNode(picked.id);
    return { ...result(runId, run.run_status, 'taken-by-other'), node: nodeView(runId, cur) };
  }

  const ctx = {
    runId,
    run,
    rootRunId: run.root_run_id,
    companyId: run.company_id,
    nodeRunId: picked.id,
    defNodeId: picked.def_node_id,
    nodeName,
    systemAgentId: defNode.systemAgentId || null,
    nodeKind: defNode.nodeKind || 'agent_call',
    attempt: picked.attempt,
    task: opts.task || describeTask(run),
    remediationNote: latestRemediationNote(runId, picked.id),
    executionMode: run.execution_mode,
  };
  // 产物目录必须**进提示词**：读的一侧（readNodeProduct）按 §四.A.5 的路径去读，
  // 写的一侧不告诉 agent 往哪写、文件名怎么起，这份文件在生产里就根本不会被写出来，
  // 于是 fail 边 / 回退 / infra_flake 全都触发不了（R3-F1 缺陷 3a）。
  if (ctx.nodeKind !== 'subgraph') ctx.task += describeArtifactContract(ctx);

  // 抢到节点即记账：一次调用一定发生了（哪怕它随后炸掉），预算「只增不减」（§四.A.3）。
  // subgraph 节点这一次花的是 workgraph-planning 的调用，同样按一次计。
  spendBudget(run, { agentCalls: 1 });

  sse.emitEvent(
    runId,
    'agent.call',
    `${ctx.nodeKind === 'subgraph' ? '规划子图' : '调用 agent'} ${ctx.systemAgentId || nodeName}（attempt #${picked.attempt + 1}）`,
    {
      node_run_id: picked.id, node_name: nodeName, node_kind: ctx.nodeKind,
      system_agent_id: ctx.systemAgentId, attempt: picked.attempt,
      remediation_note: ctx.remediationNote,
    },
    { ref: picked.id },
  );

  let outcome;
  try {
    outcome = ctx.nodeKind === 'subgraph' ? await executeSubgraph(ctx) : await agentExecutor(ctx);
  } catch (err) {
    // executor 自己炸了（规划失败 / 配置缺失等）算节点失败，不带 failReason，走兜底边
    outcome = { status: 'fail', output: err.message, failReason: null };
  }

  // executor 报了 token 用量就一并扣（没报就只扣调用次数，wallclock 由 spent.started_at 兜）
  if (outcome && outcome.tokens) spendBudget(run, { tokens: outcome.tokens });

  sse.emitEvent(
    runId,
    'node.output',
    `${nodeName} → ${(outcome && outcome.status) || 'unknown'}${outcome && outcome.failReason ? `（${outcome.failReason}）` : ''}`,
    {
      node_run_id: picked.id, node_name: nodeName, attempt: picked.attempt,
      status: outcome && outcome.status, fail_reason: (outcome && outcome.failReason) || null,
      output_ref: (outcome && outcome.outputRef) || null,
      child_run_id: (outcome && outcome.childRunId) || null,
    },
    { ref: picked.id },
  );

  if (outcome && outcome.status === 'pending') {
    // subgraph：父节点继续停在 running，等子 run 收敛回写
    return {
      ...result(runId, run.run_status, 'awaiting-child-run'),
      ok: true,
      node: nodeView(runId, getRunNode(picked.id)),
      childRunId: outcome.childRunId || null,
    };
  }

  return settleNode(runId, picked.id, outcome);
}

/**
 * 节点执行结束后的落地：写终态 → 评估出边 → 收敛判定。
 * 也是子 run 收敛时回写父节点走的同一条路（propagateToParent）。
 *
 * @param {object} outcome {status, failReason?, remediationNote?, outputRef?, agentRunId?, childRunId?}
 */
function settleNode(runId, nodeRunId, outcome = {}) {
  const to = outcome.status === 'success' ? 'success' : 'fail';
  const changed = sse.transitionNode(runId, nodeRunId, 'running', to, {
    failReason: outcome.failReason === undefined ? null : outcome.failReason,
    outputRef: outcome.outputRef,
    agentRunId: outcome.agentRunId,
    childRunId: outcome.childRunId,
    completedAt: now(),
  }, { attempt: (getRunNode(nodeRunId) || {}).attempt, remediation_note: outcome.remediationNote || null });
  // 0 行 = 这个节点已经被别人结算过了（§四.A.6），原样返回当前状态
  if (!changed) {
    return {
      ...result(runId, (getRun(runId) || {}).run_status, 'taken-by-other'),
      node: nodeView(runId, getRunNode(nodeRunId)),
    };
  }

  const nodeRun = getRunNode(nodeRunId);
  const run = getRun(runId);
  const def = defIndex(run);
  const out = { activated: [], backwardEdgeId: null, loop: null, contractDiff: null };

  if (to === 'success') {
    // 人工「重试」留下的契约快照在这里判：契约变了才级联下游（§四.C 末段）
    out.contractDiff = applyRetryCascade(runId, def, nodeRun);
    activateSuccessTargets(runId, def, nodeRun.def_node_id, out);
  } else {
    routeFailure(runId, def, nodeRun, out, outcome.remediationNote || null);
  }

  return {
    ok: true,
    runId,
    runStatus: settleRun(runId),
    reason: null,
    node: nodeView(runId, getRunNode(nodeRunId)),
    activated: out.activated,
    backwardEdgeId: out.backwardEdgeId,
    loop: out.loop,
    contractDiff: out.contractDiff,
    childRunId: nodeRun.child_run_id || null,
  };
}

// ────────────────────────────────────────────────────────────────
// §四.D.1 主流程：success 后的出边评估
// ────────────────────────────────────────────────────────────────

/**
 * 遍历 condition='success' 的出边，目标节点若**所有**入边条件都满足则置 ready。
 * 「全部满足」= 该目标所有 forward + success 的入边，其源节点都已 success（汇合点语义）。
 * fail 边和 backward 边不参与这道闸——它们是异常流程，各自单独把目标点亮。
 */
function activateSuccessTargets(runId, def, defNodeId, out) {
  for (const e of def.edges) {
    if (e.from !== defNodeId || e.condition !== 'success') continue;
    if (!allIncomingSatisfied(runId, def, e.to)) continue;
    const activated = markReady(runId, def, e.to);
    if (activated) out.activated.push(activated);
  }
}

function allIncomingSatisfied(runId, def, toDefNodeId) {
  return def.edges
    .filter((e) => e.to === toDefNodeId && e.condition === 'success' && e.edgeKind !== 'backward')
    .every((e) => {
      const src = getRunNodeByDefId(runId, e.from);
      return src && src.status === 'success';
    });
}

/** pending → ready；已经不是 pending（被别人点亮或已跑过）就跳过，返回 null */
function markReady(runId, def, defNodeId) {
  const rn = getRunNodeByDefId(runId, defNodeId);
  if (!rn || rn.status !== 'pending') return null;
  const d = def.nodes.get(defNodeId) || {};
  const name = d.name || defNodeId;
  if (!sse.transitionNode(runId, rn.id, 'pending', 'ready', {}, { node_name: name, attempt: rn.attempt })) {
    return null;
  }
  return { id: rn.id, defNodeId, name };
}

// ────────────────────────────────────────────────────────────────
// §四.D.2 异常流程骨架：fail 后选边
// ────────────────────────────────────────────────────────────────

/**
 * 选边规则（§四.D.2.2，引擎只做字符串匹配，不理解取值语义）：
 *   先找 fail_reason 精确匹配的边 → 无匹配退到 fail_reason IS NULL 的兜底边 → 再无则死路。
 * 取值域不在引擎里硬编码，由图的规划者与对应 validator 约定。
 */
function selectFailEdge(def, defNodeId, failReason) {
  const candidates = def.edges.filter((e) => e.from === defNodeId && e.condition === 'fail');
  if (candidates.length === 0) return null;
  if (failReason) {
    const exact = candidates.find((e) => e.failReason === failReason);
    if (exact) return exact;
  }
  return candidates.find((e) => e.failReason === null || e.failReason === undefined || e.failReason === '') || null;
}

function routeFailure(runId, def, nodeRun, out, remediationNote = null) {
  // infra_flake 特例（§四.D.2.1 末段）：基础设施抖动直接**原地重跑**，不选边、不写
  // loop record、不加 backward_count、不扣该节点的 retry 预算（attempt 也不加一）。
  // 终止性由 §四.A.3 的任务级总预算兜底——每次重跑仍算一次 agent 调用，池子空了整棵树挂起。
  if (nodeRun.fail_reason === INFRA_FLAKE_REASON) {
    const d = def.nodes.get(nodeRun.def_node_id) || {};
    const name = d.name || nodeRun.def_node_id;
    // 原地重跑 = 一轮新的执行：subgraph 节点必须**清空 child_run_id**，否则下一次
    // executeSubgraph 会拿着上一轮的旧子 run 直接返回 pending（缺陷 1 的另一条入口）。
    if (
      sse.transitionNode(runId, nodeRun.id, 'fail', 'ready', { childRunId: null, failReason: null },
        { node_name: name, attempt: nodeRun.attempt, summary: `node ${name}: infra_flake 原地重跑（不烧计数）` })
    ) {
      out.activated.push({ id: nodeRun.id, defNodeId: nodeRun.def_node_id, name });
      out.infraRerun = true;
    }
    return;
  }

  const edge = selectFailEdge(def, nodeRun.def_node_id, nodeRun.fail_reason);

  // 死路：没有可走的 fail 边 → 节点 blocked，等人工介入（区别于人工主动「停止」）
  if (!edge) return blockNode(runId, nodeRun.id);

  if (edge.edgeKind === 'backward') {
    out.backwardEdgeId = edge.id;
    const handled = handleBackwardEdge(runId, def, nodeRun, edge, out, remediationNote);
    // 目标节点不在本 run（定义层被改过）等异常情形：源节点 blocked，不静默留在 fail 态
    if (!handled) blockNode(runId, nodeRun.id);
    return;
  }

  // forward 的 fail 边 = 普通失败分支（如「失败转人工审核节点」），直接点亮目标
  const activated = markReady(runId, def, edge.to);
  if (activated) out.activated.push(activated);
}

/**
 * 沿一条 backward 边回退 —— §四.D.2.1 的三步，**无条件按顺序执行**：
 *
 *   0. 越顶即收敛：整棵 run 树的 backward_count 已经越过 run 级硬顶时，**连备案都不再做**，
 *      直接 blockNode。少了这一条，撞顶之后每多退一次就多一条 force_terminated 记录，
 *      而回退目标仍会被重新点亮（第 3 步），熔断只是「每轮记一笔」而不是「停下来」。
 *   1. 备案：insertLoopRecord(verdict='registered')，loop_count = 该 (run_id, edge_id) 已有
 *      记录数 + 1（由 work-graph-db 算）；同时 backward_count += 1。
 *      「无条件」指两个计数在任何回退上都写，先记账再判，不能先判后记。
 *   2. 双层硬顶：loop_count > 3（单边）**或** backward_count > 5（run 树级）→ 改判
 *      force_terminated，源节点 blocked，链路收敛等人工。
 *      **这一步是纯整数比较，不调用任何 LLM**——单边计数防同一路径死循环，外层计数防
 *      「换个借口继续退」（R-16：多条回退边各自计数会把硬顶架空成 3×N）。
 *   3. 回退：把 to_node 置 ready、attempt+1，remediation_note 落在 loop record 里，
 *      由 latestRemediationNote 在下一轮取给回退目标当上下文补充。
 *
 * @returns {boolean} 是否已经处理掉这次失败
 */
function handleBackwardEdge(runId, def, nodeRun, edge, out, remediationNote = null) {
  const target = getRunNodeByDefId(runId, edge.to);
  if (!target) return false; // 边指向的节点不在本 run（定义层被换过）→ 交给调用方 blockNode

  const rootRunId = (getRun(runId) || {}).root_run_id || runId;

  // ⓪ 整棵树已经越顶：不再登记新记录，直接收敛。
  //    子层每重跑一轮就是一张新子 run，若只按「本 run」判，越顶后的每一轮都会重新开一条
  //    记录并把目标重新点亮（父层因此烧满 4 轮），熔断形同虚设。
  const already = treeBackwardCount(rootRunId);
  if (already > LOOP_HARD_CAP_RUN) {
    out.loop = {
      edgeId: edge.id, recordId: null, loopCount: null, backwardCount: already,
      forceTerminated: true, cap: 'run',
    };
    sse.emitEvent(runId, 'loop.terminated', `熔断：run 树级硬顶已撞（${already}/${LOOP_HARD_CAP_RUN}），不再备案，直接收敛等人工`, {
      edge_id: edge.id, backward_count: already, root_run_id: rootRunId,
      verdict: 'force_terminated', cap: 'run', registered: false,
      from_node_run_id: nodeRun.id, to_node_run_id: target.id,
    }, { ref: edge.id });
    blockNode(runId, nodeRun.id);
    return true;
  }

  // ① 备案：两个计数一起加，谁都不许跳过
  const rec = insertLoopRecord({
    runId,
    edgeId: edge.id,
    fromNodeRunId: nodeRun.id,
    toNodeRunId: target.id,
    verdict: 'registered',
    remediationNote,
  });
  const backwardCount = bumpTreeBackwardCount(runId, rootRunId);
  sse.emitEvent(runId, 'loop.registered', `回退备案 loop_count=${rec.loopCount}/${LOOP_HARD_CAP_EDGE} · backward_count=${backwardCount}/${LOOP_HARD_CAP_RUN}`, {
    record_id: rec.id, edge_id: edge.id, loop_count: rec.loopCount, backward_count: backwardCount,
    from_node_run_id: nodeRun.id, to_node_run_id: target.id, verdict: 'registered',
    remediation_note: remediationNote,
  }, { ref: edge.id });
  out.loop = {
    edgeId: edge.id,
    recordId: rec.id,
    loopCount: rec.loopCount,
    backwardCount,
    forceTerminated: false,
    cap: null,
  };

  // ② 双层硬顶：纯整数比较，零 LLM 参与（backwardCount 是整棵 run 树的共池计数）
  if (rec.loopCount > LOOP_HARD_CAP_EDGE || backwardCount > LOOP_HARD_CAP_RUN) {
    updateLoopVerdict(rec.id, 'force_terminated');
    out.loop.forceTerminated = true;
    out.loop.cap = rec.loopCount > LOOP_HARD_CAP_EDGE ? 'edge' : 'run';
    sse.emitEvent(runId, 'loop.terminated', `熔断：${out.loop.cap === 'edge' ? '单边' : 'run 级'}硬顶已撞，链路收敛等人工`, {
      record_id: rec.id, edge_id: edge.id, loop_count: rec.loopCount, backward_count: backwardCount,
      verdict: 'force_terminated', cap: out.loop.cap,
    }, { ref: edge.id });
    blockNode(runId, nodeRun.id);
    return true;
  }

  // ③ 回退。先把目标的 forward 下游打回 pending，再点亮目标本身。
  out.reset = resetForwardDownstream(runId, def, edge.to);
  if (target.status !== 'ready' && target.status !== 'running' && canTransitionNode(target.status, 'ready')) {
    const d = def.nodes.get(edge.to) || {};
    const name = d.name || edge.to;
    // 回退目标重新变 ready = 新一轮：subgraph 节点的 child_run_id 必须清空，
    // 否则 executeSubgraph 会复用上一轮的终态子 run 并永远返回 pending（缺陷 1 主入口）。
    if (
      sse.transitionNode(runId, target.id, target.status, 'ready', { childRunId: null, failReason: null },
        { node_name: name, attempt: target.attempt + 1, backward_edge_id: edge.id,
          summary: `node ${name}: 沿回退边 ${edge.id} 重新点亮（attempt #${target.attempt + 2}）` })
    ) {
      bumpNodeAttempt(target.id);
      out.activated.push({ id: target.id, defNodeId: edge.to, name });
    }
  }
  return true;
}

/**
 * 外层熔断计数的**共池读**：整棵 run 树的总回退次数，记在 root run 那一行。
 *
 * 为什么不按当前 run 读：subgraph 节点每重跑一轮都新建一张子 run（executeSubgraph），
 * 子层的计数会每轮从 0 重新开始 —— 27-E2E §7.1「每一级都重置下一级的计数器，整个循环就
 * 没有终止条件」。预算（§四.A.3）早就是按 root_run_id 共池的，这里对齐同一个口径。
 */
function treeBackwardCount(rootRunId) {
  const root = getRun(rootRunId);
  return root ? root.backward_count || 0 : 0;
}

/**
 * 共池写：root run 那一行是**全树**总数（熔断判据），子 run 自己那一行仍记本层次数
 * （控制台按 run 展示用）。root 与当前 run 是同一行时只加一次。
 * work-graph-db.bumpBackwardCount 不归本文件改，这里直接按 root_run_id 寻址它。
 * @returns {number} 加完之后的**全树**总数
 */
function bumpTreeBackwardCount(runId, rootRunId) {
  if (runId !== rootRunId) bumpBackwardCount(runId);
  return bumpBackwardCount(rootRunId);
}

/**
 * 把某节点 forward 可达的全部下游打回 pending。
 *
 * 回退必须做这件事：下游节点此刻多半已经是 success/fail 终态，而 markReady 只认 pending，
 * 不重置的话第二轮永远点不亮，run 会假收敛成 completed。
 * 这与「重试默认不级联下游」不矛盾——那条讲的是**人工重试**，判据是 contract_hash
 * （见 applyRetryCascade）；自动回退的语义本来就是「上游要重做，下游全部作废」。
 *
 * ponytail: NODE_TRANSITIONS 里没有 *→pending 这一条（work-graph-db 不归本文件改），
 * 所以这里走一次带 status 条件的直接 UPDATE，仍然是「受影响行数 = 是否我改的」的幂等语义。
 * 升级路径：等 work-graph-db 在转换表里补上 success/fail/blocked/skipped → pending。
 */
function resetForwardDownstream(runId, def, fromDefNodeId) {
  const db = getDb();
  const reset = [];
  for (const defNodeId of forwardReachable(def, fromDefNodeId)) {
    const rn = getRunNodeByDefId(runId, defNodeId);
    // running 的不动（它正在跑，结算时自己会落地）；pending 的本来就是 pending
    if (!rn || rn.status === 'pending' || rn.status === 'running') continue;
    const d = def.nodes.get(defNodeId) || {};
    const name = d.name || defNodeId;
    // child_run_id 一起清：这个节点下一轮会被重新点亮，subgraph 节点必须起**新**子 run，
    // 不能复用上一轮那张已终态的图（缺陷 1 的第二条入口）。旧子 run 行留着不动，
    // 控制台按 parent_node_run_id 仍查得到历史子图。
    const changed = sse.emitTransition(
      runId,
      {
        event: 'node.status',
        ref: rn.id,
        summary: `node ${name}: ${rn.status} → pending（上游回退，本轮产出作废）`,
        data: {
          node_run_id: rn.id, node_name: name, from: rn.status, to: 'pending',
          attempt: rn.attempt, reset_by: fromDefNodeId,
        },
      },
      () => db
        .prepare(`
          UPDATE graph_run_nodes
          SET status = 'pending', fail_reason = NULL, child_run_id = NULL,
              started_at = NULL, completed_at = NULL, updated_at = ?
          WHERE id = ? AND status IN ('ready','success','fail','blocked','skipped')
        `)
        .run(now(), rn.id).changes,
    );
    if (changed) {
      // 被作废的节点下一次执行是**新的一轮**，attempt 必须往前走。
      // 不加的话它的 attempt 永远停在上一轮：产物文件名 attempt-{n}.json 不变，
      // readNodeProduct 每轮都精确命中**上一轮**那份归因文件，节点被旧判词反复判失败
      // 直到撞硬顶（R3-F1 缺陷 4 的主入口；兜底取最大编号只是它的第二条入口）。
      // 回退目标自己的 attempt 由 handleBackwardEdge 加，两边不重叠（forwardReachable 不含起点）。
      bumpNodeAttempt(rn.id);
      reset.push(rn.id);
    }
  }
  return reset;
}

/** 从某节点出发、沿非 backward 边可达的下游集合（不含自己）。图小，直接 BFS */
function forwardReachable(def, fromDefNodeId) {
  const seen = new Set();
  const queue = [fromDefNodeId];
  while (queue.length) {
    const cur = queue.shift();
    for (const e of def.edges) {
      if (e.from !== cur || e.edgeKind === 'backward') continue;
      if (e.to === fromDefNodeId || seen.has(e.to)) continue;
      seen.add(e.to);
      queue.push(e.to);
    }
  }
  return seen;
}

function blockNode(runId, nodeRunId) {
  sse.transitionNode(runId, nodeRunId, 'fail', 'blocked', {}, {
    summary: 'fail → blocked：没有可走的 fail 边或已撞熔断硬顶，等人工介入',
  });
}

// ────────────────────────────────────────────────────────────────
// run 收敛（§四.D.1 末两行）
// ────────────────────────────────────────────────────────────────

/**
 * 收敛判定：所有节点到达终态且无 pending/ready 悬空 → completed；有 blocked → blocked。
 * 收敛后若本 run 有 parent_node_run_id，回写父节点（三值透传见 deriveChildOutcome）。
 * @returns {string} 判定后的 run_status
 */
function settleRun(runId) {
  const run = getRun(runId);
  if (!run || TERMINAL_RUN_STATUS.has(run.run_status)) return run ? run.run_status : null;

  const counts = {};
  for (const r of getDb()
    .prepare(`SELECT status, COUNT(*) AS n FROM graph_run_nodes WHERE run_id = ? GROUP BY status`)
    .all(runId)) {
    counts[r.status] = r.n;
  }

  // 还有活跃节点（含停在 running 等子 run 的 subgraph 节点）→ 没收敛
  if (counts.ready || counts.running) return run.run_status;

  let target;
  if (counts.blocked) {
    // 需要人工介入：保留 pending 节点原样，人工「重试」blocked 节点后链路还能继续
    target = 'blocked';
  } else {
    // 没有 blocked 也没有活跃节点 → 剩下的 pending 再也不会被点亮（点亮只发生在节点结算时），
    // 批量 skipped，避免「所有可达节点都完成了但 run 永远不 completed」
    for (const r of getDb()
      .prepare(`SELECT id FROM graph_run_nodes WHERE run_id = ? AND status = 'pending'`)
      .all(runId)) {
      sse.transitionNode(runId, r.id, 'pending', 'skipped', {}, {
        summary: 'pending → skipped：本 run 已收敛，这条分支再也不会被点亮',
      });
    }
    target = 'completed';
  }

  if (target !== run.run_status && canTransitionRun(run.run_status, target)) {
    sse.transitionRun(runId, run.run_status, target, {}, {
      summary: `run ${runId}: ${run.run_status} → ${target}（收敛判定）`, counts,
    });
  }
  const settled = (getRun(runId) || {}).run_status;
  if (TERMINAL_RUN_STATUS.has(settled) || settled === 'blocked') propagateToParent(runId);
  return settled;
}

/**
 * 子 run 收敛 → 回写父节点。**必须回写三个值**（§四.D.0 的 ⚠️，R-16 硬要求）：
 *   success/fail        ← 子 run 的 run_status
 *   fail_reason         ← 子图归因节点的产出，父层 §四.D.2.2 靠它选边
 *   remediation_note    ← 同上，随回退传给目标节点，成为它下一次规划的输入
 * 少了后两个，org 层选边就无据可依，只能走兜底边。
 */
function propagateToParent(childRunId) {
  const child = getRun(childRunId);
  if (!child || !child.parent_run_id || !child.parent_node_run_id) return;
  const parentNode = getRunNode(child.parent_node_run_id);
  if (!parentNode || parentNode.status !== 'running') return; // 已经结算过 → 幂等返回

  const outcome = deriveChildOutcome(child);
  // remediationNote 不是 graph_run_nodes 的列，随 outcome 一起交给父层的选边/回退用
  settleNode(child.parent_run_id, child.parent_node_run_id, outcome);
}

/** 从子 run 提取三值透传的内容 */
function deriveChildOutcome(childRun) {
  const status = childRun.run_status === 'completed' ? 'success' : 'fail';

  const def = defIndex(childRun);
  const rows = getDb()
    .prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? ORDER BY updated_at DESC, id DESC`)
    .all(childRun.id);

  // 归因优先取 validator-reflection 节点的产出；没有这个节点时退到最近一个失败/阻塞节点
  // // 假设(R-16待定案): 归因保持 org node 粒度封装 —— 父层只拿到「子图整体失败 + 根因」，
  //    org 层回退只退到 org node，不跨层寻址子图里的具体节点。
  const attribution = rows.find((r) => {
    const d = def.nodes.get(r.def_node_id);
    return d && d.name === ATTRIBUTION_NODE_NAME && r.fail_reason;
  });
  const lastBad = rows.find((r) => (r.status === 'fail' || r.status === 'blocked') && r.fail_reason);
  const failReason = status === 'fail' ? (attribution || lastBad || {}).fail_reason || null : null;

  const records = listLoopRecords(childRun.id);
  const remediationNote = (records.find((r) => r.remediation_note) || {}).remediation_note || null;

  return { status, failReason, remediationNote, childRunId: childRun.id };
}

// ────────────────────────────────────────────────────────────────
// §四.A.3 任务级总预算：只挂 root run，全树按 root_run_id 扣同一个池
// ────────────────────────────────────────────────────────────────

/**
 * budget_json 的形状（§四.A.3）：
 *   { total_agent_calls, total_tokens, wallclock_ms, spent: {agent_calls, tokens, started_at} }
 * 三个上限都可选，缺了就是这一维不限额；整个 budget_json 为空 = 完全不限额。
 * 子 run 的 budget_json 恒为 null（createRun 保证），所以永远上溯到 root_run_id 那一行读写。
 */
function readBudget(run) {
  // 必须重新读一次 root 行：调用方手里的 run 往往是 advanceOne 开头取的快照，
  // 拿它的 budget_json 做读-改-写会把同一次推进里前一笔扣减覆盖掉。
  const root = getRun(run.root_run_id);
  if (!root || !root.budget_json) return null;
  try {
    return { rootRunId: root.id, budget: JSON.parse(root.budget_json) };
  } catch {
    return null; // 预算字段坏了不该把执行卡死，按不限额处理并留给控制台暴露
  }
}

/**
 * 余额检查。任何一层发起 agent 调用前都要过这一关（advanceOne 拾取节点之后立刻查）。
 * @returns {{ok: boolean, exhausted: string[], rootRunId?: string, budget?: object}}
 */
function checkBudget(run) {
  const held = readBudget(run);
  if (!held) return { ok: true, exhausted: [], budget: null };

  const b = held.budget;
  const spent = b.spent || {};
  const exhausted = [];
  if (b.total_agent_calls != null && (spent.agent_calls || 0) >= b.total_agent_calls) {
    exhausted.push('total_agent_calls');
  }
  if (b.total_tokens != null && (spent.tokens || 0) >= b.total_tokens) exhausted.push('total_tokens');
  if (
    b.wallclock_ms != null && spent.started_at &&
    Date.now() - Date.parse(spent.started_at) >= b.wallclock_ms
  ) {
    exhausted.push('wallclock_ms');
  }
  return { ok: exhausted.length === 0, exhausted, rootRunId: held.rootRunId, budget: b };
}

/**
 * 扣减（「只增不减」，§四.A.3）。始终写 root run 那一行。
 * ponytail: 读-改-写不是原子的，当前单进程串行驱动（runAuto 一次只推一个节点）下够用；
 * 升级路径是把自增下沉成一条 `UPDATE ... SET budget_json = json_set(...)` 的 SQL。
 */
function spendBudget(run, { agentCalls = 0, tokens = 0 } = {}) {
  const held = readBudget(run);
  if (!held) return null;

  const b = held.budget;
  if (!b.spent) b.spent = {};
  b.spent.agent_calls = (b.spent.agent_calls || 0) + agentCalls;
  b.spent.tokens = (b.spent.tokens || 0) + tokens;
  if (!b.spent.started_at) b.spent.started_at = now(); // 墙钟从第一次真花钱开始算
  getDb()
    .prepare(`UPDATE graph_runs SET budget_json = ?, updated_at = ? WHERE id = ?`)
    .run(JSON.stringify(b), now(), held.rootRunId);
  // 预算是控制台顶栏那三个进度条的数据源，扣一次播一次（前端监听 budget.spent）
  sse.emitEvent(run.id, 'budget.spent', `预算扣减 +${agentCalls} 次调用 / +${tokens} tokens`, {
    root_run_id: held.rootRunId, delta: { agent_calls: agentCalls, tokens },
    spent: b.spent,
    total_agent_calls: b.total_agent_calls ?? null,
    total_tokens: b.total_tokens ?? null,
  }, { ref: held.rootRunId });
  return b;
}

/**
 * 预算耗尽 → **整棵 run 树挂起等人工**，不是失败（§四.A.3）：
 * root run → blocked，树上所有非终态 run 一律置 pause_requested=1。
 * 同样只做一次 listRunTree 索引查询。
 */
function suspendRunTree(run) {
  const rootRunId = run.root_run_id;
  const tree = listRunTree(rootRunId);
  for (const r of tree) {
    if (TERMINAL_RUN_STATUS.has(r.run_status)) continue;
    setRunFlags(r.id, { pauseRequested: true });
  }
  const root = getRun(rootRunId);
  if (root && canTransitionRun(root.run_status, 'blocked')) {
    sse.transitionRun(rootRunId, root.run_status, 'blocked', {}, {
      summary: '预算耗尽：整棵 run 树挂起等人工（不是失败）',
      reason: 'budget-exhausted',
    });
  }
  return (getRun(run.id) || {}).run_status;
}

// ────────────────────────────────────────────────────────────────
// §四.C 五个控制操作（手动 / 自动共用上面的 advanceOne）
// ────────────────────────────────────────────────────────────────

/** planning → ready（图规划完成 + 通过写入前五项校验之后调，校验在 work-graph-db） */
function markRunReady(runId) {
  const run = getRun(runId);
  if (!run) throw new Error(`graph_runs 不存在: ${runId}`);
  if (run.run_status === 'ready') return { ok: true, runStatus: 'ready' };
  sse.transitionRun(runId, 'planning', 'ready', {}, { summary: '图规划完成并通过写入前校验：planning → ready' });
  return { ok: true, runStatus: 'ready' };
}

/**
 * 入口节点 = 没有任何 forward 入边的节点（不分 success / fail），run 起跑时点亮。
 *
 * **fail 入边同样算入边**：只有 fail 入边的节点（人工审核 / escalate 那一类）是失败分支的
 * 落点，它要等失败真的发生、routeFailure 选中那条边时才被 markReady 点亮。按 condition 过滤
 * 会让它在 run 一起跑就被点亮执行 —— 在任何失败发生之前先真调一次 LLM 并计入预算
 * （spendBudget 在 advanceOne 抢到节点时就扣），随后又被 resetForwardDownstream 打回 pending。
 *
 * backward 入边仍要排除：回退目标（如 coder）除了回退边可能没有别的入边，
 * 不排除它就永远起不来。
 */
function seedEntryNodes(runId) {
  const run = getRun(runId);
  const def = defIndex(run);
  const hasIncoming = new Set(def.edges.filter((e) => e.edgeKind !== 'backward').map((e) => e.to));
  const seeded = [];
  for (const [defNodeId] of def.nodes) {
    if (hasIncoming.has(defNodeId)) continue;
    const activated = markReady(runId, def, defNodeId);
    if (activated) seeded.push(activated);
  }
  return seeded;
}

/**
 * 「启动」：run_status ∈ {ready, paused, blocked} → running。
 * 从 ready 起跑时顺带点亮入口节点；恢复（paused/blocked）时不动已有节点状态。
 * 同步部分单独抽出来，供 executeSubgraph 起子 run 时复用。
 */
function startRunSync(runId, { mode } = {}) {
  const run = getRun(runId);
  if (!run) throw new Error(`graph_runs 不存在: ${runId}`);

  const fresh = run.run_status === 'ready';
  if (run.run_status !== 'running') {
    sse.transitionRun(runId, run.run_status, 'running', {}, {
      summary: `启动：${run.run_status} → running${mode ? `（${mode} 模式）` : ''}`,
      execution_mode: mode || run.execution_mode,
    });
  }
  // 「启动」清掉软信号，否则 advanceOne 一进来就被上一次的暂停挡住
  setRunFlags(runId, { pauseRequested: false, stopRequested: false });
  if (mode) {
    getDb().prepare(`UPDATE graph_runs SET execution_mode = ?, updated_at = ? WHERE id = ?`)
      .run(mode, now(), runId);
  }

  const seeded = fresh ? seedEntryNodes(runId) : [];
  return { ok: true, runId, runStatus: 'running', seeded };
}

/**
 * 「启动」（对外）。auto 模式立刻挂一个自我调度的循环上去；manual 模式只标 running，
 * 停在原地等「下一步」。返回里的 driver 是 auto 循环的 promise（路由可以不管它，测试可以 await）。
 */
function startRun(runId, opts = {}) {
  const res = startRunSync(runId, opts);
  const mode = opts.mode || (getRun(runId) || {}).execution_mode;
  if (mode !== 'auto') return res;
  // 自动模式不是另一套推进逻辑，就是一个不停调 advanceOne 的循环（§四.C 末段）
  const driver = runAuto(runId, opts).catch((err) => {
    console.error(`[graph-engine] auto 驱动异常 run=${runId}: ${err.message}`);
    return { ok: false, error: err.message };
  });
  return { ...res, driver };
}

/**
 * 「下一步」：拾取一个 ready 节点执行完就停，不继续往下跑。
 * 已 paused 时也允许单步（§四.C 表格「可用条件」那一列）——单步临时把 run 拉回 running，
 * 步完只要 pause_requested 还在就自动落回 paused。
 * **「下一步」只作用于当前层级，不级联**（§四.C 定案：暂停是全局意图，单步是局部操作）。
 */
function nextStep(runId, opts = {}) {
  const run = getRun(runId);
  if (!run) throw new Error(`graph_runs 不存在: ${runId}`);
  if (run.run_status === 'paused') {
    sse.transitionRun(runId, 'paused', 'running', {}, { summary: '「下一步」：paused → running（单步）' });
  }

  const cur = getRun(runId);
  if (cur.run_status !== 'running') {
    return Promise.resolve(result(runId, cur.run_status, 'run-not-running'));
  }
  return advanceOne(runId, { ...opts, ignorePause: true }).then((res) => {
    landPause(runId);
    return { ...res, runStatus: (getRun(runId) || {}).run_status };
  });
}

/**
 * 「暂停」：软信号，不打断正在 running 的节点，下一个节点边界生效。
 * **沿 parent_run_id 向下级联到所有 active 子 run**（§四.C 三层级联定案）——
 * 层级是三层（root → org node → 嵌套子图），不级联就会出现「点了暂停但子图还在偷偷跑」。
 */
function pauseRun(runId) {
  const run = getRun(runId);
  if (!run) throw new Error(`graph_runs 不存在: ${runId}`);

  const targets = [run, ...descendantRuns(run)].filter((r) => !TERMINAL_RUN_STATUS.has(r.run_status));
  for (const r of targets) {
    setRunFlags(r.id, { pauseRequested: true });
    landPause(r.id); // 该 run 手头没有 running 节点就立刻落 paused，有就等它到节点边界
  }
  return {
    ok: true,
    runId,
    pauseRequested: 1,
    runStatus: (getRun(runId) || {}).run_status,
    cascaded: targets.map((r) => r.id).filter((id) => id !== runId),
  };
}

/**
 * running → paused 的落库时机 = **节点边界**（§四.C：暂停不打断当前节点）。
 * 手头还有 running 节点就先不落，等 advanceOne 下一次被 pause_requested 挡住时再落。
 */
function landPause(runId) {
  const run = getRun(runId);
  if (!run || !run.pause_requested || run.run_status !== 'running') return null;
  const busy = getDb()
    .prepare(`SELECT 1 FROM graph_run_nodes WHERE run_id = ? AND status = 'running' LIMIT 1`)
    .get(runId);
  if (busy) return null;
  sse.transitionRun(runId, 'running', 'paused', {}, { summary: '到达节点边界，暂停落库：running → paused' });
  return 'paused';
}

/**
 * 「停止」：人工主动终止，**同样向下级联**。每一层：置 stop_requested → 仍 pending/ready
 * 的节点批量 skipped → run → stopped。
 *
 * 正在 running 的节点：§3.6 的进程级硬取消调用链由 T9 核实，当前是文档给的降级路径——
 * 「等它跑完、记录结果、但不再调度后续节点」（stop_requested 让 advanceOne 不再拾取新节点）。
 */
function stopRun(runId) {
  const run = getRun(runId);
  if (!run) throw new Error(`graph_runs 不存在: ${runId}`);

  const targets = [run, ...descendantRuns(run)];
  for (const r of targets) {
    setRunFlags(r.id, { stopRequested: true });
    for (const n of getDb()
      .prepare(`SELECT id, status FROM graph_run_nodes WHERE run_id = ? AND status IN ('pending','ready')`)
      .all(r.id)) {
      sse.transitionNode(r.id, n.id, n.status, 'skipped', {}, { summary: '人工「停止」：未开跑的节点批量 skipped' });
    }
    const fresh = getRun(r.id);
    if (fresh && canTransitionRun(fresh.run_status, 'stopped')) {
      sse.transitionRun(r.id, fresh.run_status, 'stopped', {}, {
        summary: `人工「停止」：${fresh.run_status} → stopped`,
        cascaded_from: runId,
      });
    }
  }
  return {
    ok: true,
    runId,
    runStatus: (getRun(runId) || {}).run_status,
    cascaded: targets.map((r) => r.id).filter((id) => id !== runId),
  };
}

/**
 * 一个 run 的全部后代（不含自己）。
 * `root_run_id` 冗余列的用处就在这里：**一次索引查询**把整棵树拉出来，再在内存里沿
 * parent_run_id 走一遍取子树，不做递归 SQL（§四.C）。
 */
function descendantRuns(run) {
  const all = listRunTree(run.root_run_id);
  const byParent = new Map();
  for (const r of all) {
    if (!byParent.has(r.parent_run_id)) byParent.set(r.parent_run_id, []);
    byParent.get(r.parent_run_id).push(r);
  }
  const out = [];
  const stack = [run.id];
  while (stack.length) {
    for (const child of byParent.get(stack.pop()) || []) {
      out.push(child);
      stack.push(child.id);
    }
  }
  return out;
}

/**
 * 「重试」：节点 status ∈ {fail, blocked} → ready，attempt + 1。
 *
 * **默认不级联重置下游**（§四.C 末段）：级联与否不由重试这个动作决定，而由**产出契约是否
 * 真的变了**决定。判定发生在重跑成功之后（applyRetryCascade），这里只把当前契约拍个快照。
 */
function retryNode(runId, nodeRunId) {
  const nodeRun = getRunNode(nodeRunId);
  if (!nodeRun || nodeRun.run_id !== runId) throw new Error(`节点不属于该 run: ${nodeRunId}`);
  // 人工重试同样是「新的一轮」：subgraph 节点清 child_run_id，重跑时起新子 run
  // （§四.D.3；不清的话 executeSubgraph 会拿旧的终态子 run 返回 pending → 永久卡 running）
  if (
    !sse.transitionNode(runId, nodeRunId, nodeRun.status, 'ready', { childRunId: null }, {
      attempt: nodeRun.attempt + 1,
      summary: `人工「重试」：${nodeRun.status} → ready（attempt #${nodeRun.attempt + 2}）`,
    })
  ) {
    return { ok: false, reason: 'taken-by-other', nodeStatus: (getRunNode(nodeRunId) || {}).status };
  }
  bumpNodeAttempt(nodeRunId);

  const run = getRun(runId);
  const snapshot = snapshotContract(run, nodeRunId);
  if (run.run_status === 'blocked') {
    sse.transitionRun(runId, 'blocked', 'running', {}, { summary: '人工重试解除阻塞：blocked → running' });
  }
  return {
    ok: true,
    runId,
    nodeRunId,
    nodeStatus: 'ready',
    contractSnapshot: snapshot,
    runStatus: (getRun(runId) || {}).run_status,
  };
}

// ────────────────────────────────────────────────────────────────
// §四.C 末段：重试后级联下游的判据 = contract_hash 变没变
// ────────────────────────────────────────────────────────────────

/** 节点私有产物目录（§四.A.5）：data/companies/{cid}/runs/{rootRunId}/nodes/{nodeRunId}/ */
function nodeArtifactDir(run, nodeRunId) {
  return path.join(REPO_ROOT, 'data', 'companies', run.company_id, 'runs', run.root_run_id, 'nodes', nodeRunId);
}

function readJsonOrNull(file) {
  try {
    return JSON.parse(fs.readFileSync(file, 'utf8'));
  } catch {
    return null;
  }
}

/** 重试前把节点当前的 contract.json 拍成 contract.prev.json；没有契约就不拍（=不会级联） */
function snapshotContract(run, nodeRunId) {
  const dir = nodeArtifactDir(run, nodeRunId);
  const cur = path.join(dir, 'contract.json');
  if (!fs.existsSync(cur)) return null;
  const prev = path.join(dir, 'contract.prev.json');
  fs.copyFileSync(cur, prev);
  return prev;
}

/**
 * 重跑成功后判级联：有快照（说明这个节点被人工重试过）才判，没快照什么都不做。
 * 四分类（added / changed / removed / unchanged，同 loops-l3.md 的增量算法）：
 * **unchanged 保持原状不重跑**；只要出现任一 added/changed/removed 就把 forward 下游打回
 * pending，等本节点的 success 出边重新点亮它们。
 *
 * ponytail: 引擎不知道「哪个 export 被哪个下游节点消费」（那是图规划者的知识），所以四分类
 * 算出来之后退化成一个整体的 dirty 位。升级路径是让 workgraph-planning 在边上声明消费的
 * export key，届时按 key 精确挑下游即可，diffContracts 的输出形状不用改。
 */
function applyRetryCascade(runId, def, nodeRun) {
  const run = getRun(runId);
  const dir = nodeArtifactDir(run, nodeRun.id);
  const prevFile = path.join(dir, 'contract.prev.json');
  if (!fs.existsSync(prevFile)) return null;

  const prev = readJsonOrNull(prevFile);
  const next = readJsonOrNull(path.join(dir, 'contract.json'));
  fs.unlinkSync(prevFile); // 快照一次性，判完即弃

  const diff = diffContracts(prev, next);
  diff.reset = diff.dirty ? resetForwardDownstream(runId, def, nodeRun.def_node_id) : [];
  return diff;
}

/**
 * 按 contract_hash 把两份契约的 exports 分成四类。
 * 单个 export 的 hash 复用 devloop/doc-store.js:382 的 hashContract()
 * （对 exports 做递归 key 排序的规范化序列化后 sha256 取前 16 位），不另造一套。
 */
function diffContracts(prev, next) {
  const a = exportsOf(prev);
  const b = exportsOf(next);
  const res = { added: [], changed: [], removed: [], unchanged: [] };
  for (const key of new Set([...Object.keys(a), ...Object.keys(b)])) {
    if (!(key in a)) res.added.push(key);
    else if (!(key in b)) res.removed.push(key);
    else if (hashExport(a[key]) === hashExport(b[key])) res.unchanged.push(key);
    else res.changed.push(key);
  }
  res.dirty = res.added.length + res.changed.length + res.removed.length > 0;
  return res;
}

function exportsOf(contract) {
  if (!contract || typeof contract !== 'object') return {};
  const e = contract.exports || contract;
  return e && typeof e === 'object' && !Array.isArray(e) ? e : { _: e };
}

function hashExport(value) {
  // 懒 require：doc-store 会拉起 path-resolver 等一串 DevLoop 依赖，不该在本模块加载时就付这个钱
  const { hashContract } = require('./devloop/doc-store');
  return hashContract({ exports: { value } });
}

// ────────────────────────────────────────────────────────────────
// 自动模式 = 不停调 advanceOne 的循环（不是另一套实现）
// ────────────────────────────────────────────────────────────────

/**
 * 在一棵 run 树里推进一步：优先推最深的 active 子 run（父节点正卡在它上面）。
 * 单张图时退化成 advanceOne(rootRunId)。
 */
async function advanceTree(rootRunId, opts = {}) {
  const tree = listRunTree(rootRunId).filter((r) => r.run_status === 'running');
  // 后建的 run 一定更深（子 run 总是在父节点 running 之后建的），倒序即「最深优先」
  for (const r of tree.reverse()) {
    const res = await advanceOne(r.id, opts);
    if (res.ok) return res;
  }
  return result(rootRunId, (getRun(rootRunId) || {}).run_status, 'no-ready-node');
}

/**
 * 自动驱动：只要 pause_requested / stop_requested 都是假就在上一次 resolve 后立刻发起下一次。
 * maxSteps 只是防跑飞的安全阀，真正的终止条件是预算（§四.A.3）与熔断（§四.D.2.1）。
 * ponytail: 串行单步循环，一次只跑一个节点，并行分支不会并发执行；上限是吞吐，
 * 升级路径是按 ready 集合并发 advanceOne（节点状态机已经是幂等条件更新，能直接支撑）。
 */
async function runAuto(rootRunId, opts = {}) {
  const maxSteps = opts.maxSteps || 1000;
  let steps = 0;
  for (; steps < maxSteps; steps++) {
    const run = getRun(rootRunId);
    if (!run || run.pause_requested || run.stop_requested) break;
    if (TERMINAL_RUN_STATUS.has(run.run_status) || run.run_status === 'blocked') break;
    const res = await advanceTree(rootRunId, opts);
    if (!res.ok) break;
  }
  return { ok: true, runId: rootRunId, steps, runStatus: (getRun(rootRunId) || {}).run_status };
}

// ────────────────────────────────────────────────────────────────
// 内部小工具
// ────────────────────────────────────────────────────────────────

/** 定义层索引：nodes 用 Map(defNodeId → node)，edges 原样（图规模小，不做二次索引） */
function defIndex(run) {
  const def = readDefGraph(run.graph_kind, run.graph_id);
  return { nodes: new Map(def.nodes.map((n) => [n.defNodeId, n])), edges: def.edges };
}

function result(runId, runStatus, reason) {
  return {
    ok: false,
    runId,
    runStatus: runStatus || null,
    reason,
    node: null,
    activated: [],
    backwardEdgeId: null,
    loop: null,
    contractDiff: null,
    childRunId: null,
  };
}

function nodeView(runId, r) {
  if (!r) return null;
  const run = getRun(runId);
  const d = run ? defIndex(run).nodes.get(r.def_node_id) : null;
  return {
    id: r.id,
    defNodeId: r.def_node_id,
    name: d ? d.name : null,
    kind: d ? d.nodeKind : null,
    status: r.status,
    attempt: r.attempt,
    failReason: r.fail_reason,
    outputRef: r.output_ref,
    agentRunId: r.agent_run_id,
    childRunId: r.child_run_id,
  };
}

/**
 * 拼进提示词的产物目录约定（§四.A.5 + validator-reflection/agent.md「产出格式」）。
 *
 * 路径与 readNodeProduct 读的**是同一个**（nodeArtifactDir），文件名给绝对路径，
 * 不留 0 基 / 1 基的解释空间——模板里 `attempt-{n}.json` 的 n 没定死基数，
 * 这里直接把本轮该写的文件名算好告诉它。
 */
function describeArtifactContract(ctx) {
  const dir = safeNodeArtifactDir(ctx);
  if (!dir) return ''; // run 缺 company_id/root_run_id：拼不出目录就不编一个假的塞给 agent
  return [
    '',
    '',
    '[节点产物目录]',
    `你的私有目录（可读可写）：${dir}`,
    `本轮的结构化产出/归因写到：${path.join(dir, `attempt-${ctx.attempt}.json`)}`,
    '格式见 system/templates/validator-reflection/agent.md「产出格式」'
      + '（schema=validator-reflection/v1；引擎只读 verdict / fail_reason / remediation_note 三个字段，'
      + 'verdict ∈ attributed|inconclusive|no_prior_failure，fail_reason ∈ '
      + `${[...ATTRIBUTION_REASONS].join('|')} 或 null）。`,
    `任务级共享产物区（只读）：${path.join(REPO_ROOT, 'data', 'companies', ctx.companyId || '', 'runs', ctx.rootRunId || '', 'artifacts')}`,
    '不要读写其它节点的私有目录（§四.A.5 的隔离规则）。',
    '判词类节点（validator / reflection）：**没有这份文件 = 无法判定**，引擎会按失败结算走兜底边。',
  ].join('\n');
}

/** nodeArtifactDir 的不抛版本（run 缺列时返回 null） */
function safeNodeArtifactDir(ctx) {
  try {
    return ctx && ctx.run ? nodeArtifactDir(ctx.run, ctx.nodeRunId) : null;
  } catch {
    return null;
  }
}

/** 没有显式任务描述时给 executor 的兜底文案（goal/task/okr 三者基数不同，见 27-NAMING §四） */
function describeTask(run) {
  const parts = [run.goal_id && `goal=${run.goal_id}`, run.task_id && `task=${run.task_id}`,
    run.okr_node_id && `okr=${run.okr_node_id}`].filter(Boolean);
  return parts.length ? `执行图节点任务（${parts.join(', ')}）` : '执行图节点任务';
}

/**
 * 「应该如何修复」，作为下一轮的上下文补充（§四.D.2.1 第 3 步）。
 *
 * **必须按 to_node_run_id 过滤**：§四.D.2.1 第 3 步说的是把判词带给**回退目标**。
 * 同一个 run 里多条回退边并存时（验收 9b 那种图），不过滤会把给 a 的修复建议原样塞进
 * b/c 的提示词 —— 判词串台比没有判词更糟，下一轮那个 agent 会照着别人的活干。
 */
function latestRemediationNote(runId, toNodeRunId) {
  const rec = listLoopRecords(runId)
    .find((r) => r.remediation_note && (!toNodeRunId || r.to_node_run_id === toNodeRunId));
  return rec ? rec.remediation_note : null;
}

module.exports = {
  // 核心原语
  advanceOne,
  advanceTree,
  runAuto,
  settleNode,
  settleRun,

  // 五个控制操作
  startRun,
  nextStep,
  pauseRun,
  stopRun,
  retryNode,
  descendantRuns,
  landPause,

  // 任务级总预算（§四.A.3）
  checkBudget,
  spendBudget,
  suspendRunTree,

  // 重试级联判据（§四.C 末段）
  diffContracts,
  applyRetryCascade,
  snapshotContract,

  // run 生命周期辅助
  markRunReady,
  seedEntryNodes,

  // executor 与注入点
  executeSubgraph,
  defaultAgentExecutor,
  setAgentExecutor,
  setPlanWorkGraph,

  // 选边 / 熔断 / 透传
  selectFailEdge,
  handleBackwardEdge,
  resetForwardDownstream,
  deriveChildOutcome,
  ATTRIBUTION_NODE_NAME,
  INFRA_FLAKE_REASON,
  LOOP_HARD_CAP_EDGE,
  LOOP_HARD_CAP_RUN,
};