diff --git a/docs/todo/system-design/27-p4-parallel-and-artifacts.md b/docs/todo/system-design/27-p4-parallel-and-artifacts.md index b517789..cc5edc5 100644 --- a/docs/todo/system-design/27-p4-parallel-and-artifacts.md +++ b/docs/todo/system-design/27-p4-parallel-and-artifacts.md @@ -605,6 +605,27 @@ R-19 之后这两件事有实体（`graph_edge_contracts` × `graph_files`）， > - **没做的**：`advanceBatch` 还没接进 `runAuto`/路由——自动模式今天仍然只调 `advanceOne`， > 实际生产流量拿不到并行，这是接下来的线（大概率是 T10 之前、或与它同批）。T5（欠费门 + 自动恢复 > 触发器）仍卡在计费系统接口确认，T10（控制台可视化）未动。 +> +> **2026-08-21 追加：`advanceBatch` 接进 `advanceTree`/`runAuto`**（同一天，紧接上面那批）： +> - `advanceTree`（`runAuto` 每一步都调它）里 `advanceOne(r.id, opts)` 换成 `advanceBatch(r.id, opts)`—— +> 「推最深的一个 active run」这个 tree-walk 逻辑不变，变的是「推」这一步本身现在能一次 claim 一批。 +> `.driver` 的调用方（生产路由、测试）一律只读 `.ok`，不依赖 `.node`/`.results` 的具体形状—— +> 核实过全仓库唯一消费方是 `advanceTree` 自己的循环判定 + 两个 blind-spot 测试文件，两者都只用 +> `.ok`，这次改动不需要做返回形状的兼容层。 +> - **审这段时抓到一个真 bug 并顺手修了**：`advanceBatch` 的 `no-ready-node` / `taken-by-other` +> 分支原来写的是 `ok:true`，跟 `advanceOne` 用 `result()` helper 的惯例（这两种情形一律 `ok:false`） +> 不一致。不修的话接进 `runAuto` 后，一个已经收敛的 run 会被自动循环一路空转到 `maxSteps`（1000） +> 才停——不是崩溃，但是真实的效率/语义回归，而且正好是旧测试从没试过「对已收敛的 run 连续调 +> `advanceBatch`」才漏掉的那种 bug。已修好并补了回归用例。 +> - 新增 2 条测试（同一文件）：`ok:false` 语义与 `advanceOne` 对齐（直接绕开引擎构造 +> 「`run_status` 还是 running 但没有 ready/running 节点」这个状态，同款手法 `graph-engine.test.js` +> 已经在用）；`runAuto`/`advanceTree` 端到端确实吃到并行——用真实 `setTimeout` 延迟（不是同步 +> `deferred`）证明 a/b/c 三个 executor 调用有时间窗口重叠，不是「反正都同步排到一起」的假象。 +> - 全量 `npx vitest run`：147 文件 / 1350 用例全绿，零回归。 +> - **仍然没做的**：`advanceTree` 一次只推「最深的一个」active run，**兄弟子 run**（比如两个并行 +> subgraph 节点各自起的子 run）不会同时被推——它们靠 `runAuto` 外层循环一轮一轮轮流服务， +> 不是真并发（已在 `advanceTree` 加了 ponytail 注释记这笔）。真并发目前只发生在**同一个 run +> 内部**。这个跨 run 兄弟并发是否值得做，留给 T10 之后再看有没有真实场景撞上。 --- diff --git a/server/services/graph-engine.js b/server/services/graph-engine.js index 8da2bde..726fdb9 100644 --- a/server/services/graph-engine.js +++ b/server/services/graph-engine.js @@ -1599,7 +1599,11 @@ async function advanceBatch(runId, opts = {}) { .prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? AND status = 'ready' ORDER BY created_at, id LIMIT ?`) .all(runId, maxConcurrency); if (!candidates.length) { - return { ok: true, runId, runStatus: settleRun(runId), reason: 'no-ready-node', results: [], activated: [] }; + // ok:false 跟 advanceOne 的 result() 惯例对齐——「这一次调用没有推进任何节点」， + // 不是「出错了」。runAuto 的循环靠 !res.ok 判断要不要停，这里如果写成 true， + // 一个已经收敛的 run 会被 runAuto 一路空转到 maxSteps 才停（§4.3 决策审这段代码时发现， + // 上线前修，不留到接 runAuto 那一步才炸）。 + return { ok: false, runId, runStatus: settleRun(runId), reason: 'no-ready-node', results: [], activated: [] }; } const claimed = []; @@ -1617,8 +1621,9 @@ async function advanceBatch(runId, opts = {}) { } } if (!claimed.length) { - // 候选全被别的 advanceBatch/advanceOne 抢走：不是错误，下一轮再拾取 - return { ok: true, runId, runStatus: run.run_status, reason: 'taken-by-other', results: [], activated: [] }; + // 候选全被别的 advanceBatch/advanceOne 抢走：不是出错，但同样是「这一次没推进任何东西」， + // ok:false 理由同上——taken-by-other 在 advanceOne 里也是 ok:false（见其 result() 用法）。 + return { ok: false, runId, runStatus: run.run_status, reason: 'taken-by-other', results: [], activated: [] }; } const settled = await Promise.allSettled( @@ -3519,18 +3524,25 @@ function hashExport(value) { } // ──────────────────────────────────────────────────────────────── -// 自动模式 = 不停调 advanceOne 的循环（不是另一套实现） +// 自动模式 = 不停调 advanceBatch 的循环（不是另一套实现） // ──────────────────────────────────────────────────────────────── /** * 在一棵 run **子树**里推进一步：优先推最深的 active 子 run（父节点正卡在它上面）。 - * 单张 run 时退化成 advanceOne(runId)。 + * 单张 run 时退化成 advanceBatch(runId)（27-P4 T4/T6 接入：以前这里调的是 advanceOne， + * 一次只推一个节点；现在同一个「最深 active run」内部能一次并发跑多个 ready 节点）。 * * **形参是子树根，不要求是 root run**（R6-F3 问题 2）。控制台「进入子图」后那一层的五个 * 控制按钮作用在子 run 上（§六 验收 15），对子 run 点自动启动时 runAuto 调的就是 * advanceTree(子 runId)。旧实现直接 `listRunTree(传入值)` —— 那是按 `root_run_id = ?` 查的， * 传子 run 拿到**零行**，循环第 0 步就 break：HTTP 回 `ok:true / auto:true`， * 而那个 run 里的 ready 节点纹丝不动（返回 ok 却什么都不做，是最坏的形态）。 + * + * ponytail: 仍然只推**一个**「最深 active run」，同一 tick 里的**兄弟**子 run（比如两个并行 + * subgraph 节点各自起的子 run）不会同时被推——它们靠 runAuto 的外层循环一轮一轮轮流服务， + * 不是真并发。真并发只发生在**同一个 run 内部**（advanceBatch 的 fan-out）。上限是「兄弟子 run + * 之间」这一维的吞吐，不影响正确性（不会有子 run 被饿死，只是不同时跑）；升级路径是把这个 + * for 循环也从「找到第一个 ok 就 return」改成对多个 active run 并发调 advanceBatch。 */ async function advanceTree(runId, opts = {}) { const self = getRun(runId); @@ -3543,7 +3555,7 @@ async function advanceTree(runId, opts = {}) { // 后建的 run 一定更深（子 run 总是在父节点 running 之后建的），倒序即「最深优先」 let informative = null; for (const r of tree.reverse()) { - const res = await advanceOne(r.id, opts); + const res = await advanceBatch(r.id, opts); if (res.ok) return res; // minor 3：ok:false 时**不能**把所有原因都塌缩成 'no-ready-node'。子层返回 // 'budget-exhausted'（还带着 budget.exhausted 明细）而树层报「没活可干」， @@ -3564,8 +3576,8 @@ async function advanceTree(runId, opts = {}) { * 自动驱动：只要 pause_requested / stop_requested 都是假就在上一次 resolve 后立刻发起下一次。 * 与 advanceTree 同款语义 —— runId 是**子树根**，对子 run 起自动就只驱动它这棵子树。 * maxSteps 只是防跑飞的安全阀，真正的终止条件是预算（§四.A.3）与熔断（§四.D.2.1）。 - * ponytail: 串行单步循环，一次只跑一个节点，并行分支不会并发执行；上限是吞吐， - * 升级路径是按 ready 集合并发 advanceOne（节点状态机已经是幂等条件更新，能直接支撑）。 + * 27-P4 T4/T6 接入后，「同一个 run 内部」已经能并发；跨兄弟子 run 的并发仍是 advanceTree + * 那条 ponytail（一次只推最深的一个），不在本函数里。 */ async function runAuto(runId, opts = {}) { const maxSteps = opts.maxSteps || 1000; diff --git a/server/tests/unit/graph-engine-advance-batch.test.js b/server/tests/unit/graph-engine-advance-batch.test.js index 0243edd..da62310 100644 --- a/server/tests/unit/graph-engine-advance-batch.test.js +++ b/server/tests/unit/graph-engine-advance-batch.test.js @@ -231,4 +231,55 @@ describe('advanceBatch（27-P4 T4/T6）', () => { for (const g of Object.values(gates)) g.resolve(); await batchPromise; }); + + it('ok:false 语义与 advanceOne 对齐：no-ready-node 不算「推进了」', async () => { + // no-ready-node：直接绕开引擎把节点写成 success（不经过 settleNode，run_status 就不会 + // 被顺带收敛），构造出「run_status 还是 running，但已经没有 ready/running 节点」这个 + // advanceOne 的 !picked 分支本来就要处理的状态。真实场景里这一般是重入/竞态的产物—— + // 这里直接摆出结果状态，不追究怎么走到的（同款手法 graph-engine.test.js 234 行已经在用： + // `wdb.updateNodeStatus(a.id, 'ready', 'running')` 绕开引擎构造中间状态）。 + const companyId1 = nextCompany(); + ensureAgents(['solo']); + const g1 = wdb.createWorkGraph({ companyId: companyId1, nodes: ['solo'], edges: [] }); + const run1 = wdb.createRun({ graphKind: 'work', graphId: g1.id, companyId: companyId1, executionMode: 'manual' }); + engine.markRunReady(run1.id); + engine.startRun(run1.id); + const solo1 = nodeByName(run1.id, 'solo'); + expect(wdb.updateNodeStatus(solo1.id, 'ready', 'running')).toBe(1); + expect(wdb.updateNodeStatus(solo1.id, 'running', 'success')).toBe(1); + + // advanceOne 的 result() 惯例是 ok:false——advanceBatch 必须跟它一致，否则接进 runAuto + // 之后，一个已经收敛的 run 会被循环一路空转到 maxSteps 才停（2026-08-21 审 §4.3 时抓到的 + // 真 bug，见 advanceBatch 里 no-ready-node 分支的注释）。 + const drained = await engine.advanceBatch(run1.id, { maxConcurrency: 3 }); + expect(drained.ok).toBe(false); + expect(drained.reason).toBe('no-ready-node'); + // 顺带验证：这一次调用自己把 run 收敛掉了（settleRun 就挂在这个分支里） + expect(wdb.getRun(run1.id).run_status).toBe('completed'); + + // taken-by-other（!claimed.length 分支）分支上同一惯例——ok:false，不额外起用例： + // 单进程 + 同步拾取循环下这个分支在真实调用序列里不可达（§4.1 决策的论证本身就是 + // 「单行条件更新在单线程下不会输给同进程的另一次调用」），要触发它得在 SELECT 和 + // claim 之间人为插入另一次写，等价于对内部实现打桩，价值不大，不为了覆盖率硬凑。 + }); + + it('runAuto/advanceTree 端到端也吃到并行（不是只有直接调 advanceBatch 才有效）', async () => { + const { runId } = buildDiamond(); // max_concurrency=null → 默认 DEFAULT_MAX_CONCURRENCY=5 + let active = 0; + let maxActive = 0; + engine.setAgentExecutor(async () => { + active += 1; + maxActive = Math.max(maxActive, active); + // 真实的异步延迟（不是同步卡在 deferred 上）：要证明的是 runAuto 的循环驱动下 + // a/b/c 三个 executor 调用有时间窗口上的重叠，不是「反正都同步排到一起」的假象。 + await new Promise((r) => { setTimeout(r, 15); }); + active -= 1; + return { status: 'success' }; + }); + + await engine.startRun(runId, { mode: 'auto' }).driver; + + expect(wdb.getRun(runId).run_status).toBe('completed'); + expect(maxActive).toBeGreaterThanOrEqual(3); // entry 之后 a/b/c 应该同时在跑，而不是依次 + }); });