<?php

declare(strict_types=1);

namespace Drupal\flowdrop_workflow_executor\Service;

use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Logger\LoggerChannelFactoryInterface;
use Drupal\Core\Queue\QueueFactory;
use Drupal\Core\Queue\QueueInterface;
use Drupal\Core\State\StateInterface;
use Drupal\flowdrop_job\Constants\JobStatus;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_workflow_executor\DTO\WorkflowExecutionResult;
use Drupal\flowdrop_workflow_executor\Enum\WorkflowExecutionMode;
use Drupal\flowdrop_workflow_executor\Enum\WorkflowExecutionStatus;
use Drupal\flowdrop\Service\EntitySerializer;
use Drupal\flowdrop_session\Constants\MessageRole;
use Drupal\flowdrop_session\Constants\MessageStatus;
use Drupal\flowdrop_session\Constants\SessionStatus;
use Drupal\flowdrop_session\Entity\FlowDropSessionInterface;
use Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface;
use Drupal\flowdrop_session\Service\MessageProcessorInterface;
use Drupal\flowdrop_session\Service\SessionService;
use Drupal\flowdrop_workflow\WorkflowDefinitionInterface;
use Drupal\flowdrop_interrupt\DTO\ExecutionStamp;
use Drupal\flowdrop_interrupt\Enum\InterruptType;
use Drupal\flowdrop_interrupt\Message\InwardSignalMessage;
use Drupal\flowdrop_interrupt\Service\InterruptManagerInterface;
use Drupal\flowdrop_interrupt\Service\InterruptMessageBusInterface;
use Psr\Log\LoggerInterface;

/**
 * Service for executing FlowDrop workflows programmatically.
 *
 * This service provides methods to execute workflows with different modes:
 * - job: Create session, trigger execution, wait for completion
 * - job_async: Create session, trigger execution, return immediately
 * - job_fire_forget: Queue for later execution (truly async, no blocking)
 *
 * @internal This class is not part of the public API and may change without notice.
 */
class WorkflowExecutionService implements WorkflowExecutionServiceInterface {

  /**
   * Maximum execution depth to prevent infinite recursion.
   */
  private const MAX_EXECUTION_DEPTH = 5;

  /**
   * Metadata key for workflow executor input data.
   */
  private const INPUT_METADATA_KEY = "workflow_executor_input";

  /**
   * Metadata key for parent context.
   */
  private const PARENT_CONTEXT_KEY = "workflow_executor_parent_context";

  /**
   * Queue name for fire-and-forget workflow execution.
   */
  public const QUEUE_NAME = "workflow_executor_queue";

  /**
   * The logger instance.
   *
   * @var \Psr\Log\LoggerInterface
   */
  protected LoggerInterface $logger;

  /**
   * The queue for fire-and-forget execution.
   *
   * @var \Drupal\Core\Queue\QueueInterface
   */
  protected QueueInterface $queue;

  /**
   * Constructs a WorkflowExecutionService object.
   *
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   The entity type manager.
   * @param \Drupal\flowdrop_session\Service\SessionService $sessionService
   *   The playground service.
   * @param \Drupal\flowdrop\Service\EntitySerializer $entitySerializer
   *   The entity serializer.
   * @param \Drupal\flowdrop_session\Service\MessageProcessorInterface $messageProcessor
   *   The message processor for workflow execution.
   * @param \Drupal\Core\State\StateInterface $state
   *   The state service.
   * @param \Drupal\Core\Queue\QueueFactory $queueFactory
   *   The queue factory.
   * @param \Drupal\Core\Logger\LoggerChannelFactoryInterface $loggerFactory
   *   The logger factory.
   * @param \Drupal\flowdrop_interrupt\Service\InterruptManagerInterface|null $interruptManager
   *   The interrupt manager, if available.
   * @param \Drupal\flowdrop_interrupt\Service\InterruptMessageBusInterface|null $messageBus
   *   The interrupt message bus, if available.
   */
  public function __construct(
    protected readonly EntityTypeManagerInterface $entityTypeManager,
    protected readonly SessionService $sessionService,
    protected readonly EntitySerializer $entitySerializer,
    protected readonly MessageProcessorInterface $messageProcessor,
    protected readonly StateInterface $state,
    QueueFactory $queueFactory,
    LoggerChannelFactoryInterface $loggerFactory,
    protected readonly ?InterruptManagerInterface $interruptManager = NULL,
    protected readonly ?InterruptMessageBusInterface $messageBus = NULL,
  ) {
    $this->queue = $queueFactory->get(self::QUEUE_NAME);
    $this->logger = $loggerFactory->get("workflow_executor");
  }

