<?php

declare(strict_types=1);

namespace Drupal\flowdrop_stategraph\Service;

use Drupal\flowdrop\DTO\ExecutionEdge;
use Drupal\flowdrop\DTO\JsApi\Edge as TransportEdge;
use Drupal\flowdrop\Exception\NodeProcessingException;
use Drupal\flowdrop\Exception\WorkflowStopException;
use Drupal\flowdrop\Utility\EdgePortClassifier;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionResult;
use Drupal\flowdrop_job\Constants\JobStatus;
use Drupal\flowdrop_job\FlowDropJobInterface;
use Drupal\flowdrop_pipeline\Constants\PausedReason;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface;
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\Compiler\DependencyGraph;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext;
use Drupal\flowdrop_runtime\Service\Runtime\ToolProjectorInterface;
use Drupal\flowdrop_node_type\Service\NodeMetadataResolver;
use Drupal\flowdrop_workflow\DTO\WorkflowDTO;
use Drupal\flowdrop_orchestration\DTO\Snapshot\NodeSnapshot;
use Drupal\flowdrop_orchestration\DTO\Snapshot\WorkflowSnapshot;
use Drupal\flowdrop_orchestration\Exception\InterruptExceptionInterface;
use Drupal\flowdrop_orchestration\Service\Orchestrator\OrchestratorHelperInterface;
use Drupal\flowdrop_runtime\Exception\OrchestrationException;
use Drupal\flowdrop_runtime\Service\Orchestrator\AbstractOrchestratorBase;
use Drupal\flowdrop_runtime\Service\Orchestrator\PipelineExecutorInterface;
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\PipelineStorageRegistryInterface;
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\MaxIterationsException;
use Drupal\flowdrop_stategraph\Plugin\FlowDropNodeProcessor\StateAwareProcessorInterface;
use Drupal\flowdrop_stategraph\Routing\ConditionalEdgeEvaluator;
use Drupal\flowdrop_runtime\Event\PipelineCompletedEvent;
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 implements PipelineExecutorInterface {

  /**
   * Execution-context key holding the namespaced orchestrator type.
   */
  public const CONTEXT_ORCHESTRATOR_TYPE = "orchestrator_type";

  /**
   * Execution-context key holding the stamped stategraph ExecutionConfig.
   */
  public const CONTEXT_STATEGRAPH_CONFIG = "stategraph_config";

  /**
   * Constructs a new StateGraphOrchestrator.
   *
   * @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_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).
   * @param \Drupal\flowdrop_runtime\Service\Runtime\ToolProjectorInterface|null $toolProjector
   *   The tool projector, used to expose wired tool nodes to the agent loop
   *   (optional).
   * @param \Drupal\flowdrop_node_type\Service\NodeMetadataResolver|null $nodeMetadataResolver
   *   Re-injects fresh node metadata from the live node type before the
   *   projected tool graph is built, so it reflects the authoritative
   *   definition rather than the stored metadata cache (optional).
   */
  public function __construct(
    PipelineStorageRegistryInterface $pipelineStorageRegistry,
    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,
    protected readonly ?ToolProjectorInterface $toolProjector = NULL,
    protected readonly ?NodeMetadataResolver $nodeMetadataResolver = NULL,
  ) {
    parent::__construct(
      $pipelineStorageRegistry,
      $jobStorage,
      $nodeRuntime,
      $jobGenerationService,
      $eventDispatcher,
      $logger,
      $workflowStateManager,
      $realTimeManager,
    );
  }

  /**
   * Per-workflow cache of dependency graphs built for tool projection.
   *
   * @var array<string, \Drupal\flowdrop_runtime\DTO\Compiler\DependencyGraph>
   */
  private array $toolGraphCache = [];

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

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

    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");
      }

      // Use the pipeline entity ID as the execution ID so the frontend can call
      // GET /api/flowdrop/pipeline/{id} to fetch node statuses.
      $executionId = (string) $pipeline->id();

      // Stamp the effective stategraph config (with the resolved thread id)
      // onto the pipeline so a later executePipeline() — deferred launch or
      // interrupt resume — can rebuild the exact same configuration.
      $this->stampStateGraphContext($pipeline, $config->withThreadId($state->threadId));

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

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

      // 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 = [];
      $persistedInterrupt = NULL;
      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->getPipelineStorage()->save($pipeline);

        // Resolve the persisted interrupt entity (handles both entity- and
        // message-shape exceptions) so the response metadata carries the
        // entity data downstream callers expect.
        $persistedInterrupt = $this->resolvePersistedInterrupt($e, $pipeline);
      }

      return $this->handleInterruptException(
        $e,
        $executionId,
        $request->getWorkflowId(),
        $startTime,
        $partialResults,
        [
          "checkpoint_id" => NULL,
          "thread_id" => NULL,
        ],
        $persistedInterrupt,
      );
    }
    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) {
      // Dispatch the event so subscribers add a failure notice to the session.
      // The happy-path dispatch inside executePipelineWithState() is bypassed
      // when an exception escapes the job-execution loop.
      if (isset($pipeline)) {
        $this->failPipelineAndDispatch($pipeline, $executionId, $e, $startTime);
      }
      else {
        $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,
      "message_history" => TRUE,
      "interrupt_support" => TRUE,
    ];
  }

  /**
   * {@inheritdoc}
   *
   * Executes (or resumes) a pre-existing stategraph pipeline — the entry
   * point used by the deferred launch path and the interrupt-resume
   * subscribers, where the pipeline was created earlier and no
   * OrchestrationRequest exists.
   *
   * The ExecutionConfig is rebuilt from the config stamped on the pipeline
   * at creation time (see {@see stampStateGraphContext()}), falling back to
   * the workflow's declared orchestrator settings. A stamped/declared
   * configuration wins over the $maxIterations argument — the workflow
   * author's intent is authoritative; the argument only bounds pipelines
   * that carry no configuration of their own.
   *
   * State restore: when the thread's latest checkpoint is available (entity
   * checkpointer), execution continues from it; with the default memory
   * checkpointer nothing survives the original request, so state is rebuilt
   * from the pipeline's input data — job statuses on the pipeline still
   * carry run progress, so completed nodes are not re-executed.
   *
   * Workflow snapshots are not generated on this path; they are an artifact
   * of the request/response chat flow and nothing on the deferred or resume
   * paths consumes them.
   */
  public function executePipeline(
    FlowDropPipelineRuntimeInterface $pipeline,
    ?int $maxIterations = 100,
    ?float $maxExecutionTime = NULL,
    ?string $executionId = NULL,
  ): OrchestrationResponse {
    $startTime = microtime(TRUE);
    $executionId = $executionId ?? (string) $pipeline->id();

    $config = $this->resolveConfigForPipeline($pipeline, $maxIterations);
    $checkpointer = $this->checkpointerFactory->create($config->checkpointerType);
    $state = $this->initializeState($config, $checkpointer, $pipeline->getInputData());

    // Re-stamp with the effective thread id so subsequent resumes of this
    // pipeline land on the same thread even if the original stamp is absent.
    $this->stampStateGraphContext($pipeline, $config->withThreadId($state->threadId));

    $this->logger->info("StateGraph pipeline execution started", [
      "@execution_id" => $executionId,
      "@pipeline_id" => $pipeline->id(),
      "@thread_id" => $state->threadId,
    ]);

    try {
      return $this->executePipelineWithState(
        $pipeline,
        $state,
        $config,
        $checkpointer,
        $executionId,
        $startTime,
      );
    }
    catch (InterruptExceptionInterface $e) {
      $pipeline = $this->reloadPipeline($pipeline);
      $partialResults = $this->collectPartialResultsFromPipeline($pipeline);

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

      $persistedInterrupt = $this->resolvePersistedInterrupt($e, $pipeline);

      return $this->handleInterruptException(
        $e,
        $executionId,
        $pipeline->getWorkflowId(),
        $startTime,
        $partialResults,
        [
          "checkpoint_id" => NULL,
          "thread_id" => $state->threadId,
        ],
        $persistedInterrupt,
      );
    }
    catch (\Exception $e) {
      $this->failPipelineAndDispatch($pipeline, $executionId, $e, $startTime);

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

  /**
   * Stamps the effective stategraph configuration onto the pipeline.
   *
   * Written at pipeline creation (orchestrate()) and refreshed by
   * executePipeline(); read back by {@see resolveConfigForPipeline()} so
   * deferred execution and interrupt resume reconstruct the configuration
   * the run started with. Also records the namespaced orchestrator type for
   * executor resolution.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline to stamp.
   * @param \Drupal\flowdrop_stategraph\DTO\ExecutionConfig $config
   *   The effective configuration (thread id already resolved).
   */
  protected function stampStateGraphContext(
    FlowDropPipelineRuntimeInterface $pipeline,
    ExecutionConfig $config,
  ): void {
    $context = $pipeline->getExecutionContext();
    $stamped = $config->toArray();
    if (($context[self::CONTEXT_ORCHESTRATOR_TYPE] ?? NULL) === OrchestratorHelperInterface::STATEGRAPH
      && ($context[self::CONTEXT_STATEGRAPH_CONFIG] ?? NULL) === $stamped) {
      return;
    }
    $context[self::CONTEXT_ORCHESTRATOR_TYPE] = OrchestratorHelperInterface::STATEGRAPH;
    $context[self::CONTEXT_STATEGRAPH_CONFIG] = $stamped;
    $pipeline->setExecutionContext($context);
    $this->getPipelineStorage()->save($pipeline);
  }

  /**
   * Rebuilds the ExecutionConfig for a pre-existing pipeline.
   *
   * Resolution order: config stamped on the pipeline's execution context →
   * the workflow's declared orchestrator settings → defaults bounded by the
   * caller's $fallbackMaxIterations. The thread id, when still unresolved,
   * is derived via {@see resolveThreadId()}.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline being executed or resumed.
   * @param int|null $fallbackMaxIterations
   *   Iteration bound to apply only when no stamped or declared
   *   configuration provides one.
   *
   * @return \Drupal\flowdrop_stategraph\DTO\ExecutionConfig
   *   The effective configuration.
   */
  protected function resolveConfigForPipeline(
    FlowDropPipelineRuntimeInterface $pipeline,
    ?int $fallbackMaxIterations,
  ): ExecutionConfig {
    $context = $pipeline->getExecutionContext();

    $stamped = $context[self::CONTEXT_STATEGRAPH_CONFIG] ?? NULL;
    if (is_array($stamped)) {
      $config = ExecutionConfig::fromArray($stamped);
    }
    else {
      // Launcher-created or legacy pipelines: fall back to the workflow's
      // declared orchestrator settings (the author's configuration).
      $settings = [];
      $workflow = $pipeline->getWorkflow();
      if ($workflow !== NULL) {
        $metadata = $workflow->getMetadata();
        $orchestratorSettings = $metadata["orchestrator_settings"] ?? [];
        $declared = is_array($orchestratorSettings)
          ? ($orchestratorSettings["stategraph"] ?? [])
          : [];
        if (is_array($declared)) {
          $settings = $declared;
        }
        // The common `timeout` setting lives at the settings root and
        // doubles as the loop's execution-time budget.
        if (is_array($orchestratorSettings)
          && isset($orchestratorSettings["timeout"])
          && !isset($settings["timeout"])) {
          $settings["timeout"] = $orchestratorSettings["timeout"];
        }
      }
      if ($fallbackMaxIterations !== NULL
        && !isset($settings["maxIterations"])
        && !isset($settings["max_iterations"])) {
        $settings["maxIterations"] = $fallbackMaxIterations;
      }
      $config = ExecutionConfig::fromArray($settings);
    }

    if ($config->threadId === NULL) {
      $config = $config->withThreadId($this->resolveThreadId($pipeline, $context));
    }

    return $config;
  }

  /**
   * Resolves the checkpoint thread for a pipeline with no explicit thread.
   *
   * Layered resolution:
   * an author-pinned thread id never reaches this method (it arrives via
   * config); a session-attached run shares the chat convention; everything
   * else gets an isolated per-pipeline thread that a later resume can
   * re-derive from the pipeline alone.
   *
   * Session:thread is N:N in real life ("clear conversation" moves a
   * session to a fresh thread; an author pin shares one thread across
   * sessions) — this helper is the single owner of the naming convention,
   * so a future decoupling changes only this method.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline.
   * @param array<string, mixed> $context
   *   The pipeline's execution context.
   *
   * @return string
   *   The resolved thread id.
   */
  protected function resolveThreadId(
    FlowDropPipelineRuntimeInterface $pipeline,
    array $context,
  ): string {
    $sessionId = $context["playground"]["session_id"] ?? NULL;
    if (($sessionId === NULL || $sessionId === "")
      && $pipeline instanceof FlowDropPipelineInterface) {
      $sessionId = $pipeline->getSessionId();
    }
    if ($sessionId !== NULL && $sessionId !== "") {
      return "playground_session_{$sessionId}";
    }
    return "pipeline_" . $pipeline->id();
  }

  /**
   * Marks the pipeline failed and dispatches PipelineCompletedEvent.
   *
   * Shared by the failure paths of orchestrate() and executePipeline(); the
   * happy-path dispatch inside executePipelineWithState() is bypassed when
   * an exception escapes the job-execution loop.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline that failed.
   * @param string $executionId
   *   The execution id.
   * @param \Exception $exception
   *   The escaping exception.
   * @param float $startTime
   *   The execution start time.
   */
  protected function failPipelineAndDispatch(
    FlowDropPipelineRuntimeInterface $pipeline,
    string $executionId,
    \Exception $exception,
    float $startTime,
  ): void {
    try {
      $executionTimeUs = (int) round((microtime(TRUE) - $startTime) * 1000000);
      $pipeline->markAsFailed($exception->getMessage());
      $context = $pipeline->getExecutionContext();
      $context["execution_time_us"] = $executionTimeUs;
      unset($context[PausedReason::CONTEXT_KEY]);
      $pipeline->setExecutionContext($context);
      $this->getPipelineStorage()->save($pipeline);
      $this->eventDispatcher->dispatch(
        new PipelineCompletedEvent($context, $executionId, PipelineStatus::FAILED, $executionTimeUs, (string) $pipeline->id()),
        PipelineCompletedEvent::NAME,
      );
    }
    catch (\Throwable $dispatchError) {
      $this->logger->error("Failed to dispatch PipelineCompletedEvent after failure: @msg", [
        "@msg" => $dispatchError->getMessage(),
      ]);
    }

    $this->logger->error("StateGraph orchestration failed: @message", [
      "@message" => $exception->getMessage(),
      "@execution_id" => $executionId,
    ]);
  }

  /**
   * {@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\FlowDropPipelineRuntimeInterface $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(
    FlowDropPipelineRuntimeInterface $pipeline,
    GraphState $state,
    ExecutionConfig $config,
    CheckpointerInterface $checkpointer,
    string $executionId,
    float $startTime,
  ): OrchestrationResponse {
    if ($pipeline->isReadyToStart() || $pipeline->isPaused()) {
      // Re-entry clears a stale budget-pause reason from the previous run
      // segment — otherwise a resume that completes (or fails) would still
      // report the old reason on the status endpoint.
      $context = $pipeline->getExecutionContext();
      if (isset($context[PausedReason::CONTEXT_KEY])) {
        unset($context[PausedReason::CONTEXT_KEY]);
        $pipeline->setExecutionContext($context);
      }
      $pipeline->markAsStarted();
      $this->getPipelineStorage()->save($pipeline);
    }

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

    while (TRUE) {
      // Poll for inward Cancel/Pause signals before each iteration. The
      // flowdrop_interrupt subscriber answers via PipelineSignalPollEvent;
      // a non-null response means the pipeline was cancelled or paused
      // and we should stop here. Mirrors SynchronousPipelineOrchestrator.
      $signalResponse = $this->pollPipelineSignal($pipeline, $executionId, $startTime);
      if ($signalResponse !== NULL) {
        return $signalResponse;
      }

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

      // Enforce the time budget (the orchestrator settings' `timeout`).
      // Breaking here lands in the post-loop status handling: remaining
      // ready jobs pause the pipeline, same as the max-iterations break.
      // The stamped reason turns the pause into a pending system Pause
      // signal (BudgetPauseSignalSubscriber), so `POST .../resume` and
      // inbox resolution can continue the run with a fresh budget window.
      if ($config->maxExecutionTime !== NULL
        && (microtime(TRUE) - $startTime) >= $config->maxExecutionTime) {
        $pauseReason = PausedReason::TIME_BUDGET;
        $this->logger->info("Reached max execution time (@time seconds)", [
          "@time" => round($config->maxExecutionTime, 2),
        ]);
        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
        );

        $resultKey = $nodeIteration > 0 ? "{$nodeId}:{$nodeIteration}" : $nodeId;
        try {
          $jobResult = $this->executeJobWithState($job, $pipeline, $state, $executionId);
        }
        catch (WorkflowStopException $e) {
          // A terminal node ended the run cleanly. Record its payload and stop
          // scheduling; downstream and unprocessed sibling jobs are abandoned.
          $results[$resultKey] = $e->getResult();
          $executedJobs[] = $job->id();
          $stopRequested = TRUE;
          break;
        }

        $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);
        $loopbackTargetIds = $this->getLoopbackTargetIds($nodeId, $pipeline);

        foreach ($nextNodes as $nextNodeId) {
          if (
            isset($loopbackTargetIds[$nextNodeId])
            && !$this->shouldFollowLoopbackEdge(
              $state,
              $nextNodeId,
              $nodeIterationCounts,
              $config->maxIterations
            )
          ) {
            continue;
          }

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

      if ($stopRequested) {
        break;
      }

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

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

    // Only a budget pause with ready jobs remaining stamps a reason; the
    // HITL (interrupted-jobs) pause deliberately carries none — the reason
    // is what gates the automatic Pause-signal creation downstream.
    $stampedPauseReason = NULL;

    if ($this->hasUnhandledFailedJobs($pipeline)) {
      $pipeline->markAsFailed("Some jobs failed during execution");
      $state = $this->stateManager->markError($state, "Pipeline execution failed");
      $status = PipelineStatus::FAILED;
      $this->markRemainingJobsAsSkipped($pipeline);
    }
    elseif ($pipeline->hasInterruptedJobs()) {
      $pipeline->pause();
      $status = PipelineStatus::PAUSED;
      $this->logger->info("Pipeline @id paused due to interrupted jobs", [
        "@id" => $pipeline->id(),
      ]);
    }
    elseif (!$stopRequested && !empty($this->jobGenerationService->getReadyJobs($pipeline))) {
      $pipeline->pause();
      $status = PipelineStatus::PAUSED;
      $stampedPauseReason = $pauseReason;
      $this->logger->info("Pipeline @id paused on @reason with ready jobs remaining; resumable via Pause-signal resolution", [
        "@id" => $pipeline->id(),
        "@reason" => $pauseReason ?? "unknown budget",
      ]);
    }
    else {
      $pipeline->markAsCompleted($results);
      $state = $this->stateManager->markComplete($state);
      $status = PipelineStatus::COMPLETED;
      $this->markRemainingJobsAsSkipped($pipeline);
    }

    $executionTimeUs = (int) round($executionTime * 1000000);
    $pipelineContext = $pipeline->getExecutionContext();
    $pipelineContext["execution_time_us"] = $executionTimeUs;
    if ($stampedPauseReason !== NULL) {
      $pipelineContext[PausedReason::CONTEXT_KEY] = $stampedPauseReason;
    }
    else {
      unset($pipelineContext[PausedReason::CONTEXT_KEY]);
    }
    $pipeline->setExecutionContext($pipelineContext);
    $this->getPipelineStorage()->save($pipeline);

    $this->eventDispatcher->dispatch(
      new PipelineCompletedEvent($pipelineContext, $executionId, $status, $executionTimeUs, (string) $pipeline->id()),
      PipelineCompletedEvent::NAME,
    );

    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\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline to collect results from.
   *
   * @return array<string, mixed>
   *   The partial results keyed by nodeId or nodeId:iteration.
   */
  private function collectPartialResultsFromPipeline(
    FlowDropPipelineRuntimeInterface $pipeline,
  ): array {
    $completedJobs = $pipeline->getJobsByStatus(JobStatus::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\FlowDropPipelineRuntimeInterface $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,
    FlowDropPipelineRuntimeInterface $pipeline,
    GraphState $state,
    FlowDropJobInterface $sourceJob,
    int $loopIteration,
  ): void {
    $idleJobs = $pipeline->getJobsByStatus(JobStatus::IDLE);
    foreach ($idleJobs as $idleJob) {
      if ($idleJob->getNodeId() === $nodeId) {
        return;
      }
    }

    $pendingJobs = $pipeline->getJobsByStatus(JobStatus::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\FlowDropPipelineRuntimeInterface $pipeline
   *   The pipeline.
   *
   * @return \Drupal\flowdrop_job\FlowDropJobInterface|null
   *   The original job or NULL.
   */
  private function findOriginalJobForNode(
    string $nodeId,
    FlowDropPipelineRuntimeInterface $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\FlowDropPipelineRuntimeInterface $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,
    FlowDropPipelineRuntimeInterface $pipeline,
    FlowDropJobInterface $originalJob,
    FlowDropJobInterface $sourceJob,
    int $loopIteration,
    GraphState $state,
  ): void {
    $originalMetadata = $originalJob->getMetadata();
    // A clone is a new execution: per-execution identity must not leak across
    // loop-iteration boundaries. An inherited interrupt_id would route the
    // clone to resume() with the original iteration's (already resolved)
    // answer — silently auto-approving every HITL gate after iteration 1.
    // execution_time_us is cosmetic: the clone has not run yet.
    unset($originalMetadata["interrupt_id"], $originalMetadata["execution_time_us"]);
    $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" => JobStatus::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->getPipelineStorage()->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\FlowDropPipelineRuntimeInterface $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,
    FlowDropPipelineRuntimeInterface $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";

      // Project tools wired to this node (empty when it has none); executeNode
      // injects them only if the processor is ToolsAware. Without this, a
      // ToolsAware node running under StateGraph (the agent-loop path) would
      // receive no tools.
      $tools = NULL;
      $toolGraph = $this->toolGraphFor($pipeline);
      if ($this->toolProjector !== NULL && $toolGraph !== NULL) {
        $tools = $this->toolProjector->project(
          $job->getNodeId(),
          $toolGraph,
          $context,
          $executionId,
          // The LIVE pipeline instance: tool calls are recorded as normal
          // pipeline job items appended here; the end-of-run save persists
          // the references.
          $pipeline,
        );
      }

      // Execute the node, retrying transient failures in place per the job's
      // retry budget. The thunk preserves this orchestrator's executeNode
      // argument shape (tool collection + graph-state injector).
      $result = $this->executeNodeWithRetry(
        $job,
        function () use ($executionId, $job, $nodeTypeId, $runtimeInputData, $configData, $context, $tools, $state): NodeExecutionResult {
          return $this->nodeRuntime->executeNode(
            $executionId,
            $job->getNodeId(),
            $nodeTypeId,
            $runtimeInputData,
            $configData,
            $context,
            $tools,
            // Inject the graph state into processors that declare the
            // capability, mirroring the ToolsAware pattern. State-heavy
            // processors implement StateAwareProcessorInterface instead of
            // reading reserved __*__ input parameters.
            static function (object $processor) use ($state): void {
              if ($processor instanceof StateAwareProcessorInterface) {
                $processor->setGraphState($state);
              }
            }
          );
        }
      );

      // First-class node failure: an error-status Output must fail the job,
      // not silently complete. Node processor failures are swallowed into the
      // Output by NodeRuntimeService. 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 continue
      //   the run so the error handler can pick it up.
      // - No error edge → re-raise to fail the run via the catch path below
      //   (status-quo behaviour).
      $output = $result->getOutput();
      if ($output->isError()) {
        if ($this->nodeHasErrorEdge($job)) {
          $errorOutput = $this->routeNodeError($job, $pipeline, $output, $result->getExecutionTime(), $executionId);
          return [
            "output" => $errorOutput,
            "stateUpdate" => NULL,
          ];
        }
        throw new NodeProcessingException(
          $output->getErrorMessage() ?? "Node {$job->getNodeId()} reported an error"
        );
      }

      $outputData = $output->toArray();
      $job->markAsCompleted($outputData);

      // Stamp execution_order (and execution_time_us) so that multi-source
      // input ports resolve to the latest executor — see
      // buildStateGraphRuntimeInputData(). Each loop re-execution completes
      // later and so receives a higher order, which is exactly the resolution
      // we want inside loops.
      $this->updateJobExecutionMetadata($job, $pipeline, $result->getExecutionTime());
      $this->jobStorage->save($job);

      // Dispatch here rather than via AbstractOrchestratorBase::executeJob()
      // because this orchestrator bypasses that method entirely and manages its
      // own execution loop via executeJobWithState().
      $this->dispatchJobCompleted($job, $pipeline, $executionId);

      $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) {
      // For message-shape interrupts the flowdrop_interrupt subscriber has
      // already persisted the entity and called markAsInterrupted on the
      // job by the time we land here — so `$e->getInterruptId()` would
      // throw a LogicException (no entity on the exception). Reload the
      // job and trust its metadata; only fall back to the exception for
      // legacy entity-shape interrupts that the subscriber didn't touch.
      $reloadedJob = $this->jobStorage->load($job->id());
      if ($reloadedJob instanceof FlowDropJobInterface) {
        $job = $reloadedJob;
      }

      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 interrupted. Interrupt ID: @interrupt_id", [
        "@job" => $job->id(),
        "@interrupt_id" => $interruptId,
      ]);

      throw $e;
    }
    catch (WorkflowStopException $e) {
      // A terminal node signals a clean stop, not a job failure. Mark this job
      // completed with the terminal payload, then re-throw — before the generic
      // \Exception catch below — so the execution 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 stopped the workflow", [
        "@job" => $job->id(),
      ]);

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

      // See comment in the success branch above.
      $this->dispatchJobCompleted($job, $pipeline, $executionId);

      $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\FlowDropPipelineRuntimeInterface $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,
    FlowDropPipelineRuntimeInterface $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(JobStatus::COMPLETED);

    // For each source node keep its latest completed job. Inside loops a node
    // runs multiple times; the highest job ID is the most recent run.
    $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;
        }
      }
    }

    // Map each source node to its execution order so multi-source ports can be
    // resolved last-executor-wins, consistent with the other orchestrators.
    $executionOrderMap = [];
    foreach ($completedJobsMap as $sourceNodeId => $completedJob) {
      $order = $completedJob->getMetadata()["execution_order"] ?? 0;
      if ($order > 0) {
        $executionOrderMap[$sourceNodeId] = $order;
      }
    }

    // Convert to typed edges once, then sort latest-execution-order first so
    // the first value written to a port (below) comes from the most recently
    // executed source.
    $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());
    });

    // Handled failures (FAILED + error_routed) feed their error payload to an
    // error handler; a routed-failure source is never in completedJobsMap.
    $errorRoutedMap = $this->buildErrorRoutedMap($pipeline);

    // Last-executor-wins: keep the first value written to each port (the latest
    // executor) and ignore earlier-executed sources targeting the same port —
    // one value per port, never an aggregated array.
    $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();

      $sourceHandle = $edge->getSourceHandle();
      $targetHandle = $edge->getTargetHandle();

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

      if ($sourcePortName !== NULL && $targetPortName !== NULL && isset($sourceOutput[$sourcePortName])) {
        if (isset($portSources[$targetPortName])) {
          // Parity with the sync/async orchestrators: surface the discarded
          // source so a fan-in into a single-value port is debuggable.
          $this->logger->warning(
            "Multiple sources target port '@port' on node '@node'; keeping value " .
            "from latest executor '@kept', ignoring '@ignored'.",
            [
              "@port" => $targetPortName,
              "@node" => $edge->getTarget(),
              "@kept" => $portSources[$targetPortName],
              "@ignored" => $sourceNodeId,
            ]
          );
          continue;
        }
        $runtimeData[$targetPortName] = $sourceOutput[$sourcePortName];
        $portSources[$targetPortName] = $sourceNodeId;
      }
    }

    return $runtimeData;
  }

  /**
   * Builds (and caches) the dependency graph used for tool projection.
   *
   * StateGraph runs off a pipeline rather than a compiled workflow, so it has
   * no DependencyGraph on hand. We derive one from the pipeline's workflow
   * graph data — the same source getLoopbackTargetIds() reads — and memoise
   * it per workflow, since the execution loop revisits nodes (loops
   * re-dispatch).
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The running pipeline.
   *
   * @return \Drupal\flowdrop_runtime\DTO\Compiler\DependencyGraph|null
   *   The dependency graph, or NULL when no projector is wired or the pipeline
   *   has no workflow.
   */
  private function toolGraphFor(FlowDropPipelineRuntimeInterface $pipeline): ?DependencyGraph {
    if ($this->toolProjector === NULL) {
      return NULL;
    }
    $workflow = $pipeline->getWorkflow();
    if ($workflow === NULL) {
      return NULL;
    }
    $key = $pipeline->getWorkflowId();
    if (!isset($this->toolGraphCache[$key])) {
      $graphData = $workflow->getGraphData();
      // Re-inject fresh metadata from the live node type so the projected tool
      // graph reflects the authoritative definition, not the stored cache.
      if ($this->nodeMetadataResolver !== NULL) {
        $graphData = $this->nodeMetadataResolver->enrichWorkflow($graphData);
      }
      $this->toolGraphCache[$key] = DependencyGraph::fromWorkflow(
        WorkflowDTO::fromArray($graphData)
      );
    }
    return $this->toolGraphCache[$key];
  }

  /**
   * Gets the IDs of nodes reached by a loopback edge from a node.
   *
   * @param string $sourceNodeId
   *   The source node ID.
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineRuntimeInterface $pipeline
   *   The running pipeline.
   *
   * @return array<string, true>
   *   Set of target node IDs reachable by a loopback edge, keyed by ID for
   *   O(1) membership checks.
   */
  private function getLoopbackTargetIds(
    string $sourceNodeId,
    FlowDropPipelineRuntimeInterface $pipeline,
  ): array {
    $loopbackTargetIds = [];

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

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

    foreach ($edges as $edge) {
      if (!is_array($edge)) {
        continue;
      }
      $transportEdge = TransportEdge::fromArray($edge);
      if ($transportEdge->getSource() !== $sourceNodeId) {
        continue;
      }

      if (EdgePortClassifier::isLoopbackTarget($transportEdge->getTargetHandle())) {
        $loopbackTargetIds[$transportEdge->getTarget()] = TRUE;
      }
    }

    return $loopbackTargetIds;
  }

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

  /**
   * Decides whether a loopback edge should be followed (target re-dispatched).
   *
   * Two kinds of loop are supported:
   * - Iterator-driven loops (e.g. ForEach): re-dispatch only while the active
   *   iterator reports more items, preserving the existing semantics via
   *   {@see self::shouldContinueLoop()}.
   * - Branch-driven loops (e.g. an agent reason/act loop): no iterator exists.
   *   Continuation is decided upstream by a gateway, so the loopback edge only
   *   surfaces as an active next-node when the loop should continue; the loop
   *   ends naturally once the gateway stops routing into it. Follow the edge,
   *   bounded by the per-node iteration count as a runaway backstop (the global
   *   max-iterations guard on the outer loop is the final safety net).
   *
   * @param \Drupal\flowdrop_stategraph\DTO\GraphState $state
   *   The current graph state.
   * @param string $targetNodeId
   *   The loopback target — the node that would be re-dispatched.
   * @param array<string, int> $nodeIterationCounts
   *   Per-node execution counts so far, keyed by node ID.
   * @param int $maxIterations
   *   The configured iteration cap.
   *
   * @return bool
   *   TRUE if the loopback target should be re-dispatched.
   */
  private function shouldFollowLoopbackEdge(
    GraphState $state,
    string $targetNodeId,
    array $nodeIterationCounts,
    int $maxIterations,
  ): bool {
    // Iterator-driven loop: keep the existing ForEach behavior exactly.
    if ($state->iterator !== NULL) {
      return $this->shouldContinueLoop($state);
    }

    // Branch-driven loop: the edge is already an active next-node, so follow it
    // up to the per-node iteration cap.
    return ($nodeIterationCounts[$targetNodeId] ?? 0) < $maxIterations;
  }

  /**
   * 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: NodeSnapshot::STATUS_COMPLETED,
        output: $output,
        error: NULL,
        injected: FALSE,
        executionOrder: $executionOrder++,
        metadata: [],
      );
    }

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

    $status = WorkflowSnapshot::STATUS_RUNNING;
    if ($graphState->error !== NULL) {
      $status = WorkflowSnapshot::STATUS_FAILED;
    }
    elseif ($graphState->isComplete) {
      $status = WorkflowSnapshot::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);
  }

}
