<?php

declare(strict_types=1);

namespace Drupal\flowdrop_runtime\Service\Orchestrator;

use Drupal\flowdrop\DTO\ExecutionEdge;
use Drupal\flowdrop\DTO\OutputInterface;
use Drupal\flowdrop\Exception\NodeProcessingException;
use Drupal\flowdrop\Exception\WorkflowStopException;
use Drupal\flowdrop\Utility\EdgePortClassifier;
use Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface;
use Drupal\flowdrop_interrupt\Exception\InterruptRequiredException;
use Drupal\flowdrop_job\Constants\JobStatus;
use Drupal\flowdrop_job\FlowDropJobInterface;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface;
use Drupal\flowdrop_pipeline\Service\JobGenerationService;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionResult;
use Drupal\flowdrop_orchestration\DTO\Snapshot\NodeSnapshot;
use Drupal\flowdrop_orchestration\DTO\Snapshot\WorkflowSnapshot;
use Drupal\flowdrop_orchestration\Exception\InterruptExceptionInterface;
use Drupal\flowdrop_runtime\Service\RealTime\RealTimeManagerInterface;
use Drupal\flowdrop_runtime\Service\Runtime\NodeRuntimeServiceInterface;
use Drupal\flowdrop_runtime\Service\Storage\JobStorageInterface;
use Drupal\flowdrop_runtime\Service\Storage\PipelineStorageInterface;
use Drupal\flowdrop_runtime\Service\Storage\PipelineStorageRegistryInterface;
use Drupal\flowdrop_runtime\Service\WorkflowState\WorkflowStateManagerInterface;
use Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationResponse;
use Drupal\flowdrop_runtime\Event\ExecutionCancelledEvent;
use Drupal\flowdrop_runtime\Event\ExecutionPausedEvent;
use Drupal\flowdrop_runtime\Event\JobCompletedEvent;
use Drupal\flowdrop_runtime\Event\PipelineSignalPollEvent;
use Drupal\flowdrop_runtime\Service\Orchestrator\Trait\JobSkippingTrait;
use Drupal\flowdrop_runtime\Service\Orchestrator\Trait\PipelineCreationTrait;
use Drupal\flowdrop_runtime\Service\Orchestrator\Trait\PortNameExtractionTrait;
use Drupal\flowdrop_orchestration\Service\Orchestrator\OrchestratorInterface;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;
use Symfony\Component\EventDispatcher\GenericEvent;

/**
 * Abstract base class for all orchestrators.
 *
 * This class provides common functionality shared across all orchestrator
 * implementations, including:
 * - Pipeline and job storage abstraction
 * - Common execution methods (executeJob, buildRuntimeInputData)
 * - Interrupt handling
 * - Snapshot generation
 * - Event dispatching.
 *
 * All orchestrators should extend this class and implement:
 * - getType(): Returns the orchestrator type identifier
 * - orchestrate(): Main orchestration logic
 * - getCapabilities(): Returns capability flags
 *
 * This class implements PipelineAwareOrchestratorInterface directly, meaning
 * all subclasses formally declare a preferred pipeline bundle. The default
 * implementation returns PipelineStorageRegistryInterface::BUNDLE_DEFAULT.
 * Override getPreferredPipelineBundle() to select a different backend.
 *
 * @api
 */
abstract class AbstractOrchestratorBase implements OrchestratorInterface, PipelineAwareOrchestratorInterface {

  use InterruptHandlingTrait;
  use JobSkippingTrait;
  use PortNameExtractionTrait;
  use PipelineCreationTrait;

  /**
   * The pipeline storage registry.
   */
  private readonly PipelineStorageRegistryInterface $pipelineStorageRegistry;

  /**
   * The pipeline storage for the current bundle.
   *
   * Lazily initialized on first call to getPipelineStorage(). Private to
   * enforce use of getPipelineStorage() so the lazy-init invariant holds.
   */
  private PipelineStorageInterface $pipelineStorage;

  /**
   * Constructs a new AbstractOrchestratorBase.
   *
   * @param \Drupal\flowdrop_runtime\Service\Storage\PipelineStorageRegistryInterface $pipelineStorageRegistry
   *   The pipeline storage registry.
   * @param \Drupal\flowdrop_runtime\Service\Storage\JobStorageInterface $jobStorage
   *   The job storage service.
   * @param \Drupal\flowdrop_runtime\Service\Runtime\NodeRuntimeServiceInterface $nodeRuntime
   *   The node runtime service.
   * @param \Drupal\flowdrop_pipeline\Service\JobGenerationService $jobGenerationService
   *   The job generation service.
   * @param \Symfony\Component\EventDispatcher\EventDispatcherInterface $eventDispatcher
   *   The event dispatcher.
   * @param \Psr\Log\LoggerInterface $logger
   *   The logger service.
   * @param \Drupal\flowdrop_runtime\Service\WorkflowState\WorkflowStateManagerInterface|null $workflowStateManager
   *   The workflow state manager (optional).
   * @param \Drupal\flowdrop_runtime\Service\RealTime\RealTimeManagerInterface|null $realTimeManager
   *   The real-time manager (optional).
   */
  public function __construct(
    PipelineStorageRegistryInterface $pipelineStorageRegistry,
    protected readonly JobStorageInterface $jobStorage,
    protected readonly NodeRuntimeServiceInterface $nodeRuntime,
    protected readonly JobGenerationService $jobGenerationService,
    protected readonly EventDispatcherInterface $eventDispatcher,
    protected readonly LoggerInterface $logger,
    protected readonly ?WorkflowStateManagerInterface $workflowStateManager = NULL,
    protected readonly ?RealTimeManagerInterface $realTimeManager = NULL,
  ) {
    $this->pipelineStorageRegistry = $pipelineStorageRegistry;
  }

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