  /**
   * {@inheritdoc}
   */
  public function execute(
    WorkflowDefinitionInterface $workflow,
    array $inputData = [],
    WorkflowExecutionMode $mode = WorkflowExecutionMode::Job,
    int $timeout = 300,
    int $pollIntervalMs = 2000,
    array $parentContext = [],
  ): WorkflowExecutionResult {
    $startTime = microtime(TRUE);
    $workflowId = $workflow->id();

    try {
      // Check execution depth to prevent infinite recursion.
      $currentDepth = ($parentContext["execution_depth"] ?? 0) + 1;
      if ($currentDepth > self::MAX_EXECUTION_DEPTH) {
        return WorkflowExecutionResult::failure(
          "Maximum execution depth (" . self::MAX_EXECUTION_DEPTH . ") exceeded. Possible infinite recursion detected."
        );
      }

      // Check for self-reference.
      $workflowChain = $parentContext["workflow_chain"] ?? [];
      if (in_array($workflowId, $workflowChain, TRUE)) {
        return WorkflowExecutionResult::failure(
          "Circular workflow reference detected: workflow '{$workflowId}' is already in the execution chain."
        );
      }

      // For fire-and-forget mode, queue the execution and return immediately.
      // This avoids blocking because Flowdrop uses synchronous_pipeline.
      if ($mode === WorkflowExecutionMode::JobFireForget) {
        return $this->queueForExecution($workflowId, $inputData, $parentContext, $currentDepth);
      }

      // For job and job_async modes, execute immediately.
      return $this->executeImmediate($workflow, $inputData, $mode, $timeout, $pollIntervalMs, $parentContext, $currentDepth, $startTime);
    }
    catch (\Exception $e) {
      $this->logger->error("Workflow execution failed: @message", [
        "@message" => $e->getMessage(),
      ]);
      return WorkflowExecutionResult::failure($e->getMessage());
    }
  }

  /**
   * Queues a workflow for later execution (fire-and-forget).
   *
   * @param string $workflowId
   *   The workflow ID.
   * @param array<string, mixed> $inputData
   *   Input data for the workflow.
   * @param array<string, mixed> $parentContext
   *   Parent context.
   * @param int $currentDepth
   *   Current execution depth.
   *
   * @return \Drupal\flowdrop_workflow_executor\DTO\WorkflowExecutionResult
   *   The execution result (pending status).
   */
  protected function queueForExecution(
    string $workflowId,
    array $inputData,
    array $parentContext,
    int $currentDepth,
  ): WorkflowExecutionResult {
    $queueItem = [
      "workflow_id" => $workflowId,
      "input_data" => $inputData,
      "parent_context" => [
        "execution_depth" => $currentDepth,
        "workflow_chain" => array_merge($parentContext["workflow_chain"] ?? [], [$workflowId]),
      ],
    ];

    $this->queue->createItem($queueItem);

    $this->logger->info("Queued workflow '@workflow' for fire-and-forget execution.", [
      "@workflow" => $workflowId,
    ]);

    // Return pending with a generated reference ID (no session yet).
    $queueRef = "queued_" . uniqid();
    return WorkflowExecutionResult::pending($queueRef);
  }

