<?php

declare(strict_types=1);

namespace Drupal\flowdrop_session\Service;

use Drupal\Component\Datetime\TimeInterface;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\Core\Logger\LoggerChannelFactoryInterface;
use Drupal\flowdrop\Utility\NodeAnchor;
use Drupal\flowdrop_runtime\Service\Storage\PipelineStorageRegistryInterface;
use Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationRequest;
use Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationResponse;
use Drupal\flowdrop_orchestration\Plugin\Orchestrator\OrchestratorPluginInterface;
use Drupal\flowdrop_orchestration\DTO\Snapshot\WorkflowSnapshot;
use Drupal\flowdrop_orchestration\Service\Orchestrator\OrchestratorHelperInterface;
use Drupal\flowdrop_job\Constants\JobStatus;
use Drupal\flowdrop_pipeline\Constants\PausedReason;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface;
use Drupal\flowdrop_session\Constants\MessageAnnotation;
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\Event\SessionTurnCompletedEvent;
use Drupal\flowdrop_session\Event\SessionTurnStartedEvent;
use Drupal\flowdrop_workflow\WorkflowDefinitionInterface;
use Drupal\flowdrop_workflow\Service\WorkflowSchemaResolverInterface;
use Psr\Log\LoggerInterface;
use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;

/**
 * Handles workflow execution for session messages.
 *
 * This service contains the execution logic extracted from SessionService:
 * - Processing user messages by executing workflows
 * - Building orchestration requests and handling responses
 * - Extracting message content from workflow results
 * - Handling interrupted executions (human-in-loop)
 *
 * @internal This class is not part of the public API and may change without notice.
 */
class SessionExecutionService implements MessageProcessorInterface {

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

  /**
   * The pipeline storage registry.
   *
   * @var \Drupal\flowdrop_runtime\Service\Storage\PipelineStorageRegistryInterface
   */
  protected readonly PipelineStorageRegistryInterface $pipelineStorageRegistry;

  /**
   * The time service.
   *
   * @var \Drupal\Component\Datetime\TimeInterface
   */
  protected readonly TimeInterface $time;

  /**
   * Constructs a new SessionExecutionService.
   *
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   The entity type manager.
   * @param \Drupal\Core\Logger\LoggerChannelFactoryInterface $loggerFactory
   *   The logger factory.
   * @param \Drupal\flowdrop_orchestration\Service\Orchestrator\OrchestratorHelperInterface $orchestratorHelper
   *   The orchestrator helper service.
   * @param \Drupal\flowdrop_session\Service\SessionService $sessionService
   *   The session service for message/session CRUD operations.
   * @param \Drupal\flowdrop_runtime\Service\Storage\PipelineStorageRegistryInterface $pipelineStorageRegistry
   *   The pipeline storage registry.
   * @param \Drupal\Component\Datetime\TimeInterface $time
   *   The time service.
   * @param \Symfony\Contracts\EventDispatcher\EventDispatcherInterface $eventDispatcher
   *   The event dispatcher, for the session turn lifecycle events.
   * @param \Drupal\flowdrop_workflow\Service\WorkflowSchemaResolverInterface|null $schemaResolver
   *   The workflow schema resolver, if available.
   */
  public function __construct(
    protected readonly EntityTypeManagerInterface $entityTypeManager,
    LoggerChannelFactoryInterface $loggerFactory,
    protected readonly OrchestratorHelperInterface $orchestratorHelper,
    protected readonly SessionService $sessionService,
    PipelineStorageRegistryInterface $pipelineStorageRegistry,
    TimeInterface $time,
    protected readonly EventDispatcherInterface $eventDispatcher,
    protected readonly ?WorkflowSchemaResolverInterface $schemaResolver = NULL,
  ) {
    $this->pipelineStorageRegistry = $pipelineStorageRegistry;
    $this->time = $time;
    $this->logger = $loggerFactory->get('flowdrop_session');
  }

  /**
   * {@inheritdoc}
   */
  public function processMessage(FlowDropSessionMessageInterface $message): void {
    $session = $message->getSession();
    if ($session === NULL) {
      throw new \RuntimeException('Message has no associated session');
    }

    $workflow = $session->getWorkflow();
    if ($workflow === NULL) {
      throw new \RuntimeException('Session has no associated workflow');
    }

    $this->executeWorkflowOnSession($message, $session, $workflow);
  }

