<?php

declare(strict_types=1);

namespace Drupal\flowdrop_pipeline\Service;

use Drupal\flowdrop\Constants\ReservedName;
use Drupal\flowdrop\DTO\ExecutionEdge;
use Drupal\flowdrop\DTO\JsApi\Edge as TransportEdge;
use Drupal\flowdrop\Utility\NodeAnchor;
use Drupal\flowdrop_job\Constants\JobStatus;
use Drupal\flowdrop_job\FlowDropJobInterface;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface;
use Drupal\flowdrop_node_type\FlowDropNodeTypeInterface;
use Drupal\flowdrop_workflow\Exception\WorkflowExecutionException;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\NonExecutableNodeProcessorInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\TriggerNodeProcessorInterface;
use Drupal\flowdrop\Service\FlowDropNodeProcessorPluginManager;
use Drupal\flowdrop\Utility\EdgePortClassifier;
use Drupal\flowdrop_node_type\Service\NodeMetadataResolver;
use Drupal\flowdrop_workflow\DTO\WorkflowDTO;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Logger\LoggerChannelFactoryInterface;
use Drupal\flowdrop_pipeline\Event\JobCreatedEvent;
use Drupal\flowdrop_pipeline\Event\PipelineJobsGeneratedEvent;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

/**
 * Service for generating jobs from pipeline workflows.
 *
 * @internal This class is not part of the public API and may change without notice.
 */
class JobGenerationService {

  public function __construct(
    protected EntityTypeManagerInterface $entityTypeManager,
    protected LoggerChannelFactoryInterface $loggerFactory,
    protected EventDispatcherInterface $eventDispatcher,
    protected FlowDropNodeProcessorPluginManager $nodeProcessorManager,
    protected NodeMetadataResolver $metadataResolver,
  ) {}

