宜博 / lemonboss

/**
 * Task Scheduler — 公司级任务队列的出队器（27-P3C 批 3，§五.1）
 *
 * 一个函数就是全部：`tick(companyId)` —— 幂等、可重入、同步（better-sqlite3 是同步 API）。
 *
 * 它把 v1 那条硬约束从「拒绝」改成「排队」：
 *   下达任务时不再因为「公司已有 active root run」吃 409，而是建一条 `run_status='queued'`
 *   的 run（不占唯一索引的坑）+ 一条 `status='queued'` 的 task；坑一空，这里把它拉起来。
 *
 * 三件事必须一起成立，少一件排队就是「静默丢任务」（正是当初否决排队的那条理由）：
 *   1. **出队是原子的** —— `updateRunStatus(runId,'queued','ready')` 一条 CAS，且「数坑 + 占坑」
 *      整段包在 `BEGIN IMMEDIATE` 里（守卫在 `updateRunStatus`，不在这里 —— 能把 root run
 *      推进活跃集的路径有 6 条，枚举一次漏一次）；`idx_graph_runs_one_active_root` 留作兜底；
 *   2. **触发不靠枚举** —— 挂在 `graph-run-sse.transitionRun` 这个唯一漏斗上
 *      （`stopRun` 与「预算耗尽 → blocked」都不走 `settleRun`，枚举一次漏一次）；
 *   3. **躺在队里的任务不许假装在排队** —— 依赖已不可能满足的，就地推导成 blocked 并写明原因。
 *
 * 本模块不调 LLM、不花 token，所以没有记账路径（§九「花掉的 token 即使失败也要记账」在这里无对象）。
 */

/**
 * 并行度上限的唯一持有者在 `work-graph-db.MAX_ACTIVE_ROOT_RUNS`（挨着唯一索引与准入事务）。
 * 这里**不重新写一个 1** —— 两个持有者迟早分叉，那正是 `canDispatch` 那次事故的形状。
 * 惰性取值（模块加载期 require work-graph-db 会成环）。
 */
const maxParallelTasks = () => wdb().MAX_ACTIVE_ROOT_RUNS;

/**
 * 依赖「已经死了」的上游状态。
 * `blocked` 也算 —— A 失败 → B blocked → 依赖 B 的 C 同样永远等不到 completed，
 * 级联要一次算到底（§五.1.2）。而 blocked 本身是**每次 tick 重算**出来的、不落死，
 * 所以 A 重跑成功后 B 回 queued、下一轮 C 也跟着回 queued，级联能原路解开。
 */
const DEAD_DEP_STATUS = new Set(['failed', 'cancelled', 'blocked']);

/** 兜底扫描周期：前面的钩子全漏了也能在一分钟内自愈 */
const SWEEP_INTERVAL_MS = 60 * 1000;

/**
 * per-company 重入闸门。
 *
 * **真实语义：只在本进程内有效**，它是一个模块级 Set，不是锁、不跨进程、也不跨 worker。
 * 别把它当并发正确性的防线 —— 两个进程同时 tick 同一家公司时它一个人都挡不住。
 * 跨进程的防线只有两道，都在 DB 里：`updateRunStatus` 的 `BEGIN IMMEDIATE` 准入事务
 * （数坑 + 占坑一步做完）和 `claimQueuedTask` 的条件更新（CAS）。这个 Set 只是省掉
 * 同进程内那部分注定要失败的 CAS，纯属省事。
 *
 * 为什么还是必须有：tick 会起 run → 起 run 走 `transitionRun` → 那里的钩子又调 tick。
 * `scheduleTick` 的 `setImmediate` 只保证「出栈后再跑」，而 `startRun(auto)` 的驱动循环
 * 有一段是**同步**跑完的（advanceOne 在挑到节点之前没有 await），一条零入口节点的 run
 * 会在 startRun 返回之前就转 completed → 同步回到钩子。钩子里也走 setImmediate，
 * 于是不会栈溢出；这个 Set 挡的是另一半：同一家公司的两次 tick 交叠着各出一条队。
 */
const inFlight = new Set();

let sweepTimer = null;