  /**
   * {@inheritdoc}
   */
  public function executeWorkflowOnSession(
    FlowDropSessionMessageInterface $message,
    FlowDropSessionInterface $session,
    WorkflowDefinitionInterface $workflow,
    bool $nested = FALSE,
    array $parentContext = [],
  ): void {
    // Compute the attributed workflow id: only non-null when the
    // executing workflow differs from the session's primary workflow.
    // This is the signal the UI uses to render a workflow-attribution
    // badge on messages from embedded sub-workflows.
    $sessionWorkflow = $session->getWorkflow();
    $attributedWorkflowId = NULL;
    if ($sessionWorkflow === NULL || (string) $sessionWorkflow->id() !== (string) $workflow->id()) {
      $attributedWorkflowId = (string) $workflow->id();
    }

    // Hierarchy seed: inherit from parent (sub-workflow case) and append
    // the current workflow as the next crumb. $workflow is already in
    // scope as a loaded entity, so label() does not trigger a load.
    $hierarchy = $this->buildHierarchy($parentContext, $workflow);

    // Update status to processing.
    $message->setStatus(MessageStatus::PROCESSING)->save();
    // Nested invocations (sub-workflow inside a parent's session) leave the
    // outer session status alone — the parent's processMessage owns the
    // RUNNING→IDLE transitions. Touching it here would briefly flip the
    // session to IDLE between sub-completion and parent-continuation and a
    // polling frontend could see a false terminal state.
    if (!$nested) {
      $session->setStatus(SessionStatus::RUNNING)->save();
    }

    try {
      // Find ChatInput node.
      $chatInputNode = $this->sessionService->findChatInputNode($workflow);

      // Build initialData (same pattern as triggers).
      $initialData = $this->buildInitialData($message, $chatInputNode, $session, $workflow);

      // Get orchestrator settings (session overrides workflow).
      $orchestratorSettings = $this->getOrchestratorSettings($session, $workflow);
      $orchestrator = $this->getOrchestrator($orchestratorSettings);

      // Build orchestration options.
      $orchestrationOptions = $this->buildOrchestrationOptions($orchestratorSettings, $session, $message, $hierarchy);

      // Resolve the bundle and storage via the registry.
      $bundle = $this->pipelineStorageRegistry->getBundleForOrchestrator($orchestrator);
      $storage = $this->pipelineStorageRegistry->getStorage($bundle);

      // Create the pipeline entity upfront so its ID is the executionId for
      // all messages in this run. Use gmdate() so the label is always UTC.
      $timestamp = $this->time->getCurrentTime();
      $pipeline = $storage->create([
        'bundle'      => $bundle,
        'label'       => 'Session ' . $session->id() . ' - ' . gmdate('Y-m-d H:i:s', $timestamp),
        'workflow_id' => (string) $workflow->id(),
        'input_data'  => $initialData,
        'status'      => PipelineStatus::PENDING,
        'session_id'  => $session->id(),
        // Lineage. For a nested (shared-session sub-workflow) run, inherit the
        // spawning pipeline and the run's root from parentContext (stamped by
        // WorkflowNode). For a top-level run both are NULL and the pipeline
        // self-roots in FlowDropPipeline::postSave() once its id is assigned.
        'parent_pipeline_id' => $parentContext['parent_pipeline_id'] ?? NULL,
        'root_pipeline_id'   => $parentContext['root_pipeline_id'] ?? NULL,
      ]);
      // Set playground context so event subscribers (JobLogSubscriber,
      // PipelineStatusSubscriber) can look up the session. The
      // pipeline_entity_id option causes the orchestrator trait to reuse this
      // pre-created entity and skip buildPipelineExecutionContext(), so the
      // context must be set here before the entity is saved.
      if (!empty($orchestrationOptions['playground'])) {
        $pipeline->setExecutionContext(['playground' => $orchestrationOptions['playground']]);
      }
      $storage->save($pipeline);
      $executionId = (string) $pipeline->id();

      // executionId is now known — add the started notice for top-level
      // runs only. Nested invocations consolidate into a single card at
      // completion (PipelineStatusSubscriber), so a separate "started"
      // row would duplicate the hierarchy chip the card already carries.
      if (!$nested) {
        $startedMetadata = [
          'level' => 'info',
          MessageAnnotation::METADATA_KEY_DISPLAY => MessageAnnotation::DISPLAY_NOTICE,
        ];
        if ($attributedWorkflowId !== NULL) {
          $startedMetadata['workflow_id'] = $attributedWorkflowId;
        }
        if ($hierarchy !== []) {
          $startedMetadata[MessageAnnotation::METADATA_KEY_HIERARCHY] = $hierarchy;
        }
        $this->sessionService->addSystemMessage(
          $session,
          'started',
          $startedMetadata,
          $executionId,
        );
      }
      $message->setExecutionId($executionId)->save();

      // The turn id is the trigger message id; the pipeline carries it in
      // its execution context (playground.message_id, set above) so
      // pipeline-side subscribers can map back to the turn. Top-level only:
      // nested sub-workflow invocations do not start a turn.
      if (!$nested) {
        $this->eventDispatcher->dispatch(
          new SessionTurnStartedEvent(
            sessionId: (string) $session->id(),
            turnId: (string) $message->id(),
            pipelineId: $executionId,
            workflowId: (string) $workflow->id(),
          ),
          SessionTurnStartedEvent::NAME,
        );
      }

      // Build orchestration request. Pass pipeline_entity_id so the
      // orchestrator's trait guard reuses this entity instead of creating
      // a second one.
      $request = new OrchestrationRequest(
        workflowId: (string) $workflow->id(),
        pipelineId: $executionId,
        workflow: $this->workflowToArray($workflow),
        initialData: $initialData,
        options: array_merge($orchestrationOptions, ['pipeline_entity_id' => $pipeline->id()]),
      );

      // Execute (direct return, like triggers).
      $response = $orchestrator->orchestrate($request);

      // Process results FIRST - this creates chat messages for any chat_output
      // nodes that executed before an interrupt. Ensures users see messages
      // from nodes that ran even if the workflow is interrupted. Pass
      // $workflow explicitly — on shared sessions the session's primary
      // workflow is the parent's, but the output nodes we need to map are
      // the sub-workflow's.
      $this->processExecutionResults($message, $response, $attributedWorkflowId, $workflow);

      // Check for interrupted status (human-in-loop).
      if ($response->getStatus() === WorkflowSnapshot::STATUS_INTERRUPTED) {
        $this->handleInterruptedExecution($message, $session, $response);
        $this->dispatchTurnCompleted($message, $session, $nested, SessionTurnCompletedEvent::STATUS_AWAITING_INPUT, $executionId);
        return;
      }

      // StateGraph surfaces HITL as a graceful pause: interrupted jobs
      // pause the pipeline and the response status is 'paused' — no
      // STATUS_INTERRUPTED, no interrupt metadata. Without this branch the
      // code below marks the session IDLE while a question is pending.
      // Discriminator (see StateGraphOrchestrator): only a budget/signal
      // pause stamps a PausedReason on the pipeline context; the HITL
      // pause deliberately carries none.
      if ($response->getStatus() === PipelineStatus::PAUSED) {
        $pausedPipeline = $this->entityTypeManager
          ->getStorage('flowdrop_pipeline')
          ->loadUnchanged($pipeline->id());
        $pausedReason = $pausedPipeline instanceof FlowDropPipelineInterface
          ? ($pausedPipeline->getExecutionContext()[PausedReason::CONTEXT_KEY] ?? NULL)
          : NULL;
        if ($pausedReason === NULL) {
          // HITL pause: same session bookkeeping as the interrupted path.
          $message->setStatus(MessageStatus::AWAITING_INPUT)->save();
          if (!$nested) {
            $session->setStatus(SessionStatus::AWAITING_INPUT)->save();
          }
          $this->dispatchTurnCompleted($message, $session, $nested, SessionTurnCompletedEvent::STATUS_AWAITING_INPUT, $executionId);
        }
        else {
          // Reasoned (budget/signal) pause: the turn ends without
          // expecting a user message — the resume is a Pause-signal
          // resolution, not chat input — so the session is released.
          $message->setStatus(MessageStatus::COMPLETED)->save();
          if (!$nested) {
            $session->setStatus(SessionStatus::IDLE)->save();
          }
          $this->dispatchTurnCompleted($message, $session, $nested, SessionTurnCompletedEvent::STATUS_PAUSED, $executionId, $pausedReason);
        }
        return;
      }

      // The synchronous pipeline orchestrator surfaces job failures by
      // returning a 'failed' response (no throw). Without an explicit
      // branch here the code below would mark the message COMPLETED and
      // the session IDLE, silently hiding the failure from callers that
      // wrap this in another workflow (WorkflowNode → executeImmediate).
      if ($response->getStatus() === PipelineStatus::FAILED) {
        $reason = $this->extractFailureReason($pipeline) ?? 'Workflow execution failed';
        throw new \RuntimeException($reason);
      }

      // Update message and session status.
      $message->setStatus(MessageStatus::COMPLETED)->save();
      // See note above about $nested — the outer session is still running
      // when this is a sub-invocation; only the top-level call may set IDLE.
      if (!$nested) {
        $session->setStatus(SessionStatus::IDLE)->save();
      }
      $this->dispatchTurnCompleted($message, $session, $nested, SessionTurnCompletedEvent::STATUS_COMPLETED, $executionId);

      $this->logger->info('Message @id processed successfully', [
        '@id' => $message->id(),
      ]);
    }
    catch (\Throwable $e) {
      $message->setStatus(MessageStatus::FAILED)->save();

      if (isset($pipeline, $storage)) {
        try {
          $pipeline->markAsFailed($e->getMessage());
          $storage->save($pipeline);
        }
        catch (\Throwable $cleanupException) {
          $this->logger->warning('Pipeline cleanup failed after orchestration error: @message', [
            '@message' => $cleanupException->getMessage(),
          ]);
        }
      }

      // Top-level only: surface the failure as a chat error and mark the
      // session IDLE before rethrowing. For nested calls the parent's
      // WorkflowNode converts the throw into a sub-workflow failure that
      // its own catch will report — emitting an error bubble here would
      // duplicate the parent's failure path.
      if (!$nested) {
        // Add the error message before setting the session to idle so that a
        // concurrent frontend poll cannot receive the terminal status before
        // the message is persisted and miss the failure notice.
        $this->createErrorMessage($session, $message, $e, $attributedWorkflowId);
        $session->setStatus(SessionStatus::IDLE)->save();
      }
      // pipelineId is NULL when execution failed before the pipeline
      // entity was created.
      $this->dispatchTurnCompleted($message, $session, $nested, SessionTurnCompletedEvent::STATUS_FAILED, $executionId ?? NULL);
      throw $e;
    }
  }

