宜博 / lemonboss

/**
 * Unit Tests — advanceBatch（27-P4 T4/T6）
 *
 * 覆盖 27-P4 §六 验收标准：
 *   2. 批量拾取幂等：候选里已经被别人抢先拾取的节点被跳过，不重复认领、不报错
 *   3. fan-out 真并行：一张有 3 个并行分支的图，同一批 advanceBatch 里三个节点同时 running
 *   4. 部分失败停等：3 个并行分支 1 个失败（无 fail 边 → blocked）时，fan-in 节点保持 pending，
 *      run 收敛到 blocked（不是 completed）
 *   5. 额度生效：`max_concurrency` 建 run 时持久化，`advanceBatch` 不传 opts 时读它
 *   7. 手动模式不变：`maxConcurrency<=1` 直接委托 `advanceOne`，一次只推进一个节点
 *
 * 「fan-out 真并行」这条不需要 setImmediate/sleep 等计时手段就能确定性断言：
 * `advanceBatch` 在 `Promise.allSettled(...)` 之前的整段（run 级检查 → checkBudget →
 * SELECT 候选 → 逐行 claim）全程同步，`.map()` 调 `runClaimedNode` 时，每个节点的
 * ctx 构建/记账/事件都在各自 hit 到 `await agentExecutor(ctx)` 之前同步跑完——
 * 所以调用 `advanceBatch(...)`（还没 await 它的返回值）这一行语句执行完，三个节点
 * 已经全部处于 running、执行器也都已经被调用过一次。这正是 27-P4 §4.3 那条并发正确性
 * 论证（Node 单线程事件循环）的直接体现，写成测试顺带验证了那条论证站得住。
 *
 * 隔离手法与 graph-engine.test.js 一致：独立 GRAPH_DB_PATH / GRAPH_RUNS_DIR，
 * agent 调用全部走注入的假 executor，不起真实 Claude CLI。
 */
import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
import fs from 'fs';
import os from 'os';
import path from 'path';