function wdb() {
  // 惰性 require：work-graph-db → graph-run-sse → 本模块，顶层 require 会成环
  return require('./work-graph-db');
}
function engine() {
  return require('./graph-engine');
}
function sse() {
  return require('./graph-run-sse');
}

/** depends_on 是 JSON 数组文本，坏数据一律当「没有依赖」（与 company.js 的 blockersOf 同口径） */
function dependsOn(task) {
  try {
    const v = JSON.parse(task.depends_on || '[]');
    return Array.isArray(v) ? v.map(String) : [];
  } catch {
    return [];
  }
}

/**
 * 找出这条 task 身上第一个「已经不可能变成 completed」的上游。
 * **查不到的依赖也算死的**：依赖的 task 被归档/删号时宁可挡下也不要放行 —— 放行等于
 * 把「上游产物已就绪」这个前提凭空当真（company.js 的 blockersOf 是同一条理由）。
 * @returns {string|null} 死掉的那个上游 id
 */
function deadDependency(task) {
  for (const id of dependsOn(task)) {
    const dep = wdb().getTask(id);
    if (!dep || DEAD_DEP_STATUS.has(dep.status)) return id;
  }
  return null;
}

/** 依赖全部 completed 才够格出队（§五.1 第 4 步）。查不到的一律不算数。 */
function dependenciesSatisfied(task) {
  return dependsOn(task).every((id) => (wdb().getTask(id) || {}).status === 'completed');
}

/**
 * 队列的**排序口径持有者**：`priority → seq → created_at`（27-P3C §六）。
 *
 * 全仓库只有这一处排队列。`queuePosition` 就是这个列表里的下标 + 1，端点必须调它来算，
 * 不许自己再排一遍 —— `canDispatch` 那次事故就是取值域被第二个持有者复述了一遍。
 * `listCompanyTasks` / `listGoalTasks` 的排序都**故意**不是这个口径（那是显示顺序），
 * 它们的注释里写着「要位次找调度器要」。
 *
 * ponytail: priority 是 TEXT（'P0'|'P1'|…），这里靠字典序恰好与优先级同向。
 * 一旦出现 'P10' 就会排到 'P2' 前面 —— 那时把 priority 换成整数列，或在这里加一张映射表。
 *
 * @returns {object[]} 该公司全部 queued task（含 manual 的：它们在界面上同样占位次）
 */
function listQueue(companyId) {
  return wdb()
    .getDb()
    .prepare(`
      SELECT * FROM tasks
       WHERE company_id = @companyId AND status = 'queued' AND archived_at IS NULL
       ORDER BY priority, seq, created_at
    `)
    .all({ companyId });
}

/** 这条 task 排在第几位（1-based）；不在队里返回 0 */
function queuePosition(companyId, taskId) {
  return listQueue(companyId).findIndex((t) => t.id === taskId) + 1;
}

/**
 * 当前占着坑的 root run 数。
 *
 * 计数 SQL 本身在 `work-graph-db.countActiveRootRuns`（与唯一索引、与准入事务同源），
 * 这里只是取来用 —— 调度器这一份**只是给界面看的快照**，不是准入判据：
 * 真正决胜的是 `updateRunStatus` 里那个 BEGIN IMMEDIATE 的事务（见下方第 3 步注释）。
 */
function activeRootRunCount(companyId) {
  return wdb().countActiveRootRuns(companyId);
}

/**
 * 依赖推导（§五.1.2）：queued ⇄ blocked 双向重算。
 *
 * **两个方向都必须做**。只做 queued → blocked 那一半的话，「blocked 只是推导结果、不落成
 * 不可逆状态」就成了空头支票：上游重跑成功后下游永远解不开。
 * 反向那一跳的闸门在 `requeueDerivedBlockedTask` 里 —— 只有 `blocked_reason` 以
 * `upstream_failed:` 开头的才是本函数标的，预算耗尽那种 blocked 绝不能被自动重排。
 *
 * 跑到不动点为止（A→B→C 的级联要一次算到底）。
 * ponytail: 最坏 O(n²) 次 getTask，n 是一家公司的在途任务数（个位到几十），够用；
 * 真到几千条时改成一次拉全表在内存里做拓扑排序。
 */