  /**
   * Generate jobs for a pipeline based on its workflow.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline to generate jobs for.
   *
   * @return array
   *   Array of created job entities.
   *
   * @throws \Exception
   *   If job generation fails.
   */
  public function generateJobs(FlowDropPipelineRuntimeInterface $pipeline): array {
    $logger = $this->loggerFactory->get('flowdrop_pipeline');

    try {
      $logger->info('Starting job generation for pipeline @id', [
        '@id' => $pipeline->id(),
      ]);

      // Get the workflow.
      $workflow = $pipeline->getWorkflow();
      if (!$workflow) {
        throw new \RuntimeException('Workflow not found for pipeline ' . $pipeline->id());
      }

      // Get nodes and edges from workflow.
      $nodes = $workflow->getNodes();
      $edges = $workflow->getEdges();

      if (empty($nodes)) {
        throw new \RuntimeException('No nodes found in workflow ' . $workflow->id());
      }

      // Build detailed edge information for each node.
      $edge_info = $this->buildEdgeInformation($nodes, $edges);

      // Build dependency graph.
      $dependency_graph = $this->buildDependencyGraph($nodes, $edges);

      // Validate dependency graph for cycles.
      $this->validateDependencyGraph($dependency_graph);

      // Calculate execution order and priorities.
      $execution_order = $this->calculateExecutionOrder($dependency_graph);

      // Filter trigger nodes: only create job for the trigger node that
      // initiated this workflow. Skip all other trigger nodes in the
      // workflow.
      $input_data = $pipeline->getInputData();
      $trigger_node_id = $input_data['trigger_node_id'] ?? NULL;

      // Create jobs first (without dependencies).
      $created_jobs = [];
      $job_storage = $this->entityTypeManager->getStorage('flowdrop_job');
      // Map node IDs to job entities.
      $node_to_job_map = [];

      // Enrich the stored (slim) nodes from the live node type, then build a
      // WorkflowDTO. Enrichment re-injects each type's authoritative metadata
      // (executor plugin, config defaults, ports) — the same step every
      // orchestrator runs before constructing a DTO (see
      // WorkflowCompiler::compile()). The DTO then owns config-merge (node-type
      // defaults under the user's data.config) and node lookup, so this service
      // no longer carries a second copy of that logic. Edge classification
      // stays on EdgePortClassifier below: the DTO models neither loopback/tool
      // edges nor branch names, and the job-metadata edge shape is a persisted
      // contract.
      $enriched = $this->metadataResolver->enrichWorkflow([
        'id' => (string) $workflow->id(),
        'nodes' => $nodes,
        'edges' => $edges,
      ]);
      $workflowDTO = WorkflowDTO::fromArray($enriched);

      // Tripwire: by job-generation time the workflow has passed save-time R8,
      // so the DTO's repair-on-read should find nothing to prune. A non-empty
      // set means a dangling edge slipped past validation — log it rather than
      // silently diverge from the edges this service still reads raw below.
      if ($workflowDTO->getDroppedEdges() !== []) {
        $logger->warning('Workflow @id reached job generation with @count dangling edge(s) the DTO pruned on read; none expected after R8 validation.', [
          '@id' => $workflow->id(),
          '@count' => count($workflowDTO->getDroppedEdges()),
        ]);
      }

      // The job metadata persists the slim node_data verbatim (BC), so keep the
      // original stored node arrays addressable by id — enrichment must not
      // leak into what gets written back.
      $slimNodesById = [];
      foreach ($nodes as $slimNode) {
        if (is_array($slimNode) && isset($slimNode['id'])) {
          $slimNodesById[(string) $slimNode['id']] = $slimNode;
        }
      }

      // Typed transport edges for tool-only classification (camelCase graph
      // shape). Built once; the classifier reads the typed getters.
      $transportEdges = array_map(
        static fn(array $edge): TransportEdge => TransportEdge::fromArray($edge),
        array_values(array_filter($edges, 'is_array')),
      );

      foreach ($workflowDTO->getNodes() as $node_id => $nodeDTO) {
        $node_id = (string) $node_id;
        $rawNode = $slimNodesById[$node_id] ?? [];
        $node_type_id = NodeAnchor::read($rawNode);

        // Tool-only nodes (every outgoing edge supplies a tool to a
        // consumer) execute inline via the tool invoker when the consumer
        // calls them — scheduling them as ordinary jobs would run them once
        // with empty args. Mirrors WorkflowCompiler::shouldExcludeNode().
        if (EdgePortClassifier::isToolOnlyNode($node_id, $transportEdges)) {
          $logger->debug('Skipping tool-only node @node_id from job generation', [
            '@node_id' => $node_id,
          ]);
          continue;
        }

        // Non-executable nodes (e.g. Note) declare via a marker interface that
        // they have no runtime behaviour. They never become jobs, regardless of
        // wiring. Mirrors WorkflowCompiler::shouldExcludeNode(). The processor
        // plugin ID is namespaced (e.g. "flowdrop_node_processor:note"); it is
        // re-resolved from the node type (the source of truth), with the bare
        // node-type id as a last resort when the type cannot be resolved.
        $executor_plugin_id = ((string) $nodeDTO->getMetadataValue('executor_plugin', '')) ?: $node_type_id;
        if ($this->isNonExecutableNodeProcessor($executor_plugin_id)) {
          $logger->debug('Skipping non-executable node @node_id from job generation', [
            '@node_id' => $node_id,
          ]);
          continue;
        }

        // Filter trigger nodes: only include the triggering trigger node.
        // Uses the namespaced executor plugin ID (see $executor_plugin_id
        // above): metadata.node_type_id is the node-type entity ID and does
        // not resolve against the processor manager.
        if ($this->isTriggerNodeProcessor($executor_plugin_id)) {
          if ($trigger_node_id !== NULL && $node_id === $trigger_node_id) {
            // Include the triggering trigger node.
            $logger->debug('Including triggering trigger node @node_id in job generation', [
              '@node_id' => $node_id,
            ]);
          }
          else {
            // Skip all other trigger nodes.
            $logger->debug('Skipping non-triggering trigger node @node_id from job generation', [
              '@node_id' => $node_id,
            ]);
            continue;
          }
        }

        // Calculate priority based on execution order.
        $priority = $this->calculateJobPriority($node_id, $execution_order, $dependency_graph);

        // Merged config (node-type defaults under the user's values) is owned
        // by the WorkflowNodeDTO, built from the enriched node above — keeping
        // the merge in a single place shared with the compiler/orchestrators.
        // node_data stays the slim stored shape for BC.
        $nodeData = $rawNode['data'] ?? [];
        $mergedConfig = $nodeDTO->getConfig();

        // Prepare job metadata with edge information and config.
        // Metadata contains design-time configuration:
        // - node_id: The workflow node ID
        // - node_type_id: The node type entity ID
        // - config: Merged parameters (defaults + user-configured values)
        // - node_data: Original node data for reference
        // - incoming_edges/outgoing_edges: Edge connections.
        $metadata = [
          'node_id' => $node_id,
          'node_type_id' => $node_type_id,
          'config' => $mergedConfig,
          'node_data' => $nodeData,
          'incoming_edges' => $edge_info[$node_id]['incoming'] ?? [],
          'outgoing_edges' => $edge_info[$node_id]['outgoing'] ?? [],
        ];

        // Prepare job data (dependencies will be added as
        // entity references later).
        // input_data is initialized empty - it will be populated with
        // runtime inputs from upstream nodes during execution.
        // Jobs start in 'idle' status - they will be transitioned to 'pending'
        // when their dependencies are met and they are ready to run.
        $job_data = [
          'label' => $nodeData['label'] ?? 'Job ' . $node_id,
          'bundle' => 'default',
          'metadata' => json_encode($metadata),
          'node_type_id' => $node_type_id,
          'status' => JobStatus::IDLE,
          'priority' => $priority,
          'input_data' => json_encode([]),
          'output_data' => json_encode([]),
          'max_retries' => $this->getMaxRetriesForNode($rawNode),
        ];

        // Create the job.
        $job = $job_storage->create($job_data);
        $job->save();

        // Store mapping for dependency resolution.
        $node_to_job_map[$node_id] = $job;

        // Add job to pipeline's job_id field.
        if (!$job instanceof FlowDropJobInterface) {
          continue;
        }
        $pipeline->addJob($job);

        $created_jobs[] = $job;

        // Dispatch job created event.
        $this->eventDispatcher->dispatch(new JobCreatedEvent(
          job: $job,
          pipelineId: $pipeline->id(),
          nodeId: $node_id,
          dependencies: $dependency_graph[$node_id] ?? [],
        ), JobCreatedEvent::NAME);

        $logger->info('Created job @job_id for node @node_id in pipeline @pipeline_id', [
          '@job_id' => $job->id(),
          '@node_id' => $node_id,
          '@pipeline_id' => $pipeline->id(),
        ]);
      }

      // Now establish entity reference dependencies between jobs.
      foreach ($dependency_graph as $node_id => $dependency_node_ids) {
        if (empty($dependency_node_ids)) {
          // No dependencies for this node.
          continue;
        }

        $job = $node_to_job_map[$node_id];
        if (!$job instanceof FlowDropJobInterface) {
          continue;
        }

        // Add each dependency as an entity reference.
        foreach ($dependency_node_ids as $dependency_node_id) {
          if (isset($node_to_job_map[$dependency_node_id])) {
            $dependency_job = $node_to_job_map[$dependency_node_id];
            if (!$dependency_job instanceof FlowDropJobInterface) {
              continue;
            }
            $job->addDependentJob($dependency_job);
          }
        }

        // Save the job with its new dependencies.
        $job->save();
      }

      // Save pipeline with all added jobs.
      $pipeline->save();

      // Dispatch pipeline jobs generated event.
      $this->eventDispatcher->dispatch(new PipelineJobsGeneratedEvent(
        pipeline: $pipeline,
        jobsCreated: count($created_jobs),
        jobIds: array_map(fn($job) => $job->id(), $created_jobs),
      ), PipelineJobsGeneratedEvent::NAME);

      $logger->info('Successfully generated @count jobs for pipeline @id', [
        '@count' => count($created_jobs),
        '@id' => $pipeline->id(),
      ]);

      return $created_jobs;

    }
    catch (\Exception $e) {
      $logger->error('Failed to generate jobs for pipeline @id: @message', [
        '@id' => $pipeline->id(),
        '@message' => $e->getMessage(),
      ]);
      throw $e;
    }
  }