  /**
   * Executes a workflow immediately (job or job_async mode).
   *
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The workflow.
   * @param array<string, mixed> $inputData
   *   Input data.
   * @param \Drupal\flowdrop_workflow_executor\Enum\WorkflowExecutionMode $mode
   *   Execution mode.
   * @param int $timeout
   *   Timeout in seconds.
   * @param int $pollIntervalMs
   *   Poll interval in milliseconds.
   * @param array<string, mixed> $parentContext
   *   Parent context.
   * @param int $currentDepth
   *   Current execution depth.
   * @param float $startTime
   *   Start time for timing.
   *
   * @return \Drupal\flowdrop_workflow_executor\DTO\WorkflowExecutionResult
   *   The execution result.
   */
  protected function executeImmediate(
    WorkflowDefinitionInterface $workflow,
    array $inputData,
    WorkflowExecutionMode $mode,
    int $timeout,
    int $pollIntervalMs,
    array $parentContext,
    int $currentDepth,
    float $startTime,
  ): WorkflowExecutionResult {
    $workflowId = $workflow->id();

    // Build metadata for the session.
    $metadata = $this->buildSessionMetadata($inputData, $parentContext, $workflowId, $currentDepth);

    // Resolve target session: reuse the parent's when one was plumbed
    // through parentContext (synchronous sub-workflow on shared session),
    // otherwise spin up a fresh playground session. The parent session id
    // is stamped by WorkflowNode::process() from the parent pipeline's
    // typed session_id column. Async dispatches don't get session-sharing
    // yet — their later resume goes through a queued message that still
    // resolves the workflow from the session.
    $parentSessionId = $this->resolveParentSessionId($parentContext);
    $sharedSession = NULL;
    if ($parentSessionId !== NULL && $mode === WorkflowExecutionMode::Job) {
      $sharedSession = $this->loadSession($parentSessionId);
      if ($sharedSession === NULL) {
        $this->logger->warning(
          "Parent session @id not found; falling back to a new session for sub-workflow @workflow.",
          ["@id" => $parentSessionId, "@workflow" => $workflowId]
        );
      }
    }

    if ($sharedSession !== NULL) {
      $session = $sharedSession;
    }
    else {
      $sessionName = "WorkflowExecutor: {$workflow->label()}";
      $session = $this->createSession($workflow, $sessionName, $metadata);
      if ($session === NULL) {
        return WorkflowExecutionResult::failure("Failed to create playground session");
      }
    }

    $sessionId = (string) $session->id();

    // Create the trigger message. On a shared session the sub-invocation is
    // a system message so the parent's chat does not show a fake user turn
    // for each sub-workflow fire.
    $message = $this->createTriggerMessage($session, $inputData, $sharedSession !== NULL);

    $this->logger->info("Triggered workflow '@workflow' in mode '@mode' (session: @session, shared: @shared)", [
      "@workflow" => $workflow->label(),
      "@mode" => $mode,
      "@session" => $sessionId,
      "@shared" => $sharedSession !== NULL ? 'yes' : 'no',
    ]);

    // Handle execution modes.
    if ($mode === WorkflowExecutionMode::JobAsync) {
      // Return immediately with session ID for later retrieval.
      return WorkflowExecutionResult::pending($sessionId);
    }

    // Job mode: process message synchronously if available.
    if ($message !== NULL) {
      try {
        // Shared session → workflow-explicit entry point that does not
        // resolve the workflow from $session->getWorkflow() and does not
        // touch the outer session status. Standalone session → the
        // original processMessage() entry point.
        if ($sharedSession !== NULL) {
          $this->messageProcessor->executeWorkflowOnSession($message, $session, $workflow, TRUE, $parentContext);
        }
        else {
          $this->messageProcessor->processMessage($message);
        }

        // Reload session to get updated status after processing.
        $session = $this->loadSession($sessionId);
        if ($session === NULL) {
          return WorkflowExecutionResult::failure("Session lost after processing", $sessionId);
        }

        // The sub-pipeline id is what the cascade subscriber matches
        // sentinels on. processMessage stamps it on the message's
        // execution_id and on the session's current_pipeline_id.
        $subPipelineId = (string) ($message->getExecutionId() ?? '');
        if ($subPipelineId === '') {
          $subPipelineId = (string) ($session->getCurrentPipelineId() ?? '');
        }

        // On a shared session the outer session status reflects the
        // parent's lifecycle, not the sub's — read the sub trigger
        // message's status instead. Standalone sessions use the
        // historical session-status signal.
        if ($sharedSession !== NULL) {
          $messageStatus = $message->getStatus();

          if ($messageStatus === MessageStatus::COMPLETED) {
            $outputs = $this->extractOutputs($session, $subPipelineId);
            $executionTimeMs = (int) ((microtime(TRUE) - $startTime) * 1000);
            return WorkflowExecutionResult::success(
              sessionId: $sessionId,
              outputs: $outputs,
              executionTimeMs: $executionTimeMs,
              pipelineId: $subPipelineId,
            );
          }

          if ($messageStatus === MessageStatus::FAILED) {
            return WorkflowExecutionResult::failure("Workflow execution failed", $sessionId);
          }

          if ($messageStatus === MessageStatus::AWAITING_INPUT) {
            return WorkflowExecutionResult::paused($sessionId, $subPipelineId);
          }
        }
        else {
          $sessionStatus = $session->getStatus();

          if ($sessionStatus === SessionStatus::IDLE) {
            // Workflow completed successfully. processMessage() sets the
            // session back to IDLE on the happy path, so reaching here in
            // sync mode means the sub-pipeline finished.
            $outputs = $this->extractOutputs($session, $subPipelineId);
            $executionTimeMs = (int) ((microtime(TRUE) - $startTime) * 1000);
            return WorkflowExecutionResult::success(
              sessionId: $sessionId,
              outputs: $outputs,
              executionTimeMs: $executionTimeMs,
              pipelineId: $subPipelineId,
            );
          }

          if ($sessionStatus === SessionStatus::FAILED) {
            return WorkflowExecutionResult::failure("Workflow execution failed", $sessionId);
          }

          if ($sessionStatus === SessionStatus::AWAITING_INPUT) {
            // Sub-workflow paused on a HITL interrupt. Don't poll — the
            // caller (WorkflowNode) is responsible for converting this
            // into a linked interrupt on the parent pipeline so the
            // parent pauses too instead of blocking the HTTP request
            // until the user responds.
            return WorkflowExecutionResult::paused($sessionId, $subPipelineId);
          }
        }
      }
      catch (\Exception $e) {
        return WorkflowExecutionResult::failure($e->getMessage(), $sessionId);
      }
    }

    // Fallback: wait for completion (message creation failure path).
    return $this->waitForCompletion($sessionId, $timeout, $pollIntervalMs, $startTime);
  }

