feat(graph-engine): T5 欠费门可插拔桩子 + 自动恢复（27-P4）

先搭桩子，不接真实计费接口（用户明确选择：真实欠费判据依赖计费系统那边确认，现在只把可插拔的挂载点和恢复机制先落地，接口确认后再换真实实现）。新增： - setBillingGate(fn)：与 setAgentExecutor/setPlanWorkGraph 同款注入模式，未注入时 checkBilling 恒返回「未欠费」，对现有行为零影响。 - checkBilling 挂进 advanceOne 与 advanceBatch，紧跟 checkBudget 之后——两者是不同维度的门（任务级预算池 vs 账户级欠费），不合并成一个判据。 - 欠费挂起复用现成的 suspendRunTree（预算耗尽走的同一条路），只是 reason 换成 'billing-arrears'。suspendRunTree 加 reason/summary 参数，并把 reason 写进新增的 graph_runs.block_reason 列（欠费清零后的自动恢复扫描要按它过滤，不用解析 timeline 文本）；startRunSync「启动」时清空这一列。 - resumeBillingClearedRuns()：欠费清零后的自动恢复触发器——今天从 blocked 恢复靠人手点「启动」，欠费清零不会有人去点，这个函数补上「扫一遍 block_reason= billing-arrears 的 blocked root run，欠费真清了就代它点启动」。调度触发源（webhook/轮询）留给计费接口接通之后再定。顺手修了一个 advanceBatch 的排序 bug：budget/billing 门必须在「确认真有 ready 候选」之后才查，跟 advanceOne 的既有顺序一致——反过来会把一个已经收敛、只是恰好预算也花完了的 run 误判成「挂起」而不是正确收敛到 completed。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