  /**
   * Generates a unique execution ID.
   *
   * @return string
   *   The execution ID.
   */
  protected function generateExecutionId(): string {
    return "exec_" . time() . "_" . uniqid();
  }

  /**
   * Reloads a pipeline to get fresh data from storage.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline to reload.
   *
   * @return \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface
   *   The reloaded pipeline.
   *
   * @throws \Drupal\flowdrop_runtime\Exception\OrchestrationException
   *   If the pipeline cannot be reloaded.
   */
  protected function reloadPipeline(FlowDropPipelineRuntimeInterface $pipeline): FlowDropPipelineRuntimeInterface {
    $pipelineId = $pipeline->id();
    if ($pipelineId === NULL) {
      return $pipeline;
    }

    // Reset caches to ensure fresh data.
    $this->getPipelineStorage()->resetCache([(string) $pipelineId]);
    $this->jobStorage->resetCache();

    $reloaded = $this->getPipelineStorage()->load($pipelineId);
    if ($reloaded === NULL) {
      throw new \RuntimeException("Failed to reload pipeline: {$pipelineId}");
    }

    return $reloaded;
  }

  /**
   * Executes a single job.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to execute.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline this job belongs to.
   * @param string $executionId
   *   The execution ID.
   *
   * @return array
   *   The job output data.
   *
   * @throws \Drupal\flowdrop_orchestration\Exception\InterruptExceptionInterface
   *   If the job is interrupted for human input.
   * @throws \Exception
   *   If job execution fails.
   */
  protected function executeJob(
    FlowDropJobInterface $job,
    FlowDropPipelineRuntimeInterface $pipeline,
    string $executionId,
  ): array {
    $this->logger->info("Executing job @job_id (node: @node_id)", [
      "@job_id" => $job->id(),
      "@node_id" => $job->getNodeId(),
      "@execution_id" => $executionId,
    ]);

    try {
      // Mark job as started.
      $job->markAsStarted();
      $this->jobStorage->save($job);

      // Get config from metadata.
      $metadata = $job->getMetadata();
      $configData = $metadata["config"] ?? [];

      // Build runtime input data from upstream jobs.
      $runtimeInputData = $this->buildRuntimeInputData($job, $pipeline);

      // Inject interrupt ID for resume support.
      $interruptId = $job->getMetadataValue("interrupt_id", NULL);
      if ($interruptId !== NULL) {
        $runtimeInputData["__interrupt_id__"] = $interruptId;
      }

      // Update job's input data.
      if (!empty($runtimeInputData)) {
        $job->setInputData($runtimeInputData);
        $this->jobStorage->save($job);
      }

      // Build execution context.
      $context = $this->buildNodeExecutionContext($job, $pipeline);
      $nodeTypeId = $job->getMetadataValue("node_type_id", "default");

      // Execute the node, retrying transient failures in place per the job's
      // retry budget.
      $result = $this->executeNodeWithRetry(
        $job,
        fn(): NodeExecutionResult => $this->nodeRuntime->executeNode(
          $executionId,
          $job->getNodeId(),
          $nodeTypeId,
          $runtimeInputData,
          $configData,
          $context
        )
      );

      // First-class node failure: a node that reported error status must fail
      // the job, not silently complete. Node processor failures are swallowed
      // into an error-status Output by NodeRuntimeService, so there is no
      // exception to catch here. Two outcomes once retries are exhausted:
      // - The node has an outgoing error edge → route the failure down it
      //   (mark FAILED + error_routed, emit the error payload) and return so
      //   the run continues and the error handler can pick it up.
      // - No error edge → re-raise to fail the run via the catch path below.
      $output = $result->getOutput();
      if ($output->isError()) {
        if ($this->nodeHasErrorEdge($job)) {
          return $this->routeNodeError($job, $pipeline, $output, $result->getExecutionTime(), $executionId);
        }
        throw new NodeProcessingException(
          $output->getErrorMessage() ?? "Node {$job->getNodeId()} reported an error"
        );
      }

      // Mark job as completed.
      $outputData = $output->toArray();
      $job->markAsCompleted($outputData);

      // Update execution metadata.
      $this->updateJobExecutionMetadata($job, $pipeline, $result->getExecutionTime());
      $this->jobStorage->save($job);

      $this->dispatchJobCompleted($job, $pipeline, $executionId);

      $this->logger->info("Job @job_id completed successfully", [
        "@job_id" => $job->id(),
        "@execution_id" => $executionId,
      ]);

      return $outputData;
    }
    catch (InterruptExceptionInterface $e) {
      // The flowdrop_interrupt subscriber may have already marked this job
      // as part of descriptor-shape persistence (it owns both entity
      // creation and job marking for that path). For legacy-shape
      // exceptions the entity is on the exception and we mark here.
      $reloadedJob = $this->jobStorage->load($job->id());
      if ($reloadedJob instanceof FlowDropJobInterface) {
        $job = $reloadedJob;
      }

      $interruptId = '';
      if ($job->getStatus() === JobStatus::INTERRUPTED) {
        $interruptId = (string) ($job->getMetadata()['interrupt_id'] ?? '');
      }
      else {
        $interruptId = $this->getInterruptIdFromException($e);
        $job->markAsInterrupted($interruptId, $e->getMessage());
        $this->jobStorage->save($job);
      }

      $this->logger->info("Job @job_id interrupted. Interrupt ID: @interrupt_id", [
        "@job_id" => $job->id(),
        "@interrupt_id" => $interruptId,
        "@execution_id" => $executionId,
      ]);

      throw $e;
    }
    catch (WorkflowStopException $e) {
      // A terminal node signals a clean stop, not a job failure. Mark this job
      // completed with the terminal payload (its intent was to stop), then
      // re-throw — before the generic \Exception catch below — so the
      // orchestration loop halts scheduling and completes the run.
      $job->markAsCompleted($e->getResult());
      $this->jobStorage->save($job);

      $this->dispatchJobCompleted($job, $pipeline, $executionId);

      $this->logger->info("Job @job_id stopped the workflow", [
        "@job_id" => $job->id(),
        "@execution_id" => $executionId,
      ]);

      throw $e;
    }
    catch (\Exception $e) {
      $job->markAsFailed($e->getMessage());
      $this->jobStorage->save($job);

      $this->dispatchJobCompleted($job, $pipeline, $executionId);

      $this->logger->error("Job @job_id failed: @message", [
        "@job_id" => $job->id(),
        "@message" => $e->getMessage(),
        "@execution_id" => $executionId,
      ]);

      throw $e;
    }
  }

