《互联网产品的 AI 架构·多能力编排与任务闭环》3.2 拓扑执行与并行汇合

// file: src/orch/schedule.ts
// tier: L2
// verify: node --test test/orch/graph.test.ts

import { AiError } from "../core/errors.ts";
import {
  deriveIdempotencyKey,
  incomingEdges,
  levelize,
  nodeMap,
  validateGraph,
  type Checkpoint,
  type ExecutionGraph,
  type GraphNode,
  type NodeRecord,
} from "./graph.ts";

export type StopReason = "complete" | "failed" | "halted" | "budget";

export interface RunOptions {
  rootIdempotencyKey: string;
  /** 覆盖节点的失败注入，用于模拟“故障已修复”后再恢复。 */
  failAttempts?: Readonly<Record<string, readonly number[]>>;
  /** 执行完该层后主动中断，模拟进程崩溃；层号从 0 起。 */
  haltAfterLevel?: number;
  /** 恢复时传入上一次的检查点。 */
  checkpoint?: Checkpoint;
}

export interface RunResult {
  graphId: string;
  status: "succeeded" | "failed" | "stopped";
  stopReason: StopReason;
  levels: string[][];
  /** 本次真正执行的节点，按层内 id 顺序。 */
  ranNodeIds: string[];
  /** 本次从检查点恢复、未重跑的节点。 */
  restoredNodeIds: string[];
  records: NodeRecord[];
  /** 本次执行的现金成本；累计成本看 checkpoint.spentCostMicros。 */
  costMicros: number;
  /** 本次执行的关键路径，不等于端到端累计延迟。 */
  latencyMs: number;
  handoffCount: number;
  checkpoint: Checkpoint;
}

function fail(message: string): never {
  throw new AiError("invalid-request", `schedule: ${message}`);
}

function skippedRecord(nodeId: string, level: number, rootKey: string): NodeRecord {
  return {
    nodeId,
    status: "skipped",
    attempts: 0,
    output: "",
    costMicros: 0,
    latencyMs: 0,
    idempotencyKey: deriveIdempotencyKey(rootKey, nodeId),
    level,
  };
}

/**
 * 汇合只看直接前驱的成功数：all 要求全成功，any 允许一个，
 * quorum 要求达到票数。失败和跳过都不算成功票。
 */
export function joinSatisfied(node: GraphNode, deps: readonly string[], records: Map<string, NodeRecord>): boolean {
  if (deps.length === 0) return true;
  const successes = deps.filter((dep) => records.get(dep)?.status === "succeeded").length;
  switch (node.join ?? "all") {
    case "any":
      return successes >= 1;
    case "quorum":
      return successes >= (node.quorum ?? deps.length);
    default:
      return successes === deps.length;
  }
}

function executeNode(
  node: GraphNode,
  level: number,
  rootKey: string,
  override: readonly number[] | undefined,
): NodeRecord {
  const failedAttempts = new Set(override ?? node.failAttempts ?? []);
  const maxAttempts = node.maxAttempts ?? 1;
  let attempts = 0;
  let costMicros = 0;
  let latencyMs = 0;
  let ok = false;
  for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
    attempts = attempt;
    costMicros += node.costMicros;
    latencyMs += node.latencyMs;
    if (!failedAttempts.has(attempt)) {
      ok = true;
      break;
    }
  }
  return {
    nodeId: node.id,
    status: ok ? "succeeded" : "failed",
    attempts,
    output: `${node.capability}#${node.id}:${ok ? "ok" : "failed"}`,
    costMicros,
    latencyMs,
    idempotencyKey: deriveIdempotencyKey(rootKey, node.id),
    level,
  };
}

/**
 * 按层执行：同层节点并行，层内延迟取最大值、成本取总和。
 * 失败不打断同层兄弟节点，只影响下游的汇合判定。
 */
