宜博 / lemonboss

/**
 * 批 3 · 排队 —— §八 验收清单的单测投影（27-P3C）
 *
 * 与 `task-scheduler.test.js` 的分工：那份钉的是**单个原语**的形状（准入守卫的判据、
 * CAS 的 WHERE、唯一索引的取值域）；这一份钉的是**端到端的那句话**——
 * 「人不再干预，队伍自己往前走」。所以这里几乎不直接调 `updateRunStatus` / `blockQueuedTask`，
 * 一律推真实入口（`engine.stopRun` / 真的烧完预算 / `sse.transitionRun` 这个唯一漏斗），
 * 然后只看结果：下一条起没起来。
 *
 * 覆盖的验收条目（编号同 §八）：
 *   2  T2/T3 依赖 T1 → T1 收敛后 T2 自动起跑，全程无人工干预
 *   3  max=1 下第二条排队而不是 409
 *   7  取消 queued → run queued→stopped，后面的顶上来
 *   8  failed 的 task rerun → attempt+1、新 run、旧 run 仍在历史里
 *   9  进程重启后 sweep 把队列捡起来
 *   13 stopRun（不走 settleRun）后队列自动前进，不靠 60s 兜底
 *   14 预算耗尽 → blocked → 同样自动前进
 *   15 排队期间 replan → 出队后按新图跑（tick 里真的调了 syncRunNodesToDef）
 *   16 manual 不被自动拉起、不堵队；人点「开始」才占坑
 *   23 上游 failed → 下游 blocked:upstream_failed；上游 rerun 成功后自动回 queued 并起跑
 *   ── 唯一索引的 WHERE 里没有 queued（出队原子性的全部依赖）
 *
 * 隔离手法与 graph-engine.test.js / task-scheduler.test.js 一致：**先设 env 再 require**
 * （graph-run-sse 在模块加载时快照 GRAPH_RUNS_DIR，先 require 再设会往仓库 data/companies/ 写
 * 而且不报错）。DB 路径与那两份**故意不同**：vitest 默认按文件并行，共用一个 sqlite 文件
 * 会让两份用例互相 DELETE 对方的行。
 */
import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
import fs from 'fs';
import os from 'os';
import path from 'path';