  /**
   * Executes a node, retrying retryable failures in place.
   *
   * Node processor failures surface as an error-status Output rather than a
   * thrown exception (see NodeRuntimeService); a failure the processor marked
   * retryable additionally carries the `error_retryable` output-metadata flag.
   * While that flag is set and the job has retry budget remaining
   * ({@see FlowDropJobInterface::canRetry()}), the node is re-run from the top
   * immediately. The retry counter is persisted each attempt for visibility.
   *
   * Retry re-runs process() wholesale, so it is only safe on idempotent
   * nodes — the node author's contract, not an engine guarantee. Retry is
   * opt-in: a node's max_retries defaults to 0, so this loops exactly once
   * unless the workflow author configured retries.
   *
   * Attempts are immediate: there is NO delay or backoff between them. This
   * fits failures that clear on a same-instant re-run (optimistic-concurrency
   * conflicts, transient locks/races); for failures that only clear after a
   * wait (network timeouts, upstream 5xx, rate limiting) route the node's
   * error port to a Timer/Delay node instead — see RetryableExceptionInterface.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job being executed; carries the retry budget (max_retries) and
   *   counter (retry_count).
   * @param callable():\Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionResult $runNode
   *   Thunk that invokes the node runtime and returns its result. Supplied by
   *   the caller so each orchestrator keeps its own executeNode argument shape
   *   (e.g. StateGraph passes a tool collection and a state injector).
   *
   * @return \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionResult
   *   The successful result, or the final error result once retries are
   *   exhausted (the caller turns an error result into a failed job).
   */
  protected function executeNodeWithRetry(FlowDropJobInterface $job, callable $runNode): NodeExecutionResult {
    while (TRUE) {
      $result = $runNode();
      $output = $result->getOutput();

      // Stop on success, on a non-retryable failure, or when out of budget.
      if (
        !$output->isError()
        || $output->getMetadata('error_retryable') !== TRUE
        || !$job->canRetry()
      ) {
        return $result;
      }

      $job->incrementRetryCount();
      $this->jobStorage->save($job);
      $this->logger->info(
        "Retrying node @node (retry @n of @max) after error: @error",
        [
          "@node" => $job->getNodeId(),
          "@n" => $job->getRetryCount(),
          "@max" => $job->getMaxRetries(),
          "@error" => $output->getErrorMessage() ?? '',
        ]
      );
    }
  }