export function runGraph(graph: ExecutionGraph, options: RunOptions): RunResult {
  const issues = validateGraph(graph);
  if (issues.length > 0) {
    fail(issues.map((issue) => `${issue.code}: ${issue.message}`).join("; "));
  }
  if (options.rootIdempotencyKey.trim() === "") fail("rootIdempotencyKey 不能为空");

  const levels = levelize(graph);
  const nodes = nodeMap(graph);
  const records = new Map<string, NodeRecord>();
  const restoredNodeIds: string[] = [];

  if (options.checkpoint) {
    if (options.checkpoint.graphId !== graph.id) {
      fail(`检查点属于图 ${options.checkpoint.graphId}，不能恢复到 ${graph.id}`);
    }
    if (options.checkpoint.rootIdempotencyKey !== options.rootIdempotencyKey) {
      fail("恢复必须沿用同一个 rootIdempotencyKey");
    }
    for (const record of options.checkpoint.records) {
      if (record.status === "succeeded" && nodes.has(record.nodeId)) {
        records.set(record.nodeId, record);
        restoredNodeIds.push(record.nodeId);
      }
    }
    restoredNodeIds.sort();
  }

  const ranNodeIds: string[] = [];
  let costMicros = 0;
  let nodeRuns = 0;
  let latencyMs = 0;
  let handoffCount = 0;
  let levelsDone = 0;
  let stopReason: StopReason = "complete";
  let stopped = false;

  for (let level = 0; level < levels.length && !stopped; level += 1) {
    let levelLatency = 0;
    let ranInLevel = false;
    for (const nodeId of levels[level]) {
      if (records.get(nodeId)?.status === "succeeded") continue;
      const node = nodes.get(nodeId);
      if (!node) continue;
      const deps = incomingEdges(graph, nodeId).map((edge) => edge.from);
      if (!joinSatisfied(node, deps, records)) {
        records.set(nodeId, skippedRecord(nodeId, level, options.rootIdempotencyKey));
        continue;
      }
      if (nodeRuns >= graph.limits.maxNodeRuns || costMicros >= graph.limits.maxCostMicros) {
        stopReason = "budget";
        stopped = true;
        break;
      }
      const override = options.failAttempts?.[nodeId];
      const record = executeNode(node, level, options.rootIdempotencyKey, override);
      records.set(nodeId, record);
      ranNodeIds.push(nodeId);
      ranInLevel = true;
      nodeRuns += record.attempts;
      costMicros += record.costMicros;
      levelLatency = Math.max(levelLatency, record.latencyMs);
    }

    if (level > 0 && ranInLevel) {
      handoffCount += 1;
      latencyMs += graph.handoffLatencyMs;
    }
    latencyMs += levelLatency;

    if (stopped) break;
    levelsDone = level + 1;
    if (options.haltAfterLevel === level) {
      stopReason = "halted";
      stopped = true;
      break;
    }
  }

  if (stopReason === "complete") {
    stopReason = records.get(graph.terminal)?.status === "succeeded" ? "complete" : "failed";
  }
  const status = stopReason === "complete" ? "succeeded" : stopReason === "failed" ? "failed" : "stopped";
  const orderedRecords = [...records.values()].sort(
    (left, right) => left.level - right.level || left.nodeId.localeCompare(right.nodeId),
  );
  const checkpoint: Checkpoint = {
    graphId: graph.id,
    rootIdempotencyKey: options.rootIdempotencyKey,
    records: orderedRecords,
    spentCostMicros: (options.checkpoint?.spentCostMicros ?? 0) + costMicros,
    nodeRuns: (options.checkpoint?.nodeRuns ?? 0) + nodeRuns,
    levelsDone,
  };

  return {
    graphId: graph.id,
    status,
    stopReason,
    levels,
    ranNodeIds,
    restoredNodeIds,
    records: orderedRecords,
    costMicros,
    latencyMs,
    handoffCount,
    checkpoint,
  };
}