function deriveQueueBlocks(companyId) {
  const { listCompanyTasks, blockQueuedTask, requeueDerivedBlockedTask, UPSTREAM_BLOCK_PREFIX } = wdb();
  const blocked = [];
  const requeued = [];

  for (let pass = 0; ; pass++) {
    const rows = listCompanyTasks(companyId).filter((t) => t.status === 'queued' || t.status === 'blocked');
    let changed = 0;
    for (const t of rows) {
      const dead = deadDependency(t);
      if (t.status === 'queued' && dead) {
        if (blockQueuedTask(t.id, `${UPSTREAM_BLOCK_PREFIX}${dead}`)) {
          blocked.push({ taskId: t.id, upstream: dead });
          changed++;
        }
      } else if (t.status === 'blocked' && !dead) {
        if (requeueDerivedBlockedTask(t.id)) {
          requeued.push(t.id);
          changed++;
        }
      }
    }
    // pass 上限 = 参与推导的行数：级联最长就是一条链，多一轮都是白跑
    if (!changed || pass >= rows.length) break;
  }
  return { blocked, requeued };
}

/**
 * 出队竞争的正常结局，不是错误 —— 认错了就会把一次正常的竞争当成故障去回滚
 * （`abandonRun` 把 run 收成 stopped、task 收成 cancelled，那是丢任务）。
 *
 * **判据本身不在这里**：唯一持有者是 `work-graph-db.isSlotRaceError`（挨着它判的那两个
 * 错误类型）。这里以前自己写了一份，按 `/UNIQUE constraint failed/i` 嗅报错文本，而
 * `routes/company.js#isSlotRace` 那份按 `err.code` —— 同一件事两个判据，且已经分叉：
 * 谁顺手改一句报错文案，调度器这边就会把「没抢到」当成故障丢任务，路由那边照常 409。
 */
function isRaceError(err) {
  return wdb().isSlotRaceError(err);
}

/**
 * 出队失败后的收尾：把这条 run 收回 stopped。
 *
 * 复用 `engine.stopRun` 而不是自己写一遍状态转换：它顺带把没开跑的节点批量 skipped、
 * 落 timeline、并经由漏斗钩子让 `task-settle` 把 task 收成 cancelled —— 于是「起不来」
 * 在界面上是一条有终态、有时间线的任务，而不是一条标着 running 却没有活 run 的幽灵。
 */
function abandonRun(runId, why) {
  console.error(`[task-scheduler] 出队回滚 run=${runId}: ${why}`);
  try {
    engine().stopRun(runId);
  } catch (err) {
    console.error(`[task-scheduler] 回滚 run=${runId} 也失败了: ${err.message}`);
  }
}

/**
 * 调度一次（§五.1 六步）。**绝不 throw**：它挂在状态转换的尾巴上，抛出去会把一次正常的
 * run 收敛也带崩（调用方拿到异常，但状态和 SSE 事件已经发出去了）。
 *
 * @returns {{companyId:string, skipped?:string, started:object[], blocked:object[], requeued:string[], active?:number}}
 */
function tick(companyId) {
  if (!companyId) return { companyId, skipped: 'no-company', started: [], blocked: [], requeued: [] };
  if (inFlight.has(companyId)) {
    // 重入不是错误：外层那次 tick 还在跑。注意它**不保证**代跑这一轮 —— 外层可能已经过了
    // 第 4 步的候选快照，看不到刚空出来的坑。真正兜住这一格的是漏斗钩子上的下一次
    // scheduleTick，以及 60s 的 sweep（见 sweep 的取值域注释）。
    return { companyId, skipped: 'reentrant', started: [], blocked: [], requeued: [] };
  }
  inFlight.add(companyId);
  try {
    return tickInner(companyId);
  } catch (err) {
    console.error(`[task-scheduler] tick 异常 company=${companyId}: ${err.stack || err.message}`);
    return { companyId, skipped: 'error', error: err.message, started: [], blocked: [], requeued: [] };
  } finally {
    inFlight.delete(companyId);
  }
}