  /**
   * Dispatches the turn-completed event for a top-level execution stop.
   *
   * A turn is the user-message-initiated unit and ends at any pipeline
   * stop; the status says why. Nested sub-workflow invocations are not
   * turns and dispatch nothing.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface $message
   *   The trigger message (its id is the turn id).
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session the turn ran on.
   * @param bool $nested
   *   Whether this is a nested sub-workflow invocation.
   * @param string $status
   *   One of the SessionTurnCompletedEvent::STATUS_* constants.
   * @param string|null $pipelineId
   *   The pipeline entity id, or NULL when execution failed before the
   *   pipeline entity was created.
   * @param string|null $pausedReason
   *   The PausedReason value for STATUS_PAUSED; NULL otherwise.
   */
  protected function dispatchTurnCompleted(
    FlowDropSessionMessageInterface $message,
    FlowDropSessionInterface $session,
    bool $nested,
    string $status,
    ?string $pipelineId,
    ?string $pausedReason = NULL,
  ): void {
    if ($nested) {
      return;
    }
    $assistantMessageIds = $pipelineId !== NULL
      ? $this->sessionService->getAssistantMessageIdsForExecution((string) $session->id(), $pipelineId)
      : [];
    $this->eventDispatcher->dispatch(
      new SessionTurnCompletedEvent(
        sessionId: (string) $session->id(),
        turnId: (string) $message->id(),
        pipelineId: $pipelineId,
        status: $status,
        assistantMessageIds: $assistantMessageIds,
        pausedReason: $pausedReason,
      ),
      SessionTurnCompletedEvent::NAME,
    );
  }

  /**
   * Finds the chat input node and its primary input parameter.
   *
   * Looks for nodes that act as chat inputs (typically nodes with specific
   * types like 'chat_input', 'text_input', etc.) and determines the correct
   * parameter name to use for injecting the session input.
   *
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The workflow entity.
   *
   * @return array{node_id: string, parameter_name: string,
   *   node_type_id: string}|null
   *   Array with node_id, parameter_name, and node_type_id, or NULL if not
   *   found.
   */
  protected function findChatInputNodeWithParameter(WorkflowDefinitionInterface $workflow): ?array {
    $nodes = $workflow->getNodes();

    // Patterns that indicate a chat/text input node.
    $input_patterns = [
      'chat_input',
      'chatinput',
      'text_input',
      'textinput',
      'user_input',
      'message_input',
      'prompt_input',
    ];

    foreach ($nodes as $node) {
      if (!is_array($node) || !isset($node['id'])) {
        continue;
      }

      $node_id = $node['id'];
      $node_type = strtolower((string) ($node['type'] ?? ''));
      $metadata = $node['data']['metadata'] ?? [];
      $node_type_id = NodeAnchor::readMetadata($metadata);
      $node_type_id_lower = strtolower($node_type_id);
      $node_name = strtolower((string) ($metadata['name'] ?? ''));
      $category = strtolower((string) ($metadata['category'] ?? ''));

      $is_input_node = FALSE;

      // Check node type ID first (most reliable - e.g., 'text_input').
      foreach ($input_patterns as $pattern) {
        if (str_contains($node_type_id_lower, $pattern)) {
          $is_input_node = TRUE;
          break;
        }
      }

      // Check node type and name.
      if (!$is_input_node) {
        foreach ($input_patterns as $pattern) {
          if (str_contains($node_type, $pattern) || str_contains($node_name, $pattern)) {
            $is_input_node = TRUE;
            break;
          }
        }
      }

      // Check category for input nodes.
      if (!$is_input_node && ($category === 'inputs' || $category === 'input')) {
        foreach ($input_patterns as $pattern) {
          if (str_contains($node_name, $pattern) || str_contains($node_type_id_lower, $pattern)) {
            $is_input_node = TRUE;
            break;
          }
        }
      }

      if ($is_input_node) {
        // Determine the primary input parameter for this node type.
        $parameter_name = $this->getPrimaryInputParameter($node_type_id, $node);

        $this->logger->debug('Found input node @id (@type_id), using parameter "@param"', [
          '@id' => $node_id,
          '@type_id' => $node_type_id,
          '@param' => $parameter_name,
        ]);

        return [
          'node_id' => $node_id,
          'parameter_name' => $parameter_name,
          'node_type_id' => $node_type_id,
        ];
      }
    }

    $this->logger->warning('No chat/text input node found in workflow @id', [
      '@id' => $workflow->id(),
    ]);

    return NULL;
  }

  /**
   * Gets the primary input parameter name for a node type.
   *
   * This method queries the node type entity to find the first connectable
   * parameter, providing a standardized way to inject initial data.
   *
   * @param string $node_type_id
   *   The node type entity ID.
   * @param array<string, mixed> $node
   *   The workflow node data (for fallback extraction).
   *
   * @return string
   *   The parameter name to use for initial input.
   */
  protected function getPrimaryInputParameter(string $node_type_id, array $node): string {
    // Try to load the node type entity.
    if (!empty($node_type_id)) {
      try {
        $node_type_storage = $this->entityTypeManager->getStorage('flowdrop_node_type');
        $node_type = $node_type_storage->load($node_type_id);

        if ($node_type !== NULL && method_exists($node_type, 'getParameters')) {
          $parameters = $node_type->getParameters();

          // Find the first connectable parameter.
          foreach ($parameters as $param_name => $param_config) {
            if (!empty($param_config['connectable'])) {
              $this->logger->debug('Found connectable parameter "@param" for node type @type', [
                '@param' => $param_name,
                '@type' => $node_type_id,
              ]);
              return $param_name;
            }
          }

          // If no connectable parameter found, look for common input
          // parameter names.
          $common_input_params = ['data', 'value', 'input', 'text', 'message', 'content'];
          foreach ($common_input_params as $param_name) {
            if (isset($parameters[$param_name])) {
              return $param_name;
            }
          }
        }
      }
      catch (\Exception $e) {
        $this->logger->warning('Failed to load node type @type: @message', [
          '@type' => $node_type_id,
          '@message' => $e->getMessage(),
        ]);
      }
    }

    // Fallback: try to extract from node's input schema in metadata.
    $input_schema = $node['data']['metadata']['inputSchema']['properties'] ?? [];
    if (!empty($input_schema)) {
      // Return the first input port name.
      $first_key = array_key_first($input_schema);
      if ($first_key !== NULL && $first_key !== 'trigger') {
        return $first_key;
      }
    }

    // Final fallback: use 'data' as the default parameter name.
    // This matches the TextInput node's expected parameter.
    return 'data';
  }

