scheduler-gateway/src/scheduler-core.js

/**
 * 调度层（H3/H7 + 加分：能力感知调度）。
 * - nextDispatchable：优先级最高、依赖已满足、过审批门、有匹配节点的任务
 * - pickNode：能力匹配 + 在线 + 有余量并发 + 负载最低（inFlight/容量、成本）；
 *   执行失败熔断：连续失败的节点短期冷却不接新活（全部冷却时回退，防饿死）
 * - result：成功回写；失败重试→重投其它节点；超限进死信
 */
import { TASK_STATE } from "./protocol.js";

// ---- 执行失败熔断（circuit breaker）----
// 背景：本机 codex/gemini CLI 100% 执行失败但传输在线，旧调度只看负载不看成败，
// 导致无能力约束的任务把有限重试次数全部烧在坏节点上（历史教训见死信 task-mt9jxqnt-dm5ne8）。
// GW_BREAKER=0 整体关闭；GW_FAIL_STREAK=0 也关闭；冷却期满自动恢复。
const BREAKER_ON = process.env.GW_BREAKER !== "0";
const BR_STREAK = Math.max(0, Number(process.env.GW_FAIL_STREAK ?? 1));
const BR_COOL = Math.max(0, Number(process.env.GW_COOLDOWN_MS ?? 60_000));
// 连续失败时冷却指数升级（×2/次），封顶 BR_MAX：对确定性坏执行器，
// 固定短冷却会在「其它节点执行耗时 > 冷却期」时过期回流，再次烧掉宝贵重试次数
const BR_MAX = Math.max(BR_COOL, Number(process.env.GW_COOLDOWN_MAX_MS ?? 600_000));

export function breakerActive() {
  return BREAKER_ON && BR_STREAK > 0 && BR_COOL > 0;
}

/** 节点是否处于熔断冷却期。 */
export function cooling(n, now = Date.now()) {
  return breakerActive() && (n.coolUntil || 0) > now;
}

export function depsReady(task, tasks) {
  return task.dependencies.every((d) => tasks.get(d)?.state === TASK_STATE.DONE);
}

export function nodeMatches(n, caps) {
  if (!caps || caps.length === 0) return true;
  return caps.every((c) => n.capabilities.includes(c));
}

/** 对候选节点统一排序：冷却剔除（全冷却回退）→ 成熟优先 → 负载 → 连击 → 完成 → 成本。
 *  pickNode 与 server._pickAmong 共用此实现——历史上两处各写一份导致行为漂移。 */
export function rankCandidates(candidates, { now = Date.now(), provenFirst = false } = {}) {
  let pool = candidates.filter((n) => !cooling(n, now));
  if (pool.length === 0 && candidates.length > 0) pool = candidates; // 全冷却回退，防饿死
  if (provenFirst) {
    const proven = pool.filter((n) => (n.completed || 0) > 0);
    if (proven.length) pool = proven;
  }
  return pool.sort((a, b) => {
    const la = a.inFlight / a.maxConcurrency;
    const lb = b.inFlight / b.maxConcurrency;
    if (la !== lb) return la - lb;
    const fa = a.failStreak || 0;
    const fb = b.failStreak || 0;
    if (fa !== fb) return fa - fb;
    if (a.completed !== b.completed) return a.completed - b.completed;
    return a.cost - b.cost;
  });
}

export function pickNode(store, caps, excludeNodeId, requiredRole, provenFirst = false) {
  const base = [...store.nodes.values()].filter(
    (n) =>
      n.online &&
      n.nodeId !== excludeNodeId &&
      n.inFlight < n.maxConcurrency &&
      (!requiredRole || (n.role || "member") === requiredRole) &&
      nodeMatches(n, caps),
  );
  return rankCandidates(base, { provenFirst })[0] || null;
}

/** 找一个可派发任务并返回 {task,node}，不修改状态。 */
export function nextDispatchable(store) {
  const now = Date.now();
  const queued = [...store.tasks.values()]
    .filter(
      (t) =>
        (t.state === TASK_STATE.QUEUED || t.state === TASK_STATE.REWORK) &&
        (!t.notBefore || t.notBefore <= now) &&
        depsReady(t, store.tasks),
    )
    // 需要审批且未批准：跳过（不派发）
    .filter((t) => !(t.requiresApproval && !t.approved))
    .sort((a, b) => b.priority - a.priority || a.createdAt - b.createdAt);

  for (const task of queued) {
    const node = pickNode(store, task.capabilities, task.lastNodeId, task.requiredRole, task.attempts >= 2);
    if (node) return { task, node };
  }
  return null;
}