  /**
   * Reads the parent session id from parentContext when shareable.
   *
   * Returns NULL when no parent session is available — the caller falls
   * back to creating a fresh playground session.
   *
   * @param array<string, mixed> $parentContext
   *   Parent context plumbed in by WorkflowNode.
   */
  private function resolveParentSessionId(array $parentContext): ?string {
    $candidate = $parentContext["session_id"] ?? NULL;
    if (!is_string($candidate)) {
      return NULL;
    }
    return $candidate !== "" ? $candidate : NULL;
  }

  /**
   * {@inheritdoc}
   */
  public function getResults(
    string $sessionId,
    bool $waitForCompletion = TRUE,
    int $timeout = 300,
    int $pollIntervalMs = 2000,
  ): WorkflowExecutionResult {
    $startTime = microtime(TRUE);

    try {
      if ($waitForCompletion) {
        return $this->waitForCompletion($sessionId, $timeout, $pollIntervalMs, $startTime);
      }

      // Just check current status without waiting.
      $session = $this->loadSession($sessionId);
      if ($session === NULL) {
        return WorkflowExecutionResult::failure("Session not found: {$sessionId}");
      }

      $status = $this->getSessionStatus($session);

      if ($status === WorkflowExecutionStatus::Completed) {
        $outputs = $this->extractOutputs($session);
        $executionTimeMs = (int) ((microtime(TRUE) - $startTime) * 1000);
        return WorkflowExecutionResult::success($sessionId, $outputs, $executionTimeMs);
      }

      if ($status === WorkflowExecutionStatus::Failed) {
        return WorkflowExecutionResult::failure("Workflow execution failed", $sessionId);
      }

      return WorkflowExecutionResult::running($sessionId);
    }
    catch (\Exception $e) {
      return WorkflowExecutionResult::failure($e->getMessage(), $sessionId);
    }
  }

  /**
   * {@inheritdoc}
   */
  public function isComplete(string $sessionId): bool {
    $session = $this->loadSession($sessionId);
    if ($session === NULL) {
      return FALSE;
    }

    $status = $this->getSessionStatus($session);
    return $status === WorkflowExecutionStatus::Completed
      || $status === WorkflowExecutionStatus::Failed;
  }

  /**
   * {@inheritdoc}
   */
  public function getStatus(string $sessionId): WorkflowExecutionStatus {
    $session = $this->loadSession($sessionId);
    if ($session === NULL) {
      return WorkflowExecutionStatus::Failed;
    }

    return $this->getSessionStatus($session);
  }

