<?php

declare(strict_types=1);

namespace Drupal\flowdrop_runtime\Service\Orchestrator;

use Drupal\Core\Logger\LoggerChannelFactoryInterface;
use Drupal\flowdrop\Constants\ReservedName;
use Drupal\flowdrop\Exception\WorkflowStopException;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\TriggerNodeProcessorInterface;
use Drupal\flowdrop\Service\FlowDropNodeProcessorPluginManager;
use Drupal\flowdrop_node_type\Service\NodeMetadataResolver;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionResult;
use Drupal\flowdrop_runtime\DTO\Compiler\DependencyGraph;
use Drupal\flowdrop_runtime\DTO\Compiler\DependencyEdge;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext;
use Drupal\flowdrop_orchestration\DTO\Snapshot\WorkflowSnapshot;
use Drupal\flowdrop_workflow\DTO\WorkflowDTO;
use Drupal\flowdrop_workflow\WorkflowDefinitionInterface;
use Drupal\flowdrop_runtime\Service\Runtime\NodeRuntimeServiceInterface;
use Drupal\flowdrop_runtime\Service\Runtime\ToolProjectorInterface;
use Drupal\flowdrop_runtime\Service\Runtime\ExecutionContext;
use Drupal\flowdrop_runtime\Service\RealTime\RealTimeManagerInterface;
use Drupal\flowdrop_runtime\Service\Compiler\WorkflowCompilerInterface;
use Drupal\flowdrop_runtime\Service\Logging\ExecutionLoggerInterface;
use Drupal\flowdrop_runtime\Service\WorkflowState\WorkflowStateManagerInterface;
use Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationRequest;
use Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationResponse;
use Drupal\flowdrop_orchestration\Exception\InterruptExceptionInterface;
use Drupal\flowdrop_runtime\Exception\OrchestrationException;
use Drupal\flowdrop_orchestration\Service\Orchestrator\OrchestratorHelperInterface;
use Drupal\flowdrop_orchestration\Service\Orchestrator\OrchestratorInterface;
use Psr\Log\LoggerInterface;
use Drupal\flowdrop_runtime\Event\NodePostExecuteEvent;
use Drupal\flowdrop_runtime\Event\NodePreExecuteEvent;
use Drupal\flowdrop_runtime\Event\NodeSkippedEvent;
use Drupal\flowdrop_runtime\Event\WorkflowCompletedEvent;
use Drupal\flowdrop_runtime\Service\Orchestrator\Trait\PipelineCreationTrait;
use Drupal\flowdrop_runtime\Service\Storage\PipelineStorageInterface;
use Drupal\flowdrop_runtime\Service\Storage\PipelineStorageRegistryInterface;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

/**
 * Synchronous workflow orchestrator with real-time updates.
 *
 * Executes workflows directly without requiring pipeline entities. Uses the
 * ExecutionGraph for determining execution order and the DependencyGraph for
 * data resolution at runtime.
 *
 * Workflow and node execution logging is handled via ExecutionLoggerInterface
 * for centralized verbosity control. Debug logs use a regular PSR logger.
 *
 * For pipeline-based execution with job tracking, use
 * SynchronousPipelineOrchestrator.
 *
 * @internal This class is not part of the public API and may change without notice.
 */
class SynchronousOrchestrator implements OrchestratorInterface, PipelineAwareOrchestratorInterface {

  use InterruptHandlingTrait;
  use PipelineCreationTrait;

  /**
   * PSR logger for pipeline-specific and debug logs.
   *
   * @var \Psr\Log\LoggerInterface
   */
  private readonly LoggerInterface $logger;

  /**
   * Pipeline storage, lazily resolved from the registry on first access.
   */
  private PipelineStorageInterface $pipelineStorage;