  /**
   * Polls for inward Cancel/Pause signals on the pipeline.
   *
   * Subclasses call this between job iterations. Dispatches a
   * {@see PipelineSignalPollEvent} which the flowdrop_interrupt subscriber
   * answers with any pending inward signal. Returns an OrchestrationResponse
   * if the pipeline should stop (signal attached); NULL to continue.
   *
   * For Cancel: marks pipeline cancelled, dispatches ExecutionCancelledEvent,
   * returns a cancelled response. Pending outward interrupts on the pipeline
   * were already cancelled by the subscriber before it attached.
   *
   * For Pause: marks pipeline paused, dispatches ExecutionPausedEvent,
   * returns a paused response. The signal itself stays pending — resolving
   * it later resumes the pipeline through the standard InterruptResolved path.
   */
  protected function pollPipelineSignal(
    FlowDropPipelineRuntimeInterface $pipeline,
    string $executionId,
    float $startTime,
  ): ?OrchestrationResponse {
    $pipelineId = (string) ($pipeline->id() ?? '');
    if ($pipelineId === '') {
      return NULL;
    }
    $workflow = $pipeline->getWorkflow();
    $workflowId = $workflow !== NULL ? (string) ($workflow->id() ?? '') : '';

    $event = new PipelineSignalPollEvent(
      pipelineId: $pipelineId,
      executionId: $executionId,
      workflowId: $workflowId,
    );
    $this->eventDispatcher->dispatch($event, PipelineSignalPollEvent::NAME);

    $signal = $event->getSignal();
    if ($signal === NULL) {
      return NULL;
    }

    return match ($signal['action']) {
      PipelineSignalPollEvent::ACTION_CANCEL => $this->applyCancelSignal(
        $pipeline, $executionId, $workflowId, $startTime, $signal,
      ),
      PipelineSignalPollEvent::ACTION_PAUSE => $this->applyPauseSignal(
        $pipeline, $executionId, $workflowId, $startTime, $signal,
      ),
      default => NULL,
    };
  }

  /**
   * Marks the pipeline cancelled and builds the OrchestrationResponse.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline runtime entity to cancel.
   * @param string $executionId
   *   The execution identifier.
   * @param string $workflowId
   *   The owning workflow identifier.
   * @param float $startTime
   *   The execution start time, used to compute total duration.
   * @param array{action: string, signal_id: string, reason: ?string, initiator_uid: ?int} $signal
   *   The incoming cancel signal payload.
   */
  private function applyCancelSignal(
    FlowDropPipelineRuntimeInterface $pipeline,
    string $executionId,
    string $workflowId,
    float $startTime,
    array $signal,
  ): OrchestrationResponse {
    $pipeline->markAsCancelled();
    $this->getPipelineStorage()->save($pipeline);

    $this->logger->info('Pipeline @pipeline cancelled by signal @signal (reason: @reason)', [
      '@pipeline' => $pipeline->id(),
      '@signal' => $signal['signal_id'],
      '@reason' => $signal['reason'] ?? '(none)',
    ]);

    $this->eventDispatcher->dispatch(
      new ExecutionCancelledEvent(
        executionId: $executionId,
        workflowId: $workflowId,
        pipelineId: (string) ($pipeline->id() ?? ''),
        signalId: $signal['signal_id'],
        reason: $signal['reason'],
        initiatorUid: $signal['initiator_uid'],
      ),
      ExecutionCancelledEvent::NAME,
    );

    return new OrchestrationResponse(
      executionId: $executionId,
      status: WorkflowSnapshot::STATUS_CANCELLED,
      results: [],
      executionTime: microtime(TRUE) - $startTime,
      metadata: [
        'orchestrator_type' => $this->getType(),
        'signal_id' => $signal['signal_id'],
        'reason' => $signal['reason'],
      ],
    );
  }

  /**
   * Marks the pipeline paused and builds the OrchestrationResponse.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline runtime entity to pause.
   * @param string $executionId
   *   The execution identifier.
   * @param string $workflowId
   *   The owning workflow identifier.
   * @param float $startTime
   *   The execution start time, used to compute total duration.
   * @param array{action: string, signal_id: string, reason: ?string, initiator_uid: ?int} $signal
   *   The incoming pause signal payload.
   */
  private function applyPauseSignal(
    FlowDropPipelineRuntimeInterface $pipeline,
    string $executionId,
    string $workflowId,
    float $startTime,
    array $signal,
  ): OrchestrationResponse {
    $pipeline->pause();
    $this->getPipelineStorage()->save($pipeline);

    $this->logger->info('Pipeline @pipeline paused by signal @signal (reason: @reason)', [
      '@pipeline' => $pipeline->id(),
      '@signal' => $signal['signal_id'],
      '@reason' => $signal['reason'] ?? '(none)',
    ]);

    $this->eventDispatcher->dispatch(
      new ExecutionPausedEvent(
        executionId: $executionId,
        workflowId: $workflowId,
        pipelineId: (string) ($pipeline->id() ?? ''),
        signalId: $signal['signal_id'],
        reason: $signal['reason'],
        initiatorUid: $signal['initiator_uid'],
      ),
      ExecutionPausedEvent::NAME,
    );

    return new OrchestrationResponse(
      executionId: $executionId,
      status: WorkflowSnapshot::STATUS_PAUSED,
      results: [],
      executionTime: microtime(TRUE) - $startTime,
      metadata: [
        'orchestrator_type' => $this->getType(),
        'signal_id' => $signal['signal_id'],
        'reason' => $signal['reason'],
      ],
    );
  }

  /**
   * Returns the interrupt UUID from a caught exception, or empty string.
   *
   * For legacy InterruptRequiredException the entity lives on the exception
   * and we read it via getInterrupt()->uuid(). For descriptor-shape, the
   * entity lives on the dispatched JobInterruptCaughtEvent — by the time
   * the orchestrator catch runs, the subscriber has already persisted the
   * entity AND called markAsInterrupted on the job, so the job's
   * metadata.interrupt_id is the authoritative source.
   *
   * This helper is only used when the job is NOT already marked — which
   * only happens for legacy-shape exceptions in pipeline-based orchestrators.
   */
  protected function getInterruptIdFromException(InterruptExceptionInterface $e): string {
    if ($e instanceof InterruptRequiredException) {
      $entity = $e->getInterrupt();
      if ($entity !== NULL) {
        return (string) ($entity->uuid() ?? '');
      }
    }
    // Stateless orchestrator path or non-interrupt-module exception: fall
    // back to the interface method. May throw for descriptor-shape — caller
    // shouldn't reach here in that case.
    return $e->getInterruptId();
  }

