<?php

declare(strict_types=1);

namespace Drupal\flowdrop_stategraph\Service;

use Drupal\flowdrop_job\FlowDropJobInterface;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface;
use Drupal\flowdrop_pipeline\Service\JobGenerationService;
use Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationRequest;
use Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationResponse;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext;
use Drupal\flowdrop_orchestration\DTO\Snapshot\NodeSnapshot;
use Drupal\flowdrop_orchestration\DTO\Snapshot\WorkflowSnapshot;
use Drupal\flowdrop_orchestration\Exception\InterruptExceptionInterface;
use Drupal\flowdrop_runtime\Exception\OrchestrationException;
use Drupal\flowdrop_runtime\Service\Orchestrator\AbstractOrchestratorBase;
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\WorkflowState\WorkflowStateManagerInterface;
use Drupal\flowdrop_stategraph\Checkpointer\CheckpointerInterface;
use Drupal\flowdrop_stategraph\DTO\ExecutionConfig;
use Drupal\flowdrop_stategraph\DTO\GraphState;
use Drupal\flowdrop_stategraph\DTO\StateUpdate;
use Drupal\flowdrop_stategraph\Exception\ApprovalRequiredException;
use Drupal\flowdrop_stategraph\Exception\MaxIterationsException;
use Drupal\flowdrop_stategraph\Routing\ConditionalEdgeEvaluator;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

/**
 * StateGraph Orchestrator - stateful graph execution with checkpointing.
 *
 * This orchestrator provides:
 * - Typed state with reducer-based updates.
 * - Checkpointing for durable execution.
 * - Conditional edge routing for loops.
 * - Iteration limits for safety.
 * - Thread-based conversation continuity.
 *
 * @see \Drupal\flowdrop_runtime\Service\Orchestrator\AbstractOrchestratorBase
 *
 * @internal This class is not part of the public API and may change without notice.
 */
class StateGraphOrchestrator extends AbstractOrchestratorBase {

  /**
   * Constructs a new StateGraphOrchestrator.
   *
   * @param \Drupal\flowdrop_runtime\Service\Storage\PipelineStorageInterface $pipelineStorage
   *   The pipeline storage service.
   * @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_stategraph\Service\StateManagerInterface $stateManager
   *   The state manager service.
   * @param \Drupal\flowdrop_stategraph\Service\CheckpointerFactory $checkpointerFactory
   *   The checkpointer factory.
   * @param \Drupal\flowdrop_stategraph\Routing\ConditionalEdgeEvaluator $edgeEvaluator
   *   The conditional edge evaluator.
   * @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(
    PipelineStorageInterface $pipelineStorage,
    JobStorageInterface $jobStorage,
    NodeRuntimeServiceInterface $nodeRuntime,
    JobGenerationService $jobGenerationService,
    EventDispatcherInterface $eventDispatcher,
    LoggerInterface $logger,
    protected readonly StateManagerInterface $stateManager,
    protected readonly CheckpointerFactory $checkpointerFactory,
    protected readonly ConditionalEdgeEvaluator $edgeEvaluator,
    ?WorkflowStateManagerInterface $workflowStateManager = NULL,
    ?RealTimeManagerInterface $realTimeManager = NULL,
  ) {
    parent::__construct(
      $pipelineStorage,
      $jobStorage,
      $nodeRuntime,
      $jobGenerationService,
      $eventDispatcher,
      $logger,
      $workflowStateManager,
      $realTimeManager,
    );
  }

  /**
   * {@inheritdoc}
   */
  public function getType(): string {
    return "flowdrop_stategraph:stategraph";
  }