  /**
   * Constructs a new SynchronousOrchestrator.
   *
   * @param \Drupal\flowdrop_runtime\Service\Runtime\NodeRuntimeServiceInterface $nodeRuntime
   *   The node runtime service.
   * @param \Drupal\flowdrop_runtime\Service\Runtime\ExecutionContext $executionContext
   *   The execution context service.
   * @param \Drupal\flowdrop_runtime\Service\RealTime\RealTimeManagerInterface $realTimeManager
   *   The real-time manager.
   * @param \Drupal\flowdrop_runtime\Service\Compiler\WorkflowCompilerInterface $workflowCompiler
   *   The workflow compiler.
   * @param \Drupal\flowdrop_runtime\Service\Logging\ExecutionLoggerInterface $executionLogger
   *   The centralized execution logger.
   * @param \Symfony\Component\EventDispatcher\EventDispatcherInterface $eventDispatcher
   *   The event dispatcher.
   * @param \Drupal\flowdrop\Service\FlowDropNodeProcessorPluginManager $nodeProcessorManager
   *   The node processor plugin manager.
   * @param \Drupal\Core\Logger\LoggerChannelFactoryInterface $loggerFactory
   *   The logger factory for pipeline/debug logs.
   * @param \Drupal\flowdrop_runtime\Service\Storage\PipelineStorageRegistryInterface $pipelineStorageRegistry
   *   The pipeline storage registry.
   * @param \Drupal\flowdrop_runtime\Service\WorkflowState\WorkflowStateManagerInterface|null $workflowStateManager
   *   The workflow state manager (optional for backward compatibility).
   * @param \Drupal\flowdrop_runtime\Service\Runtime\ToolProjectorInterface|null $toolProjector
   *   Projects tools wired to a node, for processors that consume them.
   * @param \Drupal\flowdrop_runtime\Service\Orchestrator\IteratorExecutorInterface|null $iteratorExecutor
   *   The iterator executor (optional, provided by flowdrop_iterator module).
   * @param \Drupal\flowdrop_node_type\Service\NodeMetadataResolver|null $nodeMetadataResolver
   *   Re-injects fresh node metadata from the live node type before the
   *   workflow is parsed, so execution is driven by the authoritative
   *   definition rather than the stored metadata cache. Optional; when omitted
   *   (dependency-light tests) the stored metadata is used as-is.
   */
  public function __construct(
    private readonly NodeRuntimeServiceInterface $nodeRuntime,
    private readonly ExecutionContext $executionContext,
    private readonly RealTimeManagerInterface $realTimeManager,
    private readonly WorkflowCompilerInterface $workflowCompiler,
    private readonly ExecutionLoggerInterface $executionLogger,
    private readonly EventDispatcherInterface $eventDispatcher,
    private readonly FlowDropNodeProcessorPluginManager $nodeProcessorManager,
    LoggerChannelFactoryInterface $loggerFactory,
    private readonly PipelineStorageRegistryInterface $pipelineStorageRegistry,
    private readonly ?WorkflowStateManagerInterface $workflowStateManager = NULL,
    private readonly ?ToolProjectorInterface $toolProjector = NULL,
    private readonly ?IteratorExecutorInterface $iteratorExecutor = NULL,
    private readonly ?NodeMetadataResolver $nodeMetadataResolver = NULL,
  ) {
    $this->logger = $loggerFactory->get("flowdrop_runtime");
  }

  /**
   * {@inheritdoc}
   */
  public function getType(): string {
    return OrchestratorHelperInterface::SYNCHRONOUS;
  }

  /**
   * {@inheritdoc}
   */
  public function getPreferredPipelineBundle(): string {
    return PipelineStorageRegistryInterface::BUNDLE_IN_MEMORY;
  }

  /**
   * Executes a workflow using WorkflowInterface.
   *
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The workflow to execute.
   * @param array<string, mixed> $initialContext
   *   Initial context data for the workflow execution.
   *
   * @return \Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationResponse
   *   The orchestration response.
   *
   * @throws \Drupal\flowdrop_runtime\Exception\OrchestrationException
   *   When orchestration fails.
   */
  public function executeWorkflow(WorkflowDefinitionInterface $workflow, array $initialContext = []): OrchestrationResponse {
    // Convert workflow interface to array format for orchestration.
    $workflowData = [
      "id" => $workflow->id(),
      "label" => $workflow->getLabel(),
      "description" => $workflow->getDescription(),
      "nodes" => $workflow->getNodes(),
      "edges" => $workflow->getEdges(),
      "metadata" => $workflow->getMetadata(),
    ];

    // Create orchestration request.
    $request = new OrchestrationRequest(
      workflowId: $workflow->id(),
      pipelineId: "pipeline_" . $workflow->id() . "_" . time(),
      workflow: $workflowData,
      initialData: $initialContext,
      options: [],
    );

    return $this->orchestrate($request);
  }

