From e95e77f0a3a0749984098876cec121610d00d2f6 Mon Sep 17 00:00:00 2001 From: yibo Date: Fri, 21 Aug 2026 13:35:41 +0800 Subject: [PATCH] feat(graph-engine): T5 欠费门可插拔桩子 + 自动恢复（27-P4） --- server/services/graph-engine.js | 148 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------- server/services/work-graph-db.js | 17 +++++++++++++++++ server/tests/unit/graph-engine-advance-batch.test.js | 9 +++++++-- 3 files changed, 158 insertions(+), 16 deletions(-) diff --git a/server/services/graph-engine.js b/server/services/graph-engine.js index 726fdb9..3a2b4d9 100644 --- a/server/services/graph-engine.js +++ b/server/services/graph-engine.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 | + * {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, diff --git a/server/services/work-graph-db.js b/server/services/work-graph-db.js index a6296c2..e79b6b4 100644 --- a/server/services/work-graph-db.js +++ b/server/services/work-graph-db.js @@ -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', }; /** diff --git a/server/tests/unit/graph-engine-advance-batch.test.js b/server/tests/unit/graph-engine-advance-batch.test.js index da62310..ebafef7 100644 --- a/server/tests/unit/graph-engine-advance-batch.test.js +++ b/server/tests/unit/graph-engine-advance-batch.test.js @@ -143,9 +143,13 @@ describe('advanceBatch（27-P4 T4/T6）', () => { 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('advanceBatch（27-P4 T4/T6）', () => { 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 -- libgit2 0.21.4