  /**
   * Build detailed edge information for each node.
   *
   * This stores full edge details including sourceHandle and targetHandle,
   * which are needed to implement gateway nodes and trigger inputs.
   *
   * @param array $nodes
   *   The workflow nodes.
   * @param array $edges
   *   The workflow edges.
   *
   * @return array
   *   Edge information keyed by node ID with 'incoming' and 'outgoing' arrays.
   */
  protected function buildEdgeInformation(array $nodes, array $edges): array {
    $edge_info = [];

    // Initialize empty edge arrays for all nodes.
    foreach ($nodes as $node) {
      $edge_info[$node['id']] = [
        'incoming' => [],
        'outgoing' => [],
      ];
    }

    // Process each edge.
    foreach ($edges as $edge) {
      $transportEdge = TransportEdge::fromArray($edge);
      $source_id = $transportEdge->getSource();
      $target_id = $transportEdge->getTarget();
      $source_handle = $transportEdge->getSourceHandle();
      $target_handle = $transportEdge->getTargetHandle();

      // Classify by reserved input ports (exact-suffix rules shared with
      // the compiler via EdgePortClassifier):
      // - Trigger inputs: {nodeId}-input-trigger.
      // - Loopback inputs: {nodeId}-input-loop_back (loop continuation;
      //   excluded from cycle detection, not regular dependencies).
      // - Tool inputs: {nodeId}-input-tool (tool supply; not dependencies,
      //   tool nodes are not jobs).
      $is_trigger = EdgePortClassifier::isTriggerTarget($target_handle);
      $is_loopback = EdgePortClassifier::isLoopbackTarget($target_handle);
      $is_tool = EdgePortClassifier::isToolEdge($transportEdge);

      // Extract the branch name from sourceHandle if present.
      // Format: {nodeId}-output-{branchName} (e.g., "node-123-output-True")
      // Branch names are used for gateway nodes (e.g., If-Else) that have
      // conditional outputs like "True", "False", "default".
      // The "trigger" output port is NOT a branch - it's a standard trigger
      // signal that doesn't require branch matching.
      $branch_name = '';
      $candidate_branch = EdgePortClassifier::portName($source_handle, 'output');
      if ($candidate_branch !== '' && $candidate_branch !== 'trigger') {
        $branch_name = $candidate_branch;
      }

      // The execution record owns the snake_case job-metadata serialisation;
      // building it once keeps the persisted key list in a single place.
      $executionEdge = new ExecutionEdge(
        source: $source_id,
        target: $target_id,
        sourceHandle: $source_handle,
        targetHandle: $target_handle,
        isTrigger: $is_trigger,
        isLoopback: $is_loopback,
        isTool: $is_tool,
        branchName: $branch_name,
        edgeId: $transportEdge->getId() !== '' ? $transportEdge->getId() : uniqid('edge_'),
      );

      // Store incoming edge for target node.
      if (isset($edge_info[$target_id])) {
        $edge_info[$target_id]['incoming'][] = $executionEdge->toMetadata();
      }

      // Store outgoing edge for source node.
      if (isset($edge_info[$source_id])) {
        $edge_info[$source_id]['outgoing'][] = $executionEdge->toMetadata();
      }
    }

    return $edge_info;
  }

