宜博 / lemonboss

/**
 * Unit Tests — 批 4 · 「一家公司同时只有一个 active root run」解除之后的执行位（27-P3C 批 4 / 27-P4 §一.3）
 *
 * 与既有三份的分工（都不重复它们已经钉住的形状）：
 *   · `task-scheduler.test.js`        —— 单个原语的形状（准入判据、CAS 的 WHERE、per-company 额度的回落）
 *   · `task-queue-acceptance.test.js` —— 端到端「队伍自己往前走」
 *   · `task-queue-routes.test.js`     —— HTTP 端点的字段形状
 * 这一份钉的是批 4 拆掉唯一索引之后**只剩一道守卫**的那几条性质，每一条都在「不成立时会静默」的位置上：
 *   ① max=1（缺省）时行为与批 3 逐字一致 —— 这是回归防线：绝大多数公司没写 max_parallel_tasks，
 *      回落算错的话不是报错，而是「并行度悄悄变了」；
 *   ② max=3 时三条真的同时在跑、第四条排队、收敛一条自动补位（补位那一跳靠漏斗钩子，漏了就是永远排队）；
 *   ③ **超发拦得住**。索引没了，`updateRunStatus` 的 BEGIN IMMEDIATE 是唯一的守卫，所以这里不满足于
 *      串行连打，还越过**进程边界**打一次：8 个独立进程越过同一个时间栅栏同时抢 3 个位。
 *      同进程内 better-sqlite3 是同步单写者，怎么打都串行 —— 那测不到这把锁存在的理由；
 *   ④ 坑满时人点「启动」拿到的是 409（不是 500）、点得出**全部**占位者，位子一空同一个请求就成功；
 *   ⑤ 配置非法（0 / 负数 / 字符串 / 缺失）一律回落 1 **并留声**，绝不回落成 0（永远不出队 = 静默全停）
 *      或无限（超发）；
 *   ⑥ DROP 迁移：带旧索引的存量库启动一次索引就没了，重复启动、多进程同时启动都不打架。
 *
 * 隔离手法同 task-scheduler.test.js：**先设 GRAPH_DB_PATH / GRAPH_RUNS_DIR 再 require**
 * （graph-run-sse 在模块加载时快照 GRAPH_RUNS_DIR，反过来会往仓库 data/companies/ 写而且不报错）。
 * DB 路径与其它测试文件故意不同：vitest 默认按文件并行，共用一个 sqlite 文件会互相 DELETE 对方的行。
 */
import { describe, it, expect, beforeAll, afterAll, beforeEach, vi } from 'vitest';
import fs from 'fs';
import os from 'os';
import path from 'path';
import { spawn } from 'child_process';
import express from 'express';

const TEST_DB_PATH = '/tmp/test-task-slots-parallel.sqlite';
const DB_FILES = [TEST_DB_PATH, `${TEST_DB_PATH}-shm`, `${TEST_DB_PATH}-wal`];
const MODULES = [
  '../../services/graph-db',
  '../../services/org-graph-service',
  '../../services/work-graph-db',
  '../../services/company-paths',
  '../../services/graph-run-sse',
  '../../services/graph-engine',
  '../../services/task-settle',
  '../../services/task-scheduler',
  '../../routes/graph-run',
];

/** 存量库里那条索引的**原样** SQL（从批 4 的 diff 里逐字抄来，用于伪造「升级前的库」） */
const LEGACY_INDEX_SQL = `
  CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_runs_one_active_root
    ON graph_runs(company_id)
    WHERE parent_run_id IS NULL
      AND run_status IN ('ready','running','paused')
`;
const INDEX_EXISTS_SQL = `
  SELECT name FROM sqlite_master WHERE type = 'index' AND name = 'idx_graph_runs_one_active_root'
`;

/**
 * 子进程脚本：本文件唯一能真正测到「跨进程」的手段。
 * 只依赖 work-graph-db 的公开 API（路径由 WGDB_MODULE 传进来），不碰任何私有函数。
 */