feat(graph-engine): T5 欠费门可插拔桩子 + 自动恢复（27-P4）
先搭桩子，不接真实计费接口（用户明确选择：真实欠费判据依赖计费系统那边确认，现在只把可插拔的挂载点和恢复机制先落地，接口确认后再换真实实现）。新增： - setBillingGate(fn)：与 setAgentExecutor/setPlanWorkGraph 同款注入模式，未注入时 checkBilling 恒返回「未欠费」，对现有行为零影响。 - checkBilling 挂进 advanceOne 与 advanceBatch，紧跟 checkBudget 之后——两者是不同维度的门（任务级预算池 vs 账户级欠费），不合并成一个判据。 - 欠费挂起复用现成的 suspendRunTree（预算耗尽走的同一条路），只是 reason 换成 'billing-arrears'。suspendRunTree 加 reason/summary 参数，并把 reason 写进新增的 graph_runs.block_reason 列（欠费清零后的自动恢复扫描要按它过滤，不用解析 timeline 文本）；startRunSync「启动」时清空这一列。 - resumeBillingClearedRuns()：欠费清零后的自动恢复触发器——今天从 blocked 恢复靠人手点「启动」，欠费清零不会有人去点，这个函数补上「扫一遍 block_reason= billing-arrears 的 blocked root run，欠费真清了就代它点启动」。调度触发源（webhook/轮询）留给计费接口接通之后再定。顺手修了一个 advanceBatch 的排序 bug：budget/billing 门必须在「确认真有 ready 候选」之后才查，跟 advanceOne 的既有顺序一致——反过来会把一个已经收敛、只是恰好预算也花完了的 run 误判成「挂起」而不是正确收敛到 completed。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
yibo
1 parent 8a6dbc8c
Showing 3 changed files with 158 additions and 16 deletions
server/services/graph-engine.js
server/services/work-graph-db.js
server/tests/unit/graph-engine-advance-batch.test.js
@@ -1370,6 +1370,14 @@ async function advanceOne(runId, opts = {}) {
     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;
@@ -1585,15 +1593,11 @@ async function advanceBatch(runId, opts = {}) {
     return { ok: false, runId, runStatus: run.run_status, reason: 'run-not-running', 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,
-    };
-  }
-
+  // 候选先查、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 ?`)
@@ -1606,6 +1610,26 @@ async function advanceBatch(runId, opts = {}) {
     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) || {};
@@ -2541,6 +2565,82 @@ function checkBudget(run) {
 }
  
 /**
+ * 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 };
+}
+
+/**
  * 扣减（「只增不减」，§四.A.3）。始终写 root run 那一行。
  * ponytail: 读-改-写不是原子的，当前单进程串行驱动（runAuto 一次只推一个节点）下够用；
  * 升级路径是把自增下沉成一条 `UPDATE ... SET budget_json = json_set(...)` 的 SQL。
@@ -2568,11 +2668,22 @@ function spendBudget(run, { agentCalls = 0, tokens = 0 } = {}) {
 }
  
 /**
- * 预算耗尽 → **整棵 run 树挂起等人工**，不是失败（§四.A.3）：
+ * 资源类原因（预算耗尽 / 账户欠费）→ **整棵 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) {
+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) {
@@ -2581,9 +2692,8 @@ function suspendRunTree(run) {
   }
   const root = getRun(rootRunId);
   if (root && canTransitionRun(root.run_status, 'blocked')) {
-    sse.transitionRun(rootRunId, root.run_status, 'blocked', {}, {
-      summary: '预算耗尽：整棵 run 树挂起等人工（不是失败）',
-      reason: 'budget-exhausted',
+    sse.transitionRun(rootRunId, root.run_status, 'blocked', { blockReason: reason }, {
+      summary, reason,
     });
   }
   return (getRun(run.id) || {}).run_status;
@@ -2726,6 +2836,11 @@ function startRunSync(runId, { 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) : [];
@@ -4207,6 +4322,11 @@ module.exports = {
   wallclockElapsedMs,
   suspendRunTree,
  
+  // 27-P4 T5：账户欠费门（可插拔桩子）+ 自动恢复
+  checkBilling,
+  setBillingGate,
+  resumeBillingClearedRuns,
+
   // 重试级联判据（§四.C 末段）
   diffContracts,
   applyRetryCascade,
@@ -326,6 +326,13 @@ function getDb() {
       -- DEFAULT_MAX_CONCURRENCY（引擎侧解析，不在 DB 层给默认值，理由同其它可空列——
       -- 应用层的默认值改了不用迁移存量行）。
       max_concurrency    INTEGER,
+      -- 27-P4 T5：root run 挂起（run_status='blocked'）的机器可读原因，只有
+      -- suspendRunTree 写它（'budget-exhausted' | 'billing-arrears'），startRunSync
+      -- 「启动」时无条件清空。只在 root run 上有意义——suspendRunTree 只把树的 root
+      -- 转成 blocked，子 run 只收到 pause_requested，不动它们的 block_reason（它们本来就没有）。
+      -- 存在的意义是让「欠费清零后自动恢复」这类扫描能按原因过滤（run_status/block_reason
+      -- 复合索引），不用去解析 timeline 里的 event 文本。
+      block_reason       TEXT,
       created_at         TEXT NOT NULL,
       updated_at         TEXT NOT NULL
     );
@@ -649,6 +656,15 @@ function getDb() {
   if (!hasColumn(db, 'graph_runs', 'max_concurrency')) {
     db.exec(`ALTER TABLE graph_runs ADD COLUMN max_concurrency INTEGER`);
   }
+  // 27-P4 T5：graph_runs 追加 block_reason，可空，旧行 NULL = 「不是因为欠费/预算挂起的
+  // blocked run」（批 3 时代 blocked 只有熔断/死路一种成因，回填 NULL 如实反映当时没有这个概念）。
+  if (!hasColumn(db, 'graph_runs', 'block_reason')) {
+    db.exec(`ALTER TABLE graph_runs ADD COLUMN block_reason TEXT`);
+  }
+  // 欠费自动恢复的扫描入口：`WHERE run_status='blocked' AND block_reason='billing-arrears'
+  // AND parent_run_id IS NULL`，加个复合索引配这个查询。
+  db.exec(`CREATE INDEX IF NOT EXISTS idx_graph_runs_block_reason
+    ON graph_runs(run_status, block_reason) WHERE parent_run_id IS NULL`);
   if (!hasColumn(db, 'goals', 'reported_at')) {
     db.exec(`ALTER TABLE goals ADD COLUMN reported_at TEXT`);
     db.exec(`
@@ -1355,6 +1371,7 @@ const RUN_PATCH_COLUMNS = {
   stopRequested: 'stop_requested',
   executionMode: 'execution_mode',
   budgetJson: 'budget_json',
+  blockReason: 'block_reason',
 };
  
 /**
@@ -143,9 +143,13 @@ describe(&#39;advanceBatch（27-P4 T4/T6）&#39;, () =&gt; {
     await engine.advanceOne(runId); // entry → success，点亮 a/b/c
     calls.length = 0; // 只看这一批 advanceBatch 触发的调用，entry 那次不算
  
-    // 刻意不 await：advanceBatch 在自己第一次真正挂起（Promise.allSettled）之前全程同步，
-    // 所以这一行语句本身执行完，三个节点的 claim + executor 调用就都已经发生了。
+    // advanceBatch 内部在拾取候选之后、正式 claim 之前会 `await checkBilling(run)`
+    // （27-P4 T5 接入的欠费门）——即使没注入 billingGateFn、内部零真实 await，`await` 本身
+    // 也至少让出一次 microtask。所以这里不能再指望「调用这一行语句本身执行完就已经 claim 好了」，
+    // 用一次 microtask flush 让它跨过那一次 await，claim 循环仍然是同步的，跨过去之后
+    // 三个节点必然已经全部 claim + 调用完（不依赖真实定时器，不是计时手段）。
     const batchPromise = engine.advanceBatch(runId, { maxConcurrency: 3 });
+    await Promise.resolve();
  
     expect(calls.sort()).toEqual(['a', 'b', 'c']);
     expect(statusMap(runId)).toMatchObject({ a: 'running', b: 'running', c: 'running' });
@@ -225,6 +229,7 @@ describe(&#39;advanceBatch（27-P4 T4/T6）&#39;, () =&gt; {
     await engine.advanceOne(runId); // entry → success
  
     const batchPromise = engine.advanceBatch(runId); // 不传 maxConcurrency，读 run.max_concurrency=2
+    await Promise.resolve(); // 跨过 checkBilling 那次 await（见上一条用例的注释）
     const running = Object.entries(statusMap(runId)).filter(([, s]) => s === 'running');
     expect(running).toHaveLength(2); // 三个 ready 里只拾取了两个，第三个仍是 ready