  /**
   * Build dependency graph from workflow nodes and edges.
   *
   * This method excludes loopback edges from the dependency graph.
   * Loopback edges (targeting loop_back ports) are special edges that:
   * - Create intentional cycles for loop iteration.
   * - Should not be treated as regular dependencies.
   * - Are handled separately by the orchestrator for loop execution.
   *
   * @param array $nodes
   *   The workflow nodes.
   * @param array $edges
   *   The workflow edges.
   *
   * @return array
   *   Dependency graph where keys are node IDs and values are arrays of
   *   dependency node IDs.
   */
  protected function buildDependencyGraph(array $nodes, array $edges): array {
    $dependencies = [];

    // Initialize empty dependencies for all nodes.
    foreach ($nodes as $node) {
      $dependencies[$node['id']] = [];
    }

    // Build dependencies from edges.
    foreach ($edges as $edge) {
      $transportEdge = TransportEdge::fromArray($edge);
      $source_id = $transportEdge->getSource();
      $target_id = $transportEdge->getTarget();
      $target_handle = $transportEdge->getTargetHandle();

      // Skip loopback edges - they create intentional cycles for loops.
      // Loopback inputs have format: {nodeId}-input-loop_back.
      if (EdgePortClassifier::isLoopbackTarget($target_handle)) {
        continue;
      }

      // Skip tool edges - a consumer must not depend on its tools
      // "completing": tool nodes are not jobs, they execute inline via the
      // tool invoker when the consumer calls them.
      if (EdgePortClassifier::isToolEdge($transportEdge)) {
        continue;
      }

      // Target depends on source.
      if (isset($dependencies[$target_id])) {
        $dependencies[$target_id][] = $source_id;
      }
    }

    return $dependencies;
  }

  /**
   * Validate dependency graph for cycles.
   *
   * @param array $dependency_graph
   *   The dependency graph to validate.
   *
   * @throws \RuntimeException
   *   If a cycle is detected.
   */
  protected function validateDependencyGraph(array $dependency_graph): void {
    $visited = [];
    $rec_stack = [];

    foreach (array_keys($dependency_graph) as $node_id) {
      if (!isset($visited[$node_id])) {
        if ($this->hasCycleDfs($dependency_graph, $node_id, $visited, $rec_stack)) {
          throw new \RuntimeException('Circular dependency detected in workflow');
        }
      }
    }
  }

  /**
   * Depth-first search to detect cycles.
   *
   * @param array $dependency_graph
   *   The dependency graph.
   * @param string $node_id
   *   Current node ID.
   * @param array $visited
   *   Visited nodes tracker.
   * @param array $rec_stack
   *   Recursion stack tracker.
   *
   * @return bool
   *   TRUE if cycle is detected.
   */
  protected function hasCycleDfs(array $dependency_graph, string $node_id, array &$visited, array &$rec_stack): bool {
    $visited[$node_id] = TRUE;
    $rec_stack[$node_id] = TRUE;

    // Check all dependencies (reverse direction for cycle detection)
    foreach ($dependency_graph as $target_id => $dependencies) {
      if (in_array($node_id, $dependencies, TRUE)) {
        if (!isset($visited[$target_id])) {
          if ($this->hasCycleDfs($dependency_graph, $target_id, $visited, $rec_stack)) {
            return TRUE;
          }
        }
        elseif (isset($rec_stack[$target_id]) && $rec_stack[$target_id]) {
          return TRUE;
        }
      }
    }

    $rec_stack[$node_id] = FALSE;
    return FALSE;
  }

  /**
   * Calculate execution order using topological sort.
   *
   * @param array $dependency_graph
   *   The dependency graph.
   *
   * @return array
   *   Execution order where keys are node IDs and values are order indices.
   */
  protected function calculateExecutionOrder(array $dependency_graph): array {
    $in_degree = [];
    $execution_order = [];
    $order_index = 0;

    // Calculate in-degree for each node.
    foreach (array_keys($dependency_graph) as $node_id) {
      $in_degree[$node_id] = count($dependency_graph[$node_id]);
    }

    // Queue nodes with no dependencies.
    $queue = [];
    foreach ($in_degree as $node_id => $degree) {
      if ($degree === 0) {
        $queue[] = $node_id;
      }
    }

    // Process nodes in topological order.
    while (!empty($queue)) {
      $current_node = array_shift($queue);
      $execution_order[$current_node] = $order_index++;

      // Reduce in-degree for nodes that depend on current node.
      foreach ($dependency_graph as $node_id => $dependencies) {
        if (in_array($current_node, $dependencies, TRUE)) {
          $in_degree[$node_id]--;
          if ($in_degree[$node_id] === 0) {
            $queue[] = $node_id;
          }
        }
      }
    }

    return $execution_order;
  }

  /**
   * Calculate job priority based on execution order and dependencies.
   *
   * @param string $node_id
   *   The node ID.
   * @param array $execution_order
   *   The execution order.
   * @param array $dependency_graph
   *   The dependency graph.
   *
   * @return int
   *   The calculated priority (lower number = higher priority).
   */
  protected function calculateJobPriority(string $node_id, array $execution_order, array $dependency_graph): int {
    $base_priority = ($execution_order[$node_id] ?? 1000) * 10;

    // Adjust priority based on number of dependencies.
    $dependency_count = count($dependency_graph[$node_id] ?? []);
    $dependency_adjustment = $dependency_count * 5;

    return $base_priority + $dependency_adjustment;
  }

  /**
   * Get the executor plugin ID for a node.
   *
   * @param array $node
   *   The workflow node.
   *
   * @return string
   *   The executor plugin ID.
   */
  protected function getExecutorPluginIdForNode(array $node): string {
    $node_type_id = NodeAnchor::read($node) ?: 'unknown';

    // Try to load the node type entity to get the executor plugin.
    $node_type = $this->entityTypeManager->getStorage('flowdrop_node_type')->load($node_type_id);

    if ($node_type instanceof FlowDropNodeTypeInterface && $node_type->getExecutorPlugin()) {
      return $node_type->getExecutorPlugin();
    }

    // Fall back to a default or throw exception.
    throw new WorkflowExecutionException(
      "No executor plugin ID found for node type: $node_type_id"
    );
  }