  /**
   * {@inheritdoc}
   */
  public function orchestrate(OrchestrationRequest $request): OrchestrationResponse {
    $workflowData = $request->getWorkflow();

    $pipeline = $this->createPipelineFromWorkflow(
      $request->getWorkflow(),
      $request->getInitialData(),
      $request->getOptions(),
    );
    $pipeline->markAsStarted();
    $this->getPipelineStorage()->save($pipeline);
    $executionId = (string) $pipeline->id();

    try {
      // Re-inject fresh metadata from the live node type so the DTO (and the
      // compiler it is handed to) resolve behaviour from the authoritative
      // definition, not the stored metadata cache. Idempotent.
      if ($this->nodeMetadataResolver !== NULL) {
        $workflowData = $this->nodeMetadataResolver->enrichWorkflow($workflowData);
      }
      // Convert to WorkflowDTO for normalized access to properties.
      $workflowDTO = WorkflowDTO::fromArray($workflowData);

      // Compile workflow - produces DependencyGraph and ExecutionGraph.
      $compiledWorkflow = $this->workflowCompiler->compile($workflowDTO);
      $dependencyGraph = $compiledWorkflow->getDependencyGraph();
      $executionGraph = $compiledWorkflow->getExecutionGraph();
      $nodeMappings = $compiledWorkflow->getNodeMappings();

      // Log workflow start via centralized logger.
      $this->executionLogger->logWorkflowStart(
        $executionId,
        $request->getWorkflowId(),
        $executionGraph->getNodeCount()
      );

      // Start real-time monitoring.
      $this->realTimeManager->startMonitoring($executionId, $workflowDTO->toCompilerArray());

      // Update execution status to running.
      $this->realTimeManager->updateExecutionStatus($executionId, WorkflowSnapshot::STATUS_RUNNING, [
        "workflow_id" => $request->getWorkflowId(),
        "pipeline_id" => $request->getPipelineId(),
        "executable_nodes" => $executionGraph->getNodeCount(),
        "excluded_nodes" => $executionGraph->getExcludedCount(),
        "start_time" => time(),
      ]);

      // Create execution context.
      $context = $this->executionContext->createContext(
        $request->getWorkflowId(),
        $request->getPipelineId(),
        $request->getInitialData()
      );

      // Get execution order from ExecutionGraph.
      $executionOrder = $executionGraph->getExecutionOrder();

      // Filter trigger nodes: only include the trigger node that
      // initiated this workflow. Exclude all other trigger nodes in
      // the workflow.
      $initialData = $request->getInitialData();
      $triggerNodeId = $initialData["trigger_node_id"] ?? NULL;
      if ($triggerNodeId !== NULL) {
        $filteredOrder = [];
        foreach ($executionOrder as $nodeId) {
          $nodeMapping = $nodeMappings[$nodeId] ?? NULL;
          $isTriggerNode = $nodeMapping !== NULL && $this->isTriggerNodeProcessor($nodeMapping->getProcessorId());

          // Include the triggering trigger node, exclude all other
          // trigger nodes.
          if ($isTriggerNode) {
            if ($nodeId === $triggerNodeId) {
              $filteredOrder[] = $nodeId;
              $this->logger->debug("Including triggering trigger node @node_id in execution order", [
                "@node_id" => $triggerNodeId,
              ]);
            }
            else {
              $this->logger->debug("Excluding non-triggering trigger node @node_id from execution order", [
                "@node_id" => $nodeId,
              ]);
            }
          }
          else {
            // Include all non-trigger nodes.
            $filteredOrder[] = $nodeId;
          }
        }
        $executionOrder = $filteredOrder;
      }

      $results = [];
      $skippedNodes = [];
      $gatewayOutputs = [];
      $startTime = microtime(TRUE);

      // Execute nodes in order determined by ExecutionGraph.
      foreach ($executionOrder as $nodeId) {
        $execNode = $executionGraph->getNode($nodeId);
        $nodeMapping = $nodeMappings[$nodeId] ?? NULL;
        $nodeDTO = $workflowDTO->getNode($nodeId);

        if ($execNode === NULL || $nodeMapping === NULL || $nodeDTO === NULL) {
          throw new OrchestrationException("Node $nodeId not found in compiled workflow");
        }

        // NodeRuntimeService::executeNode() loads the node type from storage
        // by its config-entity ID, then derives the executor plugin from it —
        // so it needs the node-type entity ID (the metadata.node_type_id
        // anchor), NOT the executor plugin ID that getProcessorId() returns.
        // Passing the processor ID here makes every entity load fail with
        // "Node type '<executor-plugin-id>' not found". (The trigger-node
        // check above correctly keeps using getProcessorId() — it resolves a
        // processor plugin definition, not a node-type entity.)
        $nodeType = $nodeMapping->getNodeTypeId();

        // Check if this node should execute based on gateway outputs.
        $executedNodeIds = array_keys($results);
        $shouldExecute = $this->shouldExecuteNode(
          $nodeId,
          $gatewayOutputs,
          $dependencyGraph,
          $executedNodeIds
        );

        if (!$shouldExecute) {
          // Log via centralized logger.
          $this->executionLogger->logNodeSkipped(
            $executionId,
            $nodeId,
            $nodeType,
            "branch not active"
          );
          $skippedNodes[] = $nodeId;

          // Emit node skipped event.
          $this->eventDispatcher->dispatch(
            new NodeSkippedEvent(
              nodeData: $nodeDTO->toArray(),
              executionId: $executionId,
              workflowId: $request->getWorkflowId(),
              reason: "branch_not_active",
            ),
            NodeSkippedEvent::NAME
          );
          continue;
        }

        // Node start/complete logging is handled by NodeRuntimeService.
        // Emit pre-execute event.
        $this->eventDispatcher->dispatch(
          new NodePreExecuteEvent(
            nodeData: $nodeDTO->toArray(),
            executionId: $executionId,
            workflowId: $request->getWorkflowId(),
            context: $context,
          ),
          NodePreExecuteEvent::NAME
        );

        try {
          // Resolve inputs using DependencyGraph.
          // Context is passed separately via NodeExecutionContext.
          $nodeInputs = $this->resolveNodeInputs(
            $nodeId,
            $dependencyGraph,
            $context,
            $executionId,
            $request->getWorkflowId(),
            $request->getPipelineId()
          );
          $nodeConfig = $nodeMapping->getConfig();

          // Execute based on node type (check special handling).
          if ($execNode->isIterator()) {
            $result = $this->executeIteratorNode(
              $executionId,
              $nodeId,
              $nodeInputs,
              $nodeConfig,
              $workflowDTO,
              $request->getPipelineId(),
              $context,
              $dependencyGraph
            );
          }
          else {
            // Project tools wired to this node (empty/NULL when it has none);
            // executeNode injects them only if the processor is ToolsAware.
            $tools = $this->toolProjector?->project(
              $nodeId,
              $dependencyGraph,
              $context,
              $executionId
            );
            $result = $this->nodeRuntime->executeNode(
              $executionId,
              $nodeId,
              $nodeType,
              $nodeInputs,
              $nodeConfig,
              $context,
              $tools
            );
          }

          $results[$nodeId] = $result;

          // Update context with result.
          // The execution counter is automatically incremented and
          // stored for conflict resolution when multiple sources target
          // the same input port.
          $this->executionContext->updateContext(
            $context,
            $nodeId,
            $result->getOutput()
          );

          // Capture gateway outputs for downstream branch decisions.
          $outputData = $result->getOutput()->toArray();
          if (isset($outputData[ReservedName::ACTIVE_BRANCHES])) {
            $gatewayOutputs[$nodeId] = $outputData[ReservedName::ACTIVE_BRANCHES];
          }

          // Node completion is logged by NodeRuntimeService.
          // Emit post-execute event.
          $this->eventDispatcher->dispatch(
            new NodePostExecuteEvent(
              nodeData: $nodeDTO->toArray(),
              executionId: $executionId,
              workflowId: $request->getWorkflowId(),
              context: $context,
              result: $result,
            ),
            NodePostExecuteEvent::NAME
          );

        }
        catch (WorkflowStopException $e) {
          // A terminal node reached: stop the run cleanly, abandoning any
          // nodes after it in the execution order. Emit a neutral post-execute
          // event (this is a completion, not an error) and break so the normal
          // completion path below runs and returns STATUS_COMPLETED.
          $this->eventDispatcher->dispatch(
            new NodePostExecuteEvent(
              nodeData: $nodeDTO->toArray(),
              executionId: $executionId,
              workflowId: $request->getWorkflowId(),
              context: $context,
            ),
            NodePostExecuteEvent::NAME
          );
          $this->logger->debug("Workflow @workflow stopped by terminal node @node", [
            "@workflow" => $request->getWorkflowId(),
            "@node" => $nodeId,
          ]);
          break;
        }
        catch (\Exception $e) {
          // Node failure is logged by NodeRuntimeService.
          // Emit post-execute event with error.
          $this->eventDispatcher->dispatch(
            new NodePostExecuteEvent(
              nodeData: $nodeDTO->toArray(),
              executionId: $executionId,
              workflowId: $request->getWorkflowId(),
              context: $context,
              error: $e->getMessage(),
            ),
            NodePostExecuteEvent::NAME
          );

          throw $e;
        }
      }

      $totalTime = microtime(TRUE) - $startTime;

      // Update execution status to completed.
      $this->realTimeManager->updateExecutionStatus($executionId, WorkflowSnapshot::STATUS_COMPLETED, [
        "total_execution_time" => $totalTime,
        "nodes_executed" => count($results),
        "nodes_skipped" => count($skippedNodes),
        "end_time" => time(),
      ]);

      // Log workflow completion via centralized logger.
      $this->executionLogger->logWorkflowComplete(
        $executionId,
        $request->getWorkflowId(),
        count($results),
        count($skippedNodes),
        $totalTime
      );

      // Generate workflow snapshot for potential handover.
      $snapshot = $this->generateSnapshot(
        $executionId,
        $request,
        $workflowData,
        $results,
        $skippedNodes,
        WorkflowSnapshot::STATUS_COMPLETED
      );

      // Save the snapshot to persistent storage.
      if ($snapshot !== NULL && $this->workflowStateManager !== NULL) {
        try {
          $snapshot = $this->workflowStateManager->save($snapshot);
          $this->logger->debug("Saved workflow snapshot @exec for workflow @workflow", [
            "@exec" => $snapshot->executionId,
            "@workflow" => $snapshot->workflowId,
          ]);
        }
        catch (\Exception $e) {
          $this->logger->error("Failed to save workflow snapshot: @error", [
            "@error" => $e->getMessage(),
            "@exec" => $executionId,
          ]);
        }
      }

      // Emit workflow completed event.
      $this->eventDispatcher->dispatch(
        new WorkflowCompletedEvent(
          workflowData: $workflowDTO->toCompilerArray(),
          executionId: $executionId,
          workflowId: $request->getWorkflowId(),
          results: $results,
          skippedNodes: $skippedNodes,
          executionTime: $totalTime,
          context: $context,
          snapshot: $snapshot,
        ),
        WorkflowCompletedEvent::NAME
      );

      $pipeline->markAsCompleted(['results' => array_keys($results)]);
      $this->getPipelineStorage()->save($pipeline);

      return new OrchestrationResponse(
        executionId: $executionId,
        status: WorkflowSnapshot::STATUS_COMPLETED,
        results: $results,
        executionTime: $totalTime,
        metadata: [
          "orchestrator_type" => $this->getType(),
          "nodes_executed" => count($results),
          "nodes_skipped" => count($skippedNodes),
          "nodes_excluded" => $executionGraph->getExcludedCount(),
          "skipped_node_ids" => $skippedNodes,
          "excluded_node_ids" => array_keys($executionGraph->getExcludedNodes()),
          "compiled_workflow_id" => $compiledWorkflow->getWorkflowId(),
          "snapshot" => $snapshot?->toArray(),
        ]
      );

    }
    catch (InterruptExceptionInterface $e) {
      // Use the InterruptHandlingTrait for standardized interrupt handling.
      return $this->handleInterruptException(
        $e,
        $executionId,
        $request->getWorkflowId(),
        $startTime ?? microtime(TRUE),
        $results ?? [],
      );
    }
    catch (\Exception $e) {
      if (isset($pipeline)) {
        $pipeline->markAsFailed($e->getMessage());
        $this->getPipelineStorage()->save($pipeline);
      }

      // Update execution status to failed.
      $this->realTimeManager->updateExecutionStatus($executionId, WorkflowSnapshot::STATUS_FAILED, [
        "error" => $e->getMessage(),
        "end_time" => time(),
      ]);

      // Log workflow failure via centralized logger.
      $this->executionLogger->logWorkflowFailure(
        $executionId,
        $request->getWorkflowId(),
        $e
      );

      throw new OrchestrationException(
        "Synchronous orchestration failed: " . $e->getMessage(),
        0,
        $e
      );
    }
  }