  /**
   * Resolves the persisted interrupt entity for a caught interrupt.
   *
   * Entity-shape exceptions carry the entity directly. Message-shape
   * exceptions have been persisted by JobInterruptCaughtSubscriber before
   * reaching the orchestrate-level catch, and the resulting interrupt UUID
   * is stamped on the most-recently-interrupted job. Use that to load the
   * entity so callers can hand it to
   * {@see InterruptHandlingTrait::handleInterruptException()} — without
   * this lookup the response metadata would lack the `interrupt.id`
   * field that downstream callers like SessionExecutionService read.
   *
   * Returns NULL when no entity can be resolved (e.g. flowdrop_interrupt
   * not installed, or stateless orchestrator path).
   */
  protected function resolvePersistedInterrupt(
    InterruptExceptionInterface $e,
    ?FlowDropPipelineRuntimeInterface $pipeline,
  ): ?FlowDropInterruptInterface {
    if ($e instanceof InterruptRequiredException) {
      $entity = $e->getInterrupt();
      if ($entity !== NULL) {
        return $entity;
      }
    }
    if ($pipeline === NULL) {
      return NULL;
    }
    // Find the most recently interrupted job on this pipeline and load its
    // persisted interrupt entity.
    $latestId = NULL;
    foreach ($pipeline->getJobsByStatus(JobStatus::INTERRUPTED) as $job) {
      if (!method_exists($job, 'getMetadata') || !method_exists($job, 'id')) {
        continue;
      }
      $interruptId = (string) ($job->getMetadata()['interrupt_id'] ?? '');
      if ($interruptId === '') {
        continue;
      }
      // getJobsByStatus order isn't guaranteed across storages; pick the
      // last-touched job by id (largest id = newest).
      if ($latestId === NULL || (int) $job->id() > $latestId['job_id']) {
        $latestId = ['job_id' => (int) $job->id(), 'uuid' => $interruptId];
      }
    }
    if ($latestId === NULL) {
      return NULL;
    }
    try {
      // Static service locator is used deliberately here: this defensive
      // fallback runs only for message-shape interrupts that bubble up
      // without being attached to the exception. Injecting
      // EntityTypeManager would ripple through every orchestrator
      // subclass + service definition for what is purely a degraded path.
      // @phpstan-ignore-next-line globalDrupalDependencyInjection.useDependencyInjection
      $storage = \Drupal::entityTypeManager()->getStorage('flowdrop_interrupt');
      $matches = $storage->loadByProperties(['uuid' => $latestId['uuid']]);
      $entity = $matches !== [] ? reset($matches) : NULL;
      return $entity instanceof
      FlowDropInterruptInterface ? $entity
        : NULL;
    }
    catch (\Throwable) {
      return NULL;
    }
  }

  /**
   * Dispatches the JobCompletedEvent for a job.
   *
   * Centralises the dispatch call so subclasses don't duplicate it.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job entity after execution.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline this job belongs to.
   * @param string $executionId
   *   The execution ID for this pipeline run.
   */
  protected function dispatchJobCompleted(FlowDropJobInterface $job, FlowDropPipelineRuntimeInterface $pipeline, string $executionId = ''): void {
    $this->eventDispatcher->dispatch(
      new JobCompletedEvent($job, $pipeline->getExecutionContext(), $executionId),
      JobCompletedEvent::NAME,
    );
  }