const WORKER_SRC = `
const wdb = require(process.env.WGDB_MODULE);
const [mode, ...args] = process.argv.slice(2);
const out = (o) => console.log(JSON.stringify(o));
try {
  if (mode === 'boot') {
    // 「一个进程启动了一次」= 跑一遍 getDb()（建表 + 迁移块，DROP INDEX 就在里面）
    const db = wdb.getDb();
    out({ index: !!db.prepare(${JSON.stringify(INDEX_EXISTS_SQL)}).get() });
  } else if (mode === 'seed') {
    const [cid, n] = args;
    const db = wdb.getDb();
    const now = new Date().toISOString();
    db.prepare(
      "INSERT OR IGNORE INTO system_agents (id,name,scope,owner_company_id,source_kind,legacy_path,status,created_at,updated_at)"
      + " VALUES ('solo','solo','public',NULL,'global_agent','system/agents/solo','active',?,?)"
    ).run(now, now);
    const g = wdb.createWorkGraph({ companyId: cid, nodes: ['solo'], edges: [] });
    if ((g.errors || []).length) throw new Error(JSON.stringify(g.errors));
    const runs = [];
    for (let i = 0; i < Number(n); i += 1) {
      runs.push(wdb.createRun({ graphKind: 'work', graphId: g.id, companyId: cid, runStatus: 'queued' }).id);
    }
    out({ runs });
  } else if (mode === 'race') {
    const [runId, startAt] = args;
    // 忙等到同一个时间栅栏：sleep 的精度不够，而这里要的正是「同时」
    while (Date.now() < Number(startAt)) { /* spin */ }
    out({ changes: wdb.updateRunStatus(runId, 'queued', 'ready') });
  } else if (mode === 'count') {
    out({ active: wdb.countActiveRootRuns(args[0]), max: wdb.maxActiveRootRuns(args[0]) });
  }
} catch (err) {
  out({ err: err.name, code: err.code, msg: err.message });
}
`;

