DAG 状态机与调度预算 深度解析

65function _initialState(graph: DAGGraphData, nodeId: string): NodeState {
66  const deps = _incomingAfterDeps(graph.edges, nodeId);
67  return deps.length === 0 ? "READY" : "PENDING";
68}
69
70export function createDAGRun(parsedDAG: ParsedDAG, runId: string): DAGRun {
71  const nodeStates = new Map<string, NodeState>();
72  const afterSatisfied = new Map<string, Set<string>>();
73  const inputSatisfied = new Map<string, Set<string>>();
74  const mailboxes = new Map<string, Map<string, unknown[]>>();
75
76  for (const node of parsedDAG.graph.nodes) {
77    nodeStates.set(node.node_id, _initialState(parsedDAG.graph, node.node_id));
78    afterSatisfied.set(node.node_id, new Set<string>());
79    inputSatisfied.set(node.node_id, new Set<string>());
80    mailboxes.set(node.node_id, new Map<string, unknown[]>());
81  }
82
83  return {
84    runId,
85    graph: parsedDAG.graph,
86    loopSources: new Set(parsedDAG.loop_sources),
87    nodeStates,
88    handoffedNodes: new Set<string>(),
89    afterSatisfied,
90    inputSatisfied,
91    mailboxes,
92  };
93}

234export function handoff(
235  run: DAGRun,
236  fromNode: string,
237  port: string,
238  content: unknown,
239): DAGTransitionResult {
240  if (!run.nodeStates.has(fromNode)) {
241    throw new Error(`Unknown node: ${fromNode}`);
242  }
243
244  const matchingDownstream = run.graph.edges.filter(
245    (edge) =>
246      edge.from_node === fromNode &&
247      edge.label !== "after_dep" &&
248      edge.to_node !== "" &&
249      edgeMatchesHandoff(edge, port),
250  );
251  const terminalFailure = isFailurePort(port) && matchingDownstream.length === 0;
252  run.handoffedNodes.add(fromNode);
253  run.nodeStates.set(
254    fromNode,
255    terminalFailure
256      ? "FAILED"
257      : run.loopSources.has(fromNode)
258        ? "RUNNING"
259        : "COMPLETED",
260  );
261
262  const affected = terminalFailure ? new Set<string>() : _satisfyAfterDeps(run, fromNode);
263  const mailboxReceivers = new Set<string>();
264  for (const edge of matchingDownstream) {
265    _ensureMailbox(run, edge.to_node, edge.to_port).push(content);
266    run.inputSatisfied.get(edge.to_node)?.add(fromNode);
267    affected.add(edge.to_node);
268    mailboxReceivers.add(edge.to_node);
269  }
270
271  for (const nodeId of affected) {
272    _tryPromote(run, nodeId);
273    _skipUntakenSatisfiedBranches(run, nodeId);
274  }
275  for (const nodeId of mailboxReceivers) {
276    _wakeLoopSource(run, nodeId);
277    _wakeLoopGatewayReceiver(run, fromNode, nodeId);
278  }
279  if (terminalFailure) {
280    _skipDependentNodes(run, fromNode);
281  }
282  return {
283    affectedNodes: Array.from(affected).sort(),
284    routedNodes: Array.from(mailboxReceivers).sort(),
285    terminalFailure,
286  };

1389export function dispatchReadyNodes(
1390  runId: string,
1391  dispatcher: DAGDispatcher,
1392): number {
1393  const run = store.get(runId);
1394  if (!run) return 0;
1395
1396  let count = 0;
1397  const before = _snapshotNodeStates(run);
1398  const ready = getReadyNodes(run.dagRun);
1399  for (const nodeId of ready) {
1400    const node = run.dagRun.graph.nodes.find((n) => n.node_id === nodeId);
1401    if (!node) continue;
1402    if (_isGatewayNode(node)) {
1403      if (_executeGatewayNode(runId, run, node)) count++;
1404      continue;
1405    }
1406
1407    const built = _buildDispatchEnvelope(run, nodeId);
1408    if (!built.ok) {
1409      failActiveRun(runId, nodeId, built.reason);
1410      continue;
1411    }
1412    if (run.counters.dispatches >= run.limits.max_dispatches) {
1413      abortActiveRun(runId, `max_dispatches (${run.limits.max_dispatches}) exceeded`, nodeId);
1414      break;
1415    }
1416    run.counters.dispatches++;
1417
1418    const envelope = built.envelope;
1419
1420    let result = dispatcher.dispatch(envelope);
1421    if (result.status === "failed") {
1422      const retryable = result.retryable !== false;
1423      if (retryable && recordNodeDispatchRetry(runId, nodeId, result.reason)) {
1424        if (run.counters.dispatches >= run.limits.max_dispatches) {
1425          abortActiveRun(runId, `max_dispatches (${run.limits.max_dispatches}) exceeded`, nodeId);
1426          break;
1427        }
1428        run.counters.dispatches++;
1429        result = dispatcher.dispatch({
1430          ...envelope,
1431          inputs: {
1432            ...envelope.inputs,
1433            dispatch_retry: [
1434              `Retrying DAG node ${nodeId} after transient dispatch failure: ${result.reason}`,
1435            ],
1436          },
1437        });
1438      }
1439      if (result.status === "failed") {
1440        failActiveRun(runId, nodeId, result.reason);
1441        continue;
1442      }
1443    }
1444    if (result.status !== "dispatched") continue;
1445    startNode(run.dagRun, nodeId);
1446    _markNodeSessionStatus(run, nodeId, "running");
1447    emit("dag:node_dispatched", { runId, nodeId, agentId: envelope.agentId, sessionId: envelope.sessionId });
1448    count++;
1449  }
1450  if (count > 0) {
1451    writeRunMetadata(runId, serializeRunMetadata(run));
1452    _emitNodeStateChanges(run, before);
1453  }
1454  return count;
1455}