宜博 / lemonboss

/**
 * Unit Tests — task-scheduler.js（27-P3C 批 3 · 排队）
 *
 * 逐条对应 §十 验收：
 *   2  T1 收敛后 T2 自动起跑（漏斗钩子覆盖正常收敛）
 *   13 停止即出队（stopRun 不走 settleRun，钩子必须覆盖它）
 *   15 排队期间 replan → 出队后按新图跑（syncRunNodesToDef）
 *   23 上游失败 → 下游 blocked，上游恢复后自动回 queued（blocked 是推导出来的，不落死）
 * 外加 §九.1 那张「queued 混进 run_status」的回归表里点名要钉死的三条断言：
 *   · queued 不在活跃集里（排队不占坑，出队那一跳才被准入事务数一次）
 *   · advanceOne 挡住 queued run（防止 queued run 被误推进的唯一一道闸）
 *   · startRun 拒绝 queued run（出队顺序不能反过来）
 *
 * 隔离手法与 graph-engine.test.js 一致：先设 GRAPH_DB_PATH / GRAPH_RUNS_DIR 再 require
 * （graph-run-sse 在模块加载时快照 GRAPH_RUNS_DIR，先 require 再设会往仓库 data/companies/ 写，
 * 而且不报错）。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('task-scheduler', () => {
  let wdb;
  let engine;
  let sse;
  let scheduler;
  let runsDir;
  const TEST_DB_PATH = '/tmp/test-task-scheduler.sqlite';
  /** 挂起的 executor promise，afterAll 统一放掉，免得留下永不 settle 的 driver */
  const pending = [];

  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(), 'task-scheduler-runs-'));
    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',
      '../../services/task-settle',
      '../../services/task-scheduler',
    ]) {
      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');
  });

  /**
   * `advanceOne` 的 ensureProjectRoot 会在**仓库的** data/companies/ 下给每家公司建项目根
   * （它不认 GRAPH_RUNS_DIR）。只删本文件造的那些前缀目录，且必须确认里面没有 company.json
   * —— 那是真公司，删了就是删用户数据。（写法与 goal-task-acceptance.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('sched-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, 100));
    scheduler.stopSweeper();
    cleanupLeakedCompanyDirs();
    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 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 = () => `sched-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);
  }

  /** 建一张链式 work graph 定义（校验要求单入口 + 全可达，所以多节点时串起来） */
  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;
  }

  /**
   * 「入队」——批 3 阶段一没有端点，这里手工拼出端点将来要写的那三步：
   * 建 task 行 → 建 run 行（**runStatus: 'queued'**，不占坑）→ task 标 queued + 挂 current_run_id。
   */
  function enqueue(companyId, goalId, { title, mode = 'auto', dependsOn = [], priority = 'P1', graphId } = {}) {
    const task = wdb.createTask({
      goalId, companyId, title, description: `${title} 的描述`,
      executionMode: mode, dependsOn, priority,
    });
    const run = wdb.createRun({
      graphKind: 'work',
      graphId: 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 };
  }

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

  /** 永不 resolve 的 executor：让第一条 run 稳稳停在 running，好观察队列 */
  function hangingExecutor() {
    const calls = [];
    engine.setAgentExecutor((ctx) => {
      calls.push(ctx.nodeName);
      return new Promise((resolve) => pending.push(() => resolve({ status: 'success' })));
    });
    return calls;
  }

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

  // ──────────────────────────────────────────────────────────────
  // §九.1 回归表：queued 不被误当可执行
  // ──────────────────────────────────────────────────────────────
  describe('queued 混进 run_status 的三道闸', () => {
    it('queued 不占坑：同一公司可以躺任意多条 queued root run，而出队时超发被准入事务拒', () => {
      const companyId = nextCompany();
      const goalId = newGoal(companyId);

      // 这条断言是出队原子性的**全部依赖**，钉死它
      expect(wdb.ACTIVE_ROOT_RUN_STATUS).toEqual(['ready', 'running', 'paused']);
      expect(wdb.ACTIVE_ROOT_RUN_STATUS).not.toContain('queued');

      const a = enqueue(companyId, goalId, { title: 'a' });
      const b = enqueue(companyId, goalId, { title: 'b' });
      const c = enqueue(companyId, goalId, { title: 'c' });
      for (const r of [a, b, c]) expect(wdb.getRun(r.runId).run_status).toBe('queued');

      // 第一条出队 = CAS 成功
      expect(wdb.updateRunStatus(a.runId, 'queued', 'ready')).toBe(1);
      // 第二条被**主守卫**（准入事务里数坑）拒：抛 RunSlotBusyError，**不是** changes = 0
      expect(() => wdb.updateRunStatus(b.runId, 'queued', 'ready')).toThrow(wdb.RunSlotBusyError);
      expect(wdb.getRun(b.runId).run_status).toBe('queued');
      // 批 3 这里演示的是「兜底那一层（唯一索引）也还在」：绕开 updateRunStatus 直接写库照样被拒。
      // 批 4 DROP 了那条索引，所以这一段反过来钉两件事 ——
      //   ① 索引确实没了：裸写库现在**写得进去**（这正是「删掉了最后一道 DB 兜底」的样子，
      //      要能演示出来，否则「已 DROP」只是一句注释）；
      //   ② 于是唯一的守卫只剩上面那个准入事务，它今天就得是对的。
      expect(() =>
        wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'ready' WHERE id = ?`).run(b.runId)
      ).not.toThrow();
      expect(wdb.getRun(b.runId).run_status).toBe('ready');
      // 把裸写的这条放回 queued，别让它污染后面的断言（它是绕过守卫写进去的，不是准入放行的）
      wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'queued' WHERE id = ?`).run(b.runId);
      // 同一条重复出队 = changes 0（幂等 no-op，不抛）
      expect(wdb.updateRunStatus(a.runId, 'queued', 'ready')).toBe(0);
      // queued 有出路：撤出队列
      expect(wdb.canTransitionRun('queued', 'stopped')).toBe(true);
      expect(wdb.updateRunStatus(c.runId, 'queued', 'stopped')).toBe(1);
    });

    /**
     * 准入守卫（updateRunStatus 里的 BEGIN IMMEDIATE 事务）。
     * 判据是「**从活跃集外进入活跃集**」—— 这几条把「按 to 判」那种写法钉死在墙上：
     * 按 to 判的话 ready→running 会把 run 自己算成竞争者，第一条 run 都起不来。
     */
    describe('准入守卫：从活跃集外进入活跃集才占坑', () => {
      it('集内移动不占新坑：ready→running / running→paused / paused→running 一路畅通', () => {
        const companyId = nextCompany();
        const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a' });

        expect(wdb.updateRunStatus(runId, 'queued', 'ready')).toBe(1);
        expect(wdb.countActiveRootRuns(companyId)).toBe(1); // 坑已经被它自己占着
        // 下面三跳的 from 全在活跃集里 —— 计数命中的是它自己，按 to 判就会在这里全军覆没
        expect(wdb.updateRunStatus(runId, 'ready', 'running')).toBe(1);
        expect(wdb.updateRunStatus(runId, 'running', 'paused')).toBe(1);
        expect(wdb.updateRunStatus(runId, 'paused', 'running')).toBe(1);
        expect(wdb.countActiveRootRuns(companyId)).toBe(1); // 全程一个坑
      });

      it('活跃集外进来的才拦：blocked→running 在坑满时被拒，坑空了同一跳就过', () => {
        const companyId = nextCompany();
        const goalId = newGoal(companyId);
        const a = enqueue(companyId, goalId, { title: 'a' });
        const b = enqueue(companyId, goalId, { title: 'b' });

        wdb.updateRunStatus(a.runId, 'queued', 'ready');
        wdb.updateRunStatus(a.runId, 'ready', 'running');
        // b 直接落到 blocked（活跃集外，不占坑；queued→blocked 不是合法转换，所以裸写库造状态）
        wdb.getDb().prepare(`UPDATE graph_runs SET run_status = 'blocked' WHERE id = ?`).run(b.runId);
        expect(wdb.countActiveRootRuns(companyId)).toBe(1);
        expect(() => wdb.updateRunStatus(b.runId, 'blocked', 'running')).toThrow(wdb.RunSlotBusyError);

        // a 让出坑（running → completed 离开活跃集）
        expect(wdb.updateRunStatus(a.runId, 'running', 'completed')).toBe(1);
        expect(wdb.updateRunStatus(b.runId, 'blocked', 'running')).toBe(1);
      });

      it('子 run 不占公司的坑：root 正在跑时，子 run 照样能 planning→ready', () => {
        const companyId = nextCompany();
        const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a' });
        wdb.updateRunStatus(runId, 'queued', 'ready');

        const child = wdb.createRun({
          graphKind: 'work',
          graphId: buildGraph(companyId, ['child']),
          companyId,
          parentRunId: runId,
          runStatus: 'planning',
        });
        expect(wdb.updateRunStatus(child.id, 'planning', 'ready')).toBe(1);
        // 计数只认 root：坑还是那一个
        expect(wdb.countActiveRootRuns(companyId)).toBe(1);
      });

      /**
       * 这一条以前钉的是「坑满的报错文本仍能被 company.js 的 `/UNIQUE|constraint/i` 嗅到」，
       * 理由写着「routes/company.js 那两处 catch 至今是按文本嗅探的」—— **批 3 把那两处
       * catch 整段删了**（下达不再因坑满失败，改成入队）。于是这条用例一直绿着，钉的却是
       * 一个没有任何调用方的契约，还反过来禁止清理那句报文。
       * 现在钉真正的那条：判据只有 `work-graph-db.isSlotRaceError` 一个持有者，按类型判。
       * 该认的一个都不能漏 —— 漏一种，调度器就会把一次正常竞争当故障去 abandonRun（丢任务），
       * 而路由那边照常 409，两边对同一件事给出相反结论。
       * 不该认的也一个都不能多认（批 4 起 SQLITE_CONSTRAINT_UNIQUE 就在这一侧，见下）：
       * 认宽了是把真 bug 当正常竞争 continue 掉，那是静默。
       */
      it('「没抢到坑」的判据只有一个持有者，且按类型判 —— 改一句报错文案不该改变任何分支', () => {
        const companyId = nextCompany();
        const goalId = newGoal(companyId);
        const a = enqueue(companyId, goalId, { title: 'a' });
        const b = enqueue(companyId, goalId, { title: 'b' });
        wdb.updateRunStatus(a.runId, 'queued', 'ready');

        let caught;
        try { wdb.updateRunStatus(b.runId, 'queued', 'ready'); } catch (e) { caught = e; }
        expect(caught).toBeInstanceOf(wdb.RunSlotBusyError);
        expect(wdb.isSlotRaceError(caught)).toBe(true);

        // 批 3 这里钉的是「兜底那一层（唯一索引）抛的 SqliteError 也必须被认成『没抢到』」。
        // 批 4 DROP 了索引，于是这条判据**反过来**：一个 SQLITE_CONSTRAINT_UNIQUE 再出现在
        // 这条路上，只可能是索引被旧版本进程建回来了（max>1 被悄悄压回 1）。
        // 认成「没抢到」的话调度器会 continue 掉，任务永远排队、日志一行没有 ——
        // 正是要消灭的「写不进去但不报错」。所以它必须**不**算竞争，一路抛到人能看见的地方。
        const fakeUniqueErr = Object.assign(new Error('UNIQUE constraint failed: graph_runs.company_id'), {
          code: 'SQLITE_CONSTRAINT_UNIQUE',
        });
        expect(wdb.isSlotRaceError(fakeUniqueErr)).toBe(false);

        // ⚠️ 批 3 这里断言 InvalidTransitionError 也算「没抢到」，注释写着「这条 run 已经被别人
        // 从 queued 推走了」。**探针实测证伪**：被别人推走时 updateRunStatus 走的是条件更新
        // `WHERE id=? AND run_status=@from`，**返回 0、不抛异常**；真正抛 InvalidTransitionError
        // 的只有「调用方声明了非法转换」（expected 是调用方传进来的声明值，canTransitionRun
        // judge 的是 expected→to 而不是 cur.run_status→to）。
        // 把调用方的 bug 当成正常竞争吞掉、还不留日志 —— 与上面那条 UNIQUE 同一个病。
        expect(wdb.isSlotRaceError(new wdb.InvalidTransitionError('run', 'x', 'queued', 'ready'))).toBe(false);
        // 别的错误一律不算「没抢到」—— 认宽了就是把真 bug 当成正常竞争 continue 掉
        expect(wdb.isSlotRaceError(new Error('UNIQUE constraint failed: 别处的错'))).toBe(false);
        expect(wdb.isSlotRaceError(null)).toBe(false);

        // 两个调用方都必须走这一份，不许再各写一份（已经分叉过一次了）。
        // 读源码而不是调它们的私有函数：这两个 isRaceError / isSlotRace 都没导出，
        // 而「有没有第二个持有者」本来就是源码层面的事实。
        for (const mod of ['../../services/task-scheduler.js', '../../routes/company.js']) {
          const src = fs.readFileSync(path.resolve(__dirname, mod), 'utf8');
          expect(src).toContain('isSlotRaceError');
          // 报错文本嗅探不许回来：它正是上一版分叉的那一半
          expect(src).not.toMatch(/\/UNIQUE constraint failed\/i\.test/);
        }
      });

      it('「别人抢先做完了」不许报成「坑满」：CAS 没匹配上返回 0，不抛', () => {
        const companyId = nextCompany();
        const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a' });
        expect(wdb.updateRunStatus(runId, 'queued', 'ready')).toBe(1);

        // 再来一次同样的出队：此刻 countActiveRootRuns 已经是 1，少了事务内那次重读
        // 就会把这次幂等 no-op 误报成 RunSlotBusyError —— 调度器据此会去 abandonRun，
        // 把一条刚刚成功起跑的任务收成 cancelled。
        expect(wdb.updateRunStatus(runId, 'queued', 'ready')).toBe(0);
      });
    });

    /**
     * 批 4：额度从「模块级常量 1」变成 per-company 的 `company.json.max_parallel_tasks`。
     *
     * 这组用例的重点不是「并行能跑」，而是**回落**那一条：全仓几乎所有单测的公司都没有
     * company.json（`sched-co-*` 只是个字符串），一旦回落算错，1195 条测试会一起变色 ——
     * 所以「读不到 = 1」是保护存量的那条断言，必须先钉。
     *
     * 配置写在 `company-paths.companyRootDir()` 下而不是 `data/companies/` 下：
     * 前者认 GRAPH_RUNS_DIR，本文件设了它，于是这组用例不会去碰仓库里的真公司目录
     * （准入判据一旦去读真 data/，撞上同名公司就读到真配置 —— §六 缺陷 A 同型）。
     */
    describe('每家公司自己的执行坑数（company.json.max_parallel_tasks）', () => {
      /** 给这家公司写一份 company.json；`n === null` 表示写一个没有该字段的配置 */
      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 !== null) cfg.max_parallel_tasks = n;
        fs.writeFileSync(path.join(dir, 'company.json'), JSON.stringify(cfg));
        return companyId;
      }

      it('没有 company.json 的公司回落 1 —— 存量单测与存量公司的行为一个字不变', () => {
        const companyId = nextCompany();
        expect(wdb.maxActiveRootRuns(companyId)).toBe(wdb.DEFAULT_MAX_ACTIVE_ROOT_RUNS);
        expect(wdb.DEFAULT_MAX_ACTIVE_ROOT_RUNS).toBe(1);
        // 有 company.json 但没写这个字段，同样回落
        expect(wdb.maxActiveRootRuns(setMaxParallel(nextCompany(), null))).toBe(1);
      });

      it('非法值一律回落 1，不静默当成无限：0 / 负数 / 小数 / 字符串 / 超上限', () => {
        for (const bad of [0, -3, 1.5, '3', true, null]) {
          expect(wdb.maxActiveRootRuns(setMaxParallel(nextCompany(), bad))).toBe(1);
        }
        // 超上限按上限截断（不是回落 1，也不是当真）
        const huge = setMaxParallel(nextCompany(), 100000);
        expect(wdb.maxActiveRootRuns(huge)).toBe(wdb.MAX_ACTIVE_ROOT_RUNS_CAP);
      });

      it('max=3：前三条准入放行，第 4 条被拒且 active / max / heldBy 三个字段都对', () => {
        const companyId = setMaxParallel(nextCompany(), 3);
        const goalId = newGoal(companyId);
        const runs = ['a', 'b', 'c', 'd'].map((title) => enqueue(companyId, goalId, { title }).runId);

        for (const [i, runId] of runs.slice(0, 3).entries()) {
          expect(wdb.updateRunStatus(runId, 'queued', 'ready')).toBe(1);
          expect(wdb.countActiveRootRuns(companyId)).toBe(i + 1);
        }

        let caught;
        try { wdb.updateRunStatus(runs[3], 'queued', 'ready'); } catch (e) { caught = e; }
        expect(caught).toBeInstanceOf(wdb.RunSlotBusyError);
        expect(caught.max).toBe(3);
        expect(caught.active).toBe(3);
        // heldBy 点得出**具体是谁**占着 —— 少了它 409 上只剩「坑满」，人无从判断该等谁
        expect(caught.heldBy.slice().sort()).toEqual(runs.slice(0, 3).slice().sort());
        expect(wdb.getRun(runs[3]).run_status).toBe('queued'); // 拒绝是干净的

        // 让出一个坑，第 4 条同一跳立刻过（额度是「同时几条」，不是「一共几条」）
        expect(wdb.updateRunStatus(runs[0], 'ready', 'running')).toBe(1);   // 集内移动，不占新坑
        expect(wdb.updateRunStatus(runs[0], 'running', 'completed')).toBe(1);
        expect(wdb.updateRunStatus(runs[3], 'queued', 'ready')).toBe(1);
      });

      it('额度是 per-company 的：A 公司放开到 3，B 公司照样只有 1', () => {
        const a = setMaxParallel(nextCompany(), 3);
        const b = nextCompany(); // 没配 → 1
        const aRuns = ['a1', 'a2'].map((t) => enqueue(a, newGoal(a), { title: t }).runId);
        const bRuns = ['b1', 'b2'].map((t) => enqueue(b, newGoal(b), { title: t }).runId);

        for (const runId of aRuns) expect(wdb.updateRunStatus(runId, 'queued', 'ready')).toBe(1);
        expect(wdb.updateRunStatus(bRuns[0], 'queued', 'ready')).toBe(1);
        expect(() => wdb.updateRunStatus(bRuns[1], 'queued', 'ready')).toThrow(wdb.RunSlotBusyError);
        expect(wdb.countActiveRootRuns(a)).toBe(2);
        expect(wdb.countActiveRootRuns(b)).toBe(1);
      });

      it('一次 tick 把空坑填满，不是一条一条来（max=3 → 三条同时起跑）', async () => {
        const companyId = setMaxParallel(nextCompany(), 3);
        const goalId = newGoal(companyId);
        hangingExecutor();
        const tasks = ['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(tasks.slice(0, 3).map((t) => t.taskId));
        await flush();
        expect(wdb.countActiveRootRuns(companyId)).toBe(3);
        expect(wdb.getTask(tasks[3].taskId).status).toBe('queued'); // 第 4 条还在队里
        // 再 tick 一次不该超发：坑满就原地返回
        expect(scheduler.tick(companyId).skipped).toBe('no-slot');
      });

      it('候选里有一条起不来时，不该白占一个名额 —— 剩下的坑照样被填满', () => {
        const companyId = setMaxParallel(nextCompany(), 2);
        const goalId = newGoal(companyId);
        hangingExecutor();
        // 队首这条的 run 行被抹掉 = 入队做了一半（5a 那条分支），它必须被跳过而不是吃掉一个坑
        const broken = enqueue(companyId, goalId, { title: 'broken' });
        wdb.getDb().prepare(`UPDATE tasks SET current_run_id = NULL WHERE id = ?`).run(broken.taskId);
        const ok = ['a', 'b'].map((title) => enqueue(companyId, goalId, { title }));

        const res = scheduler.tick(companyId);
        // 批 3 的 `.slice(0, max - active)` 会把 broken 算进名额，于是只起来一条
        expect(res.started.map((s) => s.taskId)).toEqual(ok.map((t) => t.taskId));
        expect(wdb.getTask(broken.taskId).status).toBe('blocked');
      });
    });

    it('advanceOne 挡住 queued run：不推进、不报错、节点一个都不点亮', async () => {
      const companyId = nextCompany();
      const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a' });
      const calls = hangingExecutor();

      const res = await engine.advanceOne(runId);
      expect(res.reason).toBe('run-not-running');
      expect(calls).toEqual([]);
      // queued run 有一整套 pending 节点行（createRun 建行时就插全了）——
      // 「靠没有节点行自然报错」那道防线不存在，闸门只有 run_status 这一句
      const nodes = wdb.getRunState(runId).nodes;
      expect(nodes.length).toBeGreaterThan(0);
      expect(nodes.every((n) => n.status === 'pending')).toBe(true);
      expect(wdb.getRun(runId).run_status).toBe('queued');
    });

    it('startRun 拒绝 queued run（出队顺序不能反过来），markRunReady 也不再假装成功', () => {
      const companyId = nextCompany();
      const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a' });

      expect(() => engine.startRun(runId)).toThrow(/还在排队中/);
      // markRunReady 的 from 写死成 planning，对 queued run 是空转 —— 以前它照样回 ok:'ready'
      expect(() => engine.markRunReady(runId)).toThrow(/未生效/);
      expect(wdb.getRun(runId).run_status).toBe('queued');
    });
  });

  // ──────────────────────────────────────────────────────────────
  // §四.3 末：排队期间 replan
  // ──────────────────────────────────────────────────────────────
  it('验收 15：排队期间图被改，出队前 syncRunNodesToDef 补齐新节点、删掉孤儿', () => {
    const companyId = nextCompany();
    const graphId = buildGraph(companyId, ['n1', 'n2']);
    const { runId } = enqueue(companyId, newGoal(companyId), { title: 'a', graphId });
    expect(wdb.getRunState(runId).nodes.map((n) => n.name).sort()).toEqual(['n1', 'n2']);

    // 「重新规划组织」：删掉 n2、加上 n3（定义层直接改，模拟 replan 落库后的结果）。
    // 关键在于 defNodeId 是节点**行 id**，所以这必须是真的删行 + 插行，改个名字不算。
    ensureAgents(['n3']);
    const db = wdb.getDb();
    const now = new Date().toISOString();
    db.prepare(`DELETE FROM work_graph_edges WHERE work_graph_id = ?`).run(graphId);
    db.prepare(`DELETE FROM work_graph_nodes WHERE work_graph_id = ? AND node_name = 'n2'`).run(graphId);
    const n1 = db.prepare(`SELECT id FROM work_graph_nodes WHERE work_graph_id = ? AND node_name = 'n1'`).get(graphId);
    const n3 = wdb.newId('wgn');
    db.prepare(`INSERT INTO work_graph_nodes (id, work_graph_id, node_name, system_agent_id, node_kind, created_at)
                VALUES (?, ?, 'n3', 'n3', 'agent_call', ?)`).run(n3, graphId, now);
    db.prepare(`INSERT INTO work_graph_edges (id, work_graph_id, from_node_id, to_node_id, condition, edge_kind, created_at)
                VALUES (?, ?, ?, ?, 'success', 'forward', ?)`).run(wdb.newId('wge'), graphId, n1.id, n3, now);

    const diff = wdb.syncRunNodesToDef(runId);
    expect(diff.added.length).toBe(1);
    expect(diff.removed.length).toBe(1);
    expect(diff.kept).toEqual([]);
    const names = wdb.getRunState(runId).nodes.map((n) => n.name).sort();
    expect(names).toEqual(['n1', 'n3']);
    expect(wdb.getRunState(runId).nodes.every((n) => n.status === 'pending')).toBe(true);
  });

  // ──────────────────────────────────────────────────────────────
  // §五.1 出队
  // ──────────────────────────────────────────────────────────────
  it('tick 一次只放一条出队（max=1），其余照排，位次按 priority → seq → created_at', () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    hangingExecutor();

    const low = enqueue(companyId, goalId, { title: 'low', priority: 'P2' });
    const high = enqueue(companyId, goalId, { title: 'high', priority: 'P0' });

    // 位次口径只有调度器一处
    expect(scheduler.listQueue(companyId).map((t) => t.id)).toEqual([high.taskId, low.taskId]);
    expect(scheduler.queuePosition(companyId, low.taskId)).toBe(2);

    const res = scheduler.tick(companyId);
    expect(res.started.map((s) => s.taskId)).toEqual([high.taskId]);
    expect(wdb.getTask(high.taskId).status).toBe('running');
    expect(wdb.getTask(high.taskId).attempt).toBe(1);
    expect(wdb.getRun(high.runId).run_status).toBe('running');
    // 后面那条一动不动
    expect(wdb.getTask(low.taskId).status).toBe('queued');
    expect(wdb.getRun(low.runId).run_status).toBe('queued');
    // 坑满了，再 tick 也不会超发
    expect(scheduler.tick(companyId).skipped).toBe('no-slot');
  });

  it('manual 任务不参与自动出队（§五.2bis：它会占着坑停在第一个节点上堵死队头）', () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    hangingExecutor();
    const m = enqueue(companyId, goalId, { title: 'm', mode: 'manual' });

    const res = scheduler.tick(companyId);
    expect(res.started).toEqual([]);
    expect(wdb.getTask(m.taskId).status).toBe('queued');
    // 但它照样占位次（界面上要显示「排队 #1」）
    expect(scheduler.queuePosition(companyId, m.taskId)).toBe(1);
  });

  it('验收 13：停止即出队 —— stopRun 不走 settleRun，靠 transitionRun 漏斗钩子把下一条拉起来', async () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    hangingExecutor();
    const a = enqueue(companyId, goalId, { title: 'a', priority: 'P0' });
    const b = enqueue(companyId, goalId, { title: 'b', priority: 'P1' });

    scheduler.tick(companyId);
    expect(wdb.getTask(a.taskId).status).toBe('running');
    expect(wdb.getTask(b.taskId).status).toBe('queued');

    engine.stopRun(a.runId); // 这条路径全程没有 settleRun
    await flush();

    expect(wdb.getTask(a.taskId).status).toBe('cancelled');
    expect(wdb.getTask(b.taskId).status).toBe('running');
    expect(wdb.getRun(b.runId).run_status).toBe('running');
  });

  it('验收 2：上一条正常收敛后，下一条自动起跑（无人工干预）', async () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    hangingExecutor();
    const a = enqueue(companyId, goalId, { title: 'a', priority: 'P0' });
    const b = enqueue(companyId, goalId, { title: 'b' });

    scheduler.tick(companyId);
    // 直接推唯一漏斗，等价于 auto driver 跑完那一跳
    expect(sse.transitionRun(a.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy();
    await flush();

    expect(wdb.getTask(a.taskId).status).toBe('completed');
    expect(wdb.getTask(b.taskId).status).toBe('running');
  });

  // ──────────────────────────────────────────────────────────────
  // §五.1.2 依赖推导
  // ──────────────────────────────────────────────────────────────
  it('验收 23：上游失败 → 下游 blocked 且带 upstream_failed:；上游恢复后自己回 queued', () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    hangingExecutor();

    const a = enqueue(companyId, goalId, { title: 'a', priority: 'P0' });
    const b = enqueue(companyId, goalId, { title: 'b', dependsOn: [] });
    const c = enqueue(companyId, goalId, { title: 'c', dependsOn: [] });
    // 依赖在 queued 状态也允许改（§六.3）：B 等 A，C 等 B —— 级联要一次算到底
    wdb.updateTask(b.taskId, { dependsOn: [a.taskId] });
    wdb.updateTask(c.taskId, { dependsOn: [b.taskId] });

    wdb.updateTask(a.taskId, { status: 'failed' });
    scheduler.tick(companyId);

    expect(wdb.getTask(b.taskId).status).toBe('blocked');
    expect(wdb.getTask(b.taskId).blocked_reason).toBe(`upstream_failed:${a.taskId}`);
    expect(wdb.getTask(c.taskId).status).toBe('blocked');
    expect(wdb.getTask(c.taskId).blocked_reason).toBe(`upstream_failed:${b.taskId}`);

    // A 重跑成功
    wdb.updateTask(a.taskId, { status: 'completed' });
    const res = scheduler.tick(companyId);
    expect(res.requeued).toContain(b.taskId);
    expect(res.requeued).toContain(c.taskId);
    expect(wdb.getTask(c.taskId).status).toBe('queued');
    expect(wdb.getTask(c.taskId).blocked_reason).toBeNull();
    // B 依赖满足了就该被拉起来，C 还得等 B
    expect(res.started.map((s) => s.taskId)).toEqual([b.taskId]);
  });

  it('别的来源的 blocked（预算耗尽那种）绝不被自动重排', () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    const t = enqueue(companyId, goalId, { title: 'x' });
    wdb.updateTask(t.taskId, { status: 'blocked', blockedReason: 'budget-exhausted' });

    const res = scheduler.tick(companyId);
    expect(res.requeued).toEqual([]);
    expect(wdb.getTask(t.taskId).status).toBe('blocked');
  });

  it('queued 但没有 run 行的任务不许假装在排队：标 blocked 并写明原因（且不会被自动重排）', () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    const t = enqueue(companyId, goalId, { title: 'x' });
    wdb.getDb().prepare(`DELETE FROM graph_run_nodes WHERE run_id = ?`).run(t.runId);
    wdb.getDb().prepare(`DELETE FROM graph_runs WHERE id = ?`).run(t.runId);

    scheduler.tick(companyId);
    const row = wdb.getTask(t.taskId);
    expect(row.status).toBe('blocked');
    expect(row.blocked_reason).toMatch(/^enqueue_incomplete:missing_run/);
    expect(scheduler.tick(companyId).requeued).toEqual([]);
  });

  it('tick 可重入且不 throw：它挂在状态转换的尾巴上，抛出去会把一次正常收敛也带崩', () => {
    expect(() => scheduler.tick(undefined)).not.toThrow();
    expect(scheduler.tick('no-such-company').started).toEqual([]);
  });
});