function tickInner(companyId) {
  const { getRun, syncRunNodesToDef, claimQueuedTask, blockQueuedTask } = wdb();

  // 1-2. 先做依赖推导：躺在队里等一个永远不会 completed 的上游，界面上和正常排队一模一样。
  //      放在选候选之前，这样被推导成 blocked 的任务这一轮就不会被当成候选。
  const { blocked, requeued } = deriveQueueBlocks(companyId);

  // 3. 坑满就不动队伍。**这一句只是省事，不是准入判据**：它读完到下面第 5c 步真正出队之间
  //    队列可以变（另一个进程/另一次 tick），check-then-act 在这里天然不成立。
  //    权威判据在 `updateRunStatus` 的准入事务里（BEGIN IMMEDIATE 内数坑 + 占坑一步做完），
  //    坑被抢走时它抛 RunSlotBusyError，由 5c 的 catch + isRaceError 当成「这轮没抢到」。
  //    ponytail: max 恒为 1（批 3 只做排队）；读 company.json.max_parallel_tasks 与
  //    DROP 唯一索引一起归批 4 / 27-P4。
  const max = maxParallelTasks();
  const active = activeRootRunCount(companyId);
  const started = [];
  if (active >= max) return { companyId, skipped: 'no-slot', active, started, blocked, requeued };

  // 4. 选候选：queued + 依赖全 completed + auto。
  //    manual **不参与自动出队**（§五.2bis）：startRun(manual) 只点亮入口节点就停下等 /next，
  //    而那时 run 已经是 running、占着坑 —— 没人点「下一步」它就把后面所有任务无限期堵死，
  //    而队列界面上看不出任何异常。manual 只能由人在控制台上显式点「开始」。
  const candidates = listQueue(companyId)
    .filter((t) => t.execution_mode === 'auto' && dependenciesSatisfied(t))
    .slice(0, max - active);

  // 5. 逐条出队，整条包在 try/catch 里
  for (const t of candidates) {
    const runId = t.current_run_id;
    try {
      // 5a. 入队时必须已经建好 queued 的 run 行；没有就是入队做了一半，
      //     这条任务永远起不来。标成 blocked（不带 upstream_failed 前缀 → 不会被自动重排）
      //     让人看见，而不是让它继续假装在排队。
      if (!runId || !getRun(runId)) {
        if (blockQueuedTask(t.id, `enqueue_incomplete:missing_run${runId ? `:${runId}` : ''}`)) {
          blocked.push({ taskId: t.id, upstream: null, reason: 'missing_run' });
        }
        console.error(`[task-scheduler] task=${t.id} 是 queued 但找不到 run（current_run_id=${runId}）`);
        continue;
      }

      // 5b. 排队期间 org graph 可能被重新规划过：节点行是入队那一刻的快照，定义是读时取最新。
      //     不对齐的话新增节点没有 run 行，出队后永远点不亮，而且全程无报错（§四.3 末）。
      syncRunNodesToDef(runId);

      // 5c. 原子出队：`WHERE id=? AND run_status='queued'` 的条件更新，整段（数坑 + 占坑）
      //     由 updateRunStatus 的 BEGIN IMMEDIATE 事务保护，超发的那条拿不到坑。
      //     走 transitionRun 而不是裸的 updateRunStatus，是为了让这一跳同样落 timeline、
      //     推 SSE（控制台要看见任务起跑）；`to='ready'` 在活跃集里，所以它不会反手再触发一次调度。
      //     三种「没抢到」的形态：
      //       · changes=0            → transitionRun 返回 null（这条 run 已被别人推走）；
      //       · RunSlotBusyError     → 主守卫在事务里判出坑满（**不是** changes=0）；
      //       · UNIQUE constraint    → 兜底索引（批 4 DROP 前一直在），
      //     后两种都是抛出来的，落到下面的 catch → isRaceError → continue。
      if (!sse().transitionRun(runId, 'queued', 'ready', {}, {
        summary: `出队：queued → ready（task ${t.id}）`,
        task_id: t.id,
      })) continue;

      // 5d. 认领 task。抢不到（被取消 / 被别的 tick 抢走 / 已归档）就把 run 收回去，
      //     否则会留下一条 ready 却没有任务的 run 白占着坑。
      if (!claimQueuedTask(t.id, runId)) {
        abandonRun(runId, `task=${t.id} 认领失败（已不是 queued 或 current_run_id 已改指）`);
        continue;
      }

      // 5e. 起跑。mode 恒为 auto（候选已过滤），description 从 task 行读回 ——
      //     排队之后请求闭包早就不在了（§四.7 禁令二说的就是这件事）。
      //     startRunSync 内部会回读确认真的转成了 running，没起来会抛，落到下面的 catch。
      engine().startRun(runId, { mode: t.execution_mode, task: t.description });
      started.push({ taskId: t.id, runId, goalId: t.goal_id });
    } catch (err) {
      if (isRaceError(err)) continue; // 没抢到，正常
      // 到这儿说明 run 可能已经出队但没能起跑：task 已标 running 而没有活的 run，
      // 这正是要消灭的那类「写了但没生效还不报错」。收回 run，让 task 落到可见的终态。
      if (runId && (getRun(runId) || {}).run_status === 'ready') {
        abandonRun(runId, `startRun 失败: ${err.message}`);
      } else {
        console.error(`[task-scheduler] 出队 task=${t.id} 失败: ${err.stack || err.message}`);
      }
    }
  }

  return { companyId, active, started, blocked, requeued };
}