describe('批 4 · 公司级执行位（max_parallel_tasks）', () => {
  let wdb;
  let engine;
  let sse;
  let scheduler;
  let runsDir;
  let workerPath;
  let server;
  let origin;
  /** 挂起的 executor promise，afterAll 统一放掉，免得留下永不 settle 的 driver */
  const pending = [];

  beforeAll(async () => {
    for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch { /* 本来就没有 */ } }
    process.env.GRAPH_DB_PATH = TEST_DB_PATH;
    runsDir = fs.mkdtempSync(path.join(os.tmpdir(), 'task-slots-parallel-'));
    process.env.GRAPH_RUNS_DIR = runsDir;
    for (const m of MODULES) delete require.cache[require.resolve(m)];

    wdb = require('../../services/work-graph-db');
    sse = require('../../services/graph-run-sse');
    engine = require('../../services/graph-engine');
    scheduler = require('../../services/task-scheduler');

    workerPath = path.join(runsDir, 'slot-worker.cjs');
    fs.writeFileSync(workerPath, WORKER_SRC);

    const app = express();
    app.use(express.json());
    app.use('/api/graph-runs', require('../../routes/graph-run'));
    await new Promise((resolve) => { server = app.listen(0, resolve); });
    origin = `http://127.0.0.1:${server.address().port}`;
  });

  /**
   * `advanceOne` 的 ensureProjectRoot 会在**仓库的** data/companies/ 下给每家公司建项目根
   * （它不认 GRAPH_RUNS_DIR）。只删本文件造的那些前缀目录，且必须确认里面没有 company.json
   * —— 那是真公司，删了就是删用户数据。（写法与 task-scheduler.test.js 一致）
   */
  function cleanupLeakedCompanyDirs() {
    const root = path.resolve(__dirname, '../../../data/companies');
    if (!fs.existsSync(root)) return;
    for (const cid of fs.readdirSync(root)) {
      if (!cid.startsWith('slots-co-')) continue;
      const dir = path.join(root, cid);
      if (fs.existsSync(path.join(dir, 'company.json'))) continue;
      try { fs.rmSync(dir, { recursive: true, force: true }); } catch { /* 删不掉就留着 */ }
    }
  }

  afterAll(async () => {
    // 先放掉挂起的 executor 再等一拍：auto driver 恢复后还会跑一轮 advanceOne，
    // 而 ensureProjectRoot 会把刚删掉的公司目录原样建回来 —— 顺序反了就等于没清理
    for (const p of pending) p();
    await new Promise((r) => setTimeout(r, 120));
    scheduler.stopSweeper();
    if (server) server.close();
    cleanupLeakedCompanyDirs();
    for (const f of DB_FILES) { try { fs.unlinkSync(f); } catch { /* 已经删了 */ } }
    delete process.env.GRAPH_DB_PATH;
    try { fs.rmSync(runsDir, { recursive: true, force: true }); } catch { /* ignore */ }
    delete process.env.GRAPH_RUNS_DIR;
    for (const m of MODULES) delete require.cache[require.resolve(m)];
  });

  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 tasks;
      DELETE FROM goals;
      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;
  /** 每个用例一家独立公司：执行位是按 company 数的 */
  const nextCompany = () => `slots-co-${++seq}`;

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

  function buildGraph(companyId, nodes = ['solo']) {
    ensureAgents(nodes);
    const edges = nodes.slice(1).map((to, i) => ({ from: nodes[i], to }));
    const g = wdb.createWorkGraph({ companyId, nodes, edges });
    expect(g.errors || []).toEqual([]);
    return g.id;
  }

  const newGoal = (companyId) => wdb.createGoal({ companyId, title: 'g', description: 'g' }).id;

  /** 入队：建 task 行 → 建 queued 的 run 行（不占位）→ task 标 queued + 挂 current_run_id */
  function enqueue(companyId, goalId, { title, mode = 'auto' } = {}) {
    const task = wdb.createTask({
      goalId, companyId, title, description: `${title} 的描述`, executionMode: mode,
    });
    const run = wdb.createRun({
      graphKind: 'work',
      graphId: buildGraph(companyId),
      companyId,
      goalId,
      taskId: task.id,
      executionMode: mode,
      runStatus: 'queued',
    });
    wdb.updateTask(task.id, { status: 'queued', currentRunId: run.id, queuedAt: new Date().toISOString() });
    return { taskId: task.id, runId: run.id };
  }

  /** 永不 resolve 的 executor：让起跑的 run 稳稳停在 running，好观察执行位 */
  function hangingExecutor(onCall) {
    engine.setAgentExecutor((ctx) => {
      if (onCall) onCall(ctx);
      return new Promise((resolve) => pending.push(() => resolve({ status: 'success' })));
    });
  }

  const flush = () => new Promise((r) => setImmediate(() => setImmediate(r)));

  /** 给这家公司写一份 company.json（走 company-paths，认 GRAPH_RUNS_DIR，绝不碰真 data/companies/） */
  function setMaxParallel(companyId, n) {
    const { companyRootDir } = require('../../services/company-paths');
    const dir = companyRootDir(companyId);
    fs.mkdirSync(dir, { recursive: true });
    const cfg = { id: companyId, name: companyId };
    if (n !== undefined) cfg.max_parallel_tasks = n;
    fs.writeFileSync(path.join(dir, 'company.json'), JSON.stringify(cfg));
    return companyId;
  }

  const statuses = (ids) => ids.map((id) => wdb.getTask(id).status);

  /** 起一个子进程跑 worker，拿回 { code, json, stdout, stderr } */
  function runWorker(args, { dbPath = TEST_DB_PATH } = {}) {
    return new Promise((resolve) => {
      const child = spawn(process.execPath, [workerPath, ...args], {
        env: {
          ...process.env,
          GRAPH_DB_PATH: dbPath,
          GRAPH_RUNS_DIR: runsDir,
          WGDB_MODULE: require.resolve('../../services/work-graph-db'),
        },
        stdio: ['ignore', 'pipe', 'pipe'],
      });
      let stdout = '';
      let stderr = '';
      child.stdout.on('data', (d) => { stdout += d; });
      child.stderr.on('data', (d) => { stderr += d; });
      child.on('close', (code) => {
        let json = null;
        try { json = JSON.parse(stdout.trim().split('\n').pop()); } catch { /* 留给断言去报 */ }
        resolve({ code, json, stdout, stderr });
      });
    });
  }

  // ────────────────────────────────────────────────────────────────
  // ① max=1：默认配置下的行为必须与批 3 逐字一致
  // ────────────────────────────────────────────────────────────────
  describe('① max=1（缺省）：批 4 之后一个字都不该变', () => {
    it('没配 company.json 的公司：一次只放一条、其余照排、收敛后自动接力，全程 active ≤ 1', async () => {
      const companyId = nextCompany();
      const goalId = newGoal(companyId);
      hangingExecutor();
      // 「读不到配置 = 1」是保护全仓存量公司与存量单测的那条断言
      expect(wdb.maxActiveRootRuns(companyId)).toBe(1);

      const ids = ['a', 'b', 'c'].map((title) => enqueue(companyId, goalId, { title }));
      const first = scheduler.tick(companyId);
      expect(first.max).toBe(1);
      expect(first.started.map((s) => s.taskId)).toEqual([ids[0].taskId]);
      await flush();
      expect(statuses(ids.map((t) => t.taskId))).toEqual(['running', 'queued', 'queued']);
      expect(wdb.countActiveRootRuns(companyId)).toBe(1);

      // 再 tick 一次绝不能多起一条（批 3 这里就是 no-slot）
      expect(scheduler.tick(companyId).skipped).toBe('no-slot');
      expect(wdb.countActiveRootRuns(companyId)).toBe(1);
      // 位次口径没变：跑着的那条不在队里，b 是第 1
      expect(scheduler.queuePosition(companyId, ids[1].taskId)).toBe(1);
      expect(scheduler.queuePosition(companyId, ids[2].taskId)).toBe(2);

      // 收敛一条 → 漏斗钩子自己把下一条拉起来，**没有人再 tick 一次**
      sse.transitionRun(ids[0].runId, 'running', 'completed', {}, { summary: '用例直推终态' });
      await flush();
      expect(statuses(ids.map((t) => t.taskId))).toEqual(['completed', 'running', 'queued']);
      expect(wdb.countActiveRootRuns(companyId)).toBe(1);

      sse.transitionRun(ids[1].runId, 'running', 'completed', {}, { summary: '用例直推终态' });
      await flush();
      expect(statuses(ids.map((t) => t.taskId))).toEqual(['completed', 'completed', 'running']);
      expect(wdb.countActiveRootRuns(companyId)).toBe(1);
    });
  });

  // ────────────────────────────────────────────────────────────────
  // ② max=3：三条同时在跑
  // ────────────────────────────────────────────────────────────────
  describe('② max=3：三条无依赖的任务同时进入 running', () => {
    it('前三条同时跑、第四条排队；收敛一条之后第四条自动补位（补位靠漏斗钩子，不靠人再点一次）', async () => {
      const companyId = setMaxParallel(nextCompany(), 3);
      const goalId = newGoal(companyId);
      hangingExecutor();
      const ids = ['a', 'b', 'c', 'd'].map((title) => enqueue(companyId, goalId, { title }));

      const res = scheduler.tick(companyId);
      expect(res.max).toBe(3);
      expect(res.started.map((s) => s.taskId)).toEqual(ids.slice(0, 3).map((t) => t.taskId));
      await flush();

      // 「同时在跑」要三条都验：任务行是 running，run 行在活跃集里，计数是 3
      expect(statuses(ids.map((t) => t.taskId))).toEqual(['running', 'running', 'running', 'queued']);
      for (const t of ids.slice(0, 3)) {
        expect(wdb.ACTIVE_ROOT_RUN_STATUS).toContain(wdb.getRun(t.runId).run_status);
      }
      expect(wdb.countActiveRootRuns(companyId)).toBe(3);
      expect(wdb.getRun(ids[3].runId).run_status).toBe('queued'); // 第四条一个位都没占
      expect(scheduler.queuePosition(companyId, ids[3].taskId)).toBe(1);
      expect(scheduler.tick(companyId).skipped).toBe('no-slot'); // 满了就是满了

      // 收敛一条 → 第四条自己顶上来，且顶上来之后仍然是 3（不是 4）
      sse.transitionRun(ids[0].runId, 'running', 'completed', {}, { summary: '用例直推终态' });
      await flush();
      expect(statuses(ids.map((t) => t.taskId))).toEqual(['completed', 'running', 'running', 'running']);
      expect(wdb.countActiveRootRuns(companyId)).toBe(3);
    });
  });

  // ────────────────────────────────────────────────────────────────
  // ③ 超发拦得住 —— 索引已 DROP，这一节是全批最重要的一条
  // ────────────────────────────────────────────────────────────────
  describe('③ 超发：唯一索引没了之后，活跃数永不超过 max', () => {
    it('8 条抢 3 个位：恰好 3 条进得去，其余全是干净的拒绝（一个字都没写进去）', () => {
      const companyId = setMaxParallel(nextCompany(), 3);
      const goalId = newGoal(companyId);
      const runs = Array.from({ length: 8 }, (_, i) => enqueue(companyId, goalId, { title: `t${i}` }).runId);

      const results = runs.map((runId) => {
        try {
          return { ok: wdb.updateRunStatus(runId, 'queued', 'ready') };
        } catch (err) {
          return { err: err.name, active: err.active, max: err.max };
        }
      });
      expect(results.filter((r) => r.ok === 1)).toHaveLength(3);
      // 被拒的每一条都必须是 RunSlotBusyError（别的错 = 别的毛病，不许混过去），
      // 而且拒绝当下报的 active 恒等于 max —— 报成别的数就说明计数和判据不同源
      for (const r of results.filter((x) => !x.ok)) {
        expect(r).toEqual({ err: 'RunSlotBusyError', active: 3, max: 3 });
      }
      expect(wdb.countActiveRootRuns(companyId)).toBe(3);
      // 拒绝是干净的：没进去的那 5 条一条都没被改写
      expect(runs.filter((id) => wdb.getRun(id).run_status === 'queued')).toHaveLength(5);
    });

    it('出队过程中反手再 tick 一次（漏斗钩子的真实形状）：活跃数一刻都没超过 max', async () => {
      const companyId = setMaxParallel(nextCompany(), 3);
      const goalId = newGoal(companyId);
      const observed = [];
      const nested = [];
      // executor 在起跑的栈里被调用 —— 这正是「tick 起 run → run 转状态 → 钩子又 tick」那条回路
      hangingExecutor(() => {
        observed.push(wdb.countActiveRootRuns(companyId));
        nested.push(scheduler.tick(companyId));
      });
      const ids = Array.from({ length: 8 }, (_, i) => enqueue(companyId, goalId, { title: `t${i}` }));

      scheduler.tick(companyId);
      await flush();
      await flush();

      expect(observed.length).toBeGreaterThan(0);
      expect(Math.max(...observed)).toBeLessThanOrEqual(3); // 中途一刻都没超发
      expect(wdb.countActiveRootRuns(companyId)).toBe(3);
      expect(statuses(ids.map((t) => t.taskId)).filter((s) => s === 'running')).toHaveLength(3);
      // 嵌套那次要么被重入闸门挡回（同一次 tick 的栈里），要么老老实实报坑满 —— 都不许多起一条
      for (const r of nested) {
        expect(['reentrant', 'no-slot', undefined]).toContain(r.skipped);
        if (!r.skipped) expect(r.started).toEqual([]);
      }
    });

    it('多进程同时抢：8 个独立进程越过同一个时间栅栏，活跃数仍然只有 3', async () => {
      // 单独一个库文件：跨进程这一节要测的是 BEGIN IMMEDIATE 那把写锁，
      // 与本文件其它用例共用一个库只会让计数互相污染。
      const dbPath = path.join(runsDir, 'mp-race.sqlite');
      const companyId = setMaxParallel(`slots-mp-${++seq}`, 3);

      const seeded = await runWorker(['seed', companyId, '8'], { dbPath });
      expect(seeded.code, seeded.stderr).toBe(0);
      const runIds = seeded.json.runs;
      expect(runIds).toHaveLength(8);

      // 同一个时间栅栏：子进程各自忙等到这一刻再打（起进程本身要几十毫秒，不留余量就变成串行）
      const startAt = Date.now() + 1500;
      const results = await Promise.all(runIds.map((runId) => runWorker(['race', runId, String(startAt)], { dbPath })));
      for (const r of results) expect(r.code, r.stderr).toBe(0);

      const admitted = results.filter((r) => r.json && r.json.changes === 1);
      const refused = results.filter((r) => r.json && r.json.err);
      expect(admitted).toHaveLength(3);
      // 拒绝的一律是 RunSlotBusyError —— 出现 SQLITE_BUSY / UNIQUE 就说明守卫不是靠事务成立的
      expect(refused.map((r) => r.json.err)).toEqual(['RunSlotBusyError', 'RunSlotBusyError',
        'RunSlotBusyError', 'RunSlotBusyError', 'RunSlotBusyError']);
      const counted = await runWorker(['count', companyId], { dbPath });
      expect(counted.json).toEqual({ active: 3, max: 3 });
    }, 60000);

    it('旧版本进程把唯一索引建回来时必须吵：SQLITE_CONSTRAINT_UNIQUE 不算「没抢到」', () => {
      const companyId = setMaxParallel(nextCompany(), 3);
      const goalId = newGoal(companyId);
      const a = enqueue(companyId, goalId, { title: 'a' });
      const b = enqueue(companyId, goalId, { title: 'b' });
      const db = wdb.getDb();

      db.exec(LEGACY_INDEX_SQL); // 伪造「旧版本进程启动了一次」
      try {
        expect(wdb.updateRunStatus(a.runId, 'queued', 'ready')).toBe(1);
        // max=3 明明还有位，索引却把第二条按 v1 的规矩拒了 —— 并行被悄悄压回 1
        let caught;
        try { wdb.updateRunStatus(b.runId, 'queued', 'ready'); } catch (e) { caught = e; }
        expect(caught).toBeDefined();
        expect(caught.code).toBe('SQLITE_CONSTRAINT_UNIQUE');
        // 关键：**不能**被当成正常竞争。当成竞争的话调度器 continue 掉，
        // 这条任务永远排队、日志一行没有、界面与正常排队一模一样。
        expect(wdb.isSlotRaceError(caught)).toBe(false);
      } finally {
        db.exec('DROP INDEX IF EXISTS idx_graph_runs_one_active_root');
      }
      // 索引一没，同一跳立刻过 —— 证明刚才那次拒绝确实是索引干的，不是额度不够
      expect(wdb.updateRunStatus(b.runId, 'queued', 'ready')).toBe(1);
      expect(wdb.countActiveRootRuns(companyId)).toBe(2);
    });
  });

  // ────────────────────────────────────────────────────────────────
  // ④ 恢复路径：坑满 → 409（不是 500）→ 位子一空同一个请求就成功
  // ────────────────────────────────────────────────────────────────
  describe('④ 位满时人点「启动」：409 且点名全部占位者，位子一空能回来', () => {
    it('max=3 全占满 → 409 + holders 三条都在 + hint 指向每一条真停得掉的；腾一个位就 200', async () => {
      const companyId = setMaxParallel(nextCompany(), 3);
      const goalId = newGoal(companyId);
      // 占位的三条：直推 running（用例不需要真跑 agent）
      const holders = ['h1', 'h2', 'h3'].map((title) => {
        const { taskId, runId } = enqueue(companyId, goalId, { title });
        wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'running' WHERE id = ?`).run(runId);
        return { taskId, runId };
      });
      expect(wdb.countActiveRootRuns(companyId)).toBe(3);

      // 被挡的那条：**没有 task** 的已收敛 run（裸 run 退不回队列，错误才会一路抛到路由层）
      const dead = wdb.createRun({ graphKind: 'work', graphId: buildGraph(companyId, ['solo2']), companyId });
      wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'completed' WHERE id = ?`).run(dead.id);
      const nodeId = wdb.getRunState(dead.id).nodes[0].id;
      wdb.getDb().prepare(`UPDATE graph_run_nodes SET status = 'fail' WHERE id = ?`).run(nodeId);

      const url = `${origin}/api/graph-runs/${dead.id}/nodes/${nodeId}/retry`;
      const r1 = await fetch(url, { method: 'POST' });
      const body = await r1.json();
      expect(r1.status).toBe(409); // 500 会让调用方去重试 / 报故障，而这里唯一有用的动作是等或停
      expect(body.reason).toBe('run-slot-busy');
      expect({ active: body.active, max: body.max }).toEqual({ active: 3, max: 3 });
      // 三条占位者一条都不能少：只点名一条的话，用户停掉它发现还是起不来
      expect(body.holders.map((h) => h.runId).sort()).toEqual(holders.map((h) => h.runId).sort());
      expect(body.holders.map((h) => h.taskId).sort()).toEqual(holders.map((h) => h.taskId).sort());
      for (const h of holders) expect(body.hint).toContain(`/api/graph-runs/${h.runId}/stop`);
      // 拒绝是干净的：节点没被动过
      expect(wdb.getRunNode(nodeId).status).toBe('fail');
      expect(wdb.getRunNode(nodeId).attempt).toBe(0);

      // 腾出一个位 → 同一个请求就成功（这才叫「回得来」）
      sse.transitionRun(holders[0].runId, 'running', 'completed', {}, { summary: '用例直推终态' });
      await flush();
      const r2 = await fetch(url, { method: 'POST' });
      const ok = await r2.json();
      expect(r2.status).toBe(200);
      expect(ok.ok).toBe(true);
      // 它真的占上了刚空出来的那个位（ready 还是已经被驱动推到 running 都算 —— 判据是活跃集，
      // 不是某个具体状态：按具体状态断言的话，哪天驱动早半拍这条用例就会假红）
      expect(wdb.ACTIVE_ROOT_RUN_STATUS).toContain(wdb.getRun(dead.id).run_status);
      expect(wdb.countActiveRootRuns(companyId)).toBe(3); // 3 个位：走了一条、进来一条
    });
  });

  // ────────────────────────────────────────────────────────────────
  // ⑤ 配置非法 → 回落 1 并留声
  // ────────────────────────────────────────────────────────────────
  describe('⑤ 配置非法：回落 1，绝不回落成 0（静默全停）或无限（超发）', () => {
    it('0 / -1 / "abc" / true / 小数 / 缺字段 / 没文件 / 坏 JSON 一律 1，且每一次都真的打了日志', () => {
      const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
      try {
        for (const bad of [0, -1, 'abc', true, 1.5, '3']) {
          const cid = setMaxParallel(nextCompany(), bad);
          warn.mockClear();
          expect(wdb.maxActiveRootRuns(cid)).toBe(1);
          // 「配置写错了就当没配」必须是能看见的，否则就是一次静默降级
          expect(warn).toHaveBeenCalledTimes(1);
          expect(String(warn.mock.calls[0][0])).toContain(cid);
        }

        // 坏掉的 JSON：不许把一次出队变成 500，但同样必须留声
        const broken = nextCompany();
        const { companyRootDir } = require('../../services/company-paths');
        fs.mkdirSync(companyRootDir(broken), { recursive: true });
        fs.writeFileSync(path.join(companyRootDir(broken), 'company.json'), '{ 这不是 JSON');
        warn.mockClear();
        expect(wdb.maxActiveRootRuns(broken)).toBe(1);
        expect(warn).toHaveBeenCalledTimes(1);

        // 没文件 / 没这个字段：这是绝大多数公司的常态，不该刷日志
        warn.mockClear();
        expect(wdb.maxActiveRootRuns(nextCompany())).toBe(1);
        expect(wdb.maxActiveRootRuns(setMaxParallel(nextCompany(), undefined))).toBe(1);
        expect(warn).not.toHaveBeenCalled();

        // 超上限截断到 CAP（不是回落 1，也不是当真）——「无限」在这里同样不许出现
        const huge = setMaxParallel(nextCompany(), 1e6);
        warn.mockClear();
        expect(wdb.maxActiveRootRuns(huge)).toBe(wdb.MAX_ACTIVE_ROOT_RUNS_CAP);
        expect(Number.isFinite(wdb.MAX_ACTIVE_ROOT_RUNS_CAP)).toBe(true);
        expect(warn).toHaveBeenCalledTimes(1);
      } finally {
        warn.mockRestore();
      }
    });

    it('同一份非法配置只吵一次（日志不刷屏），但**回落值每次都算**，不是缓存下来的', () => {
      const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
      try {
        const cid = setMaxParallel(nextCompany(), 0);
        expect(wdb.maxActiveRootRuns(cid)).toBe(1);
        expect(wdb.maxActiveRootRuns(cid)).toBe(1);
        expect(wdb.maxActiveRootRuns(cid)).toBe(1);
        expect(warn).toHaveBeenCalledTimes(1);

        // 把配置改对：**不重启、不清缓存**，下一次准入就该按新值走（额度是调用时读盘，不是模块级快照）
        setMaxParallel(cid, 2);
        expect(wdb.maxActiveRootRuns(cid)).toBe(2);
        const goalId = newGoal(cid);
        const runs = ['a', 'b', 'c'].map((t) => enqueue(cid, goalId, { title: t }).runId);
        expect(wdb.updateRunStatus(runs[0], 'queued', 'ready')).toBe(1);
        expect(wdb.updateRunStatus(runs[1], 'queued', 'ready')).toBe(1);
        expect(() => wdb.updateRunStatus(runs[2], 'queued', 'ready')).toThrow(wdb.RunSlotBusyError);

        // 再改回非法值：回落 1 立刻生效，坑满的判据跟着变（此刻 2 条在跑 > 1，新的一条必被拒）
        setMaxParallel(cid, -5);
        expect(wdb.maxActiveRootRuns(cid)).toBe(1);
        let caught;
        try { wdb.updateRunStatus(runs[2], 'queued', 'ready'); } catch (e) { caught = e; }
        expect(caught).toBeInstanceOf(wdb.RunSlotBusyError);
        expect(caught.max).toBe(1);
      } finally {
        warn.mockRestore();
      }
    });
  });

  // ────────────────────────────────────────────────────────────────
  // ⑥ DROP 迁移：存量库（带旧索引）启动一次就没了
  // ────────────────────────────────────────────────────────────────
  describe('⑥ 迁移：带 idx_graph_runs_one_active_root 的存量库', () => {
    const Database = require('better-sqlite3');
    /** 直接开一个连接看索引在不在 —— 不经过 work-graph-db，免得「自己证明自己」 */
    function indexExists(dbPath) {
      const db = new Database(dbPath, { readonly: true });
      try { return !!db.prepare(INDEX_EXISTS_SQL).get(); } finally { db.close(); }
    }
    function createLegacyIndex(dbPath) {
      const db = new Database(dbPath);
      try { db.exec(LEGACY_INDEX_SQL); } finally { db.close(); }
    }

    it('启动一次索引就没了；再启动两次不报错（DROP INDEX IF EXISTS 本身幂等）', async () => {
      const dbPath = path.join(runsDir, 'migrate-seq.sqlite');
      const first = await runWorker(['boot'], { dbPath });
      expect(first.code, first.stderr).toBe(0);

      // 伪造存量库：把批 4 删掉的那条索引原样建回去
      createLegacyIndex(dbPath);
      expect(indexExists(dbPath)).toBe(true);

      const second = await runWorker(['boot'], { dbPath });
      expect(second.code, second.stderr).toBe(0);
      expect(second.json).toEqual({ index: false }); // 进程自己看到的
      expect(indexExists(dbPath)).toBe(false);       // 库里真的没了

      for (let i = 0; i < 2; i += 1) {
        const again = await runWorker(['boot'], { dbPath });
        expect(again.code, again.stderr).toBe(0);
        expect(again.json).toEqual({ index: false });
      }
    }, 30000);

    it('多进程同时启动不打架：5 个进程一起跑迁移，全部退出码 0，索引没了', async () => {
      const dbPath = path.join(runsDir, 'migrate-parallel.sqlite');
      expect((await runWorker(['boot'], { dbPath })).code).toBe(0);
      createLegacyIndex(dbPath);
      expect(indexExists(dbPath)).toBe(true);

      const boots = await Promise.all(Array.from({ length: 5 }, () => runWorker(['boot'], { dbPath })));
      for (const b of boots) {
        expect(b.code, b.stderr).toBe(0);   // SQLITE_BUSY / "no such index" 都会在这里现形
        expect(b.json).toEqual({ index: false });
      }
      expect(indexExists(dbPath)).toBe(false);
    }, 30000);

    it('本文件这个库自己也没有那条索引（新建库走 CREATE TABLE 块，那里已经没有它了）', () => {
      expect(wdb.getDb().prepare(INDEX_EXISTS_SQL).get()).toBeUndefined();
    });
  });
});