  /**
   * Get max retries for a node based on its configuration.
   *
   * @param array $node
   *   The workflow node.
   *
   * @return int
   *   The maximum number of retries.
   */
  protected function getMaxRetriesForNode(array $node): int {
    // Read the injected `maxRetries` reserved config property (see
    // ReservedName::MAX_RETRIES / NodeMetadataResolver). The snake_case
    // `max_retries` is tolerated as a fallback for hand-authored config.
    $config = $node['data']['config'] ?? [];
    $value = $config[ReservedName::MAX_RETRIES] ?? $config['max_retries'] ?? NULL;
    if ($value !== NULL) {
      return max(0, (int) $value);
    }

    // Retry is opt-in: by default a node is not retried.
    return 0;
  }

  /**
   * Get ready jobs for a pipeline (jobs whose dependencies are completed).
   *
   * Jobs follow this status lifecycle:
   * - idle: Job is created but not yet ready to run (dependencies not met).
   * - pending: Job is ready to run (all dependencies met).
   * - running: Job is currently executing.
   * - completed/failed/cancelled/skipped: Terminal states.
   *
   * This method finds idle jobs whose dependencies are met and transitions
   * them to 'pending' status, then returns all pending jobs.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline.
   *
   * @return array
   *   Array of ready job entities (jobs in 'pending' status).
   */
  public function getReadyJobs(FlowDropPipelineRuntimeInterface $pipeline): array {
    $logger = $this->loggerFactory->get('flowdrop_pipeline');
    $completed_jobs = $pipeline->getJobsByStatus(JobStatus::COMPLETED);

    // Build a map of completed jobs by node ID for quick lookup. A node can
    // have several jobs in a loop (one per iteration) — keep the newest one
    // explicitly instead of relying on iteration order.
    $completed_jobs_map = [];
    foreach ($completed_jobs as $job) {
      $existing = $completed_jobs_map[$job->getNodeId()] ?? NULL;
      if ($existing === NULL || (int) $job->id() > (int) $existing->id()) {
        $completed_jobs_map[$job->getNodeId()] = $job;
      }
    }

    // Build a map of "handled failures" by node ID: jobs marked FAILED that
    // carry the `error_routed` flag (their failure was routed down an error
    // edge). These satisfy incoming error edges the same way a COMPLETED job
    // satisfies a data edge. Keep the newest per node (loop iterations).
    $error_routed_jobs_map = [];
    foreach ($pipeline->getJobsByStatus(JobStatus::FAILED) as $job) {
      if (($job->getMetadata()['error_routed'] ?? FALSE) !== TRUE) {
        continue;
      }
      $existing = $error_routed_jobs_map[$job->getNodeId()] ?? NULL;
      if ($existing === NULL || (int) $job->id() > (int) $existing->id()) {
        $error_routed_jobs_map[$job->getNodeId()] = $job;
      }
    }

    // Track the newest job per node across ALL statuses. A completed job
    // gates control flow only while it is the node's newest job: once a
    // newer job for the same node is in flight (a loop-iteration clone),
    // the old branch decision is stale and must not satisfy trigger edges
    // or branch gates of the new iteration.
    $latest_job_ids = [];
    foreach ($pipeline->getJobs() as $job) {
      $node_id = $job->getNodeId();
      $latest_job_ids[$node_id] = max($latest_job_ids[$node_id] ?? 0, (int) $job->id());
    }

    // Get all idle jobs and check if their dependencies are met.
    // If so, transition them to 'pending' status.
    $idle_jobs = $pipeline->getJobsByStatus(JobStatus::IDLE);
    foreach ($idle_jobs as $job) {
      if ($this->areJobDependenciesMet($job, $completed_jobs_map, $latest_job_ids, $error_routed_jobs_map)) {
        $logger->debug('Transitioning job @job_id (@label) from idle to pending', [
          '@job_id' => $job->id(),
          '@label' => $job->label(),
        ]);
        $job->markAsPending();
        $job->save();
      }
    }

    // Now get all pending jobs (including those just transitioned).
    $pending_jobs = $pipeline->getJobsByStatus(JobStatus::PENDING);

    // Sort by priority (lower number = higher priority).
    usort($pending_jobs, function ($a, $b) {
      return $a->getPriority() <=> $b->getPriority();
    });

    return $pending_jobs;
  }