describe('advanceBatch（27-P4 T4/T6）', () => {
  let wdb;
  let engine;
  let sse;
  let runsDir;
  const TEST_DB_PATH = '/tmp/test-graph-engine-advance-batch.sqlite';

  beforeAll(() => {
    for (const f of [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`]) {
      try { fs.unlinkSync(f); } catch {}
    }
    process.env.GRAPH_DB_PATH = TEST_DB_PATH;
    runsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'graph-advance-batch-runs-'));
    process.env.GRAPH_RUNS_DIR = runsDir;
    for (const m of [
      '../../services/graph-db',
      '../../services/graph-service',
      '../../services/org-graph-service',
      '../../services/work-graph-db',
      '../../services/graph-run-sse',
      '../../services/graph-engine',
    ]) {
      delete require.cache[require.resolve(m)];
    }
    wdb = require('../../services/work-graph-db');
    sse = require('../../services/graph-run-sse');
    engine = require('../../services/graph-engine');
  });

  afterAll(() => {
    for (const f of [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`]) {
      try { fs.unlinkSync(f); } catch {}
    }
    delete process.env.GRAPH_DB_PATH;
    try { fs.rmSync(runsDir, { recursive: true, force: true }); } catch {}
    delete process.env.GRAPH_RUNS_DIR;
  });

  beforeEach(() => {
    const db = wdb.getDb();
    db.pragma('foreign_keys = OFF');
    db.exec(`
      DELETE FROM graph_run_loop_records;
      DELETE FROM graph_run_nodes;
      DELETE FROM graph_runs;
      DELETE FROM work_graph_edges;
      DELETE FROM work_graph_nodes;
      DELETE FROM work_graphs;
      DELETE FROM system_agents;
    `);
    db.pragma('foreign_keys = ON');
    engine.setAgentExecutor(null);
  });

  let seq = 0;
  const nextCompany = () => `p4b-co-${++seq}`;

  function ensureAgents(ids) {
    const db = wdb.getDb();
    const ts = new Date().toISOString();
    const ins = db.prepare(`
      INSERT OR IGNORE INTO system_agents
        (id, name, scope, owner_company_id, legacy_path, status, created_at, updated_at)
      VALUES (?, ?, 'public', NULL, ?, 'active', ?, ?)
    `);
    for (const id of new Set(ids)) ins.run(id, id, `system/agents/${id}`, ts, ts);
  }

  /** entry → a,b,c（fan-out）→ join（fan-in），全部用 condition=success 的默认边 */
  function buildDiamond({ maxConcurrency = null } = {}) {
    const companyId = nextCompany();
    const nodes = ['entry', 'a', 'b', 'c', 'join'];
    ensureAgents(nodes);
    const g = wdb.createWorkGraph({
      companyId,
      nodes,
      edges: [
        { from: 'entry', to: 'a' },
        { from: 'entry', to: 'b' },
        { from: 'entry', to: 'c' },
        { from: 'a', to: 'join' },
        { from: 'b', to: 'join' },
        { from: 'c', to: 'join' },
      ],
    });
    expect(g.errors || []).toEqual([]);
    const run = wdb.createRun({
      graphKind: 'work', graphId: g.id, companyId, executionMode: 'manual', maxConcurrency,
    });
    engine.markRunReady(run.id);
    return { runId: run.id, companyId };
  }

  const nodeByName = (runId, name) =>
    wdb.getRunState(runId).nodes.find((n) => n.name === name);
  const statusMap = (runId) =>
    Object.fromEntries(wdb.getRunState(runId).nodes.map((n) => [n.name, n.status]));

  const deferred = () => {
    let resolve;
    const promise = new Promise((r) => { resolve = r; });
    return { promise, resolve };
  };

  it('fan-out 真并行（验收 3）：同一批里 a/b/c 同时处于 running，而不是依次', async () => {
    const { runId } = buildDiamond();
    const gates = { a: deferred(), b: deferred(), c: deferred() };
    const calls = [];
    engine.setAgentExecutor(async (ctx) => {
      calls.push(ctx.nodeName);
      if (gates[ctx.nodeName]) await gates[ctx.nodeName].promise;
      return { status: 'success' };
    });

    engine.startRun(runId);
    await engine.advanceOne(runId); // entry → success，点亮 a/b/c
    calls.length = 0; // 只看这一批 advanceBatch 触发的调用，entry 那次不算

    // 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' });

    gates.a.resolve();
    gates.b.resolve();
    gates.c.resolve();
    const res = await batchPromise;

    expect(res.ok).toBe(true);
    expect(res.results).toHaveLength(3);
    expect(statusMap(runId)).toMatchObject({
      a: 'success', b: 'success', c: 'success', join: 'ready',
    });
  });

  it('批量拾取跳过已被抢走的候选（验收 2）：不重复认领、不报错', async () => {
    const { runId } = buildDiamond();
    engine.setAgentExecutor(async () => ({ status: 'success' }));
    engine.startRun(runId);
    await engine.advanceOne(runId); // entry → success

    // 模拟「b 已经被别的调用抢先拾取」：直接做条件更新，绕开 advanceBatch
    const b = nodeByName(runId, 'b');
    expect(sse.transitionNode(runId, b.id, 'ready', 'running', {}, { node_name: 'b' })).toBeTruthy();

    const res = await engine.advanceBatch(runId, { maxConcurrency: 3 });
    // 只认领了 a、c 两个，b 被跳过——既没有重复认领，也没有报错
    expect(res.ok).toBe(true);
    expect(res.results.map((r) => r.node.name).sort()).toEqual(['a', 'c']);
    // b 停在「别人」占着的 running，advanceBatch 完全没碰它（没有被结算成任何终态）
    expect(statusMap(runId).b).toBe('running');
  });

  it('部分失败停等（验收 4）：c 失败（无 fail 边 → blocked）时 join 保持 pending，run 收敛到 blocked', async () => {
    const { runId } = buildDiamond();
    engine.setAgentExecutor(async (ctx) => (
      ctx.nodeName === 'c' ? { status: 'fail' } : { status: 'success' }
    ));
    engine.startRun(runId);
    await engine.advanceOne(runId); // entry → success

    const res = await engine.advanceBatch(runId, { maxConcurrency: 3 });
    expect(res.ok).toBe(true);
    expect(statusMap(runId)).toMatchObject({
      a: 'success', b: 'success', c: 'blocked', join: 'pending',
    });
    expect(res.runStatus).toBe('blocked');
    expect(wdb.getRun(runId).run_status).toBe('blocked');
  });

  it('maxConcurrency<=1 直接委托 advanceOne（验收 7）：一次只推进一个节点', async () => {
    const { runId } = buildDiamond({ maxConcurrency: 1 });
    const calls = [];
    engine.setAgentExecutor(async (ctx) => { calls.push(ctx.nodeName); return { status: 'success' }; });
    engine.startRun(runId);

    const r1 = await engine.advanceBatch(runId);
    expect(r1.results).toHaveLength(1);
    expect(r1.results[0].node.name).toBe('entry');
    expect(calls).toEqual(['entry']);

    // 只推进了一个：a/b/c 还没被拾取（委托给 advanceOne 之后，一次只认领一个 ready 节点）
    expect(statusMap(runId)).toMatchObject({ a: 'ready', b: 'ready', c: 'ready' });
  });

  it('额度生效（验收 5）：run.max_concurrency 建 run 时持久化，不传 opts 时读它', async () => {
    const { runId } = buildDiamond({ maxConcurrency: 2 });
    expect(wdb.getRun(runId).max_concurrency).toBe(2);

    const gates = { a: deferred(), b: deferred(), c: deferred() };
    engine.setAgentExecutor(async (ctx) => {
      if (gates[ctx.nodeName]) await gates[ctx.nodeName].promise;
      return { status: 'success' };
    });
    engine.startRun(runId);
    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

    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 应该同时在跑，而不是依次
  });

  it('advanceTree 真并发推进兄弟子 run（27-P4 收口：不再是「一次只推最深的一个」）', async () => {
    // entry → sub-a, sub-b 两个并行 subgraph 节点，各自规划出一张只有一个 leaf 节点的子图。
    // 两个 subgraph 节点没有依赖关系，同一批 advanceBatch 里会一起从 ready 变 running；
    // 各自的 leaf 节点则要等到**下一次** advanceTree 才会被推——这正是本用例要验的那一跳。
    const companyId = nextCompany();
    ensureAgents(['entry', 'sub-a', 'sub-b']);
    const g = wdb.createWorkGraph({
      companyId,
      nodes: ['entry', { nodeName: 'sub-a', nodeKind: 'subgraph' }, { nodeName: 'sub-b', nodeKind: 'subgraph' }],
      edges: [{ from: 'entry', to: 'sub-a' }, { from: 'entry', to: 'sub-b' }],
    });
    expect(g.errors || []).toEqual([]);
    const run = wdb.createRun({ graphKind: 'work', graphId: g.id, companyId, executionMode: 'manual' });
    engine.markRunReady(run.id);
    const runId = run.id;

    engine.setPlanWorkGraph(async (ctx) => {
      ensureAgents(['leaf']);
      const childGraph = wdb.createWorkGraph({ companyId, nodes: ['leaf'], edges: [] });
      expect(childGraph.errors || []).toEqual([]);
      return { workGraphId: childGraph.id };
    });

    let active = 0;
    let maxActive = 0;
    engine.setAgentExecutor(async (ctx) => {
      if (ctx.nodeName !== 'leaf') return { status: 'success' }; // entry：立即成功，不参与并发计数
      active += 1;
      maxActive = Math.max(maxActive, active);
      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');
    // 两个子 run 各自的 leaf 节点要在同一个时间窗口里跑——如果 advanceTree 仍然只推
    // 「最深 active run」，sub-a/sub-b 的子 run 会被 runAuto 的外层循环一轮一轮轮流服务，
    // maxActive 永远是 1。
    expect(maxActive).toBeGreaterThanOrEqual(2);

    const subA = nodeByName(runId, 'sub-a');
    const subB = nodeByName(runId, 'sub-b');
    expect(wdb.getRunState(subA.childRunId).nodes.find((n) => n.name === 'leaf').status).toBe('success');
    expect(wdb.getRunState(subB.childRunId).nodes.find((n) => n.name === 'leaf').status).toBe('success');
  });
});