  /**
   * Builds initial data for workflow execution.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface $message
   *   The user message.
   * @param array<string, mixed>|null $chatInputNode
   *   The ChatInput node or NULL.
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface|null $workflow
   *   The owning workflow, if any.
   *
   * @return array<string, mixed>
   *   The initial data array.
   */
  protected function buildInitialData(
    FlowDropSessionMessageInterface $message,
    ?array $chatInputNode,
    FlowDropSessionInterface $session,
    ?WorkflowDefinitionInterface $workflow = NULL,
  ): array {
    $initialData = [];

    // Get message metadata (may contain additional inputs).
    $metadata = $message->getMessageMetadata();
    $additionalInputs = $metadata['inputs'] ?? [];

    if ($chatInputNode !== NULL) {
      $nodeId = $chatInputNode['id'] ?? NULL;
      if ($nodeId !== NULL) {
        // Build ChatInput node data.
        $nodeData = [
          'message' => $message->getContent(),
        ];

        // Add optional conversation history.
        $history = $this->getMessageHistoryForChatInput($session);
        if (!empty($history)) {
          $nodeData['history'] = $history;
        }

        // Add session and message IDs for tracking.
        // Cast to integers to satisfy strict type validation in Resolver.
        $sessionIdValue = $session->id();
        $messageIdValue = $message->id();
        $nodeData['sessionId'] = $sessionIdValue !== NULL ? (int) $sessionIdValue : NULL;
        $nodeData['messageId'] = $messageIdValue !== NULL ? (int) $messageIdValue : NULL;

        $initialData[$nodeId] = $nodeData;
      }
    }

    // Translate named inputs from the workflow's declared input_ports map
    // into node-keyed initialData. Existing callers passing node-keyed
    // entries pass through unchanged.
    if ($workflow !== NULL && $this->schemaResolver !== NULL && $additionalInputs !== []) {
      $resolved = $this->schemaResolver->resolveInputs($workflow, $additionalInputs);
      foreach ($resolved as $resolvedNodeId => $resolvedPorts) {
        $initialData[$resolvedNodeId] = isset($initialData[$resolvedNodeId])
          ? array_merge($initialData[$resolvedNodeId], $resolvedPorts)
          : $resolvedPorts;
      }
    }
    else {
      // Legacy path: array_merge the additional inputs as-is.
      $initialData = array_merge($initialData, $additionalInputs);
    }

    return $initialData;
  }

  /**
   * Gets orchestrator settings (session overrides workflow).
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The workflow entity.
   *
   * @return array<string, mixed>
   *   The orchestrator settings.
   */
  protected function getOrchestratorSettings(
    FlowDropSessionInterface $session,
    WorkflowDefinitionInterface $workflow,
  ): array {
    // Session settings override workflow defaults.
    $sessionSettings = $session->getOrchestratorSettings();
    if (!empty($sessionSettings)) {
      return $sessionSettings;
    }

    // Fall back to workflow metadata if available.
    $workflowMetadata = $workflow->getMetadata();
    return $workflowMetadata['orchestrator_settings'] ?? [];
  }

  /**
   * Gets the orchestrator instance from settings.
   *
   * @param array<string, mixed> $settings
   *   The orchestrator settings.
   *
   * @return \Drupal\flowdrop_orchestration\Plugin\Orchestrator\OrchestratorPluginInterface
   *   The orchestrator instance.
   */
  protected function getOrchestrator(array $settings): OrchestratorPluginInterface {
    // Use StateGraph as default for session (interactive chat workflows).
    return $this->orchestratorHelper->getInstanceFromSettings(
      $settings,
      OrchestratorHelperInterface::STATEGRAPH
    );
  }

  /**
   * Builds orchestration options from settings.
   *
   * @param array<string, mixed> $orchestratorSettings
   *   The orchestrator settings.
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface $message
   *   The message entity.
   * @param array<int, array{id: string, label: string, icon?: string}> $hierarchy
   *   Resolved workflow → sub-workflow trail to expose to subscribers via
   *   PlaygroundContext. Empty for runs that should not carry a trail.
   *
   * @return array<string, mixed>
   *   The orchestration options.
   */
  protected function buildOrchestrationOptions(
    array $orchestratorSettings,
    FlowDropSessionInterface $session,
    FlowDropSessionMessageInterface $message,
    array $hierarchy = [],
  ): array {
    $options = [];

    // Add StateGraph-specific options if orchestrator is stategraph.
    $orchestratorType = $orchestratorSettings['type'] ?? OrchestratorHelperInterface::STATEGRAPH;
    if ($this->orchestratorHelper->isStateGraph($orchestratorType)) {
      // Workflow-declared settings nest StateGraph values under a
      // `stategraph` key (the plugin's configuration shape); session-level
      // settings use flat keys.
      $stategraphSettings = $orchestratorSettings['stategraph'] ?? [];
      $stategraphSettings = is_array($stategraphSettings) ? $stategraphSettings : [];
      $options['stategraph'] = [
        'thread_id' => "playground_session_{$session->id()}",
        'create_checkpoints' => TRUE,
        'max_iterations' => $stategraphSettings['max_iterations']
        ?? $orchestratorSettings['max_iterations']
        ?? 100,
        // Sessions default to the persistent (entity) checkpointer: a HITL
        // interrupt resumes in a LATER REQUEST, where memory-checkpointer
        // state no longer exists — the resumed graph would restart with an
        // empty state channel (an agent loop then re-runs its tool calls to
        // the iteration cap). Sessions are exactly the resumable use-case,
        // so persistence is the only correct default. Only an explicit
        // session-level setting overrides it; the workflow-declared nested
        // value is deliberately NOT consulted, because the StateGraph
        // plugin form persists its `memory` default into every workflow's
        // metadata — honoring it would silently reintroduce the lost-state
        // bug for any workflow saved through the form.
        'checkpointer_type' => $orchestratorSettings['checkpointer_type'] ?? 'entity',
      ];
      // The common `timeout` setting doubles as the stategraph loop's
      // execution-time budget (ExecutionConfig::maxExecutionTime).
      if (isset($orchestratorSettings['timeout']) && is_numeric($orchestratorSettings['timeout'])) {
        $options['stategraph']['timeout'] = $orchestratorSettings['timeout'];
      }
    }

    // Add session-specific metadata. `hierarchy` lets PlaygroundContext
    // expose the workflow → sub-workflow trail to event subscribers
    // without re-resolving labels.
    $playground = [
      'session_id' => $session->id(),
      'message_id' => $message->id(),
      'sequence_number' => $message->getSequenceNumber(),
    ];
    if ($hierarchy !== []) {
      $playground['hierarchy'] = $hierarchy;
    }
    $options['playground'] = $playground;

    return $options;
  }