  /**
   * Check if a job's dependencies are met.
   *
   * This method implements support for:
   * - Gateway nodes (e.g., IfElse) with conditional branches
   * - Trigger inputs that control execution flow
   * - Regular data inputs.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to check.
   * @param array $completed_jobs_map
   *   Map of completed node IDs to job entities (newest completed per node).
   * @param array<string, int> $latest_job_ids
   *   Newest job entity id per node ID, across all statuses.
   * @param array<string, \Drupal\flowdrop_job\FlowDropJobInterface> $error_routed_jobs_map
   *   Map of node IDs to their newest FAILED+`error_routed` job. These satisfy
   *   incoming error edges (a node's failure was routed to a handler).
   *
   * @return bool
   *   TRUE if all dependencies are met.
   */
  protected function areJobDependenciesMet(FlowDropJobInterface $job, array $completed_jobs_map, array $latest_job_ids = [], array $error_routed_jobs_map = []): bool {
    $logger = $this->loggerFactory->get('flowdrop_pipeline');
    $dependent_jobs = $job->getDependentJobs();

    if (empty($dependent_jobs)) {
      // No dependencies - job is ready to run.
      return TRUE;
    }

    // Get incoming edges from job metadata, hydrated into typed execution
    // records so the gating logic reads getters, not raw snake_case keys.
    $metadata = $job->getMetadata();
    $incoming_edges = array_map(
      static fn(array $edge): ExecutionEdge => ExecutionEdge::fromMetadata($edge),
      array_values(array_filter($metadata['incoming_edges'] ?? [], 'is_array')),
    );

    // Separate trigger, data, loopback, and error edges.
    // Loopback edges are excluded from dependency checking because they
    // represent loop-back connections that only have data AFTER the loop body
    // executes (e.g., ForEach node's loop_back input).
    // Error edges activate an error handler: they are satisfied when their
    // source node failed and routed its error down this edge.
    $trigger_edges = [];
    $data_edges = [];
    $loopback_edges = [];
    $error_edges = [];

    foreach ($incoming_edges as $edge) {
      if (EdgePortClassifier::isErrorEdge($edge)) {
        // Error edges gate execution like triggers, satisfied by a routed
        // failure rather than a completion. Checked first so a node-error
        // path is never mistaken for an ordinary data dependency.
        $error_edges[] = $edge;
      }
      elseif ($edge->isLoopback()) {
        // Loopback edges don't block execution - skip for dependency checking.
        $loopback_edges[] = $edge;
      }
      elseif ($edge->isTool()
        || EdgePortClassifier::isToolTarget($edge->getTargetHandle())) {
        // Tool edges supply tools, not data: tool nodes are not jobs (they
        // execute inline via the tool invoker), so they can never appear in
        // the completed map and must not gate readiness. The classifier
        // fallback covers jobs generated before the is_tool flag existed.
        continue;
      }
      elseif ($edge->isTrigger()) {
        $trigger_edges[] = $edge;
      }
      else {
        $data_edges[] = $edge;
      }
    }

    $logger->debug('Checking dependencies for job @job_id (@label): @trigger_count trigger, @data_count data, @loopback_count loopback, @error_count error edges', [
      '@job_id' => $job->id(),
      '@label' => $job->label(),
      '@trigger_count' => count($trigger_edges),
      '@data_count' => count($data_edges),
      '@loopback_count' => count($loopback_edges),
      '@error_count' => count($error_edges),
    ]);

    // RULE: An incoming error edge activates an error handler as soon as its
    // source node has failed and routed its error here (FAILED + error_routed),
    // mirroring trigger OR-semantics. This is checked before the trigger/data
    // rules so a node reachable from BOTH a success path and an error path
    // runs on whichever fires first.
    if (!empty($error_edges)) {
      foreach ($error_edges as $error_edge) {
        $source_node_id = $error_edge->getSource();
        $source_job = $error_routed_jobs_map[$source_node_id] ?? NULL;
        if ($source_job === NULL) {
          continue;
        }
        // A stale source (a newer job for the node is already in flight)
        // cannot satisfy the edge — same rule as trigger/branch gating.
        if (!$this->isNewestJobForNode($source_job, $latest_job_ids)) {
          continue;
        }
        $logger->info('Error edge satisfied for job @job_id (@label): source @source routed its failure here', [
          '@job_id' => $job->id(),
          '@label' => $job->label(),
          '@source' => $source_node_id,
        ]);
        return TRUE;
      }

      // A pure error handler (only error edges) whose source has not failed-
      // and-routed is not ready. With additional trigger/data edges, fall
      // through so the success path can still activate it.
      if (empty($trigger_edges) && empty($data_edges)) {
        $logger->debug('Job @job_id (@label) NOT ready: no error edge satisfied', [
          '@job_id' => $job->id(),
          '@label' => $job->label(),
        ]);
        return FALSE;
      }
    }

    // RULE: If a job has trigger inputs, ONLY triggers control execution.
    // Data edges are for fetching data, not for execution flow.
    if (!empty($trigger_edges)) {
      $logger->debug('Job @job_id has trigger inputs - checking ONLY triggers for execution', [
        '@job_id' => $job->id(),
      ]);

      $any_trigger_satisfied = FALSE;

      foreach ($trigger_edges as $trigger_edge) {
        $source_node_id = $trigger_edge->getSource();
        $branch_name = $trigger_edge->getBranchName();

        // Check if source node is completed.
        if (!isset($completed_jobs_map[$source_node_id])) {
          // Source not completed, this trigger not satisfied.
          $logger->debug('Trigger not satisfied for job @job_id: source @source not completed', [
            '@job_id' => $job->id(),
            '@source' => $source_node_id,
          ]);
          continue;
        }

        $source_job = $completed_jobs_map[$source_node_id];

        // A stale source cannot satisfy a trigger: when a newer job for the
        // source node is already in flight (a loop-iteration clone), the
        // completed job carries the PREVIOUS iteration's branch decision.
        // Treat it as not-completed — the consumer waits for the in-flight
        // job of its own iteration. Out-of-loop sources never have newer
        // jobs, so their (final) decision keeps satisfying triggers.
        if (!$this->isNewestJobForNode($source_job, $latest_job_ids)) {
          $logger->debug('Trigger not satisfied for job @job_id: source @source completed job is stale (newer job in flight)', [
            '@job_id' => $job->id(),
            '@source' => $source_node_id,
          ]);
          continue;
        }

        // For gateway nodes, check if the active_branches matches.
        if ($branch_name !== '') {
          $output_data = $source_job->getOutputData();
          $active_branches = $output_data['active_branches'] ?? '';

          $logger->debug('Checking branch match for job @job_id: branch @branch vs active @active', [
            '@job_id' => $job->id(),
            '@branch' => $branch_name,
            '@active' => $active_branches,
          ]);

          // Check if this branch was activated.
          // active_branches can be a single string (e.g., "true")
          // or potentially comma-separated for multiple branches.
          $active_branch_list = explode(',', strtolower($active_branches));
          $active_branch_list = array_map('trim', $active_branch_list);

          if (in_array(strtolower($branch_name), $active_branch_list, TRUE)) {
            // This trigger is satisfied!
            $logger->info('Trigger satisfied for job @job_id (@label): branch @branch matches', [
              '@job_id' => $job->id(),
              '@label' => $job->label(),
              '@branch' => $branch_name,
            ]);
            $any_trigger_satisfied = TRUE;
            break;
          }
        }
        else {
          // No branch name specified - any completion satisfies.
          $logger->info('Trigger satisfied for job @job_id (@label): no branch requirement', [
            '@job_id' => $job->id(),
            '@label' => $job->label(),
          ]);
          $any_trigger_satisfied = TRUE;
          break;
        }
      }

      // If no trigger is satisfied, job is not ready.
      if (!$any_trigger_satisfied) {
        $logger->info('Job @job_id (@label) NOT ready: no triggers satisfied', [
          '@job_id' => $job->id(),
          '@label' => $job->label(),
        ]);
        return FALSE;
      }

      // Trigger satisfied - job is ready!
      // Data edges are ignored for execution flow when triggers are present.
      $logger->info('Job @job_id (@label) is READY to execute (trigger satisfied)', [
        '@job_id' => $job->id(),
        '@label' => $job->label(),
      ]);
      return TRUE;
    }

    // No triggers - use port-aware dependency checking.
    //
    // DATA DEPENDENCY RULES:
    // - Multiple edges → SAME port:   OR logic (any one source is enough)
    // - Multiple edges → DIFFERENT ports: AND logic (all ports must have data)
    //
    // This allows branching workflows where only one branch executes,
    // but both branches connect to the same downstream input port.
    $logger->debug('Job @job_id has NO triggers - checking port-aware data dependencies', [
      '@job_id' => $job->id(),
    ]);

    // If no data edges remain after filtering (only loopback edges existed),
    // fall back to traditional dependency checking using dependent_jobs.
    // This ensures we still respect the dependency graph even when all edges
    // are loopback type.
    if (empty($data_edges)) {
      $logger->debug('Job @job_id has no data edges - falling back to dependent_jobs check', [
        '@job_id' => $job->id(),
      ]);

      foreach ($dependent_jobs as $dependent_job) {
        $dependency_node_id = $dependent_job->getNodeId();

        if (!isset($completed_jobs_map[$dependency_node_id])) {
          $logger->debug('Dependency not met for job @job_id: @dep_node_id not completed', [
            '@job_id' => $job->id(),
            '@dep_node_id' => $dependency_node_id,
          ]);
          return FALSE;
        }
      }

      $logger->info('Job @job_id (@label) is READY to execute (all dependent jobs completed)', [
        '@job_id' => $job->id(),
        '@label' => $job->label(),
      ]);
      return TRUE;
    }

    // Group data edges by target port.
    $edgesByTargetPort = [];
    foreach ($data_edges as $edge) {
      $targetHandle = $edge->getTargetHandle();
      $targetPort = $this->extractPortNameFromHandle($targetHandle, 'input');

      if ($targetPort === NULL) {
        // Fallback: use the full handle if port extraction fails.
        $targetPort = $targetHandle;
      }

      if (!isset($edgesByTargetPort[$targetPort])) {
        $edgesByTargetPort[$targetPort] = [];
      }
      $edgesByTargetPort[$targetPort][] = $edge;
    }

    $logger->debug('Job @job_id has @port_count unique target ports with data edges', [
      '@job_id' => $job->id(),
      '@port_count' => count($edgesByTargetPort),
    ]);

    // For each port, at least ONE source must be completed (OR within port).
    // All ports must have at least one completed source (AND across ports).
    foreach ($edgesByTargetPort as $targetPort => $portEdges) {
      $anySourceCompleted = FALSE;
      $sourceNodes = [];

      foreach ($portEdges as $edge) {
        $sourceNodeId = $edge->getSource();
        $sourceNodes[] = $sourceNodeId;

        if (isset($completed_jobs_map[$sourceNodeId])
          && $this->isEdgeBranchActive($edge, $completed_jobs_map, $latest_job_ids)) {
          $anySourceCompleted = TRUE;
          $logger->debug('Port @port on job @job_id: source @source is COMPLETED', [
            '@port' => $targetPort,
            '@job_id' => $job->id(),
            '@source' => $sourceNodeId,
          ]);
          // Found at least one completed source for this port - that's enough!
          break;
        }
      }

      if (!$anySourceCompleted) {
        $logger->debug('Port @port on job @job_id has NO completed sources. Sources: @sources', [
          '@port' => $targetPort,
          '@job_id' => $job->id(),
          '@sources' => implode(', ', $sourceNodes),
        ]);
        return FALSE;
      }
    }

    // All ports have at least one completed source!
    $logger->info('Job @job_id (@label) is READY to execute (all ports have data - port-aware OR logic)', [
      '@job_id' => $job->id(),
      '@label' => $job->label(),
    ]);
    return TRUE;
  }