describe('批 3 排队 · §八 验收', () => {
  let wdb;
  let engine;
  let sse;
  let scheduler;
  let validation;
  let runsDir;
  const TEST_DB_PATH = '/tmp/test-task-queue-acceptance.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(), 'task-queue-acc-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',
      '../../services/goal-validation',
    ]) {
      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');
    validation = require('../../services/goal-validation');

    // goal 级验收默认会**真起 Claude CLI**（§四.10）。本文件有几条用例会让一个 goal 下的
    // 多条 task 全部 completed，那正是它的触发条件 —— 不注入就会在单测里起子进程。
    validation.setValidateGoal(async () => ({ pass: true, verdict: '用例注入：pass', tokens: 0 }));
  });

  /**
   * `advanceOne` 的 ensureProjectRoot 会在**仓库的** data/companies/ 下给每家公司建项目根
   * （它不认 GRAPH_RUNS_DIR）。只删本文件造的那些前缀目录，且必须确认里面没有 company.json
   * —— 那是真公司，删了就是删用户数据。
   */
  function cleanupLeakedCompanyDirs() {
    const root = path.resolve(__dirname, '../../../data/companies');
    if (!fs.existsSync(root)) return;
    for (const cid of fs.readdirSync(root)) {
      if (!cid.startsWith('qacc-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 () => {
    // 等一拍再清：末尾那条用例的 setImmediate 钩子可能还没出栈，
    // 它一跑 advanceOne 就会经 ensureProjectRoot 把刚删掉的公司目录原样建回来
    await new Promise((r) => setTimeout(r, 120));
    scheduler.stopSweeper();
    validation.setValidateGoal(null);
    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);
    // 每条用例都从「没有兜底定时器」开始：本文件要证明的正是「队列靠漏斗钩子前进，
    // 不是靠 60s 扫描」。定时器开着的话，验收 13 / 14 就算钩子漏了也可能蒙混过关。
    scheduler.stopSweeper();
  });

  // ── 夹具 ──────────────────────────────────────────────────────
  let seq = 0;
  /** 每个用例一家独立公司：唯一索引是按 company 约束的 */
  const nextCompany = () => `qacc-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, 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}`, 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;
  }

  /**
   * 「重新规划组织」落库后的样子：把这张图的节点整批换掉。
   *
   * 必须是真的删行 + 插行，改个 node_name 不算 —— `graph_run_nodes.def_node_id` 认的是
   * 节点**行 id**，只有换行才会造出「run 上的节点行指向已不存在的定义」这种孤儿，
   * 而那正是 `syncRunNodesToDef` 要修的东西。
   */
  function rewriteGraph(graphId, names) {
    ensureAgents(names);
    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 = ?`).run(graphId);
    const ids = names.map((name) => {
      const id = wdb.newId('wgn');
      db.prepare(`INSERT INTO work_graph_nodes (id, work_graph_id, node_name, system_agent_id, node_kind, created_at)
                  VALUES (?, ?, ?, ?, 'agent_call', ?)`).run(id, graphId, name, name, now);
      return id;
    });
    for (let i = 1; i < ids.length; i++) {
      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, ids[i - 1], ids[i], now);
    }
  }

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

  /**
   * 入队 —— 端点还没接上（`company.js` 里 `enqueueTaskRun` 目前零调用方），这里手工拼出
   * 它承诺的那三步：建 task 行 → 建 **`runStatus:'queued'`** 的 root run（不占坑）
   * → task 翻 queued + 挂 current_run_id。顺序与 `enqueueTaskRun` 逐字一致
   * （先 run 后 task：反过来的话中间窗口里的 tick 会把一次正常入队判成 enqueue_incomplete）。
   */
  function enqueue(companyId, goalId, { title, mode = 'auto', dependsOn = [], priority = 'P1', graphId, budget } = {}) {
    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',
      budget,
    });
    wdb.updateTask(task.id, {
      status: 'queued',
      currentRunId: run.id,
      queuedAt: new Date().toISOString(),
    });
    return { taskId: task.id, runId: run.id, graphId: run.graphId };
  }

  /**
   * 任务级重跑（§六.3 的 `POST /:cid/tasks/:taskId/rerun`，端点未落地时的等价三步）。
   * **新起一条 run**，旧 run 一个字段都不动 —— 旧 run 留在历史里可点开正是验收 8 的要求。
   */
  function rerun(companyId, taskRow, { graphId } = {}) {
    const run = wdb.createRun({
      graphKind: 'work',
      graphId: graphId || buildGraph(companyId),
      companyId,
      goalId: taskRow.goal_id,
      taskId: taskRow.id,
      executionMode: taskRow.execution_mode,
      runStatus: 'queued',
    });
    wdb.updateTask(taskRow.id, {
      status: 'queued',
      currentRunId: run.id,
      startedAt: null,
      finishedAt: null,
      blockedReason: null,
    });
    return run.id;
  }

  /**
   * 人在控制台上点「开始」—— `company.js:promoteQueuedTask` 的四步，逐字同序。
   * manual 任务只有这一条起跑路径（调度器**故意**不碰它们，§五.2bis）。
   */
  function humanStart(taskRow) {
    const runId = taskRow.current_run_id;
    wdb.syncRunNodesToDef(runId);
    if (!sse.transitionRun(runId, 'queued', 'ready', {}, { summary: '人工出队', task_id: taskRow.id })) return null;
    if (!wdb.claimQueuedTask(taskRow.id, runId)) return null;
    return engine.startRun(runId, { mode: taskRow.execution_mode, task: taskRow.description });
  }

  /**
   * 永不 settle 的 executor：让 run 稳稳停在 running，好观察队列。
   *
   * **故意不留放行钩子**。放行会让 driver 在用例结束后继续跑一轮 advanceOne，而那时
   * beforeEach 已经把 graph_run_nodes 清空了 —— 于是每条用例收尾都刷一行
   * 「auto 驱动异常: graph_run_nodes 不存在」。悬着的 promise 不会拖住进程退出，
   * 也不会再碰库，比放行干净。
   */
  function hangingExecutor() {
    const calls = [];
    engine.setAgentExecutor((ctx) => {
      calls.push(ctx.nodeName);
      return new Promise(() => {});
    });
    return calls;
  }

  /**
   * 等一个条件成立。**不用固定次数的 flush**：从「漏斗钩子」到「下一条真的 running」中间
   * 隔着 setImmediate + 一整条 async driver，拍脑袋数几拍出来的用例是不稳的。
   * 超时就把最后一次的实际值一起报出来，免得只看到一句 timeout。
   */
  async function waitFor(fn, label, timeoutMs = 2000) {
    const deadline = Date.now() + timeoutMs;
    let last;
    for (;;) {
      last = fn();
      if (last) return last;
      if (Date.now() > deadline) throw new Error(`等待超时：${label}（最后一次取到 ${JSON.stringify(last)}）`);
      await new Promise((r) => setImmediate(r));
    }
  }

  const taskStatus = (id) => wdb.getTask(id).status;
  const runStatus = (id) => wdb.getRun(id).run_status;

  // ──────────────────────────────────────────────────────────────
  // 批 4：唯一索引已 DROP —— 事务准入是唯一的守卫
  // ──────────────────────────────────────────────────────────────
  /**
   * 批 3 这一条钉的是「唯一索引的 WHERE 里没有 queued」（当时出队原子性靠它兜底）。
   * 批 4 把索引 DROP 了 —— 但要钉的事情**没变**：超发必须拦得住。所以这一条整体反转：
   *   · 索引必须**不存在**（`CREATE ... IF NOT EXISTS` 对存量库不生效，所以必须显式 DROP。
   *     它要是还在，max>1 的第二条准入会在事务放行之后吃一句 SQLite 原文的 500）；
   *   · 主守卫必须**真的**是主守卫，而且此刻就在拦超发（下面直接演示）。
   * 断言索引不在，而不只是断言 JS 常量没了：两者一旦分叉，只看常量的用例照样绿。
   */
  it('唯一索引已 DROP，且主守卫独自拦得住超发 —— 删兜底之后 max=1 的行为一个字不变', () => {
    const row = wdb.getDb()
      .prepare(`SELECT sql FROM sqlite_master WHERE type = 'index' AND name = 'idx_graph_runs_one_active_root'`)
      .get();
    expect(row).toBeUndefined();

    // 取值域仍是准入判据的唯一持有者（queued 不在里面 = 排队不占坑）
    expect(wdb.ACTIVE_ROOT_RUN_STATUS).toEqual(['ready', 'running', 'paused']);
    expect(wdb.ACTIVE_ROOT_RUN_STATUS).not.toContain('queued');
    // 没配 max_parallel_tasks 的公司回落 1 —— 存量行为不变的那条保证
    const companyId = nextCompany();
    expect(wdb.maxActiveRootRuns(companyId)).toBe(1);

    // 主守卫单独上阵：第一条进坑，第二条被事务拒（RunSlotBusyError，不是 SqliteError）
    const goalId = newGoal(companyId);
    const a = enqueue(companyId, goalId, { title: 'a' });
    const b = enqueue(companyId, goalId, { title: 'b' });
    expect(wdb.updateRunStatus(a.runId, 'queued', 'ready')).toBe(1);
    expect(() => wdb.updateRunStatus(b.runId, 'queued', 'ready')).toThrow(wdb.RunSlotBusyError);
    expect(wdb.getRun(b.runId).run_status).toBe('queued'); // 拒绝是干净的：一个字都没写
    expect(wdb.countActiveRootRuns(companyId)).toBe(1);
  });

  // ──────────────────────────────────────────────────────────────
  // 验收 3：第二条排队而不是 409
  // ──────────────────────────────────────────────────────────────
  it('验收 3：max=1 时第二条任务入队不报错（这就是 409 消失的地方），两条都躺在队里各占位次', () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    hangingExecutor();

    const a = enqueue(companyId, goalId, { title: 'a' });
    const b = enqueue(companyId, goalId, { title: 'b' });

    // 建两条 queued root run 本身就不该撞唯一索引 —— 旧代码正是在这一步 409 的
    expect(runStatus(a.runId)).toBe('queued');
    expect(runStatus(b.runId)).toBe('queued');
    expect(scheduler.queuePosition(companyId, a.taskId)).toBe(1);
    expect(scheduler.queuePosition(companyId, b.taskId)).toBe(2);

    scheduler.tick(companyId);
    expect(taskStatus(a.taskId)).toBe('running');
    expect(taskStatus(b.taskId)).toBe('queued');

    // 已经有一条在跑时**再入队第三条**：这才是「下达第二个任务」的真实时序。
    // 旧代码在这里抛 UNIQUE constraint → 路由翻 409；现在必须一声不响地排上。
    let c;
    expect(() => { c = enqueue(companyId, goalId, { title: 'c' }); }).not.toThrow();
    expect(runStatus(c.runId)).toBe('queued');
    expect(wdb.countActiveRootRuns(companyId)).toBe(1); // 排队的一条都不占坑

    // ⚠️ 位次口径：listQueue 只数 `status='queued'` 的行，跑着的那条不在队里。
    // 所以此刻 b 是第 1、c 是第 2 —— §八.3 写的「queuePosition=2」是把正在跑的那条也数了进去。
    // 这里钉的是**实现现在的口径**（端点回的就是这个数），口径分歧登记在返回结论里。
    expect(scheduler.queuePosition(companyId, b.taskId)).toBe(1);
    expect(scheduler.queuePosition(companyId, c.taskId)).toBe(2);
  });

  // ──────────────────────────────────────────────────────────────
  // 验收 2：依赖链自动前进
  // ──────────────────────────────────────────────────────────────
  it('验收 2：T2/T3 依赖 T1 → T1 跑时两条都 queued；T1 收敛后 T2 自动起跑，全程无人工干预', async () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    hangingExecutor();

    const t1 = enqueue(companyId, goalId, { title: 't1' });
    const t2 = enqueue(companyId, goalId, { title: 't2' });
    const t3 = enqueue(companyId, goalId, { title: 't3' });
    wdb.updateTask(t2.taskId, { dependsOn: [t1.taskId] });
    wdb.updateTask(t3.taskId, { dependsOn: [t1.taskId] });

    // 唯一的一次人工干预就是这一下（等价于用户点「下达」）；之后一次 tick 都不许再调
    scheduler.tick(companyId);
    expect(taskStatus(t1.taskId)).toBe('running');
    expect(taskStatus(t2.taskId)).toBe('queued');
    expect(taskStatus(t3.taskId)).toBe('queued');
    // 依赖未满足的照样在队里占位次（界面上要显示「排队 #1 / #2」），只是不够格出队
    expect(scheduler.queuePosition(companyId, t2.taskId)).toBe(1);
    expect(scheduler.queuePosition(companyId, t3.taskId)).toBe(2);

    // T1 正常收敛：推唯一漏斗，等价于 auto driver 跑完最后一个节点那一跳
    expect(sse.transitionRun(t1.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy();

    await waitFor(() => taskStatus(t2.taskId) === 'running', 'T1 收敛后 T2 自动起跑');
    expect(taskStatus(t1.taskId)).toBe('completed');
    expect(runStatus(t2.runId)).toBe('running');
    // max=1：T3 依赖也满足了，但坑只有一个，它继续排
    expect(taskStatus(t3.taskId)).toBe('queued');
    expect(scheduler.queuePosition(companyId, t3.taskId)).toBe(1);

    // 再收敛 T2 → T3 接着自己起来（依赖链一路走到底，中间没有人碰过）
    expect(sse.transitionRun(t2.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy();
    await waitFor(() => taskStatus(t3.taskId) === 'running', 'T2 收敛后 T3 自动起跑');
  });

  // ──────────────────────────────────────────────────────────────
  // 验收 7：取消排队中的任务
  // ──────────────────────────────────────────────────────────────
  it('验收 7：取消一个 queued task → run queued→stopped、task cancelled，后面那条顶上来', 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' });
    const c = enqueue(companyId, goalId, { title: 'c', priority: 'P2' });

    scheduler.tick(companyId);
    expect(taskStatus(a.taskId)).toBe('running');
    expect(scheduler.queuePosition(companyId, c.taskId)).toBe(2);

    // 取消 = `company.js:cancelQueuedRun` 那一跳（走漏斗，不裸写库）。
    // task 状态**不在这里写**：由漏斗钩子上的 task-settle 按 RUN_TO_TASK_STATUS 收成 cancelled，
    // 「取消排队」在库里 / 时间线上 / 界面上因此是同一件事。
    expect(sse.transitionRun(b.runId, 'queued', 'stopped', {}, { summary: '取消排队' })).toBeTruthy();
    expect(runStatus(b.runId)).toBe('stopped');
    await waitFor(() => taskStatus(b.taskId) === 'cancelled', '取消后 task 收成 cancelled');

    // 后面那条顶上来：位次 2 → 1
    expect(scheduler.queuePosition(companyId, c.taskId)).toBe(1);

    // A 让出坑之后，接班的是 C 而不是被取消的 B
    expect(sse.transitionRun(a.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy();
    await waitFor(() => taskStatus(c.taskId) === 'running', 'C 顶上来起跑');
    expect(runStatus(b.runId)).toBe('stopped'); // 被取消的那条一动不动
  });

  // ──────────────────────────────────────────────────────────────
  // 验收 8：任务级 rerun
  // ──────────────────────────────────────────────────────────────
  it('验收 8：failed 的 task rerun → attempt 累加、新起一条 run，旧 run 仍在历史里可点开', async () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    hangingExecutor();

    const a = enqueue(companyId, goalId, { title: 'a' });
    scheduler.tick(companyId);
    expect(wdb.getTask(a.taskId).attempt).toBe(1); // attempt 由 claimQueuedTask 在出队时 +1

    // 第一次跑失败（走漏斗 → task-settle 回写）
    expect(sse.transitionRun(a.runId, 'running', 'failed', {}, { summary: '用例直推终态' })).toBeTruthy();
    await waitFor(() => taskStatus(a.taskId) === 'failed', 'run 失败回写 task');

    const oldRun = wdb.getRun(a.runId);
    const oldNodeCount = wdb.getRunState(a.runId).nodes.length;
    expect(oldNodeCount).toBeGreaterThan(0);

    // rerun：新 run + task 回 queued
    const newRunId = rerun(companyId, wdb.getTask(a.taskId));
    expect(newRunId).not.toBe(a.runId);
    scheduler.tick(companyId);

    const fresh = wdb.getTask(a.taskId);
    expect(fresh.status).toBe('running');
    expect(fresh.current_run_id).toBe(newRunId);
    expect(fresh.attempt).toBe(2); // 第二代次
    expect(runStatus(newRunId)).toBe('running');

    // 旧 run 一个字段都没被动过 —— 「点开历史」靠的就是它还在、节点行还在
    const stillThere = wdb.getRun(a.runId);
    expect(stillThere.run_status).toBe('failed');
    expect(stillThere.created_at).toBe(oldRun.created_at);
    expect(wdb.getRunState(a.runId).nodes.length).toBe(oldNodeCount);
    // 两条 run 都挂在同一条 task 下，历史列表里能同时看到
    const roots = wdb.listCompanyRuns(companyId, { rootOnly: true }).filter((r) => r.task_id === a.taskId);
    expect(roots.map((r) => r.id).sort()).toEqual([a.runId, newRunId].sort());
    // 进度也是按 run 分别算的，rerun 不会把旧 run 的进度算到新 run 头上
    const prog = wdb.taskProgress([a.runId, newRunId]);
    expect(prog.has(a.runId)).toBe(true);
    expect(prog.has(newRunId)).toBe(true);
  });

  // ──────────────────────────────────────────────────────────────
  // 验收 9：进程重启后队列被捡起来
  // ──────────────────────────────────────────────────────────────
  it('验收 9：sweep 把「所有队里有人的公司」各推一次 —— 进程重启丢掉的唤醒靠它补回来', () => {
    const co1 = nextCompany();
    const co2 = nextCompany();
    hangingExecutor();

    const a = enqueue(co1, newGoal(co1), { title: 'a' });
    const b = enqueue(co2, newGoal(co2), { title: 'b' });

    // 模拟「重启后没有任何钩子会被触发」：谁都没调过 tick，两条就那么躺着
    expect(taskStatus(a.taskId)).toBe('queued');
    expect(taskStatus(b.taskId)).toBe('queued');

    const results = scheduler.sweep();
    // 跨公司各起各的：坑是按 company 算的
    expect(taskStatus(a.taskId)).toBe('running');
    expect(taskStatus(b.taskId)).toBe('running');
    expect(results.map((r) => r.companyId).sort()).toEqual([co1, co2].sort());
  });

  it('startSweeper 幂等且 unref —— 兜底定时器绝不该拖着进程不退出', () => {
    const t1 = scheduler.startSweeper(50);
    const t2 = scheduler.startSweeper(50);
    expect(t2).toBe(t1); // 重复调只留一个，否则每次热重载都多一个扫描器
    expect(typeof t1.hasRef).toBe('function');
    expect(t1.hasRef()).toBe(false);
    scheduler.stopSweeper();
    // 停完再起要能起来（不能因为 sweepTimer 没清干净就永远起不了第二个）
    const t3 = scheduler.startSweeper(50);
    expect(t3).not.toBe(t1);
    scheduler.stopSweeper();
  });

  // ──────────────────────────────────────────────────────────────
  // 验收 13 / 14：两条不走 settleRun 的释放路径
  // ──────────────────────────────────────────────────────────────
  it('验收 13：stopRun 不走 settleRun —— 队列靠 transitionRun 漏斗自己前进，不是等 60s 兜底', 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(taskStatus(a.taskId)).toBe('running');

    // beforeEach 里已经 stopSweeper —— 这条用例全程没有任何定时器，
    // 队列还能前进就只可能是漏斗钩子干的（这正是「不是等 60s」的证明方式）
    engine.stopRun(a.runId);

    await waitFor(() => taskStatus(b.taskId) === 'running', 'stopRun 后 B 自动起跑');
    expect(taskStatus(a.taskId)).toBe('cancelled');
    expect(runStatus(a.runId)).toBe('stopped');
  });

  it('验收 14：预算耗尽 → root run blocked → 下一条同样自动起跑（这条路径也不走 settleRun）', async () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    const calls = hangingExecutor();
    // 额度直接给 0：advanceOne 拾到第一个 ready 节点、真要花钱那一刻就判耗尽
    const a = enqueue(companyId, goalId, { title: 'a', priority: 'P0', budget: { total_agent_calls: 0, spent: {} } });
    const b = enqueue(companyId, goalId, { title: 'b', priority: 'P1' });

    scheduler.tick(companyId);

    await waitFor(() => runStatus(a.runId) === 'blocked', '预算耗尽 → root run blocked');
    // 一次 agent 都没起：预算闸门在发起调用之前
    expect(calls).toEqual([]);
    // blocked 不是失败，是挂起等人加预算 —— task 跟着是 blocked 而不是 failed
    expect(taskStatus(a.taskId)).toBe('blocked');

    await waitFor(() => taskStatus(b.taskId) === 'running', '预算耗尽让出坑后 B 自动起跑');
    // A 这种 blocked 不带 upstream_failed: 前缀，绝不会被下一次 tick 自动重排回 queued
    expect(scheduler.tick(companyId).requeued).toEqual([]);
    expect(taskStatus(a.taskId)).toBe('blocked');
  });

  // ──────────────────────────────────────────────────────────────
  // 验收 15：排队期间 replan
  // ──────────────────────────────────────────────────────────────
  it('验收 15：排队期间图被重新规划 → 出队后按新图跑，没有孤儿节点也不卡死', async () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    const calls = hangingExecutor();

    const graphId = buildGraph(companyId, ['old-entry']);
    const a = enqueue(companyId, goalId, { title: 'a', priority: 'P0' });   // 占着坑，让 b 真的排队
    const b = enqueue(companyId, goalId, { title: 'b', priority: 'P1', graphId });
    scheduler.tick(companyId);
    expect(taskStatus(b.taskId)).toBe('queued');
    // 入队那一刻的节点行快照
    expect(wdb.getRunState(b.runId).nodes.map((n) => n.name)).toEqual(['old-entry']);

    // 「重新规划组织」：整批换节点。定义是读时取最新，节点行还停在旧快照上。
    rewriteGraph(graphId, ['new-entry']);

    // 让出坑，b 自动出队 —— tick 内部必须先 syncRunNodesToDef 再出队，
    // 漏了这一句：run 上只有一条指向已删定义的孤儿节点行，seedEntryNodes 点不亮任何东西，
    // run 会一路 running 到 settleRun 判「没有活跃节点」，任务无声无息地假收敛。
    expect(sse.transitionRun(a.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy();
    await waitFor(() => taskStatus(b.taskId) === 'running', 'b 出队起跑');

    const names = wdb.getRunState(b.runId).nodes.map((n) => n.name);
    expect(names).toEqual(['new-entry']);          // 按新图跑，旧节点行已被删干净（无孤儿）
    await waitFor(() => calls.includes('new-entry'), '新图的入口节点真的被点亮并调用');
    expect(calls).not.toContain('old-entry');
    expect(runStatus(b.runId)).toBe('running');    // 没卡死：真的停在新节点上跑着
  });

  // ──────────────────────────────────────────────────────────────
  // 验收 16：manual 不堵队
  // ──────────────────────────────────────────────────────────────
  it('验收 16：manual 排在队头也不被自动拉起（不堵队），auto 照常起跑；人点「开始」才占坑', async () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    hangingExecutor();

    // manual 优先级更高、排在队头 —— 它要是被自动拉起，就会占着坑停在第一个节点上等人，
    // 后面所有任务无限期堵死，而队列界面看不出任何异常（§五.2bis 的原话）
    const m = enqueue(companyId, goalId, { title: 'm', mode: 'manual', priority: 'P0' });
    const a = enqueue(companyId, goalId, { title: 'a', mode: 'auto', priority: 'P1' });

    scheduler.tick(companyId);
    expect(taskStatus(m.taskId)).toBe('queued');   // 队头的 manual 被跳过
    expect(taskStatus(a.taskId)).toBe('running');  // 后面的 auto 照样起跑 = 没被堵住
    expect(scheduler.queuePosition(companyId, m.taskId)).toBe(1); // 但位次照占，界面要显示

    // auto 收敛让出坑，再 tick 多少次 manual 也不会自己起来
    expect(sse.transitionRun(a.runId, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy();
    await waitFor(() => wdb.countActiveRootRuns(companyId) === 0, 'auto 收敛让出坑');
    scheduler.tick(companyId);
    scheduler.tick(companyId);
    expect(taskStatus(m.taskId)).toBe('queued');

    // 人点「开始」才起跑、才占坑
    expect(humanStart(wdb.getTask(m.taskId))).toBeTruthy();
    expect(taskStatus(m.taskId)).toBe('running');
    expect(runStatus(m.runId)).toBe('running');
    expect(wdb.countActiveRootRuns(companyId)).toBe(1);
  });

  it('L1：人亲手起过的 manual 被退回队列后，坑一空自己接回来（不用再点一次）', async () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    hangingExecutor();

    // 人亲手起跑一条 manual —— 这一步就是「授权」，started_at 在这里被写上
    const m = enqueue(companyId, goalId, { title: 'm', mode: 'manual', priority: 'P0' });
    expect(humanStart(wdb.getTask(m.taskId))).toBeTruthy();
    expect(wdb.getTask(m.taskId).attempt).toBeGreaterThan(0);   // 授权的事实记在 attempt 上

    // 它撞上预算/挂起被退回队列（直接用数据层那条退路，等价于 admitOrRequeue 走到的分支）
    expect(sse.transitionRun(m.runId, 'running', 'blocked', {}, { summary: '用例：预算耗尽' })).toBeTruthy();
    expect(wdb.requeueBlockedRootRun(m.runId)).toBeTruthy();
    expect(taskStatus(m.taskId)).toBe('queued');
    expect(runStatus(m.runId)).toBe('queued');

    // 坑是空的 —— 批 3 那版这里会永远躺着（调度器 filter 只认 auto），人得**另外**再点一次
    expect(wdb.countActiveRootRuns(companyId)).toBe(0);
    scheduler.tick(companyId);

    // 现在它自己回来了：人刚才那次点击被兑现，而不是被吞掉
    expect(taskStatus(m.taskId)).toBe('running');
    expect(runStatus(m.runId)).toBe('running');
  });

  it('L1 反面：从没被人起过的 manual 仍然不许自动出队（首次起跑的授权不能省）', () => {
    const companyId = nextCompany();
    const goalId = newGoal(companyId);
    hangingExecutor();
    const m = enqueue(companyId, goalId, { title: 'm', mode: 'manual', priority: 'P0' });
    expect(wdb.getTask(m.taskId).attempt).toBe(0);              // 从没被人起过
    scheduler.tick(companyId);
    scheduler.tick(companyId);
    expect(taskStatus(m.taskId)).toBe('queued');
    expect(wdb.countActiveRootRuns(companyId)).toBe(0);
  });

  // ──────────────────────────────────────────────────────────────
  // 验收 23：blocked 是推导出来的，不是落死的
  // ──────────────────────────────────────────────────────────────
  it('验收 23：上游 failed → 下游 blocked:upstream_failed；上游 rerun 成功后下游自己回 queued 并起跑', 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' });
    wdb.updateTask(b.taskId, { dependsOn: [a.taskId] });

    scheduler.tick(companyId);
    expect(taskStatus(a.taskId)).toBe('running');

    // A 真的失败（走漏斗：settle 回写 + 释放坑各一次，次序由 afterRunTransition 保证）
    expect(sse.transitionRun(a.runId, 'running', 'failed', {}, { summary: '用例直推终态' })).toBeTruthy();
    await waitFor(() => taskStatus(b.taskId) === 'blocked', '上游失败后下游被推导成 blocked');
    expect(wdb.getTask(b.taskId).blocked_reason).toBe(`upstream_failed:${a.taskId}`);
    // 关键：B **不是**躺在队里假装排队 —— 它已经不在队列里了，界面上是标红的 blocked
    expect(scheduler.queuePosition(companyId, b.taskId)).toBe(0);

    // 重跑上游（§五.1.2 给的两条出路之一）
    const aRun2 = rerun(companyId, wdb.getTask(a.taskId));
    scheduler.tick(companyId);
    expect(taskStatus(a.taskId)).toBe('running');
    // ⚠️ 这一步的口径容易想反：B 在**上游刚被重跑**时就回 queued 了，不用等 A 跑成功。
    // 因为「回 queued」的判据是 `deadDependency === null`（上游不再处于 failed/cancelled/blocked），
    // 而不是「上游已 completed」—— 后者是**够不够格出队**的判据（dependenciesSatisfied），两者正交。
    // 分开是对的：A 一重跑，B 在界面上就该从标红的 blocked 变回「排队中，等 A」，
    // 而不是等 A 跑完几分钟里一直标红说「上游失败了」。
    expect(taskStatus(b.taskId)).toBe('queued');
    expect(wdb.getTask(b.taskId).blocked_reason).toBeNull();
    // 回了队列但**不许被起跑**：坑此刻被 A 占着，且 A 还没 completed
    expect(runStatus(b.runId)).toBe('queued');

    expect(sse.transitionRun(aRun2, 'running', 'completed', {}, { summary: '用例直推终态' })).toBeTruthy();
    // 上游恢复 → B 自己回 queued 并被同一轮 tick 拉起来。
    // 这一整段就是「blocked 每次 tick 重算、不落成不可逆状态」的兑现证明：
    // 没有 requeueDerivedBlockedTask 的话，B 会永远躺在 blocked 里。
    await waitFor(() => taskStatus(b.taskId) === 'running', '上游 rerun 成功后下游自动起跑');
    expect(wdb.getTask(b.taskId).blocked_reason).toBeNull();
    expect(runStatus(b.runId)).toBe('running');
  });
});