  /**
   * Builds the hierarchy trail to stamp on the pipeline's execution context.
   *
   * Reads the parent's hierarchy from `$parentContext['hierarchy']` when
   * present (sub-workflow case) and appends an entry for `$workflow`.
   * `$workflow` is already a loaded entity supplied by the caller, so
   * resolving its label triggers no additional storage access.
   *
   * @param array<string, mixed> $parentContext
   *   Inherited execution context. May carry a `hierarchy` array.
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The workflow about to execute.
   *
   * @return array<int, array{id: string, label: string, icon?: string}>
   *   The combined trail with the current workflow as the last crumb.
   */
  private function buildHierarchy(array $parentContext, WorkflowDefinitionInterface $workflow): array {
    $hierarchy = [];
    $inherited = $parentContext['hierarchy'] ?? [];
    if (is_array($inherited)) {
      foreach ($inherited as $item) {
        if (!is_array($item)) {
          continue;
        }
        $id = $item['id'] ?? NULL;
        $label = $item['label'] ?? NULL;
        if (!is_string($id) || $id === '' || !is_string($label) || $label === '') {
          continue;
        }
        $entry = ['id' => $id, 'label' => $label];
        $icon = $item['icon'] ?? NULL;
        if (is_string($icon) && $icon !== '') {
          $entry['icon'] = $icon;
        }
        $hierarchy[] = $entry;
      }
    }

    $hierarchy[] = [
      'id' => 'wf:' . $workflow->id(),
      'label' => (string) ($workflow->label() ?? $workflow->id()),
    ];

    return $hierarchy;
  }

  /**
   * Converts workflow entity to array format.
   *
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface $workflow
   *   The workflow entity.
   *
   * @return array<string, mixed>
   *   The workflow array.
   */
  protected function workflowToArray(WorkflowDefinitionInterface $workflow): array {
    return [
      'id' => $workflow->id(),
      'label' => $workflow->label(),
      'description' => $workflow->getDescription(),
      'nodes' => $workflow->getNodes(),
      'edges' => $workflow->getEdges(),
      'metadata' => $workflow->getMetadata(),
    ];
  }

  /**
   * Gets message history for ChatInput nodes with caching.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param int $limit
   *   Maximum number of messages to return.
   *
   * @return array<int, array<string, mixed>>
   *   Array of message history items.
   */
  protected function getMessageHistoryForChatInput(
    FlowDropSessionInterface $session,
    int $limit = 10,
  ): array {
    // Check session metadata cache.
    $cache = $session->getMetadata()['message_history'] ?? NULL;
    if ($cache !== NULL &&
        (time() - ($cache['last_updated'] ?? 0)) < 300 &&
        count($cache['messages'] ?? []) >= $limit) {
      return $cache['messages'];
    }

    // Fetch from database.
    $sessionId = $session->id();
    if ($sessionId === NULL) {
      return [];
    }
    $messages = $this->sessionService->getMessages((int) $sessionId, NULL, $limit);
    $history = [];

    foreach ($messages as $message) {
      if (in_array($message->getRole(), ['user', 'assistant'], TRUE)) {
        $history[] = [
          'role' => $message->getRole(),
          'content' => $message->getContent(),
          'timestamp' => $message->getTimestamp(),
        ];
      }
    }

    // Update cache.
    $metadata = $session->getMetadata();
    $metadata['message_history'] = [
      'messages' => $history,
      'last_updated' => time(),
      'message_count' => count($history),
    ];
    $session->setMetadata($metadata)->save();

    return $history;
  }

  /**
   * Extracts message content from node output.
   *
   * Uses a simple, predictable key-based approach:
   * 1. Check standard output keys in priority order (message, text, etc.)
   * 2. Fall back to first non-metadata key in the output
   * 3. JSON encoding as last resort.
   *
   * @param mixed $output
   *   The node output data.
   * @param string|null $node_type_id
   *   Optional node type ID (reserved for future use).
   *
   * @return string|null
   *   The extracted message content, or NULL if not found.
   */
  protected function extractMessageContent(mixed $output, ?string $node_type_id = NULL): ?string {
    if (is_string($output)) {
      return $output;
    }

    if (!is_array($output)) {
      return NULL;
    }

    // Standard output keys in priority order.
    // Use explicit key names for predictable behavior rather than
    // relying on node type configuration order.
    $content_keys = [
    // ChatInput, ChatOutput use this.
      "message",
    // TextInput, TextOutput use this.
      "text",
    // General output key.
      "output",
    // General content key.
      "content",
    // General value key.
      "value",
    // General result key.
      "result",
    // AI/chat responses.
      "response",
    ];

    foreach ($content_keys as $key) {
      if (isset($output[$key])) {
        return $this->valueToString($output[$key]);
      }
    }

    // If no standard key found, try the first non-metadata key.
    foreach ($output as $key => $value) {
      // Skip internal/metadata keys.
      if (str_starts_with($key, "_") || $key === "success" || $key === "error" || $key === "format") {
        continue;
      }
      return $this->valueToString($value);
    }

    // Fallback: encode entire output as JSON.
    $encoded = json_encode($output, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES);
    return $encoded !== FALSE ? $encoded : NULL;
  }

  /**
   * Converts a value to a string representation.
   *
   * @param mixed $value
   *   The value to convert.
   *
   * @return string
   *   The string representation.
   */
  protected function valueToString(mixed $value): string {
    if (is_string($value)) {
      return $value;
    }

    if (is_numeric($value)) {
      return (string) $value;
    }

    if (is_bool($value)) {
      return $value ? 'true' : 'false';
    }

    if (is_array($value)) {
      // If it's a simple array of strings, join them.
      if (array_is_list($value) && !empty($value) && is_string($value[0])) {
        return implode("\n", array_map('strval', $value));
      }
      return json_encode($value, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES);
    }

    return (string) $value;
  }