  /**
   * Builds runtime input data for a job from upstream completed jobs.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to build input data for.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline.
   *
   * @return array
   *   Runtime input data mapped by port names.
   */
  protected function buildRuntimeInputData(
    FlowDropJobInterface $job,
    FlowDropPipelineRuntimeInterface $pipeline,
  ): array {
    $runtimeData = [];
    $metadata = $job->getMetadata();
    $incomingEdges = $metadata["incoming_edges"] ?? [];

    // Build map of completed jobs.
    $completedJobs = $pipeline->getJobsByStatus(JobStatus::COMPLETED);
    $completedJobsMap = [];
    $executionOrderMap = [];
    foreach ($completedJobs as $completedJob) {
      $completedJobsMap[$completedJob->getNodeId()] = $completedJob;
      $jobMetadata = $completedJob->getMetadata();
      $order = $jobMetadata["execution_order"] ?? 0;
      if ($order > 0) {
        $executionOrderMap[$completedJob->getNodeId()] = $order;
      }
    }

    // Build map of "handled failures": FAILED jobs that routed their error
    // down an error edge. An error handler reads its source's error payload
    // from here, since a routed-failure source is never in completedJobsMap.
    $errorRoutedMap = $this->buildErrorRoutedMap($pipeline);

    // Convert to typed edges once, then sort by execution order (latest first).
    $edges = array_map(
      static fn (array $rawEdge): ExecutionEdge => ExecutionEdge::fromMetadata($rawEdge),
      $incomingEdges,
    );
    usort($edges, function (ExecutionEdge $a, ExecutionEdge $b) use ($executionOrderMap): int {
      $orderA = $executionOrderMap[$a->getSource()] ?? 0;
      $orderB = $executionOrderMap[$b->getSource()] ?? 0;

      if ($orderA > 0 && $orderB > 0) {
        return $orderB <=> $orderA;
      }
      if ($orderA > 0) {
        return -1;
      }
      if ($orderB > 0) {
        return 1;
      }
      return strcmp($a->getSource(), $b->getSource());
    });

    // Track port sources for conflict detection.
    $portSources = [];

    foreach ($edges as $edge) {
      if ($edge->isTrigger()) {
        continue;
      }

      // Error edge: feed the source's routed error payload to the handler.
      if (EdgePortClassifier::isErrorEdge($edge)) {
        $this->mapErrorEdgeInput($edge, $errorRoutedMap, $runtimeData, $portSources);
        continue;
      }

      $sourceNodeId = $edge->getSource();
      if (!isset($completedJobsMap[$sourceNodeId])) {
        continue;
      }

      $sourceJob = $completedJobsMap[$sourceNodeId];
      $sourceOutput = $sourceJob->getOutputData();

      $sourcePortName = $this->extractPortName($edge->getSourceHandle(), "output");
      $targetPortName = $this->extractPortName($edge->getTargetHandle(), "input");

      if ($sourcePortName !== NULL && $targetPortName !== NULL && isset($sourceOutput[$sourcePortName])) {
        // Last-executor-wins: edges are 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 array.
        if (isset($portSources[$targetPortName])) {
          $this->logger->warning(
            "Multiple sources target port '@port' on job '@job'; keeping value " .
            "from latest executor '@kept', ignoring '@ignored'.",
            [
              "@port" => $targetPortName,
              "@job" => $job->id(),
              "@kept" => $portSources[$targetPortName],
              "@ignored" => $sourceNodeId,
            ]
          );
          continue;
        }
        $runtimeData[$targetPortName] = $sourceOutput[$sourcePortName];
        $portSources[$targetPortName] = $sourceNodeId;
      }
    }

    return $runtimeData;
  }

  /**
   * Builds a map of nodes whose failure was routed down an error edge.
   *
   * A "handled failure" is a job marked FAILED carrying the `error_routed`
   * metadata flag; its `error` output is the payload an error handler reads.
   * Keyed by node ID, newest job wins (loop iterations re-run a node).
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline.
   *
   * @return array<string, \Drupal\flowdrop_job\FlowDropJobInterface>
   *   Map of node ID to its newest FAILED+`error_routed` job.
   */
  protected function buildErrorRoutedMap(FlowDropPipelineRuntimeInterface $pipeline): array {
    $map = [];
    foreach ($pipeline->getJobsByStatus(JobStatus::FAILED) as $job) {
      if (($job->getMetadata()["error_routed"] ?? FALSE) !== TRUE) {
        continue;
      }
      $existing = $map[$job->getNodeId()] ?? NULL;
      if ($existing === NULL || (int) $job->id() > (int) $existing->id()) {
        $map[$job->getNodeId()] = $job;
      }
    }
    return $map;
  }

  /**
   * Maps a routed error payload onto an error handler's target input port.
   *
   * The source node emits its error on the reserved `error` output port; the
   * handler receives that payload on whatever input port the error edge lands
   * on. Honours the same last-executor-wins guard as data edges.
   *
   * @param \Drupal\flowdrop\DTO\ExecutionEdge $edge
   *   The incoming error edge (job-metadata execution record).
   * @param array<string, \Drupal\flowdrop_job\FlowDropJobInterface> $errorRoutedMap
   *   Map of node IDs to their FAILED+`error_routed` job.
   * @param array<string, mixed> $runtimeData
   *   The runtime input data being assembled (by reference).
   * @param array<string, string> $portSources
   *   Map of target port to the source already written there (by reference).
   */
  protected function mapErrorEdgeInput(
    ExecutionEdge $edge,
    array $errorRoutedMap,
    array &$runtimeData,
    array &$portSources,
  ): void {
    $sourceNodeId = $edge->getSource();
    if (!isset($errorRoutedMap[$sourceNodeId])) {
      return;
    }
    $payload = $errorRoutedMap[$sourceNodeId]->getOutputData()["error"] ?? NULL;
    if ($payload === NULL) {
      return;
    }
    $targetPortName = $this->extractPortName($edge->getTargetHandle(), "input");
    if ($targetPortName === NULL) {
      return;
    }
    if (isset($portSources[$targetPortName])) {
      $this->logger->warning(
        "Multiple sources target port '@port'; keeping value from latest " .
        "executor '@kept', ignoring error source '@ignored'.",
        [
          "@port" => $targetPortName,
          "@kept" => $portSources[$targetPortName],
          "@ignored" => $sourceNodeId,
        ]
      );
      return;
    }
    $runtimeData[$targetPortName] = $payload;
    $portSources[$targetPortName] = $sourceNodeId;
  }