  /**
   * {@inheritdoc}
   */
  public function supportsWorkflow(array $workflow): bool {
    return TRUE;
  }

  /**
   * {@inheritdoc}
   */
  public function getCapabilities(): array {
    return [
      "synchronous_execution" => TRUE,
      "parallel_execution" => FALSE,
      "async_execution" => FALSE,
      "retry_support" => TRUE,
      "error_recovery" => TRUE,
      "workflow_compilation" => TRUE,
      "pipeline_based" => FALSE,
      "direct_execution" => TRUE,
      "interrupt_support" => TRUE,
    ];
  }

  // =========================================================================
  // PipelineCreationTrait accessor implementations
  // =========================================================================

  /**
   * {@inheritdoc}
   */
  protected function getPipelineStorage(): PipelineStorageInterface {
    if (!isset($this->pipelineStorage)) {
      $this->pipelineStorage = $this->pipelineStorageRegistry->getStorage($this->getPreferredPipelineBundle());
    }
    return $this->pipelineStorage;
  }

  /**
   * {@inheritdoc}
   */
  protected function getPipelineCreationLogger(): LoggerInterface {
    return $this->logger;
  }

  /**
   * {@inheritdoc}
   */
  protected function getPipelineCreationEventDispatcher(): EventDispatcherInterface {
    return $this->eventDispatcher;
  }