  /**
   * Whether a data edge from a branch-emitting source is on an active branch.
   *
   * A source that declares `active_branches` in its output (gateways) gates
   * its BRANCH-port edges by branch membership — whatever input port the
   * edge lands on. Trigger-port targets always had this gating; without it
   * here, a gateway branch wired to a plain data port executed both branches.
   * Sources that emit no `active_branches` (ordinary nodes — their output
   * ports are values, not branches) gate nothing.
   *
   * A branch-emitting source may still expose ordinary data-output ports
   * alongside its branches — dual-mode nodes (e.g. Confirmation, which emits
   * a `confirmed` value and a true/false branch) and gateway pass-through
   * values. A source port is a genuine branch only when its name is NOT one
   * of the source's data-output keys; data ports carry values and are never
   * gated. Without this distinction the `active_branches` emission poisons the
   * node's own data edges (branch "confirmed" ∉ ["true"] → skipped downstream).
   *
   * @param \Drupal\flowdrop\DTO\ExecutionEdge $edge
   *   The incoming edge (job-metadata execution record).
   * @param array<string, \Drupal\flowdrop_job\FlowDropJobInterface> $completed_jobs_map
   *   Completed jobs keyed by node ID (newest completed per node).
   * @param array<string, int> $latest_job_ids
   *   Newest job entity id per node ID, across all statuses.
   *
   * @return bool
   *   FALSE only when the source declares branches and this edge's branch
   *   is not active.
   */
  protected function isEdgeBranchActive(ExecutionEdge $edge, array $completed_jobs_map, array $latest_job_ids = []): bool {
    $branch_name = $edge->getBranchName();
    if ($branch_name === '') {
      return TRUE;
    }

    $source_job = $completed_jobs_map[$edge->getSource()] ?? NULL;
    if (!$source_job instanceof FlowDropJobInterface) {
      // Not completed yet — the completion check owns that part.
      return TRUE;
    }

    $output_data = $source_job->getOutputData();
    $active_branches = $output_data['active_branches'] ?? NULL;
    if (!is_string($active_branches) || $active_branches === '') {
      // The source does not declare branches: its named outputs are
      // ordinary value ports.
      return TRUE;
    }

    // A branch-emitting source vouches only for its own iteration: with a
    // newer source job in flight, this decision is stale and cannot
    // activate the edge.
    if (!$this->isNewestJobForNode($source_job, $latest_job_ids)) {
      return FALSE;
    }

    // A port whose name matches a data-output key is a value carrier, not a
    // branch — it flows regardless of which branch is active. Only true
    // branch ports (names absent from the output payload, e.g. "true"/"false"
    // on a gateway) are subject to branch-membership gating.
    if (array_key_exists($branch_name, $output_data)) {
      return TRUE;
    }

    $active_branch_list = array_map('trim', explode(',', strtolower($active_branches)));
    return in_array(strtolower($branch_name), $active_branch_list, TRUE);
  }