  /**
   * Whether a job's node has an outgoing error edge.
   *
   * An error edge ("{nodeId}-output-error" → handler) means the workflow
   * author declared an alternate path for this node's failure. Read from the
   * job's stored outgoing edges (job-metadata shape).
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to check.
   *
   * @return bool
   *   TRUE when at least one outgoing edge emits from the error port.
   */
  protected function nodeHasErrorEdge(FlowDropJobInterface $job): bool {
    $outgoing = $job->getMetadata()["outgoing_edges"] ?? [];
    foreach ($outgoing as $edge) {
      if (is_array($edge) && EdgePortClassifier::isErrorEdge(ExecutionEdge::fromMetadata($edge))) {
        return TRUE;
      }
    }
    return FALSE;
  }

  /**
   * Routes a node's terminal failure down its error edge.
   *
   * Marks the job FAILED + `error_routed` (the "handled failure" state) and
   * emits the failure payload on the reserved `error` output port. The job is
   * FAILED — not COMPLETED — so its success-path successors never become ready
   * and are skipped; only an error edge is satisfied by an `error_routed` job
   * (see JobGenerationService::areJobDependenciesMet). The run continues so the
   * error handler executes; the pipeline still ends COMPLETED unless an
   * *unhandled* failure occurs elsewhere ({@see hasUnhandledFailedJobs()}).
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The failing job.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline.
   * @param \Drupal\flowdrop\DTO\OutputInterface $output
   *   The error-status output the node produced.
   * @param float $executionTime
   *   The node's execution time, stamped for last-executor-wins ordering.
   * @param string $executionId
   *   The execution ID.
   *
   * @return array<string, mixed>
   *   The routed error output ({error: {...}}) the job now carries.
   */
  protected function routeNodeError(
    FlowDropJobInterface $job,
    FlowDropPipelineRuntimeInterface $pipeline,
    OutputInterface $output,
    float $executionTime,
    string $executionId,
  ): array {
    $payload = [
      "message" => $output->getErrorMessage() ?? "Node {$job->getNodeId()} reported an error",
      "code" => $output->getErrorCode(),
      "node_id" => $job->getNodeId(),
      "retryable" => $output->getMetadata("error_retryable") === TRUE,
    ];
    $errorOutput = ["error" => $payload];

    $job->markAsFailed($payload["message"]);
    $job->setOutputData($errorOutput);
    $metadata = $job->getMetadata();
    $metadata["error_routed"] = TRUE;
    $job->setMetadata($metadata);

    // Stamp execution order so a downstream handler resolves this source
    // last-executor-wins, consistent with completed jobs.
    $this->updateJobExecutionMetadata($job, $pipeline, $executionTime);
    $this->jobStorage->save($job);

    // JobCompletedEvent is the generic "job finished" signal — fired for both
    // completed and failed jobs (see the event's docblock and the unhandled-
    // failure catch path in executeJob, which dispatches it after
    // markAsFailed). Subscribers discriminate on $job->getStatus(); this job
    // is FAILED + error_routed. Dispatching here keeps a routed failure on the
    // same per-node-result channel as every other terminal job.
    $this->dispatchJobCompleted($job, $pipeline, $executionId);

    $this->logger->info("Job @job_id failed and routed its error down the error edge: @message", [
      "@job_id" => $job->id(),
      "@message" => $payload["message"],
    ]);

    return $errorOutput;
  }

  /**
   * Whether the pipeline has a FAILED job whose failure was NOT routed.
   *
   * A node that failed and routed its error down an error edge is FAILED but
   * `error_routed` — a handled failure that must not fail the whole run.
   * Distinguishes "an error handler took over" from "the run blew up", which
   * {@see FlowDropPipelineRuntimeInterface::hasFailedJobs()} cannot.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline.
   *
   * @return bool
   *   TRUE when at least one FAILED job lacks the `error_routed` flag.
   */
  protected function hasUnhandledFailedJobs(FlowDropPipelineRuntimeInterface $pipeline): bool {
    foreach ($pipeline->getJobsByStatus(JobStatus::FAILED) as $job) {
      if (($job->getMetadata()["error_routed"] ?? FALSE) !== TRUE) {
        return TRUE;
      }
    }
    return FALSE;
  }

  /**
   * Builds node execution context.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job being executed.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline.
   *
   * @return \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext
   *   The execution context.
   */
  protected function buildNodeExecutionContext(
    FlowDropJobInterface $job,
    FlowDropPipelineRuntimeInterface $pipeline,
  ): NodeExecutionContext {
    $workflow = $pipeline->getWorkflow();
    $workflowId = $workflow !== NULL ? ($workflow->id() ?? "") : "";
    $pipelineId = $pipeline->id() ?? "";

    $metadata = [
      "job_id" => $job->id(),
      "node_id" => $job->getNodeId(),
    ];

    // Include session_id for interrupt handling.
    $pipelineContext = $pipeline->getExecutionContext();
    if (isset($pipelineContext["playground"]["session_id"])) {
      $metadata["session_id"] = $pipelineContext["playground"]["session_id"];
    }

    return new NodeExecutionContext(
      workflowId: $workflowId,
      pipelineId: $pipelineId,
      initialData: $pipeline->getInputData(),
      metadata: $metadata
    );
  }