  /**
   * Resolves inputs for a node using the DependencyGraph.
   *
   * Gets data from upstream nodes based on data edges in the
   * dependency graph. Context is passed separately via
   * NodeExecutionContext to NodeRuntimeService, which injects it into
   * processors implementing ExecutionContextAwareInterface.
   *
   * @param string $nodeId
   *   The node ID.
   * @param \Drupal\flowdrop_runtime\DTO\Compiler\DependencyGraph $dependencyGraph
   *   The dependency graph.
   * @param \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext $context
   *   The execution context.
   * @param string $executionId
   *   The current execution ID.
   * @param string $workflowId
   *   The workflow ID being executed.
   * @param string $pipelineId
   *   The pipeline ID being executed.
   *
   * @return array<string, mixed>
   *   The resolved node inputs from upstream nodes.
   */
  private function resolveNodeInputs(
    string $nodeId,
    DependencyGraph $dependencyGraph,
    NodeExecutionContext $context,
    string $executionId,
    string $workflowId,
    string $pipelineId,
  ): array {
    $inputData = [];

    // Get data dependencies from DependencyGraph.
    $dataParents = $dependencyGraph->getParents($nodeId, DependencyEdge::TYPE_DATA);

    // Track which target ports have been set and by which sources.
    // This helps detect conflicts when multiple sources target the same port.
    $portSources = [];

    // Get execution order from context metadata for sorting by "latest".
    $metadata = $context->getMetadata();
    $executionOrder = $metadata["node_execution_order"] ?? [];

    // Sort parents by execution order (latest first = higher counter),
    // then by node ID as fallback.
    // This ensures that when multiple sources target the same port,
    // the most recently executed node's value is used, which is more
    // intuitive than alphabetical ordering.
    // The execution counter increments with each node completion,
    // avoiding timestamp collisions.
    usort($dataParents, function (array $a, array $b) use ($executionOrder): int {
      $orderA = $executionOrder[$a["nodeId"]] ?? 0;
      $orderB = $executionOrder[$b["nodeId"]] ?? 0;

      // If both have execution order, sort by latest
      // (descending = higher counter first).
      if ($orderA > 0 && $orderB > 0) {
        return $orderB <=> $orderA;
      }

      // If only one has execution order, prioritize it.
      if ($orderA > 0) {
        return -1;
      }
      if ($orderB > 0) {
        return 1;
      }

      // Fallback to alphabetical by node ID if no execution order available.
      return strcmp($a["nodeId"], $b["nodeId"]);
    });

    foreach ($dataParents as $parent) {
      $parentOutput = $context->getNodeOutput($parent["nodeId"]);
      if ($parentOutput !== NULL) {
        // Merge parent output into input data.
        // Use port names for precise mapping if available.
        $outputArray = $parentOutput->toArray();
        $sourcePort = $parent["sourcePortName"] ?? "";
        $targetPort = $parent["targetPortName"] ?? "";

        if (!empty($sourcePort) && !empty($targetPort) && isset($outputArray[$sourcePort])) {
          // Last-executor-wins: $dataParents is sorted latest-execution-order
          // first, so the first value written to a port comes from the most
          // recently executed source. Keep it and ignore earlier-executed
          // sources targeting the same port — one value per port, never an
          // aggregated array.
          if (isset($portSources[$targetPort])) {
            $keptSource = $portSources[$targetPort];
            $keptOrder = $executionOrder[$keptSource] ?? 0;
            $ignoredOrder = $executionOrder[$parent["nodeId"]] ?? 0;

            $this->logger->warning(
              "Multiple sources target the same input port '@target_port' on node '@node_id'. " .
              "Keeping value from '@kept_source' (execution order: @kept_order, the most recent executor); " .
              "ignoring '@ignored_source' (execution order: @ignored_order).",
              [
                "@target_port" => $targetPort,
                "@node_id" => $nodeId,
                "@kept_source" => $keptSource,
                "@kept_order" => $keptOrder > 0 ? (string) $keptOrder : "N/A",
                "@ignored_source" => $parent["nodeId"],
                "@ignored_order" => $ignoredOrder > 0 ? (string) $ignoredOrder : "N/A",
              ]
            );
            continue;
          }
          $inputData[$targetPort] = $outputArray[$sourcePort];
          $portSources[$targetPort] = $parent["nodeId"];
        }
        else {
          // Merge all outputs.
          // When port names are not specified, merge all outputs.
          // Note: This can also cause overwrites if multiple parents have
          // the same output keys. This is expected behavior for unkeyed merges.
          $inputData = array_merge($inputData, $outputArray);
        }
      }
    }

    // Context is now passed directly via NodeExecutionContext to
    // NodeRuntimeService, which injects it into processors
    // implementing ExecutionContextAwareInterface.
    return $inputData;
  }

