Function executeNodeGraph

src/nodeGraphExecutor.ts:318–441

(request: NodeGraphExecutionRequest)

  return extractConnectionData(targetIncomingConnections[targetIncomingConnections.length - 1], executionResults, nodeMap);
};
export const executeNodeGraph = async (request: NodeGraphExecutionRequest): Promise<NodeGraphExecutionResult> => {
  const { nodes, connections, nodeConfigs, executionContext, resetCounters = true } = request;
  const startTime = performance.now();
  console.log(`Starting node graph execution: ${nodes.length} nodes, ${connections.length} connections`);
  if (resetCounters) {
    resetCounterTracking();
  }
  try {
    const nodeMap = new Map(nodes.map((node) => [node.id, node]));
    const outgoingConnectionsBySource = new Map<string, SerializedConnection[]>();
    const incomingConnectionsByTarget = new Map<string, SerializedConnection[]>();
    const incomingEdges = new Set<string>();
    for (const connection of connections) {
      const outgoingConnections = outgoingConnectionsBySource.get(connection.sourceId);
      if (outgoingConnections) {
        outgoingConnections.push(connection);
      } else {
        outgoingConnectionsBySource.set(connection.sourceId, [connection]);
      }
      const incomingConnections = incomingConnectionsByTarget.get(connection.targetId);
      if (incomingConnections) {
        incomingConnections.push(connection);
      } else {
        incomingConnectionsByTarget.set(connection.targetId, [connection]);
      }
      incomingEdges.add(connection.targetId);
    }
    const startingNodes = nodes.filter((node) => !incomingEdges.has(node.id));
    if (startingNodes.length === 0) {
      return {
        success: false,
        executionResults: new Map(),
        totalExecuted: 0,
        successCount: 0,
        failureCount: 0,
        error: "No starting nodes found in the graph",
      };
    }
    flowExecutionDebugLog(
      executionContext,
      `Graph starting nodes: ${startingNodes.map((node) => `${node.id}(${node.type})`).join(", ")}`,
    );
    const executionResults = new Map<string, NodeExecutionResult>();
    const executed = new Set<string>();
    const executionQueue = startingNodes.map((node) => ({ node, inputData: null as any }));
    let queueIndex = 0;
    while (queueIndex < executionQueue.length) {
      const { node, inputData } = executionQueue[queueIndex++];
      if (executed.has(node.id)) {
        continue;
      }
      try {
        flowExecutionDebugLog(executionContext, `Executing node ${node.id} (${node.type})`);
        const config = nodeConfigs?.[node.id];
        const textValue =
          config === undefined ? serializeNodeConfigForExecution(node) : typeof config === "string" ? config : "";
        const result = await executeNodeAction({
          nodeId: node.id,