  /**
   * {@inheritdoc}
   */
  public function orchestrate(OrchestrationRequest $request): OrchestrationResponse {
    $executionId = $this->generateExecutionId();
    $startTime = microtime(TRUE);

    $this->logger->info("StateGraph orchestration started", [
      "@execution_id" => $executionId,
      "@workflow_id" => $request->getWorkflowId(),
    ]);

    try {
      // Extract execution configuration from options.
      $options = $request->getOptions();
      $config = ExecutionConfig::fromArray($options["stategraph"] ?? []);

      // Create checkpointer based on configuration.
      $checkpointer = $this->checkpointerFactory->create($config->checkpointerType);

      // Initialize or restore state.
      $initialSnapshot = $request->getInitialSnapshot();
      if ($initialSnapshot !== NULL) {
        $this->logger->info("Initializing StateGraph from WorkflowSnapshot @id", [
          "@id" => $initialSnapshot->executionId,
        ]);
        $state = $this->convertFromWorkflowSnapshot($initialSnapshot, $config->threadId);
      }
      else {
        $state = $this->initializeState($config, $checkpointer, $request->getInitialData());
      }

      // Create pipeline from workflow using trait.
      $pipeline = $this->createPipelineFromWorkflow(
        $request->getWorkflow(),
        $request->getInitialData(),
        $options
      );

      if ($pipeline === NULL) {
        throw new OrchestrationException("Failed to create pipeline from workflow");
      }

      // Generate jobs for the pipeline.
      $this->jobGenerationService->generateJobs($pipeline, $initialSnapshot);

      // Reload pipeline to ensure fresh job references.
      $pipeline = $this->reloadPipeline($pipeline);

      // Execute the pipeline with stateful tracking.
      $response = $this->executePipelineWithState(
        $pipeline,
        $state,
        $config,
        $checkpointer,
        $executionId,
        $startTime
      );

      // Generate and attach workflow snapshot.
      $finalState = GraphState::fromArray($response->getMetadata()["final_state"] ?? []);
      $snapshot = $this->generateStateGraphSnapshot(
        $executionId,
        $request,
        $finalState,
        $response->getResults(),
      );

      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,
          ]);
        }

        $metadata = $response->getMetadata();
        $metadata["snapshot"] = $snapshot->toArray();
        $response = new OrchestrationResponse(
          executionId: $response->getExecutionId(),
          status: $response->getStatus(),
          results: $response->getResults(),
          executionTime: $response->getExecutionTime(),
          metadata: $metadata,
        );
      }

      return $response;
    }
    catch (InterruptExceptionInterface $e) {
      $partialResults = [];
      if (isset($pipeline)) {
        $pipeline = $this->reloadPipeline($pipeline);
        $partialResults = $this->collectPartialResultsFromPipeline($pipeline);

        // Pause the pipeline so resume subscribers can detect and restart it.
        $pipeline->pause();
        $this->pipelineStorage->save($pipeline);
      }

      return $this->handleInterruptException(
        $e,
        $executionId,
        $request->getWorkflowId(),
        $startTime,
        $partialResults,
        [
          "checkpoint_id" => $e instanceof ApprovalRequiredException ? $e->checkpoint->id : NULL,
          "thread_id" => $e instanceof ApprovalRequiredException ? $e->checkpoint->threadId : NULL,
        ],
      );
    }
    catch (MaxIterationsException $e) {
      $executionTime = microtime(TRUE) - $startTime;

      $this->logger->error("StateGraph execution exceeded max iterations: @message", [
        "@message" => $e->getMessage(),
        "@execution_id" => $executionId,
      ]);

      return new OrchestrationResponse(
        executionId: $executionId,
        status: "max_iterations_exceeded",
        results: [],
        executionTime: $executionTime,
        metadata: [
          "max_iterations" => $e->maxIterations,
          "current_iterations" => $e->currentIterations,
          "node_id" => $e->nodeId,
        ]
      );
    }
    catch (\Exception $e) {
      $this->logger->error("StateGraph orchestration failed: @message", [
        "@message" => $e->getMessage(),
        "@execution_id" => $executionId,
      ]);

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

  /**
   * {@inheritdoc}
   */
  public function getCapabilities(): array {
    return [
      "synchronous_execution" => TRUE,
      "parallel_execution" => FALSE,
      "async_execution" => FALSE,
      "retry_support" => TRUE,
      "error_recovery" => TRUE,
      "pipeline_based" => TRUE,
      "job_tracking" => TRUE,
      "stateful" => TRUE,
      "checkpointing" => TRUE,
      "conditional_routing" => TRUE,
      "iterator_support" => TRUE,
      "approval_gates" => TRUE,
      "message_history" => TRUE,
      "interrupt_support" => TRUE,
    ];
  }

  /**
   * {@inheritdoc}
   */
  protected function getPipelineLabelSuffix(): string {
    return " - StateGraph Pipeline " . date("Y-m-d H:i:s");
  }

  /**
   * {@inheritdoc}
   */
  protected function buildPipelineExecutionContext(array $options): array {
    $context = parent::buildPipelineExecutionContext($options);

    // Store playground metadata for interrupt handlers.
    if (isset($options["playground"])) {
      $context["playground"] = $options["playground"];
    }

    return $context;
  }

  /**
   * Initializes or restores state based on configuration.
   *
   * @param \Drupal\flowdrop_stategraph\DTO\ExecutionConfig $config
   *   The execution configuration.
   * @param \Drupal\flowdrop_stategraph\Checkpointer\CheckpointerInterface $checkpointer
   *   The checkpointer instance.
   * @param array<string, mixed> $initialData
   *   The initial data for new executions.
   *
   * @return \Drupal\flowdrop_stategraph\DTO\GraphState
   *   The initialized or restored state.
   */
  private function initializeState(
    ExecutionConfig $config,
    CheckpointerInterface $checkpointer,
    array $initialData,
  ): GraphState {
    if ($config->resumeFromCheckpoint !== NULL) {
      $checkpoint = $checkpointer->get($config->resumeFromCheckpoint);
      if ($checkpoint !== NULL) {
        $this->logger->debug("Restoring state from checkpoint @id", [
          "@id" => $config->resumeFromCheckpoint,
        ]);
        return $this->stateManager->restoreFromCheckpoint($checkpoint);
      }
      $this->logger->warning("Checkpoint @id not found, creating new state", [
        "@id" => $config->resumeFromCheckpoint,
      ]);
    }

    if ($config->threadId !== NULL) {
      $latestCheckpoint = $checkpointer->getLatest($config->threadId);
      if ($latestCheckpoint !== NULL) {
        $this->logger->debug("Resuming thread @thread from checkpoint @id", [
          "@thread" => $config->threadId,
          "@id" => $latestCheckpoint->id,
        ]);
        return $this->stateManager->restoreFromCheckpoint($latestCheckpoint);
      }
    }

    return $this->stateManager->createInitialState($config->threadId, $initialData);
  }

  /**
   * Executes a pipeline with stateful tracking.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface $pipeline
   *   The pipeline to execute.
   * @param \Drupal\flowdrop_stategraph\DTO\GraphState $state
   *   The current graph state.
   * @param \Drupal\flowdrop_stategraph\DTO\ExecutionConfig $config
   *   The execution configuration.
   * @param \Drupal\flowdrop_stategraph\Checkpointer\CheckpointerInterface $checkpointer
   *   The checkpointer.
   * @param string $executionId
   *   The execution ID.
   * @param float $startTime
   *   The start time.
   *
   * @return \Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationResponse
   *   The orchestration response.
   */
  private function executePipelineWithState(
    FlowDropPipelineInterface $pipeline,
    GraphState $state,
    ExecutionConfig $config,
    CheckpointerInterface $checkpointer,
    string $executionId,
    float $startTime,
  ): OrchestrationResponse {
    if ($pipeline->isReadyToStart() || $pipeline->isPaused()) {
      $pipeline->markAsStarted();
      $this->pipelineStorage->save($pipeline);
    }

    $results = [];
    $executedJobs = [];
    $iteration = 0;
    $lastCheckpointId = NULL;
    $nodeIterationCounts = [];

    while (TRUE) {
      if ($iteration >= $config->maxIterations) {
        $this->logger->info("Reached max iterations (@max)", [
          "@max" => $config->maxIterations,
        ]);
        break;
      }

      $pipeline = $this->reloadPipeline($pipeline);
      $readyJobs = $this->jobGenerationService->getReadyJobs($pipeline);

      if (empty($readyJobs)) {
        break;
      }

      foreach ($readyJobs as $job) {
        $nodeId = $job->getNodeId();
        $nodeIteration = $nodeIterationCounts[$nodeId] ?? 0;

        $state = $this->stateManager->applyUpdate(
          $state,
          StateUpdate::create(["currentNodeId" => $nodeId]),
          $config->maxIterations
        );

        $jobResult = $this->executeJobWithState($job, $pipeline, $state, $executionId);

        $resultKey = $nodeIteration > 0 ? "{$nodeId}:{$nodeIteration}" : $nodeId;
        $results[$resultKey] = $jobResult["output"];
        $nodeIterationCounts[$nodeId] = $nodeIteration + 1;

        if (isset($jobResult["stateUpdate"])) {
          $state = $this->stateManager->applyUpdate(
            $state,
            $jobResult["stateUpdate"],
            $config->maxIterations
          );
        }

        if ($config->createCheckpoints) {
          $checkpoint = $this->stateManager->createCheckpoint(
            $state,
            $nodeId,
            $lastCheckpointId
          );
          $checkpointer->put($checkpoint);
          $lastCheckpointId = $checkpoint->id;
        }

        $executedJobs[] = $job->id();

        $nextNodes = $this->edgeEvaluator->evaluateEdges($nodeId, $state, $pipeline);
        $loopbackEdges = $this->getLoopbackEdges($nodeId, $pipeline);

        foreach ($nextNodes as $nextNodeId) {
          if (isset($loopbackEdges[$nextNodeId]) && !$this->shouldContinueLoop($state)) {
            continue;
          }

          $this->ensureJobExistsForNode(
            $nextNodeId,
            $pipeline,
            $state,
            $job,
            $nodeIterationCounts[$nextNodeId] ?? 0
          );
        }
      }

      $state = $this->stateManager->incrementIteration($state, $config->maxIterations);
      $iteration++;
    }

    $executionTime = microtime(TRUE) - $startTime;
    $pipeline = $this->reloadPipeline($pipeline);

    if ($pipeline->hasFailedJobs()) {
      $pipeline->markAsFailed("Some jobs failed during execution");
      $state = $this->stateManager->markError($state, "Pipeline execution failed");
      $status = "failed";
      $this->markRemainingJobsAsSkipped($pipeline);
    }
    elseif ($pipeline->hasInterruptedJobs()) {
      $pipeline->pause();
      $status = "paused";
      $this->logger->info("Pipeline @id paused due to interrupted jobs", [
        "@id" => $pipeline->id(),
      ]);
    }
    elseif (!empty($this->jobGenerationService->getReadyJobs($pipeline))) {
      $pipeline->pause();
      $status = "paused";
    }
    else {
      $pipeline->markAsCompleted($results);
      $state = $this->stateManager->markComplete($state);
      $status = "completed";
      $this->markRemainingJobsAsSkipped($pipeline);
    }

    $executionTimeUs = (int) round($executionTime * 1000000);
    $pipelineContext = $pipeline->getExecutionContext();
    $pipelineContext["execution_time_us"] = $executionTimeUs;
    $pipeline->setExecutionContext($pipelineContext);
    $this->pipelineStorage->save($pipeline);

    if ($config->createCheckpoints) {
      $finalCheckpoint = $this->stateManager->createCheckpoint(
        $state,
        "workflow_end",
        $lastCheckpointId
      );
      $checkpointer->put($finalCheckpoint);
    }

    $this->logger->info("StateGraph execution @status in @time seconds", [
      "@status" => $status,
      "@time" => round($executionTime, 3),
      "@executed_jobs" => count($executedJobs),
    ]);

    return new OrchestrationResponse(
      executionId: $executionId,
      status: $status,
      results: $results,
      executionTime: $executionTime,
      metadata: [
        "pipeline_id" => $pipeline->id(),
        "executed_jobs" => $executedJobs,
        "iterations" => $iteration,
        "total_jobs" => count($pipeline->getJobs()),
        "thread_id" => $state->threadId,
        "final_state" => $state->toArray(),
        "node_iteration_counts" => $nodeIterationCounts,
      ]
    );
  }

  /**
   * Collects partial results from completed jobs in a pipeline.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface $pipeline
   *   The pipeline to collect results from.
   *
   * @return array<string, mixed>
   *   The partial results keyed by nodeId or nodeId:iteration.
   */
  private function collectPartialResultsFromPipeline(
    FlowDropPipelineInterface $pipeline,
  ): array {
    $completedJobs = $pipeline->getJobsByStatus("completed");

    usort($completedJobs, function (FlowDropJobInterface $a, FlowDropJobInterface $b): int {
      $aId = $a->id();
      $bId = $b->id();
      if ($aId === NULL || $bId === NULL) {
        return 0;
      }
      return (int) $aId <=> (int) $bId;
    });

    $partialResults = [];
    $nodeOccurrences = [];

    foreach ($completedJobs as $completedJob) {
      $nodeId = $completedJob->getNodeId();
      $outputData = $completedJob->getOutputData();

      if (empty($outputData)) {
        continue;
      }

      $metadata = $completedJob->getMetadata();
      $loopIteration = $metadata["loop_iteration"] ?? NULL;

      if (!isset($nodeOccurrences[$nodeId])) {
        $nodeOccurrences[$nodeId] = 0;
      }
      $currentOccurrence = $nodeOccurrences[$nodeId];
      $nodeOccurrences[$nodeId]++;

      if ($loopIteration !== NULL) {
        $resultKey = "{$nodeId}:{$loopIteration}";
      }
      elseif ($currentOccurrence > 0) {
        $resultKey = "{$nodeId}:{$currentOccurrence}";
      }
      else {
        $resultKey = $nodeId;
      }

      $partialResults[$resultKey] = $outputData;
    }

    return $partialResults;
  }

  /**
   * Ensures a job exists for a node, creating one if needed for loop iteration.
   *
   * @param string $nodeId
   *   The node ID to create a job for.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface $pipeline
   *   The pipeline.
   * @param \Drupal\flowdrop_stategraph\DTO\GraphState $state
   *   The current graph state.
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $sourceJob
   *   The job that triggered this.
   * @param int $loopIteration
   *   The loop iteration number.
   */
  private function ensureJobExistsForNode(
    string $nodeId,
    FlowDropPipelineInterface $pipeline,
    GraphState $state,
    FlowDropJobInterface $sourceJob,
    int $loopIteration,
  ): void {
    $idleJobs = $pipeline->getJobsByStatus("idle");
    foreach ($idleJobs as $idleJob) {
      if ($idleJob->getNodeId() === $nodeId) {
        return;
      }
    }

    $pendingJobs = $pipeline->getJobsByStatus("pending");
    foreach ($pendingJobs as $pendingJob) {
      if ($pendingJob->getNodeId() === $nodeId) {
        return;
      }
    }

    $readyJobs = $this->jobGenerationService->getReadyJobs($pipeline);
    foreach ($readyJobs as $readyJob) {
      if ($readyJob->getNodeId() === $nodeId) {
        return;
      }
    }

    $originalJob = $this->findOriginalJobForNode($nodeId, $pipeline);
    if ($originalJob === NULL) {
      $this->logger->warning("No original job found for node @node", [
        "@node" => $nodeId,
      ]);
      return;
    }

    $this->createLoopIterationJob(
      $nodeId,
      $pipeline,
      $originalJob,
      $sourceJob,
      $loopIteration,
      $state
    );
  }

  /**
   * Finds the original job created for a node.
   *
   * @param string $nodeId
   *   The node ID.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface $pipeline
   *   The pipeline.
   *
   * @return \Drupal\flowdrop_job\FlowDropJobInterface|null
   *   The original job or NULL.
   */
  private function findOriginalJobForNode(
    string $nodeId,
    FlowDropPipelineInterface $pipeline,
  ): ?FlowDropJobInterface {
    $allJobs = $pipeline->getJobs();
    foreach ($allJobs as $job) {
      if ($job->getNodeId() === $nodeId) {
        return $job;
      }
    }
    return NULL;
  }

  /**
   * Creates a new job for a loop iteration.
   *
   * @param string $nodeId
   *   The node ID.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface $pipeline
   *   The pipeline.
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $originalJob
   *   The original job.
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $sourceJob
   *   The source job.
   * @param int $loopIteration
   *   The loop iteration number.
   * @param \Drupal\flowdrop_stategraph\DTO\GraphState $state
   *   The current graph state.
   */
  private function createLoopIterationJob(
    string $nodeId,
    FlowDropPipelineInterface $pipeline,
    FlowDropJobInterface $originalJob,
    FlowDropJobInterface $sourceJob,
    int $loopIteration,
    GraphState $state,
  ): void {
    $originalMetadata = $originalJob->getMetadata();
    $loopMetadata = array_merge($originalMetadata, [
      "loop_iteration" => $loopIteration,
      "parent_job_id" => $sourceJob->id(),
      "loop_source_node" => $sourceJob->getNodeId(),
      "is_loop_iteration" => TRUE,
    ]);

    $displayIteration = $loopIteration + 1;

    $originalLabel = $originalJob->label();
    $baseLabel = is_string($originalLabel)
      ? preg_replace("/\\s*#\\d+$/", "", $originalLabel)
      : "Job";

    if ($loopIteration === 1 && is_string($originalLabel) && !preg_match("/#\\d+$/", $originalLabel)) {
      $originalJob->set("label", $baseLabel . " #1");
      $this->jobStorage->save($originalJob);
    }

    $newJob = $this->jobStorage->create([
      "label" => $baseLabel . " #" . $displayIteration,
      "bundle" => "default",
      "metadata" => json_encode($loopMetadata),
      "node_type_id" => $originalMetadata["node_type_id"] ?? "default",
      "status" => "idle",
      "priority" => $originalJob->getPriority(),
      "input_data" => json_encode([]),
      "output_data" => json_encode([]),
      "max_retries" => $originalJob->getMaxRetries(),
    ]);

    $newJob->setNodeId($nodeId);
    $newJob->addDependentJob($sourceJob);
    $this->jobStorage->save($newJob);

    $pipeline->addJob($newJob);
    $this->pipelineStorage->save($pipeline);

    $this->logger->info("Created loop iteration job @job_id for node @node (#@iter)", [
      "@job_id" => $newJob->id(),
      "@node" => $nodeId,
      "@iter" => $displayIteration,
    ]);
  }

  /**
   * Executes a job with state tracking.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job to execute.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface $pipeline
   *   The pipeline.
   * @param \Drupal\flowdrop_stategraph\DTO\GraphState $state
   *   The current state.
   * @param string $executionId
   *   The execution ID.
   *
   * @return array{output: array<string, mixed>, stateUpdate: StateUpdate|null}
   *   The job output and optional state update.
   */
  private function executeJobWithState(
    FlowDropJobInterface $job,
    FlowDropPipelineInterface $pipeline,
    GraphState $state,
    string $executionId,
  ): array {
    $this->logger->debug("Executing job @job (node: @node) with state", [
      "@job" => $job->id(),
      "@node" => $job->getNodeId(),
    ]);

    try {
      $job->markAsStarted();
      $this->jobStorage->save($job);

      $metadata = $job->getMetadata();
      $configData = $metadata["config"] ?? [];
      $runtimeInputData = $this->buildStateGraphRuntimeInputData($job, $pipeline, $state);

      $interruptId = $metadata["interrupt_id"] ?? NULL;
      if ($interruptId !== NULL) {
        $runtimeInputData["__interrupt_id__"] = $interruptId;
      }

      $filteredInputData = array_filter(
        $runtimeInputData,
        fn(string $key): bool => !str_starts_with($key, "__"),
        ARRAY_FILTER_USE_KEY
      );
      $job->setInputData($filteredInputData);
      $this->jobStorage->save($job);

      $context = $this->buildNodeExecutionContext($job, $pipeline);
      $contextMetadata = $context->getMetadata();
      $contextMetadata["graph_state"] = $state->toArray();

      $pipelineContext = $pipeline->getExecutionContext();
      if (isset($pipelineContext["playground"]["session_id"])) {
        $contextMetadata["session_id"] = $pipelineContext["playground"]["session_id"];
      }

      $context = new NodeExecutionContext(
        workflowId: $context->getWorkflowId(),
        pipelineId: $context->getPipelineId(),
        initialData: $context->getInitialData(),
        metadata: $contextMetadata
      );

      $nodeTypeId = $metadata["node_type_id"] ?? "default";

      $result = $this->nodeRuntime->executeNode(
        $executionId,
        $job->getNodeId(),
        $nodeTypeId,
        $runtimeInputData,
        $configData,
        $context
      );

      $outputData = $result->getOutput()->toArray();
      $job->markAsCompleted($outputData);

      $executionTimeUs = (int) round($result->getExecutionTime() * 1000000);
      $jobMetadata = $job->getMetadata();
      $jobMetadata["execution_time_us"] = $executionTimeUs;
      $job->setMetadata($jobMetadata);
      $this->jobStorage->save($job);

      $stateUpdate = NULL;
      if (isset($outputData["state_update"]) && is_array($outputData["state_update"])) {
        $stateUpdate = StateUpdate::create($outputData["state_update"])
          ->withSourceNode($job->getNodeId());
      }

      return [
        "output" => $outputData,
        "stateUpdate" => $stateUpdate,
      ];
    }
    catch (InterruptExceptionInterface $e) {
      $job->markAsInterrupted($e->getInterruptId(), $e->getMessage());
      $this->jobStorage->save($job);

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

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

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

      throw $e;
    }
  }

  /**
   * Builds runtime input data for a StateGraph job.
   *
   * @param \Drupal\flowdrop_job\FlowDropJobInterface $job
   *   The job.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface $pipeline
   *   The pipeline.
   * @param \Drupal\flowdrop_stategraph\DTO\GraphState $state
   *   The current state.
   *
   * @return array<string, mixed>
   *   The runtime input data.
   */
  private function buildStateGraphRuntimeInputData(
    FlowDropJobInterface $job,
    FlowDropPipelineInterface $pipeline,
    GraphState $state,
  ): array {
    $runtimeData = [];

    // Add state data to inputs.
    $runtimeData["__state__"] = $state->toArray();
    $runtimeData["__messages__"] = array_map(
      fn($m) => $m->toArray(),
      $state->messages
    );
    $runtimeData["__data__"] = $state->data;

    if ($state->iterator !== NULL) {
      $runtimeData["__iterator__"] = $state->iterator->toArray();
      $runtimeData["__current_item__"] = $state->iterator->currentItem;
    }

    $metadata = $job->getMetadata();
    $nodeTypeId = $metadata["node_type_id"] ?? "";
    if ($nodeTypeId === "foreach" && $state->iterator !== NULL) {
      $runtimeData["__iterator__"] = $state->iterator->toArray();
    }

    // Get data from completed upstream jobs.
    $incomingEdges = $metadata["incoming_edges"] ?? [];
    $completedJobs = $pipeline->getJobsByStatus("completed");

    $completedJobsMap = [];
    foreach ($completedJobs as $completedJob) {
      $sourceNodeId = $completedJob->getNodeId();
      $jobId = (int) $completedJob->id();

      if (!isset($completedJobsMap[$sourceNodeId])) {
        $completedJobsMap[$sourceNodeId] = $completedJob;
      }
      else {
        $existingJobId = (int) $completedJobsMap[$sourceNodeId]->id();
        if ($jobId > $existingJobId) {
          $completedJobsMap[$sourceNodeId] = $completedJob;
        }
      }
    }

    foreach ($incomingEdges as $edge) {
      if ($edge["is_trigger"] ?? FALSE) {
        continue;
      }

      $sourceNodeId = $edge["source"];
      if (!isset($completedJobsMap[$sourceNodeId])) {
        continue;
      }

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

      $sourceHandle = $edge["source_handle"] ?? "";
      $targetHandle = $edge["target_handle"] ?? "";

      $sourcePortName = $this->extractPortName($sourceHandle, "output");
      $targetPortName = $this->extractPortName($targetHandle, "input");

      if ($sourcePortName !== NULL && $targetPortName !== NULL) {
        if (isset($sourceOutput[$sourcePortName])) {
          $runtimeData[$targetPortName] = $sourceOutput[$sourcePortName];
        }
      }
    }

    return $runtimeData;
  }

  /**
   * Gets outgoing loopback edges from a node.
   *
   * @param string $sourceNodeId
   *   The source node ID.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface $pipeline
   *   The pipeline.
   *
   * @return array<string, array<string, mixed>>
   *   Map of target node ID to edge info for loopback edges.
   */
  private function getLoopbackEdges(
    string $sourceNodeId,
    FlowDropPipelineInterface $pipeline,
  ): array {
    $loopbackEdges = [];

    $workflow = $pipeline->getWorkflow();
    if ($workflow === NULL) {
      return $loopbackEdges;
    }

    $graphData = $workflow->getGraphData();
    $edges = $graphData["edges"] ?? [];

    foreach ($edges as $edge) {
      if (($edge["source"] ?? "") !== $sourceNodeId) {
        continue;
      }

      $targetHandle = $edge["targetHandle"] ?? "";
      $targetId = $edge["target"] ?? "";

      if (str_contains($targetHandle, "-input-loop_back")) {
        $loopbackEdges[$targetId] = [
          "target" => $targetId,
          "targetHandle" => $targetHandle,
          "sourceHandle" => $edge["sourceHandle"] ?? "",
          "edgeId" => $edge["id"] ?? "",
        ];
      }
    }

    return $loopbackEdges;
  }

  /**
   * Checks if the loop should continue based on iterator state.
   *
   * @param \Drupal\flowdrop_stategraph\DTO\GraphState $state
   *   The current graph state.
   *
   * @return bool
   *   TRUE if the loop should continue.
   */
  private function shouldContinueLoop(GraphState $state): bool {
    if ($state->iterator === NULL) {
      return FALSE;
    }
    return $state->iterator->hasMore === TRUE;
  }

  /**
   * Generates a WorkflowSnapshot from StateGraph execution.
   *
   * @param string $executionId
   *   The execution ID.
   * @param \Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationRequest $request
   *   The orchestration request.
   * @param \Drupal\flowdrop_stategraph\DTO\GraphState $state
   *   The final graph state.
   * @param array<string, mixed> $results
   *   The execution results.
   *
   * @return \Drupal\flowdrop_orchestration\DTO\Snapshot\WorkflowSnapshot|null
   *   The generated snapshot or NULL.
   */
  private function generateStateGraphSnapshot(
    string $executionId,
    OrchestrationRequest $request,
    GraphState $state,
    array $results,
  ): ?WorkflowSnapshot {
    if ($this->workflowStateManager === NULL) {
      return NULL;
    }

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

  /**
   * Converts a GraphState to a WorkflowSnapshot.
   *
   * @param \Drupal\flowdrop_stategraph\DTO\GraphState $graphState
   *   The GraphState to convert.
   * @param string $workflowId
   *   The workflow ID.
   * @param array<string, mixed> $workflowDefinition
   *   The workflow definition.
   * @param array<string, array<string, mixed>> $nodeOutputs
   *   Map of node ID to output data.
   *
   * @return \Drupal\flowdrop_orchestration\DTO\Snapshot\WorkflowSnapshot
   *   The equivalent WorkflowSnapshot.
   */
  public function convertToWorkflowSnapshot(
    GraphState $graphState,
    string $workflowId,
    array $workflowDefinition,
    array $nodeOutputs = [],
  ): WorkflowSnapshot {
    $nodeStates = [];
    $executionOrder = 1;

    foreach ($nodeOutputs as $nodeId => $output) {
      $nodeStates[$nodeId] = new NodeSnapshot(
        nodeId: $nodeId,
        status: "completed",
        output: $output,
        error: NULL,
        injected: FALSE,
        executionOrder: $executionOrder++,
        metadata: [],
      );
    }

    $workflowVersion = $this->workflowStateManager !== NULL
      ? $this->workflowStateManager->computeWorkflowVersion($workflowDefinition)
      : md5(json_encode($workflowDefinition) ?: "");

    $status = "running";
    if ($graphState->error !== NULL) {
      $status = "failed";
    }
    elseif ($graphState->isComplete) {
      $status = "completed";
    }

    $snapshotData = $graphState->toSnapshotData();

    return new WorkflowSnapshot(
      workflowId: $workflowId,
      workflowVersion: $workflowVersion,
      executionId: $graphState->threadId ?: $this->generateExecutionId(),
      status: $status,
      nodeStates: $nodeStates,
      initialInput: $snapshotData["initialInput"],
      messages: $snapshotData["messages"],
      iterator: $snapshotData["iterator"],
      iterationCount: $snapshotData["iterationCount"],
      threadId: $snapshotData["threadId"],
      error: $snapshotData["error"],
      metadata: array_merge($snapshotData["metadata"], [
        "orchestrator_type" => $this->getType(),
        "converted_from" => "GraphState",
      ]),
      createdAt: time(),
      updatedAt: time(),
    );
  }

  /**
   * Converts a WorkflowSnapshot to a GraphState.
   *
   * @param \Drupal\flowdrop_orchestration\DTO\Snapshot\WorkflowSnapshot $snapshot
   *   The WorkflowSnapshot to convert.
   * @param string|null $threadId
   *   Optional thread ID override.
   *
   * @return \Drupal\flowdrop_stategraph\DTO\GraphState
   *   The equivalent GraphState.
   */
  public function convertFromWorkflowSnapshot(
    WorkflowSnapshot $snapshot,
    ?string $threadId = NULL,
  ): GraphState {
    return GraphState::fromWorkflowSnapshot($snapshot, $threadId);
  }

}