  /**
   * Determines if a node should execute based on gateway outputs.
   *
   * Uses DependencyGraph to check trigger edges and branch activation.
   *
   * @param string $nodeId
   *   The node ID to check.
   * @param array<string, string> $gatewayOutputs
   *   Map of gateway node IDs to their active_branches output.
   * @param \Drupal\flowdrop_runtime\DTO\Compiler\DependencyGraph $dependencyGraph
   *   The dependency graph.
   * @param array<string> $executedNodeIds
   *   Array of node IDs that have been executed so far.
   *
   * @return bool
   *   TRUE if the node should execute, FALSE if it should be skipped.
   */
  private function shouldExecuteNode(
    string $nodeId,
    array $gatewayOutputs,
    DependencyGraph $dependencyGraph,
    array $executedNodeIds,
  ): bool {
    // Get trigger edges from DependencyGraph.
    $triggerEdges = $dependencyGraph->getIncomingEdges($nodeId, DependencyEdge::TYPE_TRIGGER);

    // If no trigger edges, node executes based on data dependencies only.
    if (empty($triggerEdges)) {
      return TRUE;
    }

    // With trigger edges, at least ONE trigger must be satisfied.
    foreach ($triggerEdges as $edge) {
      $sourceId = $edge->getSource();
      $branchName = $edge->getMetadataValue("branchName", "");

      // Check if source is a gateway that has been executed.
      if (isset($gatewayOutputs[$sourceId])) {
        $activeBranches = $gatewayOutputs[$sourceId];

        // If no specific branch required, any completion satisfies.
        if ($branchName === "") {
          return TRUE;
        }

        // Check if the branch name matches any active branch.
        $activeBranchList = array_map("trim", explode(",", strtolower($activeBranches)));

        if (in_array(strtolower($branchName), $activeBranchList, TRUE)) {
          $this->logger->debug("Trigger satisfied for node @node_id: branch @branch matches active @active", [
            "@node_id" => $nodeId,
            "@branch" => $branchName,
            "@active" => $activeBranches,
          ]);
          return TRUE;
        }
      }
      else {
        // Source is NOT a gateway - trigger is
        // satisfied if source was executed.
        if (in_array($sourceId, $executedNodeIds, TRUE)) {
          $this->logger->debug("Trigger satisfied for node @node_id: non-gateway source @source was executed", [
            "@node_id" => $nodeId,
            "@source" => $sourceId,
          ]);
          return TRUE;
        }
      }
    }

    // No trigger was satisfied.
    $this->logger->debug("No triggers satisfied for node @node_id", [
      "@node_id" => $nodeId,
    ]);
    return FALSE;
  }