  /**
   * Updates job execution metadata after successful execution.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline.
   * @param float $executionTime
   *   The execution time in seconds.
   */
  protected function updateJobExecutionMetadata(
    FlowDropJobInterface $job,
    FlowDropPipelineRuntimeInterface $pipeline,
    float $executionTime,
  ): void {
    // Calculate execution order.
    $allJobs = $pipeline->getJobs();
    $maxOrder = 0;
    foreach ($allJobs as $otherJob) {
      $otherMetadata = $otherJob->getMetadata();
      $maxOrder = max($maxOrder, $otherMetadata["execution_order"] ?? 0);
    }

    $jobMetadata = $job->getMetadata();
    $jobMetadata["execution_order"] = $maxOrder + 1;
    $jobMetadata["execution_time_us"] = (int) round($executionTime * 1000000);
    $job->setMetadata($jobMetadata);
  }

  /**
   * Generates a WorkflowSnapshot from pipeline state.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline.
   * @param string $executionId
   *   The execution ID.
   * @param string $status
   *   The snapshot status.
   *
   * @return \Drupal\flowdrop_orchestration\DTO\Snapshot\WorkflowSnapshot|null
   *   The snapshot, or NULL if state manager is unavailable.
   */
  protected function generateSnapshotFromPipeline(
    FlowDropPipelineRuntimeInterface $pipeline,
    string $executionId,
    string $status,
  ): ?WorkflowSnapshot {
    if ($this->workflowStateManager === NULL) {
      return NULL;
    }

    try {
      $workflow = $pipeline->getWorkflow();
      $workflowId = $workflow !== NULL ? ($workflow->id() ?? "unknown") : "unknown";
      $workflowVersion = "";

      if ($workflow !== NULL) {
        $workflowVersion = $this->workflowStateManager->computeWorkflowVersion([
          "nodes" => $workflow->getNodes(),
          "edges" => $workflow->getEdges(),
        ]);
      }

      $nodeStates = [];
      foreach ($pipeline->getJobs() as $job) {
        $nodeStates[$job->getNodeId()] = $this->createNodeSnapshotFromJob($job);
      }

      return new WorkflowSnapshot(
        workflowId: $workflowId,
        workflowVersion: $workflowVersion,
        executionId: $executionId,
        status: $status,
        nodeStates: $nodeStates,
        initialInput: $pipeline->getInputData(),
        metadata: [
          "orchestrator_type" => $this->getType(),
          "pipeline_id" => $pipeline->id(),
        ],
        createdAt: time(),
        updatedAt: time(),
      );
    }
    catch (\Exception $e) {
      $this->logger->error("Failed to generate snapshot: @error", [
        "@error" => $e->getMessage(),
      ]);
      return NULL;
    }
  }

  /**
   * Creates a NodeSnapshot from a job.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job.
   *
   * @return \Drupal\flowdrop_orchestration\DTO\Snapshot\NodeSnapshot
   *   The node snapshot.
   */
  protected function createNodeSnapshotFromJob(FlowDropJobInterface $job): NodeSnapshot {
    $jobStatus = $job->getStatus();
    $jobMetadata = $job->getMetadata();

    $nodeStatus = match ($jobStatus) {
      JobStatus::IDLE => NodeSnapshot::STATUS_IDLE,
      JobStatus::PENDING => NodeSnapshot::STATUS_PENDING,
      JobStatus::RUNNING => NodeSnapshot::STATUS_RUNNING,
      JobStatus::COMPLETED => NodeSnapshot::STATUS_COMPLETED,
      JobStatus::FAILED => NodeSnapshot::STATUS_FAILED,
      JobStatus::SKIPPED, JobStatus::CANCELLED => NodeSnapshot::STATUS_SKIPPED,
      JobStatus::INTERRUPTED => NodeSnapshot::STATUS_INTERRUPTED,
      default => NodeSnapshot::STATUS_IDLE,
    };

    return new NodeSnapshot(
      nodeId: $job->getNodeId(),
      status: $nodeStatus,
      output: $jobStatus === JobStatus::COMPLETED ? $job->getOutputData() : NULL,
      error: $jobStatus === JobStatus::FAILED ? $job->getErrorMessage() : NULL,
      injected: (bool) ($jobMetadata["injected"] ?? FALSE),
      executionOrder: (int) ($jobMetadata["execution_order"] ?? 0),
      metadata: ["job_id" => $job->id()],
    );
  }

  /**
   * Dispatches an event.
   *
   * @param string $eventName
   *   The event name.
   * @param mixed $subject
   *   The event subject.
   * @param array $context
   *   Additional context.
   */
  protected function dispatchEvent(string $eventName, mixed $subject, array $context = []): void {
    $this->eventDispatcher->dispatch(
      new GenericEvent($subject, $context),
      $eventName
    );
  }

  // =========================================================================
  // Trait accessor implementations
  // =========================================================================

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

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

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

  /**
   * {@inheritdoc}
   */
  protected function getJobSkippingStorage(): JobStorageInterface {
    return $this->jobStorage;
  }

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

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

  /**
   * {@inheritdoc}
   */
  protected function getInterruptRealTimeManager(): RealTimeManagerInterface {
    if ($this->realTimeManager === NULL) {
      throw new \RuntimeException("RealTimeManager required but not injected.");
    }
    return $this->realTimeManager;
  }

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

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

}
