宜博 / 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 条，枚举一次漏一次）。批 4 DROP 掉了 `idx_graph_runs_one_active_root`
 *      那道 DB 兜底，那个事务从此是**唯一**的守卫；
 *   2. **触发不靠枚举** —— 挂在 `graph-run-sse.transitionRun` 这个唯一漏斗上
 *      （`stopRun` 与「预算耗尽 → blocked」都不走 `settleRun`，枚举一次漏一次）；
 *   3. **躺在队里的任务不许假装在排队** —— 依赖已不可能满足的，就地推导成 blocked 并写明原因。
 *
 * 本模块不调 LLM、不花 token，所以没有记账路径（§九「花掉的 token 即使失败也要记账」在这里无对象）。
 */

/**
 * 并行度上限的唯一持有者在 `work-graph-db.maxActiveRootRuns(companyId)`（挨着准入事务）。
 * 这里**不重新写一个默认值、也不缓存** —— 两个持有者迟早分叉，那正是 `canDispatch`
 * 那次事故的形状。批 4 起它是 per-company 的（读 company.json.max_parallel_tasks），
 * 所以签名必须带 companyId：一个无参的 getter 表达不了 per-company，只会变成又一个假上限。
 * 惰性取值（模块加载期 require work-graph-db 会成环）。
 */
const maxParallelTasks = (companyId) => wdb().maxActiveRootRuns(companyId);

/**
 * 依赖「已经死了」的上游状态。
 * `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 的：它们在界面上同样占位次）
 */
// 批 5：只认 executor='graph'。agent 派单（executor='agent'）虽然也进 tasks 表，
// 但它的执行载体是「直接起一次 agent 对话」，由 /task/dispatch 自己触发 ——
// 进表只为了可见、可查、可串成派单树，**不该被调度器出队**。
// 漏了这一条的后果是静默的：调度器会拿一条根本没有 run 行的 task 去 syncRunNodesToDef，
// 5a 判空 → abandonRun → 把一条正在被 agent 执行的派单收成 cancelled。
function listQueue(companyId) {
  return wdb()
    .getDb()
    .prepare(`
      SELECT * FROM tasks
       WHERE company_id = @companyId AND status = 'queued' AND archived_at IS NULL
         AND executor = 'graph'
       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++) {
    // 同上：依赖推导也只管图任务。派单任务没有 depends_on 语义（它的先后靠 parent_task_id 表达），
    // 放进来只会让它被判成「依赖不可能满足」而无辜转 blocked。
    const rows = listCompanyTasks(companyId)
      .filter((t) => t.executor === 'graph' && (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, max?: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 当成「这轮没抢到」。
  //    批 4 起 max 是 per-company 的（company.json.max_parallel_tasks，缺省 1）。
  const max = maxParallelTasks(companyId);
  const active = activeRootRunCount(companyId);
  const started = [];
  if (active >= max) return { companyId, skipped: 'no-slot', active, max, started, blocked, requeued };

  // 4. 选候选：queued + 依赖全 completed + auto。
  //    manual **不参与自动出队**（§五.2bis）：startRun(manual) 只点亮入口节点就停下等 /next，
  //    而那时 run 已经是 running、占着坑 —— 没人点「下一步」它就把后面所有任务无限期堵死，
  //    而队列界面上看不出任何异常。manual 只能由人在控制台上显式点「开始」。
  //
  //    **不 slice**：批 3 这里是 `.slice(0, max - active)`，用的是「候选数 = 能起的条数」这个
  //    在 max=1 时看不出错的假设 —— 只要队首那条起跑失败（5a 缺 run 行 / 5d 认领失败），
  //    这一轮就白跑，空着的坑要等下一次钩子或 60s sweep 才被填上。max>1 时这类空转会成倍出现。
  //    改成按**真的起来了几条**收敛：循环里自己数，坑满就 break。
  //    ⚠️ **但 `started_at` 非空的 manual 要放进来**（批 3 复核 L1）：
  //    「manual 不自动出队」约束的是**首次起跑的授权** —— 人要盯着一步步推，不该半夜被自动拉起。
  //    而一条**已经被人亲手启动过**、跑到一半因为坑满被 `admitOrRequeue` 退回队列的 manual 任务，
  //    授权早就给过了；让它自己接回来不是新授权，是**把人刚才那次点击兑现掉**。
  //    不放进来的后果实测过：人点「启动」→ 退回队列 → 坑空了 → **没人接**，任务躺着、
  //    sweep 每 60s 空转，而 `graph-engine.js` 那句「『T1 永远回不来』这个后果因此消失」
  //    对 manual 根本不成立 —— 又一条只兑现一半的注释。
  //    判据用 **`attempt > 0`** 而不是 `started_at`：后者看着更贴切，但
  //    `requeueBlockedRootRun` 退回队列时会把 `started_at` 清成 NULL（它记的是「本次尝试何时开跑」，
  //    退回队列后本次尝试就没了），于是那个字段恰好在我需要它的那一刻是空的 —— 实测踩过。
  //    `attempt` 由 `claimQueuedTask` 每次认领 +1、退回队列时不清，它才是「这条被人起过没有」的事实。
  const candidates = listQueue(companyId)
    .filter((t) => (t.execution_mode === 'auto' || t.attempt > 0) && dependenciesSatisfied(t));

  // 5. 逐条出队，每条独立 try/catch —— 一条失败绝不能中止整轮（那等于把后面的坑白白空着）
  for (const t of candidates) {
    // 本轮已经填满就收工。`active` 是进循环前的快照、`started.length` 是本轮的战果，
    // 两者都只会低估空位（跑着的 run 可能中途收敛让出坑），保守一点即可：
    // 少起的那条由漏斗钩子上的下一次 scheduleTick 接手，而不是在这里赌一个更新的计数。
    if (active + started.length >= max) break;
    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'` 在活跃集里，所以它不会反手再触发一次调度。
      //     两种「没抢到」的形态（批 4 DROP 唯一索引后，兜底那一层没有了）：
      //       · changes=0            → transitionRun 返回 null（这条 run 已被别人推走）→ 试下一条；
      //       · RunSlotBusyError     → 主守卫在事务里判出坑满（**不是** changes=0）→ 整轮到此为止。
      //     抛出来的那种落到下面的 catch。
      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) {
      // 坑满：这一刻公司真的没位置了，后面的候选一个都进不去 —— 继续试只是白白多抛几次。
      // 与下面那种「这一条没抢到」区分开：那种要接着试下一条，否则 max>1 时一条被别人
      // 推走就会让剩下的坑空着（正是本轮 slice 去掉的那类空转）。
      if (err instanceof wdb().RunSlotBusyError) break;
      if (isRaceError(err)) continue; // 这一条没抢到（已被别人从 queued 推走），试下一条
      // 到这儿说明 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, max, 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 executor = 'graph'
           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,
  // 批 4 删掉了 `get MAX_PARALLEL_TASKS()`：它无参，表达不了 per-company 的额度，
  // 而且全仓零调用方 —— 一个「导出了、没人用、还恰好是错的」的形状。
  // 要额度就调 `work-graph-db.maxActiveRootRuns(companyId)`，那是唯一持有者。
};