  /**
   * Checks if a processor is a trigger node processor.
   *
   * @param string $processorId
   *   The processor plugin ID.
   *
   * @return bool
   *   TRUE if the processor implements TriggerNodeProcessorInterface,
   *   FALSE otherwise.
   */
  private function isTriggerNodeProcessor(string $processorId): bool {
    try {
      $definition = $this->nodeProcessorManager->getDefinition($processorId);
      if (!isset($definition["class"])) {
        return FALSE;
      }

      $class = $definition["class"];
      return is_subclass_of($class, TriggerNodeProcessorInterface::class) ||
        in_array(TriggerNodeProcessorInterface::class, class_implements($class), TRUE);
    }
    catch (\Exception $e) {
      // If we can't get the definition, assume it's not a trigger node.
      return FALSE;
    }
  }

  /**
   * Executes an Iterator node with special sub-workflow handling.
   *
   * @param string $executionId
   *   The execution ID.
   * @param string $nodeId
   *   The iterator node ID.
   * @param array<string, mixed> $inputData
   *   The input data for the iterator.
   * @param array<string, mixed> $config
   *   The iterator configuration.
   * @param \Drupal\flowdrop_workflow\DTO\WorkflowDTO $workflow
   *   The workflow DTO.
   * @param string $pipelineId
   *   The parent pipeline ID.
   * @param \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext $context
   *   The execution context.
   * @param \Drupal\flowdrop_runtime\DTO\Compiler\DependencyGraph $dependencyGraph
   *   The dependency graph.
   *
   * @return \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionResult
   *   The execution result.
   *
   * @throws \Drupal\flowdrop_runtime\Exception\OrchestrationException
   *   When iterator execution fails.
   */
  private function executeIteratorNode(
    string $executionId,
    string $nodeId,
    array $inputData,
    array $config,
    WorkflowDTO $workflow,
    string $pipelineId,
    NodeExecutionContext $context,
    DependencyGraph $dependencyGraph,
  ): NodeExecutionResult {
    $this->logger->info("Executing Iterator node @node_id with special handling", [
      "@node_id" => $nodeId,
      "@execution_id" => $executionId,
    ]);

    $startTime = microtime(TRUE);

    try {
      if ($this->iteratorExecutor === NULL) {
        throw new OrchestrationException(
          "Iterator node requires the flowdrop_iterator module to be enabled"
        );
      }

      $output = $this->iteratorExecutor->execute(
        $executionId,
        $nodeId,
        $inputData,
        $config,
        $workflow,
        $pipelineId,
      );

      $executionTime = microtime(TRUE) - $startTime;

      $this->logger->info("Iterator node @node_id completed in @time seconds", [
        "@node_id" => $nodeId,
        "@time" => round($executionTime, 3),
        "@execution_id" => $executionId,
      ]);

      return new NodeExecutionResult(
        nodeId: $nodeId,
        output: $output,
        status: "success",
        executionTime: $executionTime,
        metadata: [
          "is_iterator" => TRUE,
          "child_pipeline_id" => $output->get("_childPipelineId"),
          "total_iterations" => $output->get("total"),
          "errors" => $output->get("errors"),
        ]
      );

    }
    catch (\Exception $e) {
      $this->logger->error("Iterator node @node_id execution failed: @error", [
        "@node_id" => $nodeId,
        "@error" => $e->getMessage(),
        "@execution_id" => $executionId,
      ]);

      throw new OrchestrationException(
        "Iterator execution failed: " . $e->getMessage(),
        0,
        $e
      );
    }
  }