  /**
   * {@inheritdoc}
   */
  public function cancelExecution(
    string $sessionId,
    ?string $reason = NULL,
    ?int $initiatorUid = NULL,
  ): bool {
    if ($this->interruptManager === NULL) {
      $this->logger->warning(
        'cancelExecution called for session @session but InterruptManager is unavailable.',
        ['@session' => $sessionId],
      );
      return FALSE;
    }
    $pipelineId = $this->findActiveSubPipelineForSession($sessionId);
    if ($pipelineId === NULL) {
      $this->logger->info(
        'cancelExecution: no active sub-pipeline for session @session; nothing to cancel.',
        ['@session' => $sessionId],
      );
      return FALSE;
    }
    return $this->cancelPipeline($pipelineId, $reason, $initiatorUid);
  }

  /**
   * {@inheritdoc}
   *
   * Cascade entry point — invoked from `WorkflowNode::cancel()` when a
   * parent pipeline is cancelled and its WorkflowExecution interrupt
   * context carries `sub_pipeline_id`.
   */
  public function cancelPipeline(
    string $pipelineId,
    ?string $reason = NULL,
    ?int $initiatorUid = NULL,
  ): bool {
    if ($pipelineId === '') {
      return FALSE;
    }
    if ($this->interruptManager === NULL) {
      $this->logger->warning(
        'cancelPipeline called for pipeline @pipeline but InterruptManager is unavailable.',
        ['@pipeline' => $pipelineId],
      );
      return FALSE;
    }
    // Bail if a Cancel signal is already pending — prevents redundant
    // cascade signals when multiple parents target the same sub-pipeline.
    if ($this->interruptManager->findPendingInwardSignal($pipelineId) !== NULL) {
      $this->logger->debug(
        'cancelPipeline: pipeline @pipeline already has a pending inward signal; skipping.',
        ['@pipeline' => $pipelineId],
      );
      return FALSE;
    }
    if ($this->messageBus === NULL) {
      $this->logger->warning(
        'cancelPipeline: message bus unavailable; cannot create cascade signal.',
      );
      return FALSE;
    }

    // Look up the pipeline's workflow id for the execution stamp. Best
    // effort — the cascade still works if it can't be resolved.
    $workflowId = '';
    try {
      $pipeline = $this->entityTypeManager
        ->getStorage('flowdrop_pipeline')
        ->load($pipelineId);
      if ($pipeline !== NULL && method_exists($pipeline, 'getWorkflowId')) {
        $workflowId = (string) ($pipeline->getWorkflowId() ?? '');
      }
    }
    catch (\Exception $e) {
      // Non-fatal — fall through with empty workflowId.
    }

    $interrupt = $this->messageBus->dispatch(
      new InwardSignalMessage(
        type: InterruptType::Cancel,
        message: $reason !== NULL && $reason !== ''
          ? 'Cascaded cancel: ' . $reason
          : 'Cascaded cancel from parent workflow',
        targetPipelineId: $pipelineId,
        initiatorUid: $initiatorUid,
        reason: $reason,
      ),
      new ExecutionStamp(
        workflowId: $workflowId,
        nodeId: 'signal:cancel:cascade',
      ),
    );

    $this->logger->info(
      'Cascade Cancel signal @uuid created on sub-pipeline @pipeline.',
      [
        '@uuid' => $interrupt->uuid(),
        '@pipeline' => $pipelineId,
      ],
    );
    return TRUE;
  }

  /**
   * Finds the most-recent active sub-pipeline bound to a session.
   *
   * "Active" means not in a terminal status (completed/failed/cancelled).
   * Used as a fallback when only a session id is available — newer
   * dispatches stamp sub_pipeline_id on the interrupt context and use
   * cancelPipeline() directly.
   *
   * @param string $sessionId
   *   The session id.
   *
   * @return string|null
   *   The pipeline id, or NULL if no active sub-pipeline is bound to
   *   the session.
   */
  protected function findActiveSubPipelineForSession(string $sessionId): ?string {
    try {
      $ids = $this->entityTypeManager
        ->getStorage('flowdrop_pipeline')
        ->getQuery()
        ->accessCheck(FALSE)
        ->condition('session_id', $sessionId)
        ->condition('status', [PipelineStatus::COMPLETED, PipelineStatus::FAILED, PipelineStatus::CANCELLED], 'NOT IN')
        ->sort('id', 'DESC')
        ->range(0, 1)
        ->execute();
      if (empty($ids)) {
        return NULL;
      }
      return (string) reset($ids);
    }
    catch (\Exception $e) {
      $this->logger->warning(
        'findActiveSubPipelineForSession failed for session @id: @msg',
        ['@id' => $sessionId, '@msg' => $e->getMessage()],
      );
      return NULL;
    }
  }

