From 9891c3b71e3c6ebe87e20bbd546262e0fc6f57a7 Mon Sep 17 00:00:00 2001 From: yibo Date: Thu, 20 Aug 2026 16:44:33 +0800 Subject: [PATCH] fix(budget): 墙钟要跟着整棵 run 树，不只是 root —— 真跑撞出来的 --- server/services/graph-engine.js | 28 +++++++++++++++++++++++++--- server/services/work-graph-db.js | 38 ++++++++++++++++++++++++++++++++++++++ server/tests/unit/budget-wallclock-active.test.js | 61 ++++++++++++++++++++++++++++++++++++++++++++++++++++--------- server/tests/unit/budget-wallclock-migration.test.js | 88 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 203 insertions(+), 12 deletions(-) create mode 100644 server/tests/unit/budget-wallclock-migration.test.js diff --git a/server/services/graph-engine.js b/server/services/graph-engine.js index d1b3777..2b4ddd4 100644 --- a/server/services/graph-engine.js +++ b/server/services/graph-engine.js @@ -2281,18 +2281,33 @@ function wallclockElapsedMs(spent = {}) { */ function settleWallclock(runId, to) { const run = getRun(runId); - if (!run || run.parent_run_id) return null; // 预算只挂在 root run 那一行 + 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') { - s.active_ms = s.active_ms || 0; - s.started_at = now(); // 无条件重锚：进 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)); } @@ -2305,6 +2320,13 @@ function settleWallclock(runId, to) { 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}} diff --git a/server/services/work-graph-db.js b/server/services/work-graph-db.js index 0fe11f6..e934850 100644 --- a/server/services/work-graph-db.js +++ b/server/services/work-graph-db.js @@ -623,10 +623,48 @@ function getDb() { `); } + // 墙钟预算改成「只算活跃时长」之后的存量重锚（graph-engine.wallclockElapsedMs）。 + // + // 老形态：spent.started_at = 第一次花钱的绝对时刻，没有 active_ms。那个锚点里混着 + // **挂起等人**的时间，真实活跃时长复原不出来 —— 按 0 起算、一笔勾销。另外两维 + // （agent_calls / tokens）是精确的且没受影响，真正兜住花钱的是那两维。 + // + // **为什么必须在开库时做，而不是交给转换钩子**：钩子只在「进 / 出 running」时触发， + // 而升级那一刻正停在 running 的 run 不会再进一次 running —— 旧锚点一直挂着， + // 下一次 checkBudget 直接判耗尽。 + // + // 放在**这里**而不是 graph-db.js 的 migrate()：graph_runs 是本文件建的表， + // graph-db 的迁移块跑在它之前，写在那边会在新库上直接 `no such table: graph_runs` + // （踩过了 —— 和 §16.2 那次「加索引写进建表块」是同一类顺序问题）。 + backfillWallclockAnchor(db); + schemaEnsured = true; return db; } +/** + * 存量墙钟锚点重置：active_ms 补 0；running 的重新锚到现在，其余断开锚点。 + * 只碰配了 wallclock_ms 的行 —— 没配的本来就不限墙钟，凭空加字段只会让人以为限了。 + * 判据是 active_ms 缺席，所以第二遍是零行更新（幂等）。 + */ +function backfillWallclockAnchor(db) { + const rows = db + .prepare(`SELECT id, run_status, budget_json FROM graph_runs WHERE budget_json IS NOT NULL`) + .all(); + const upd = db.prepare(`UPDATE graph_runs SET budget_json = ? WHERE id = ?`); + const nowIso = new Date().toISOString(); + for (const r of rows) { + let b; + try { b = JSON.parse(r.budget_json); } catch { continue; } // 坏字段不该把开库卡死 + if (!b || b.wallclock_ms == null) continue; + if (!b.spent) b.spent = {}; + if (b.spent.active_ms != null) continue; // 已经是新形态 + b.spent.active_ms = 0; + b.spent.started_at = r.run_status === 'running' ? nowIso : null; + upd.run(JSON.stringify(b), r.id); + } +} + // ──────────────────────────────────────────────────────────────── // §四.A.4 状态机（模式照抄 devloop/unit-registry.js：显式转换表 + 非法转换抛错） // ──────────────────────────────────────────────────────────────── diff --git a/server/tests/unit/budget-wallclock-active.test.js b/server/tests/unit/budget-wallclock-active.test.js index 77a4f2f..ef0686e 100644 --- a/server/tests/unit/budget-wallclock-active.test.js +++ b/server/tests/unit/budget-wallclock-active.test.js @@ -73,6 +73,9 @@ describe('预算 · 墙钟只算活跃时长', () => { expect(started.started_at).toBeTruthy(); expect(started.active_ms).toBe(5 * 60_000); + // 生产里转换先落库、钩子后跑，所以这里也得先把状态改掉 —— 不改的话 + // treeHasRunningRun 仍看到 running，结算被正确地跳过（那是另一条用例在验的行为） + wdb.getDb().prepare(`UPDATE graph_runs SET run_status='blocked' WHERE id='r_settle'`).run(); const stopped = engine.settleWallclock('r_settle', 'blocked'); expect(stopped.started_at).toBeNull(); expect(stopped.active_ms).toBeGreaterThanOrEqual(5 * 60_000); // 只增不减 @@ -88,16 +91,56 @@ describe('预算 · 墙钟只算活跃时长', () => { expect(engine.checkBudget(wdb.getRun('r_legacy')).ok).toBe(true); }); - it('子 run 不动预算：结算只写 root 那一行', () => { - seedRun('r_root', { active_ms: 0, started_at: null }, 'running'); - const db = wdb.getDb(); - db.prepare(`INSERT INTO graph_runs + /** + * 这一组是**真跑撞出来的**：第一版只认 root run 的转换。 + * 而人工重试一个 work 层节点动的是**子 run**（blocked → running），root 全程 blocked —— + * 于是子 run 跑整场、墙钟一次都没开始计，旧锚点还留着，第一次 checkBudget 立刻判耗尽。 + * 单测当时全绿：它们只喂了 root。 + */ + const seedChild = (id, rootId, status) => { + wdb.getDb().prepare(`INSERT INTO graph_runs (id, graph_kind, graph_id, company_id, root_run_id, parent_run_id, execution_mode, run_status, budget_json, created_at, updated_at) - VALUES ('r_child','work','wg1','co1','r_root','r_root','manual','running', NULL, ?, ?)`) - .run('2026-08-18T15:49:00.000Z', '2026-08-18T15:49:00.000Z'); - expect(engine.settleWallclock('r_child', 'blocked')).toBeNull(); - const root = JSON.parse(wdb.getRun('r_root').budget_json); - expect(root.spent.started_at).toBeNull(); // 没被子 run 的转换改动 + VALUES (?,'work','wg1','co1',?,?,'manual',?, NULL, ?, ?)`) + .run(id, rootId, rootId, status, '2026-08-18T15:49:00.000Z', '2026-08-18T15:49:00.000Z'); + }; + + it('子 run 进 running 也开始计时：预算行仍写 root，但触发的是子 run 的转换', () => { + seedRun('r_root', { active_ms: 0, started_at: null }, 'blocked'); // root 一直 blocked + seedChild('r_child', 'r_root', 'running'); + const s = engine.settleWallclock('r_child', 'running'); + expect(s).not.toBeNull(); + expect(s.started_at).toBeTruthy(); + expect(JSON.parse(wdb.getRun('r_root').budget_json).spent.started_at).toBeTruthy(); + }); + + it('存量行由子 run 的转换重锚 —— 那条卡了两天的 run 就是这个形态', () => { + seedRun('r_root2', { agent_calls: 7, tokens: 69961, started_at: '2026-08-18T15:48:17.630Z' }, 'blocked'); + seedChild('r_child2', 'r_root2', 'running'); + engine.settleWallclock('r_child2', 'running'); + const spent = JSON.parse(wdb.getRun('r_root2').budget_json).spent; + expect(spent.active_ms).toBe(0); + expect(Date.parse(spent.started_at)).toBeGreaterThan(Date.parse('2026-08-19T00:00:00Z')); + expect(engine.checkBudget(wdb.getRun('r_child2')).ok).toBe(true); // 子 run 查的也是 root 的池子 + }); + + it('一层停了不结算 —— 树里还有别的 run 在跑，计时不能被提前掐断', () => { + seedRun('r_root3', { active_ms: 60_000, started_at: '2026-08-20T00:00:00.000Z' }, 'running'); + seedChild('r_child3', 'r_root3', 'blocked'); + expect(engine.settleWallclock('r_child3', 'blocked')).toBeNull(); + expect(JSON.parse(wdb.getRun('r_root3').budget_json).spent.started_at).toBe('2026-08-20T00:00:00.000Z'); + + // root 也停下来，树里再没有 running → 这才结算 + wdb.getDb().prepare(`UPDATE graph_runs SET run_status='blocked' WHERE id='r_root3'`).run(); + const s = engine.settleWallclock('r_root3', 'blocked'); + expect(s.started_at).toBeNull(); + expect(s.active_ms).toBeGreaterThan(60_000); + }); + + it('树里已经有 run 在跑时，另一条进 running 不重锚（否则前一段被抹掉）', () => { + seedRun('r_root4', { active_ms: 0, started_at: '2026-08-20T00:00:00.000Z' }, 'running'); + seedChild('r_child4', 'r_root4', 'running'); + expect(engine.settleWallclock('r_child4', 'running')).toBeNull(); + expect(JSON.parse(wdb.getRun('r_root4').budget_json).spent.started_at).toBe('2026-08-20T00:00:00.000Z'); }); it('挂在漏斗上：走 sse.transitionRun 就会结算，不需要调用方记得调', () => { diff --git a/server/tests/unit/budget-wallclock-migration.test.js b/server/tests/unit/budget-wallclock-migration.test.js new file mode 100644 index 0000000..b4789bf --- /dev/null +++ b/server/tests/unit/budget-wallclock-migration.test.js @@ -0,0 +1,88 @@ +/** + * 墙钟改成「只算活跃时长」之后的**存量库迁移路径**。 + * + * 单独钉这一条的理由和 tasks-executor-migration 一样：新建库那条路由单测天然覆盖， + * 「存量数据 + 重启」这条必须另外安排人走一遍。这里要守的是两点： + * + * ① **必须在开库时重锚**，不能只靠转换钩子。钩子只在「进 / 出 running」时触发， + * 而升级那一刻正停在 running 的 run 不会再进一次 running —— 旧锚点一直挂着， + * 下一次 checkBudget 直接判耗尽（= 那条 run 永远醒不过来）。 + * ② **迁移代码必须放在建 graph_runs 的那个模块里**。第一版写进了 graph-db.js 的 + * migrate()，而 graph_runs 是 work-graph-db 建的表、迁移块跑在它之前 —— + * 新库上直接 `no such table: graph_runs`，getDb() 起不来。 + */ +import { describe, it, expect } from 'vitest'; +import { createRequire } from 'module'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; +const require_ = createRequire(import.meta.url); + +/** 用独立 DB 文件跑一遍 getDb()（写法照抄 tasks-executor-migration.test.js） */ +function freshDb(dbPath) { + process.env.GRAPH_DB_PATH = dbPath; + for (const m of ['../../services/work-graph-db', '../../services/graph-db']) { + delete require_.cache[require_.resolve(m)]; + } + const wdb = require_('../../services/work-graph-db'); + wdb.getDb(); + return wdb; +} + +const OLD_ANCHOR = '2026-08-18T15:48:17.630Z'; +const legacyBudget = () => JSON.stringify({ + total_agent_calls: 100, total_tokens: 5000000, wallclock_ms: 7200000, + spent: { agent_calls: 7, tokens: 69961, started_at: OLD_ANCHOR }, // 没有 active_ms = 老形态 +}); + +function seed(db, id, status, budgetJson) { + db.prepare(`INSERT INTO graph_runs + (id, graph_kind, graph_id, company_id, root_run_id, execution_mode, run_status, budget_json, created_at, updated_at) + VALUES (?, 'org', 'g1', 'co1', ?, 'manual', ?, ?, ?, ?)`) + .run(id, id, status, budgetJson, OLD_ANCHOR, OLD_ANCHOR); +} + +describe('墙钟锚点的存量迁移', () => { + it('存量库重启：老锚点归零重锚，running 的续上、其余断开，且不炸', () => { + const dbPath = path.join(fs.mkdtempSync(path.join(os.tmpdir(), 'wc-mig-')), 'graph.sqlite'); + const wdb = freshDb(dbPath); + const db = wdb.getDb(); + + seed(db, 'r_running', 'running', legacyBudget()); + seed(db, 'r_blocked', 'blocked', legacyBudget()); + seed(db, 'r_nowc', 'running', JSON.stringify({ total_agent_calls: 10, spent: { agent_calls: 1 } })); + seed(db, 'r_broken', 'running', '{ 这不是 json'); + db.close(); + + // 重开进程 = 服务重启 + const wdb2 = freshDb(dbPath); + const get = (id) => wdb2.getRun(id).budget_json; + + const running = JSON.parse(get('r_running')).spent; + expect(running.active_ms).toBe(0); // 旧锚点一笔勾销 + expect(Date.parse(running.started_at)).toBeGreaterThan(Date.parse('2026-08-19T00:00:00Z')); + expect(running.agent_calls).toBe(7); // 精确的那两维一个字不动 + expect(running.tokens).toBe(69961); + + const blocked = JSON.parse(get('r_blocked')).spent; + expect(blocked.active_ms).toBe(0); + expect(blocked.started_at).toBeNull(); // 没在跑就不该有锚点 + + // 没配 wallclock_ms 的行不许被塞字段 —— 塞了会让人以为它限了墙钟 + expect(JSON.parse(get('r_nowc')).spent).toEqual({ agent_calls: 1 }); + // 坏字段跳过即可，不能把开库卡死（上面 getDb() 没抛就是证据） + expect(get('r_broken')).toBe('{ 这不是 json'); + }); + + it('幂等：再重启一次，已经重锚过的行一个字不动', () => { + const dbPath = path.join(fs.mkdtempSync(path.join(os.tmpdir(), 'wc-mig2-')), 'graph.sqlite'); + const wdb = freshDb(dbPath); + seed(wdb.getDb(), 'r_x', 'blocked', legacyBudget()); + wdb.getDb().close(); + + const first = freshDb(dbPath).getRun('r_x').budget_json; + const second = freshDb(dbPath).getRun('r_x').budget_json; + expect(second).toBe(first); + expect(JSON.parse(first).spent.active_ms).toBe(0); + }); +}); -- libgit2 0.21.4