  /**
   * Generates a workflow snapshot from execution results.
   *
   * Creates a WorkflowSnapshot that captures the complete state of the
   * workflow execution, including all completed node outputs. This snapshot
   * can be used for:
   * - Orchestrator handover (sync-to-async).
   * - Workflow resume from checkpoints.
   * - State inspection and debugging.
   *
   * @param string $executionId
   *   The execution ID.
   * @param \Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationRequest $request
   *   The orchestration request.
   * @param array<string, mixed> $workflowData
   *   The workflow definition.
   * @param array<string, \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionResult> $results
   *   Map of node ID to execution result.
   * @param array<string> $skippedNodes
   *   Array of skipped node IDs.
   * @param string $status
   *   The workflow status (completed|failed|paused).
   *
   * @return \Drupal\flowdrop_orchestration\DTO\Snapshot\WorkflowSnapshot|null
   *   The generated snapshot, or NULL if state manager not available.
   */
  private function generateSnapshot(
    string $executionId,
    OrchestrationRequest $request,
    array $workflowData,
    array $results,
    array $skippedNodes,
    string $status,
  ): ?WorkflowSnapshot {
    // If no state manager is available, skip snapshot generation.
    if ($this->workflowStateManager === NULL) {
      return NULL;
    }

    try {
      // Compute workflow version hash.
      $workflowVersion = $this->workflowStateManager->computeWorkflowVersion($workflowData);

      // Create the base snapshot.
      $snapshot = WorkflowSnapshot::create(
        workflowId: $request->getWorkflowId(),
        workflowVersion: $workflowVersion,
        initialInput: $request->getInitialData(),
        metadata: [
          "orchestrator_type" => $this->getType(),
          "source_execution_id" => $executionId,
          "generated_at" => time(),
        ],
      );

      // Initialize nodes from workflow definition.
      $nodes = $workflowData["nodes"] ?? [];
      $snapshot = $snapshot->withInitializedNodes($nodes);

      // Update status.
      $snapshot = $snapshot->withStatus($status);

      // Add completed nodes from results.
      $executionOrder = 1;
      foreach ($results as $nodeId => $result) {
        $outputArray = $result->getOutput()->toArray();
        $snapshot = $snapshot->withNodeCompleted(
          nodeId: $nodeId,
          output: $outputArray,
          executionOrder: $executionOrder++,
          injected: FALSE,
        );
      }

      // Mark skipped nodes.
      foreach ($skippedNodes as $nodeId) {
        $snapshot = $snapshot->withNodeSkipped($nodeId, "branch_not_active");
      }

      $this->logger->debug("Generated workflow snapshot for execution @exec", [
        "@exec" => $executionId,
        "@nodes_completed" => count($results),
        "@nodes_skipped" => count($skippedNodes),
      ]);

      return $snapshot;

    }
    catch (\Exception $e) {
      $this->logger->error("Failed to generate workflow snapshot: @error", [
        "@error" => $e->getMessage(),
        "@exec" => $executionId,
      ]);
      return NULL;
    }
  }

}