  /**
   * Builds metadata for the session.
   *
   * @param array<string, mixed> $inputData
   *   Input data for the workflow.
   * @param array<string, mixed> $parentContext
   *   Parent context for recursion tracking.
   * @param string $workflowId
   *   The workflow ID.
   * @param int $currentDepth
   *   Current execution depth.
   *
   * @return array<string, mixed>
   *   The metadata array.
   */
  protected function buildSessionMetadata(
    array $inputData,
    array $parentContext,
    string $workflowId,
    int $currentDepth,
  ): array {
    // Build workflow chain for circular reference detection.
    $workflowChain = $parentContext["workflow_chain"] ?? [];
    $workflowChain[] = $workflowId;

    return [
      self::INPUT_METADATA_KEY => $inputData,
      self::PARENT_CONTEXT_KEY => [
        "execution_depth" => $currentDepth,
        "workflow_chain" => $workflowChain,
        "parent_session_id" => $parentContext["session_id"] ?? NULL,
      ],
    ];
  }

  /**
   * Creates a playground session for workflow execution.
   *
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The workflow.
   * @param string $sessionName
   *   The session name.
   * @param array<string, mixed> $metadata
   *   Session metadata.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionInterface|null
   *   The created session or NULL on failure.
   */
  protected function createSession(
    WorkflowDefinitionInterface $workflow,
    string $sessionName,
    array $metadata,
  ): ?FlowDropSessionInterface {
    try {
      $session = $this->sessionService->createSession($workflow, $sessionName, $metadata);

      // Workaround: Re-set metadata to ensure proper JSON encoding.
      // (Same pattern as NodeSessionService).
      $session->setMetadata($metadata);
      $session->save();

      return $session;
    }
    catch (\Exception $e) {
      $this->logger->error("Failed to create session: @message", [
        "@message" => $e->getMessage(),
      ]);
      return NULL;
    }
  }

  /**
   * Creates a trigger message to start workflow execution.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session.
   * @param array<string, mixed> $inputData
   *   Input data for the workflow.
   * @param bool $asSystem
   *   When TRUE, mark the trigger message as a system message.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface|null
   *   The created message entity, or NULL on failure.
   */
  protected function createTriggerMessage(
    FlowDropSessionInterface $session,
    array $inputData,
    bool $asSystem = FALSE,
  ): ?FlowDropSessionMessageInterface {
    try {
      // Create a triggering message. For top-level invocations this is a
      // user message (the synthetic intent that kicks off the workflow).
      // For sub-invocations inside a shared session it's a system message
      // so the parent's chat does not show a fake user turn each time a
      // sub-workflow fires.
      $messageContent = !empty($inputData)
        ? "Execute workflow with input data"
        : "Execute workflow";

      // Route through SessionService so trigger rows get the same
      // entity-id-as-sequence_number treatment every other message does.
      // Going through storage directly used to leave sequence_number at
      // its field default of 0, and on a shared session (sub-pipeline
      // reusing the parent's session) those seq=0 rows accumulated at the
      // head of `sequence_number ASC` lookups, pushing recent node-status
      // logs past the playground's default 50-message page.
      $metadata = [
        "inputs" => $inputData,
        "triggered_by" => "workflow_executor",
      ];
      return $asSystem
        ? $this->sessionService->addSystemMessage($session, $messageContent, $metadata)
        : $this->sessionService->addUserMessage($session, $messageContent, $metadata);
    }
    catch (\Exception $e) {
      $this->logger->error("Failed to create trigger message: @message", [
        "@message" => $e->getMessage(),
      ]);
      return NULL;
    }
  }