  /**
   * Gets the primary output parameter name for a node type.
   *
   * @param string $node_type_id
   *   The node type entity ID.
   *
   * @return string|null
   *   The primary output parameter name, or NULL if not found.
   */
  protected function getPrimaryOutputParameter(string $node_type_id): ?string {
    if (empty($node_type_id)) {
      return NULL;
    }

    try {
      $node_type_storage = $this->entityTypeManager->getStorage('flowdrop_node_type');
      $node_type = $node_type_storage->load($node_type_id);

      if ($node_type !== NULL && method_exists($node_type, 'getOutputs')) {
        $outputs = $node_type->getOutputs();

        // First, look for an output marked as primary.
        foreach ($outputs as $output_name => $output_config) {
          if (!empty($output_config['primary'])) {
            return $output_name;
          }
        }

        // Fallback: look for common content output names (message, text, etc.)
        // to provide better defaults for chat/text output nodes.
        $contentOutputNames = ['message', 'text', 'output', 'content', 'result', 'response'];
        foreach ($contentOutputNames as $contentName) {
          if (isset($outputs[$contentName])) {
            $config = $outputs[$contentName];
            if (!isset($config['exposed']) || $config['exposed']) {
              return $contentName;
            }
          }
        }

        // Final fallback: find the first exposed output.
        foreach ($outputs as $output_name => $output_config) {
          if (!isset($output_config['exposed']) || $output_config['exposed']) {
            return $output_name;
          }
        }
      }
    }
    catch (\Exception $e) {
      // Silently fail - we'll use fallback logic.
    }

    return NULL;
  }

  /**
   * Maps a `schema_form` interrupt's schema to the legacy presentation type.
   *
   * The HITL plugins (Confirmation/Choice/TextInput/FormInput) all throw
   * `schema_form` interrupts now and encode their UI shape inside the
   * schema. The frontend dispatcher still discriminates on the older
   * type strings; this method bridges the two.
   *
   * @param array<string, mixed> $schema
   *   The schema payload from the persisted interrupt.
   *
   * @return string
   *   One of `confirmation`, `choice`, `text`, `form`, or `schema_form`
   *   when the shape doesn't match a known presentation.
   */
  private function derivePresentationType(array $schema): string {
    // ConfirmationNode marks itself explicitly.
    if (($schema['presentation'] ?? '') === 'confirmation') {
      return 'confirmation';
    }
    // FormInputNode emits a JSON-schema object.
    if (($schema['type'] ?? '') === 'object') {
      return 'form';
    }
    // ChoiceNode emits enum / items.enum depending on single vs multi.
    if (isset($schema['enum']) && is_array($schema['enum'])) {
      return 'choice';
    }
    if (($schema['type'] ?? '') === 'array' && isset($schema['items']['enum'])) {
      return 'choice';
    }
    // TextInputNode emits a plain string schema.
    if (($schema['type'] ?? '') === 'string') {
      return 'text';
    }
    return 'schema_form';
  }

  /**
   * Extracts a human-readable failure reason from the failed pipeline.
   *
   * The orchestrator stamps the pipeline's error_message and any failed
   * job's error_message before returning a 'failed' response. Prefer the
   * first failed job's message (more specific) and fall back to the
   * pipeline-level message.
   */
  protected function extractFailureReason(mixed $pipeline): ?string {
    if ($pipeline === NULL || !is_object($pipeline)) {
      return NULL;
    }
    if (method_exists($pipeline, 'getJobsByStatus')) {
      foreach ($pipeline->getJobsByStatus(JobStatus::FAILED) as $job) {
        if (method_exists($job, 'getErrorMessage')) {
          $jobError = (string) $job->getErrorMessage();
          if ($jobError !== '') {
            $nodeId = method_exists($job, 'getNodeId') ? (string) $job->getNodeId() : '';
            return $nodeId !== ''
              ? sprintf('Node "%s" failed: %s', $nodeId, $jobError)
              : $jobError;
          }
        }
      }
    }
    if (method_exists($pipeline, 'getErrorMessage')) {
      $pipelineError = (string) $pipeline->getErrorMessage();
      if ($pipelineError !== '') {
        return $pipelineError;
      }
    }
    return NULL;
  }

  /**
   * Creates an error system message for failed execution.
   *
   * The exception message is intentionally not surfaced to the user; details
   * are available in the execution log. The exception class is stored in
   * metadata for diagnostic purposes only.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface $userMessage
   *   The user message that triggered the failed execution.
   * @param \Throwable $exception
   *   The throwable used for error classification metadata.
   * @param string|null $attributedWorkflowId
   *   The owning workflow id, when known, for diagnostic metadata.
   */
  protected function createErrorMessage(
    FlowDropSessionInterface $session,
    FlowDropSessionMessageInterface $userMessage,
    \Throwable $exception,
    ?string $attributedWorkflowId = NULL,
  ): void {
    $metadata = [
      'level' => 'error',
      'error_type' => get_class($exception),
      'user_message_id' => $userMessage->id(),
    ];
    if ($attributedWorkflowId !== NULL) {
      $metadata['workflow_id'] = $attributedWorkflowId;
    }
    $this->sessionService->addSystemMessage($session, 'Execution failed', $metadata);
  }