  /**
   * Whether a job is the newest job that exists for its node.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to check.
   * @param array<string, int> $latest_job_ids
   *   Newest job entity id per node ID, across all statuses.
   *
   * @return bool
   *   TRUE when no newer job exists for the same node (or no map was given).
   */
  protected function isNewestJobForNode(FlowDropJobInterface $job, array $latest_job_ids): bool {
    $latest_id = $latest_job_ids[$job->getNodeId()] ?? 0;
    return $latest_id === 0 || (int) $job->id() >= $latest_id;
  }

  /**
   * Extracts the port name from a handle string.
   *
   * Handle format: {nodeId}-{direction}-{portName}
   * Examples:
   * - "text_input.1-output-text" → "text"
   * - "chat_model.1-input-message" → "message"
   * - "logger.1-input-trigger" → "trigger"
   *
   * @param string $handle
   *   The handle string to parse.
   * @param string $direction
   *   The direction to match ("input" or "output").
   *
   * @return string|null
   *   The extracted port name, or NULL if parsing fails.
   */
  protected function extractPortNameFromHandle(string $handle, string $direction): ?string {
    if (empty($handle)) {
      return NULL;
    }

    $pattern = "/-" . preg_quote($direction, "/") . "-/";
    $parts = preg_split($pattern, $handle, 2);

    if ($parts !== FALSE && count($parts) === 2) {
      return $parts[1];
    }

    return NULL;
  }

  /**
   * Clear all jobs for a pipeline.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline to clear jobs for.
   *
   * @return int
   *   Number of jobs deleted.
   */
  public function clearJobs(FlowDropPipelineRuntimeInterface $pipeline): int {
    $jobs = $pipeline->getJobs();
    $count = count($jobs);

    if ($count > 0) {
      $job_storage = $this->entityTypeManager->getStorage('flowdrop_job');
      $job_storage->delete($jobs);

      // Clear jobs from pipeline's job_id field.
      $pipeline->clearJobs();
      $pipeline->save();

      $this->loggerFactory->get('flowdrop_pipeline')->info('Cleared @count jobs for pipeline @id', [
        '@count' => $count,
        '@id' => $pipeline->id(),
      ]);
    }

    return $count;
  }

  /**
   * 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.
   */
  protected 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;
    }
  }

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

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

}
