graph-engine-stop-run-orphan-race.test.js 8.05 KB
/**
 * Unit Test — stopRun 不能给「正在 executeSubgraph 里 await planWorkGraph」的节点留孤儿子 run
 * (T-mtgxcnfz 复现)
 *
 * 真实事故时序:run_e4ba218e9446 于 2026-08-31T07:30:50.644Z 落 stopped,它的子 run
 * run_12aa199b07f2 created_at 是 07:30:50.607Z(早 37ms,同一次事件循环附近),但从未被置
 * stop_requested,一直跑到人工手动 /stop 才收掉。
 *
 * 根因:`stopRun`(graph-engine.js)在函数开头一次性拍 `descendantRuns(run)` 快照决定要停
 * 哪些 run;`executeSubgraph`(同文件)建子 run 之前有一次真实 `await planWorkGraph(ctx)`——
 * 如果这次 await 跨越了 stopRun 执行的时间点,子 run 在 stopRun 拍完快照**之后**才落库,
 * 永远不在 targets 里,`stopRun` 本身又是全同步函数、没有内部 await,事后补扫赶不上。
 *
 * 修法钉在 executeSubgraph:`await planWorkGraph` 之后、真正 `createRun` 之前重新读一次
 * 父 run,`stop_requested` 已经变真就放弃创建子 run,就地按节点失败结算——不产生孤儿。
 *
 * 用真实的 org run + subgraph 节点复现(不是打桩内部实现细节):`setPlanWorkGraph` 注入一个
 * 可以从测试里手动控制何时 resolve 的假规划,制造出与生产一致的 await 窗口,在窗口内调用
 * `engine.stopRun`,再放行规划,断言子 run 没有被建出来、父节点按 fail 收敛、run 落 stopped。
 */
import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
import fs from 'fs';
import os from 'os';
import path from 'path';

describe('stopRun 与 executeSubgraph 的建子 run 竞态(T-mtgxcnfz)', () => {
  let orgSvc;
  let wdb;
  let engine;
  let runsDir;
  const TEST_DB_PATH = '/tmp/test-graph-engine-stop-run-orphan-race.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-engine-stop-run-orphan-race-'));
    process.env.GRAPH_RUNS_DIR = runsDir;
    for (const m of [
      '../../services/graph-db',
      '../../services/org-graph-service',
      '../../services/work-graph-db',
      '../../services/graph-run-sse',
      '../../services/graph-engine',
    ]) {
      delete require.cache[require.resolve(m)];
    }
    orgSvc = require('../../services/org-graph-service');
    wdb = require('../../services/work-graph-db');
    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 org_graph_edges;
      DELETE FROM org_graph_nodes;
      DELETE FROM system_agents;
    `);
    db.pragma('foreign_keys = ON');
    engine.setAgentExecutor(null);
    engine.setPlanWorkGraph(null);
  });

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

  it('planWorkGraph 还没 resolve 时被 stopRun 停掉:不建子 run,节点按 fail 收敛,run 落 stopped', async () => {
    const companyId = nextCompany();
    expect(orgSvc.writeGraph(companyId, { nodes: ['org-coder'], edges: [] }).ok).toBe(true);

    let releasePlanning;
    const planningStarted = new Promise((resolveStarted) => {
      engine.setPlanWorkGraph((ctx) => {
        resolveStarted();
        return new Promise((resolve) => {
          releasePlanning = () => {
            // 造一张真实可用的子图定义——workGraphId 不能是 null,否则会在
            // executeSubgraph 里被「未返回 workGraphId」那道**无关**的校验拦下,
            // 那不是这条测试要盯的路径,会把「没建成子 run」的原因张冠李戴。
            const worker = `${ctx.nodeName}-worker`;
            const db = wdb.getDb();
            const now = new Date().toISOString();
            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', ?, ?)
            `).run(worker, worker, `system/agents/${worker}`, now, now);
            const g = wdb.createWorkGraph({ companyId, nodes: [worker], edges: [] });
            resolve({ workGraphId: g.id });
          };
        });
      });
    });

    const run = wdb.createRun({ graphKind: 'org', graphId: companyId, companyId, executionMode: 'manual' });
    engine.markRunReady(run.id);
    engine.startRun(run.id, { mode: 'manual' });

    // 不 await:advanceOne 内部会在 executeSubgraph 里卡在 planWorkGraph 那次 await 上,
    // 这条 promise 要等我们手动 releasePlanning 之后才会继续往下走。
    const stepPromise = engine.nextStep(run.id);
    await planningStarted; // 确认已经真的进了 await 窗口,不是还没排到

    // 窗口期间人工调 /stop——此时子 run 还没建出来,descendantRuns(run) 快照里只有 run 自己。
    const stopResult = engine.stopRun(run.id);
    expect(stopResult.ok).toBe(true);
    expect(wdb.getRun(run.id).run_status).toBe('stopped');
    // 这一刻整棵树只有 root 自己——旧代码的 bug 正是「这个断言此刻成立,但之后会被打破」
    expect(wdb.listRunTree(run.id)).toHaveLength(1);

    // 放行规划,让 executeSubgraph 走到「要不要建子 run」那一步
    releasePlanning();
    const stepResult = await stepPromise;
    expect(stepResult.ok).toBe(true);

    // 核心断言:没有孤儿子 run——树上除了 root 一个都没多出来
    const tree = wdb.listRunTree(run.id);
    expect(tree).toHaveLength(1);

    // 节点没有 pending 的 child_run_id,且不再停在 running 干等一个不存在的子 run——
    // 没有匹配的 fail 边时 routeFailure 落 blocked(需要人工介入),这里不咬死具体是
    // fail 还是 blocked,只钉住「不是孤儿」这条:没有子 run、没有继续假装在跑。
    const node = wdb.getRunState(run.id).nodes.find((n) => n.name === 'org-coder');
    expect(node.childRunId).toBeNull();
    expect(node.status).not.toBe('running');
    expect(['fail', 'blocked']).toContain(node.status);
  });

  it('对照组:没有并发 stop 时,同样的 await 窗口结束后子 run 正常建出来', async () => {
    const companyId = nextCompany();
    expect(orgSvc.writeGraph(companyId, { nodes: ['org-coder'], edges: [] }).ok).toBe(true);

    engine.setPlanWorkGraph(async (ctx) => {
      const worker = `${ctx.nodeName}-worker`;
      const db = wdb.getDb();
      const now = new Date().toISOString();
      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', ?, ?)
      `).run(worker, worker, `system/agents/${worker}`, now, now);
      const g = wdb.createWorkGraph({ companyId, nodes: [worker], edges: [] });
      expect(g.errors || []).toEqual([]);
      return { workGraphId: g.id };
    });

    const run = wdb.createRun({ graphKind: 'org', graphId: companyId, companyId, executionMode: 'manual' });
    engine.markRunReady(run.id);
    engine.startRun(run.id, { mode: 'manual' });

    const stepResult = await engine.nextStep(run.id);
    expect(stepResult.ok).toBe(true);
    expect(wdb.listRunTree(run.id)).toHaveLength(2);
    const node = wdb.getRunState(run.id).nodes.find((n) => n.name === 'org-coder');
    expect(node.childRunId).toBeTruthy();
    expect(node.status).toBe('running'); // 等子 run 收敛,父节点停在 running
  });
});