  /**
   * Handles interrupted workflow execution (human-in-loop).
   *
   * When a workflow is interrupted for user input, this method:
   * - Creates a system message with the interrupt details.
   * - Updates session status to 'awaiting_input'.
   * - Stores the pending interrupt ID in session metadata.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface $message
   *   The user message that triggered execution.
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session entity.
   * @param \Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationResponse $response
   *   The orchestration response.
   */
  protected function handleInterruptedExecution(
    FlowDropSessionMessageInterface $message,
    FlowDropSessionInterface $session,
    OrchestrationResponse $response,
  ): void {
    $metadata = $response->getMetadata();
    $interruptData = $metadata['interrupt'] ?? [];
    $interrupt = $interruptData['interrupt'] ?? [];

    $interruptId = $interrupt['id'] ?? NULL;
    $interruptType = $interrupt['type'] ?? 'unknown';
    $interruptMessage = $interrupt['message'] ?? 'User input required';
    $interruptSchema = $interrupt['schema'] ?? NULL;
    $interruptOptions = $interrupt['options'] ?? NULL;
    $interruptDefaultValue = $interrupt['defaultValue'] ?? NULL;
    $linkedInterruptId = $interrupt['linkedInterruptId'] ?? $interrupt['linked_interrupt_id'] ?? NULL;

    // Skip the interrupt_request message for linked interrupts: they're
    // gated on another interrupt's resolution (e.g. the parent's
    // WorkflowExecution placeholder linked to a sub-pipeline sentinel),
    // not on the user. The sub-workflow's own messages — including its
    // real HITL bubble — already land in the session stream and serve
    // the user-facing purpose. We still need to stamp pause state on the
    // pipeline / session / trigger message so cron, cancel, and a polling
    // frontend see the workflow as awaiting input — otherwise the parent
    // pipeline keeps a stale (resolved) pending_interrupt_id after a
    // cascade-driven re-pause and a fresh frontend poll never notices the
    // new interrupt.
    if (is_string($linkedInterruptId) && $linkedInterruptId !== '') {
      $executionId = $response->getExecutionId();
      if ($executionId !== NULL && $executionId !== '') {
        $pipeline = $this->entityTypeManager
          ->getStorage('flowdrop_pipeline')
          ->load($executionId);
        if ($pipeline instanceof FlowDropPipelineInterface) {
          $pipeline->setPendingInterruptId($interruptId);
          $pipeline->setPendingInterruptType($interruptType);
          $pipeline->save();
        }
      }
      $session->setStatus(SessionStatus::AWAITING_INPUT)->save();
      $message->setStatus(MessageStatus::AWAITING_INPUT)->save();
      return;
    }

    // HITL nodes consolidated their typed interrupt payloads under the single
    // `schema_form` entity type, with the original UI shape encoded inside
    // the schema (e.g. `schema.presentation = 'confirmation'`, `schema.enum`,
    // `schema.type = 'object'`, etc.). The frontend's
    // `buildInterruptConfig()` dispatcher still switches on the *legacy*
    // type strings (confirmation/choice/text/form), so without a translation
    // here the prompt renders blank — no buttons, no labels. Translate back
    // to the frontend's vocabulary based on the schema's shape.
    if ($interruptType === 'schema_form' && is_array($interruptSchema)) {
      $interruptType = $this->derivePresentationType($interruptSchema);
    }

    // Build message metadata with flat keys for the frontend library.
    // The library's extractInterruptMetadata() expects these as top-level
    // fields rather than nested inside schema/options.
    $messageMetadata = [
      'type' => 'interrupt_request',
      'interrupt_id' => $interruptId,
      'interrupt_type' => $interruptType,
      'schema' => $interruptSchema,
      'options' => $interruptOptions,
      'default_value' => $interruptDefaultValue,
      'node_id' => $interrupt['nodeId'] ?? NULL,
      // Surface the originating workflow so UIs can render attribution
      // when an interrupt comes from a sub-workflow embedded in the
      // session stream. Read by InterruptBubble's footer (TS-side
      // mapping handled in metadataToInterrupt).
      'workflow_id' => $interrupt['workflowId'] ?? NULL,
      'execution_id' => $response->getExecutionId(),
      'allow_cancel' => TRUE,
    ];

    // Confirmation labels.
    if ($interruptType === 'confirmation') {
      // New format: labels live on the schema (confirmLabel/declineLabel).
      if (is_array($interruptSchema)) {
        if (isset($interruptSchema['confirmLabel'])) {
          $messageMetadata['confirm_label'] = (string) $interruptSchema['confirmLabel'];
        }
        if (isset($interruptSchema['declineLabel'])) {
          $messageMetadata['cancel_label'] = (string) $interruptSchema['declineLabel'];
        }
      }
      // Legacy format: from options[{value, label}]. Overrides schema-derived
      // values since the explicit options form is the older, more precise API.
      if (is_array($interruptOptions)) {
        foreach ($interruptOptions as $option) {
          if (isset($option['value'], $option['label'])) {
            if ($option['value'] === 'true' || $option['value'] === TRUE) {
              $messageMetadata['confirm_label'] = $option['label'];
            }
            elseif ($option['value'] === 'false' || $option['value'] === FALSE) {
              $messageMetadata['cancel_label'] = $option['label'];
            }
          }
        }
      }
    }
    // Choice options: derive `{value, label, description?}` pairs from the
    // schema. ChoiceNode persists parallel arrays `enum`/`enumLabels`/
    // `enumDescriptions` (single-select on the schema root, multi-select
    // under `items`). Fall back to `label = value` only when labels are
    // missing — e.g. an interrupt created by something other than
    // ChoiceNode that didn't carry labels.
    if ($interruptType === 'choice' && (!is_array($interruptOptions) || $interruptOptions === [])) {
      $enum = NULL;
      $labels = NULL;
      $descriptions = NULL;
      if (is_array($interruptSchema)) {
        if (isset($interruptSchema['enum']) && is_array($interruptSchema['enum'])) {
          $enum = $interruptSchema['enum'];
          $labels = isset($interruptSchema['enumLabels']) && is_array($interruptSchema['enumLabels'])
            ? $interruptSchema['enumLabels']
            : NULL;
          $descriptions = isset($interruptSchema['enumDescriptions']) && is_array($interruptSchema['enumDescriptions'])
            ? $interruptSchema['enumDescriptions']
            : NULL;
        }
        elseif (isset($interruptSchema['items']['enum']) && is_array($interruptSchema['items']['enum'])) {
          $enum = $interruptSchema['items']['enum'];
          $labels = isset($interruptSchema['items']['enumLabels']) && is_array($interruptSchema['items']['enumLabels'])
            ? $interruptSchema['items']['enumLabels']
            : NULL;
          $descriptions = isset($interruptSchema['items']['enumDescriptions']) && is_array($interruptSchema['items']['enumDescriptions'])
            ? $interruptSchema['items']['enumDescriptions']
            : NULL;
        }
      }
      if ($enum !== NULL) {
        $derived = [];
        foreach (array_values($enum) as $i => $value) {
          $entry = [
            'value' => (string) $value,
            'label' => (string) ($labels[$i] ?? $value),
          ];
          if ($descriptions !== NULL && isset($descriptions[$i]) && $descriptions[$i] !== NULL && $descriptions[$i] !== '') {
            $entry['description'] = (string) $descriptions[$i];
          }
          $derived[] = $entry;
        }
        $messageMetadata['options'] = $derived;
      }
      if (is_array($interruptSchema)) {
        if (isset($interruptSchema['multiple'])) {
          $messageMetadata['multiple'] = (bool) $interruptSchema['multiple'];
        }
        if (isset($interruptSchema['minSelections'])) {
          $messageMetadata['min_selections'] = (int) $interruptSchema['minSelections'];
        }
        if (isset($interruptSchema['maxSelections'])) {
          $messageMetadata['max_selections'] = (int) $interruptSchema['maxSelections'];
        }
      }
    }
    if (is_array($interruptSchema)) {
      if (isset($interruptSchema['placeholder'])) {
        $messageMetadata['placeholder'] = $interruptSchema['placeholder'];
      }
      // `format: multiline` is how TextInputNode advertises multiline mode.
      // Surface both that and any explicit `multiline` boolean.
      if (isset($interruptSchema['multiline'])) {
        $messageMetadata['multiline'] = (bool) $interruptSchema['multiline'];
      }
      elseif (($interruptSchema['format'] ?? '') === 'multiline') {
        $messageMetadata['multiline'] = TRUE;
      }
      if (isset($interruptSchema['minLength'])) {
        $messageMetadata['min_length'] = (int) $interruptSchema['minLength'];
      }
      if (isset($interruptSchema['maxLength'])) {
        $messageMetadata['max_length'] = (int) $interruptSchema['maxLength'];
      }
    }

    // Create system message with interrupt info.
    $this->sessionService->addSystemMessage($session, $interruptMessage, $messageMetadata);

    // Update session status to awaiting_input.
    $session->setStatus(SessionStatus::AWAITING_INPUT);
    $session->save();

    // Stamp the pending-interrupt slot on the executing pipeline. Pipelines
    // carry their own pending state now so concurrent sub-pipelines in the
    // same session don't share one slot.
    $executionId = $response->getExecutionId();
    if ($executionId !== NULL && $executionId !== '') {
      $pipelineStorage = $this->entityTypeManager->getStorage('flowdrop_pipeline');
      $pipeline = $pipelineStorage->load($executionId);
      if ($pipeline instanceof FlowDropPipelineInterface) {
        $pipeline->setPendingInterruptId($interruptId);
        $pipeline->setPendingInterruptType($interruptType);
        $pipeline->save();
      }
      else {
        $this->logger->warning('Cannot stamp pending state — pipeline @id not found for session @session', [
          '@id' => $executionId,
          '@session' => $session->id(),
        ]);
      }
    }

    // Update message status to awaiting_input.
    $message->setStatus(MessageStatus::AWAITING_INPUT)->save();

    $this->logger->info('Workflow interrupted for session @session, interrupt @interrupt', [
      '@session' => $session->id(),
      '@interrupt' => $interruptId,
    ]);
  }