/** 节点 poll：原子地领走一个任务。 */
export function claimForNode(store, nodeId) {
  const n = store.nodes.get(nodeId);
  if (!n || !n.online || n.inFlight >= n.maxConcurrency) return null;
  const now = Date.now();
  const isCooling = cooling(n, now);
  const task = [...store.tasks.values()]
    .filter(
      (t) =>
        (t.state === TASK_STATE.QUEUED || t.state === TASK_STATE.REWORK) &&
        (!t.notBefore || t.notBefore <= now) &&
        depsReady(t, store.tasks) &&
        !(t.requiresApproval && !t.approved) &&
        nodeMatches(n, t.capabilities),
    )
    .sort((a, b) => b.priority - a.priority || a.createdAt - b.createdAt)[0];
  if (!task) return null;
  if (isCooling) {
    // 冷却节点不领新活；但若该任务没有任何「活着且能接」的其它节点，允许自救执行
    //（与 pickNode 的全冷却回退同一防饿死原则）。
    // 「活着」判据：lastPoll 5 秒内真的来领过活（真实执行器 idle 时也每 ~100-200ms poll 一次）。
    // 心跳/lastSeen 不可用 —— 只注册不出勤的僵尸登记会永远挡住自救（G 段实测踩坑）。
    const elsewhere = [...store.nodes.values()].some(
      (m) =>
        m.nodeId !== nodeId &&
        m.online &&
        m.inFlight < m.maxConcurrency &&
        !cooling(m, now) &&
        (m.lastPoll || 0) >= now - 5_000 &&
        nodeMatches(m, task.capabilities),
    );
    if (elsewhere) return null;
  }
  // 重试任务优先成熟节点：本节点从未完成任务、且存在其它可接的已完成过节点的，让位
  if (task.attempts >= 2 && (n.completed || 0) === 0) {
    const provenAlt = [...store.nodes.values()].some(
      (m) =>
        m.nodeId !== nodeId &&
        m.online &&
        m.inFlight < m.maxConcurrency &&
        !cooling(m, now) &&
        (m.completed || 0) > 0 &&
        nodeMatches(m, task.capabilities),
    );
    if (provenAlt) return null;
  }

  task.state = TASK_STATE.ASSIGNED;
  task.nodeId = nodeId;
  task.lastNodeId = nodeId;
  task.attempts += 1;
  task.assignedAt = Date.now();
  task.history.push({ at: Date.now(), note: `assigned→${nodeId} attempt#${task.attempts}` });
  n.inFlight += 1;
  store.stats.assigned += 1;
  store.walAppend?.("task.claimed", {
    taskId: task.id,
    key: `claim:${task.id}:${nodeId}:${task.attempts}`,
    data: { nodeId, attempt: task.attempts },
  });
  store.addAudit({
    kind: "task.assign",
    taskId: task.id,
    nodeId,
    attempt: task.attempts,
  });
  store.bus.emit("task", { taskId: task.id, state: task.state, nodeId });
  store.scheduleSave();
  return task;
}

/** 节点回报结果。 */
export function settleResult(store, taskId, nodeId, result) {
  const t = store.tasks.get(taskId);
  const n = store.nodes.get(nodeId);
  if (n) n.inFlight = Math.max(0, n.inFlight - 1);
  if (!t) return { ok: false, reason: "no-task" };
  if (["done", "failed", "dead", "cancelled"].includes(t.state)) return { ok: false, reason: "already-settled:" + t.state };

  if (result.ok) {
    t.state = result.state || TASK_STATE.DONE;
    t.result = {
      output: result.output ?? null,
      score: result.score ?? null,
      evidence: result.evidence || null,
      executor: result.executor || null,
      at: Date.now(),
    };
    if (n) {
      n.completed += 1;
      n.failStreak = 0;
      n.coolUntil = 0;
    }
    store.stats.completed += 1;
    store.addAudit({
      kind: "task.done",
      taskId,
      nodeId,
      executor: result.executor,
      score: result.score,
      durationMs: result.durationMs,
    });
  } else {
    if (n) {
      n.failed += 1;
      n.failStreak = (n.failStreak || 0) + 1;
      if (breakerActive() && n.failStreak >= BR_STREAK) {
        // 打开/续期熔断：冷却时长随连击指数升级（首次=BR_COOL，之后 ×2，封顶 BR_MAX）
        const trips = n.failStreak - BR_STREAK + 1;
        const coolMs = Math.min(BR_COOL * Math.pow(2, trips - 1), BR_MAX);
        n.coolUntil = Date.now() + coolMs;
        store.addAudit({ kind: "node.cool", nodeId, streak: n.failStreak, coolMs });
      }
    }
    if (t.attempts < t.maxAttempts) {
      // 重投：回到队列，短暂退避，排除刚失败的节点
      t.state = TASK_STATE.QUEUED;
      t.lastNodeId = nodeId;
      t.notBefore = Date.now() + 200 * t.attempts;
      store.stats.requeued += 1;
      store.addAudit({ kind: "task.retry", taskId, nodeId, error: String(result.error).slice(0, 200) });
    } else {
      t.state = TASK_STATE.DEAD;
      t.result = { error: String(result.error).slice(0, 500), at: Date.now() };
      store.deadLetter.push({
        taskId,
        title: t.title,
        error: String(result.error).slice(0, 500),
        attempts: t.attempts,
        at: Date.now(),
      });
      store.stats.dead += 1;
      store.addAudit({ kind: "task.dead", taskId, attempts: t.attempts });
    }
  }
  t.updatedAt = Date.now();
  store.walAppend?.("task.settled", {
    taskId: t.id,
    key: `settle:${t.id}:${nodeId}`,
    data: { ok: result.ok, state: t.state, error: result.error ? String(result.error).slice(0, 200) : null },
  });
  store.bus.emit("task", { taskId: t.id, state: t.state });
  store.scheduleSave();
  return { ok: true, state: t.state };
}

/** 死信重投。 */
export function requeueDead(store, taskId) {
  const t = store.tasks.get(taskId);
  if (!t || t.state !== TASK_STATE.DEAD) return null;
  t.state = TASK_STATE.QUEUED;
  t.attempts = 0;
  t.lastNodeId = null;
  t.notBefore = 0;
  t.result = null;
  const idx = store.deadLetter.findIndex((d) => d.taskId === taskId);
  if (idx >= 0) store.deadLetter.splice(idx, 1);
  store.addAudit({ kind: "dead.requeue", taskId });
  store.bus.emit("task", { taskId, state: t.state });
  store.scheduleSave();
  return t;
}