  /**
   * Waits for workflow completion with polling.
   *
   * @param string $sessionId
   *   The session ID.
   * @param int $timeout
   *   Timeout in seconds.
   * @param int $pollIntervalMs
   *   Polling interval in milliseconds.
   * @param float $startTime
   *   Start time for timing calculation.
   *
   * @return \Drupal\flowdrop_workflow_executor\DTO\WorkflowExecutionResult
   *   The execution result.
   */
  protected function waitForCompletion(
    string $sessionId,
    int $timeout,
    int $pollIntervalMs,
    float $startTime,
  ): WorkflowExecutionResult {
    $timeoutMs = $timeout * 1000;
    $elapsed = 0;

    while ($elapsed < $timeoutMs) {
      $session = $this->loadSession($sessionId);
      if ($session === NULL) {
        return WorkflowExecutionResult::failure("Session not found: {$sessionId}");
      }

      $status = $this->getSessionStatus($session);

      if ($status === WorkflowExecutionStatus::Completed) {
        $outputs = $this->extractOutputs($session);
        $executionTimeMs = (int) ((microtime(TRUE) - $startTime) * 1000);
        return WorkflowExecutionResult::success($sessionId, $outputs, $executionTimeMs);
      }

      if ($status === WorkflowExecutionStatus::Failed) {
        $executionTimeMs = (int) ((microtime(TRUE) - $startTime) * 1000);
        return WorkflowExecutionResult::failure("Workflow execution failed", $sessionId);
      }

      // Sleep and continue polling.
      usleep($pollIntervalMs * 1000);
      $elapsed = (int) ((microtime(TRUE) - $startTime) * 1000);
    }

    // Timeout reached.
    $executionTimeMs = (int) ((microtime(TRUE) - $startTime) * 1000);
    return WorkflowExecutionResult::timeout($sessionId, $executionTimeMs);
  }

  /**
   * Loads a session by ID.
   *
   * @param string $sessionId
   *   The session ID.
   *
   * @return \Drupal\flowdrop_session\Entity\FlowDropSessionInterface|null
   *   The session or NULL if not found.
   */
  protected function loadSession(string $sessionId): ?FlowDropSessionInterface {
    try {
      $session = $this->entityTypeManager
        ->getStorage("flowdrop_session")
        ->load($sessionId);

      if ($session instanceof FlowDropSessionInterface) {
        return $session;
      }
    }
    catch (\Exception $e) {
      $this->logger->error("Failed to load session @id: @message", [
        "@id" => $sessionId,
        "@message" => $e->getMessage(),
      ]);
    }

    return NULL;
  }

  /**
   * Gets the status of a session.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session.
   *
   * @return \Drupal\flowdrop_workflow_executor\Enum\WorkflowExecutionStatus
   *   The execution status.
   */
  protected function getSessionStatus(FlowDropSessionInterface $session): WorkflowExecutionStatus {
    $status = $session->getStatus();
    // Map session status to our status constants.
    // Note: "idle" maps to Completed because processMessage() sets the session
    // back to "idle" after successful execution. When polling via
    // waitForCompletion(), the session has already been triggered, so "idle"
    // means execution finished (e.g. after human-in-loop completion).
    return match ($status) {
      SessionStatus::IDLE, SessionStatus::COMPLETED, "finished" => WorkflowExecutionStatus::Completed,
      SessionStatus::FAILED, "error" => WorkflowExecutionStatus::Failed,
      SessionStatus::RUNNING, "processing" => WorkflowExecutionStatus::Running,
      SessionStatus::AWAITING_INPUT => WorkflowExecutionStatus::Paused,
      default => WorkflowExecutionStatus::Pending,
    };
  }

  /**
   * Extracts outputs from a completed session.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session.
   *
   * @return array<string, mixed>
   *   The outputs array.
   */

  /**
   * Finds the most recent pipeline whose execution_context links to a session.
   *
   * Pipelines are linked to sessions through the playground sub-context
   * (`execution_context.playground.session_id`) set by SessionExecutionService.
   * Until sessions reliably carry a current_pipeline_id, we look it up here.
   */
  protected function findPipelineForSession(FlowDropSessionInterface $session): ?object {
    $sessionId = $session->id();
    if ($sessionId === NULL) {
      return NULL;
    }
    try {
      $storage = $this->entityTypeManager->getStorage('flowdrop_pipeline');
      $ids = $storage->getQuery()
        ->accessCheck(FALSE)
        ->condition('execution_context', '%"session_id":"' . $sessionId . '"%', 'LIKE')
        ->sort('id', 'DESC')
        ->range(0, 1)
        ->execute();
      if (empty($ids)) {
        return NULL;
      }
      return $storage->load(reset($ids));
    }
    catch (\Exception $e) {
      $this->logger->warning('Failed to locate pipeline for session @id: @msg', [
        '@id' => $sessionId,
        '@msg' => $e->getMessage(),
      ]);
      return NULL;
    }
  }