  /**
   * Processes execution results and creates assistant messages.
   *
   * This method handles both single-execution and loop iteration results.
   * Results can have keys in two formats:
   * - Simple: "node_id" (e.g., "chat_output.1")
   * - Iteration: "node_id:n" (e.g., "chat_output.1:0", "chat_output.1:1")
   *
   * For loop iterations, a separate message is created for each iteration.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface $userMessage
   *   The user message that triggered execution.
   * @param \Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationResponse $response
   *   The orchestration response.
   * @param string|null $attributedWorkflowId
   *   When non-null, stamped onto assistant/system message metadata so
   *   the UI can render workflow attribution for embedded sub-workflows.
   * @param \Drupal\flowdrop_workflow\WorkflowDefinitionInterface|null $workflow
   *   The executed workflow. When passed, output-node discovery uses
   *   this workflow's nodes — required for shared-session sub-invocations
   *   where the session's primary workflow is the parent's, not the
   *   sub-workflow being processed. Falls back to $session->getWorkflow()
   *   for back-compat with the legacy non-nested call path.
   */
  protected function processExecutionResults(
    FlowDropSessionMessageInterface $userMessage,
    OrchestrationResponse $response,
    ?string $attributedWorkflowId = NULL,
    ?WorkflowDefinitionInterface $workflow = NULL,
  ): void {
    $session = $userMessage->getSession();
    if ($session === NULL) {
      return;
    }

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

    $results = $response->getResults();
    $nodes = $workflow->getNodes();
    $messagesCreated = FALSE;

    // Patterns that indicate a chat/text output node.
    $outputPatterns = [
      'chat_output',
      'chatoutput',
      'text_output',
      'textoutput',
      'message_output',
      'chat_message',
      'response_output',
    ];

    // Build a map of output node IDs and their types.
    $outputNodes = [];
    foreach ($nodes as $node) {
      if (!is_array($node) || !isset($node['id'])) {
        continue;
      }

      $nodeId = $node['id'];
      $metadata = $node['data']['metadata'] ?? [];
      $nodeTypeId = strtolower(NodeAnchor::readMetadata($metadata));

      // Check if this is an output node.
      foreach ($outputPatterns as $pattern) {
        if (str_contains($nodeTypeId, $pattern)) {
          $outputNodes[$nodeId] = $nodeTypeId;
          break;
        }
      }
    }

    // Process results - handle both simple keys and iteration keys.
    // Results are sorted to ensure iteration order is preserved.
    $resultKeys = array_keys($results);
    sort($resultKeys);

    foreach ($resultKeys as $resultKey) {
      // Parse the result key to extract node ID and optional iteration.
      $parsedKey = $this->parseResultKey($resultKey);
      $nodeId = $parsedKey['nodeId'];
      $iteration = $parsedKey['iteration'];

      // Check if this result is for an output node.
      if (!isset($outputNodes[$nodeId])) {
        continue;
      }

      $nodeTypeId = $outputNodes[$nodeId];
      $output = $results[$resultKey];
      $content = $this->extractMessageContent($output, $nodeTypeId);

      if ($content !== NULL && $content !== '') {
        // Sequence number is now adopted from the entity id by
        // SessionService::createMessage(), so no pre-allocation here.
        $messageMetadata = [
          'source' => 'workflow_execution',
          'node_type' => $nodeTypeId,
          'execution_id' => $userMessage->getExecutionId(),
          'result_key' => $resultKey,
        ];

        if ($iteration !== NULL) {
          $messageMetadata['loop_iteration'] = $iteration;
        }

        if ($attributedWorkflowId !== NULL) {
          $messageMetadata['workflow_id'] = $attributedWorkflowId;
        }

        $assistantMessage = $this->sessionService->addAssistantMessage(
          $session,
          $content,
          $nodeId,
          $messageMetadata
        );

        // Link assistant message to user message.
        $userMessageId = $userMessage->id();
        if ($userMessageId !== NULL) {
          $assistantMessage->setParentMessageId((int) $userMessageId);
        }
        $assistantMessage->setExecutionId($userMessage->getExecutionId());
        $assistantMessage->setStatus(MessageStatus::COMPLETED);
        $assistantMessage->save();

        $messagesCreated = TRUE;
      }
    }

    // If no output messages were created and the run actually finished
    // (not paused for HITL, not failed), surface a "no output" notice.
    // Skipping interrupted/failed avoids a misleading "completed" line
    // landing right before the interrupt UI on every HITL pause.
    if (!$messagesCreated && empty($results)) {
      $status = $response->getStatus();
      if ($status !== WorkflowSnapshot::STATUS_INTERRUPTED && $status !== PipelineStatus::FAILED) {
        $noOutputMetadata = ['level' => 'info'];
        if ($attributedWorkflowId !== NULL) {
          $noOutputMetadata['workflow_id'] = $attributedWorkflowId;
        }
        $this->sessionService->addSystemMessage(
          $session,
          'Workflow execution completed with no output.',
          $noOutputMetadata,
        );
      }
    }
  }

  /**
   * Parses a result key to extract node ID and optional iteration number.
   *
   * Result keys can be in two formats:
   * - Simple: "node_id" (e.g., "chat_output.1")
   * - Iteration: "node_id:iteration" (e.g., "chat_output.1:0")
   *
   * @param string $resultKey
   *   The result key to parse.
   *
   * @return array{nodeId: string, iteration: int|null}
   *   Array with 'nodeId' and 'iteration' (null if no iteration suffix).
   */
  protected function parseResultKey(string $resultKey): array {
    // Check for iteration suffix (colon separator).
    if (str_contains($resultKey, ':')) {
      $parts = explode(':', $resultKey, 2);
      $nodeId = $parts[0];
      $iteration = is_numeric($parts[1]) ? (int) $parts[1] : NULL;

      return [
        'nodeId' => $nodeId,
        'iteration' => $iteration,
      ];
    }

    // No iteration suffix - simple node ID.
    return [
      'nodeId' => $resultKey,
      'iteration' => NULL,
    ];
  }

}