/**
 * 异步触发。**漏斗钩子必须用这个而不是 tick**：出栈后再跑，避免在一次状态转换的栈里
 * 反手又转一次状态（tick 会起 run → 起 run 转状态 → 又回到钩子）。
 */
function scheduleTick(companyId) {
  if (!companyId) return;
  setImmediate(() => tick(companyId));
}

/**
 * 兜底扫描：所有「tick 还有事可做」的公司各 tick 一次。前面的钩子全漏了也能自愈。
 *
 * **取值域必须覆盖 tick 承诺要治的全部状态**，不只是 queued。
 * `deriveQueueBlocks` 有两个方向，反向那一半（`blocked(upstream_failed:*) → queued`）
 * 只发生在 tick 里，而这类公司**可能一条 queued 任务都没有**：
 *   A failed → B blocked:upstream_failed:A（B 是唯一在途任务）→ 进程重启
 *   → 只扫 queued 的话这家公司再也不会被 tick → 即使 A 后来被 rerun 成功，B 也永远 blocked。
 * 那样一来「blocked 是每次 tick 重算出来的、不落死」（见 DEAD_DEP_STATUS 注释）就没有兜底层，
 * 唯一触发是进程内那次 setImmediate —— 注释承诺的防线在重启后不存在。
 *
 * 前缀口径不在这里复述：跟 `requeueDerivedBlockedTask` 的 WHERE 取同一个
 * `UPSTREAM_BLOCK_PREFIX`，两个持有者迟早分叉。别的来源的 blocked（预算耗尽等）
 * 本来就不该被自动重排，也就没必要为它们唤醒 tick。
 *
 * 这也是**启动时那次全量扫描**的唯一实现（`server/index.js` 启动钩子直接调 sweep），
 * 所以启动口径与定时口径天然同源，不会一边宽一边窄。
 */
function sweep() {
  try {
    const rows = wdb()
      .getDb()
      .prepare(`
        SELECT DISTINCT company_id FROM tasks
         WHERE archived_at IS NULL
           AND (status = 'queued'
                OR (status = 'blocked' AND blocked_reason LIKE @prefix))
      `)
      .all({ prefix: `${wdb().UPSTREAM_BLOCK_PREFIX}%` });
    return rows.map((r) => tick(r.company_id));
  } catch (err) {
    console.error(`[task-scheduler] sweep 异常: ${err.message}`);
    return [];
  }
}

/**
 * 起 60s 兜底定时器。`unref()` —— 它绝不该拖着进程不退出。
 * 幂等：重复调只留一个。
 */
function startSweeper(intervalMs = SWEEP_INTERVAL_MS) {
  if (sweepTimer) return sweepTimer;
  sweepTimer = setInterval(sweep, intervalMs);
  if (typeof sweepTimer.unref === 'function') sweepTimer.unref();
  return sweepTimer;
}

function stopSweeper() {
  if (sweepTimer) clearInterval(sweepTimer);
  sweepTimer = null;
}

module.exports = {
  tick,
  scheduleTick,
  listQueue,
  queuePosition,
  sweep,
  startSweeper,
  stopSweeper,
  // 转发 work-graph-db 的那一份，别在这里存第二个副本
  get MAX_PARALLEL_TASKS() { return maxParallelTasks(); },
};