  /**
   * {@inheritdoc}
   */
  public function getPipelineOutputs(string $pipelineId): array {
    if ($pipelineId === '') {
      return [];
    }
    try {
      $pipeline = $this->entityTypeManager
        ->getStorage('flowdrop_pipeline')
        ->load($pipelineId);
    }
    catch (\Exception $e) {
      $this->logger->warning('getPipelineOutputs failed to load pipeline @id: @msg', [
        '@id' => $pipelineId,
        '@msg' => $e->getMessage(),
      ]);
      return [];
    }
    if ($pipeline === NULL) {
      return [];
    }
    // The session is needed for the assistant-message backwards-compat
    // aggregation inside extractOutputs. Resolve it from the pipeline's
    // typed session_id column (Phase A) when present.
    $session = NULL;
    if (method_exists($pipeline, 'getSessionId')) {
      $sessionId = $pipeline->getSessionId();
      if (is_string($sessionId) && $sessionId !== '') {
        $session = $this->loadSession($sessionId);
      }
    }
    if ($session === NULL) {
      // Fabricate a minimal session-less path: extractOutputs only needs
      // session->id() for the assistant-message query, and that query is
      // scoped further by execution_id when pipelineId is non-empty.
      // Returning [] here would lose the job outputs entirely; instead
      // walk the completed-jobs directly.
      $outputs = [];
      if (method_exists($pipeline, 'getJobsByStatus')) {
        foreach ($pipeline->getJobsByStatus(JobStatus::COMPLETED) as $job) {
          if (!method_exists($job, 'getNodeId') || !method_exists($job, 'getOutputData')) {
            continue;
          }
          $nodeId = (string) $job->getNodeId();
          if ($nodeId !== '') {
            $outputs[$nodeId] = $job->getOutputData();
          }
        }
      }
      return $outputs;
    }
    return $this->extractOutputs($session, $pipelineId);
  }

  /**
   * Extracts outputs from a session, scoped to a pipeline when provided.
   */
  protected function extractOutputs(FlowDropSessionInterface $session, string $pipelineId = ''): array {
    $outputs = [];

    try {
      // Per-node outputs from completed jobs. When a specific pipeline id
      // is provided (sub-pipelines sharing the parent's session), load
      // that one directly. Otherwise fall back to discovery for legacy
      // one-pipeline-per-session callers.
      $pipeline = NULL;
      if ($pipelineId !== '') {
        $loaded = $this->entityTypeManager->getStorage('flowdrop_pipeline')->load($pipelineId);
        if ($loaded !== NULL) {
          $pipeline = $loaded;
        }
      }
      if ($pipeline === NULL) {
        $pipeline = $this->findPipelineForSession($session);
      }
      if ($pipeline !== NULL && method_exists($pipeline, 'getJobsByStatus')) {
        foreach ($pipeline->getJobsByStatus(JobStatus::COMPLETED) as $job) {
          if (!method_exists($job, 'getNodeId') || !method_exists($job, 'getOutputData')) {
            continue;
          }
          $nodeId = (string) $job->getNodeId();
          if ($nodeId !== '') {
            $outputs[$nodeId] = $job->getOutputData();
          }
        }
      }

      // Backwards-compat: aggregate assistant message content and any flat
      // metadata outputs so non-schema callers (e.g. ChatService) keep
      // working. Scope to the specific pipeline when known so concurrent
      // sub-pipelines in the same session don't bleed outputs into each
      // other.
      $messageQuery = [
        "session_id" => $session->id(),
        "role" => MessageRole::ASSISTANT,
      ];
      if ($pipelineId !== '') {
        $messageQuery['execution_id'] = $pipelineId;
      }
      $messages = $this->entityTypeManager
        ->getStorage("flowdrop_session_message")
        ->loadByProperties($messageQuery);

      foreach ($messages as $message) {
        // Extract content and metadata from message.
        if (method_exists($message, "getContent")) {
          $outputs["content"] = $message->getContent();
        }
        if (method_exists($message, "getMessageMetadata")) {
          $metadata = $message->getMessageMetadata();
          if (!empty($metadata["outputs"])) {
            $outputs = array_merge($outputs, $metadata["outputs"]);
          }
        }
      }
    }
    catch (\Exception $e) {
      $this->logger->error("Failed to extract outputs: @message", [
        "@message" => $e->getMessage(),
      ]);
    }

    return $outputs;
  }

}
