<?php

declare(strict_types=1);

namespace Drupal\flowdrop_runtime\Service\Runtime;

use Drupal\flowdrop\Constants\ReservedName;
use Drupal\flowdrop\DTO\Output;
use Drupal\flowdrop\DTO\ValidationResult;
use Drupal\flowdrop\Utility\PortExposure;
use Drupal\flowdrop\Exception\RetryableExceptionInterface;
use Drupal\flowdrop\Exception\WorkflowStopException;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\flowdrop\DTO\ExecutionContextDTO;
use Drupal\flowdrop\DTO\ParameterBagInterface;
use Drupal\flowdrop\DTO\Tool\ToolCollectionInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\ExecutionContextAwareInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\ResumableNodeProcessorInterface;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\ToolsAwareInterface;
use Drupal\flowdrop\Service\FlowDropNodeProcessorPluginManager;
use Drupal\flowdrop_orchestration\DTO\ResolvedInterruptInterface;
use Drupal\flowdrop_node_type\FlowDropNodeTypeInterface;
use Drupal\flowdrop_runtime\Constants\ExecutionStatus;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionResult;
use Drupal\flowdrop_orchestration\Exception\InterruptExceptionInterface;
use Drupal\flowdrop_runtime\Exception\NonSerializableOutputException;
use Drupal\flowdrop_runtime\Exception\RuntimeException;
use Drupal\flowdrop_runtime\Service\Logging\ExecutionLoggerInterface;
use Drupal\flowdrop_runtime\Event\JobInterruptCaughtEvent;
use Drupal\flowdrop_runtime\Event\NodeCompletedEvent;
use Drupal\flowdrop_runtime\Service\ParameterResolverInterface;
use Drupal\flowdrop_runtime\Service\RealTime\RealTimeManagerInterface;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

/**
 * Core node runtime service with Unified Parameter System support.
 *
 * This service executes nodes using the Unified Parameter System:
 * 1. Loads the node type entity and its parameter configuration
 * 2. Gets the plugin's parameter schema
 * 3. Uses ParameterResolver to resolve parameters from multiple sources
 * 4. Executes the plugin with resolved parameters.
 *
 * All routine logging is handled via ExecutionLoggerInterface, which
 * provides centralized, verbosity-controlled logging.
 *
 * @internal This class is not part of the public API and may change without notice.
 *
 * @see docs/development/unified-parameter-system-spec.md
 * @see \Drupal\flowdrop_runtime\Service\Logging\ExecutionLoggerInterface
 */
class NodeRuntimeService implements NodeRuntimeServiceInterface {

  /**
   * Constructs a NodeRuntimeService.
   *
   * @param \Drupal\flowdrop_runtime\Service\Logging\ExecutionLoggerInterface $executionLogger
   *   The centralized execution logger.
   * @param \Symfony\Component\EventDispatcher\EventDispatcherInterface $eventDispatcher
   *   The event dispatcher.
   * @param \Drupal\flowdrop\Service\FlowDropNodeProcessorPluginManager $nodeProcessorManager
   *   The node processor plugin manager.
   * @param \Drupal\flowdrop_runtime\Service\RealTime\RealTimeManagerInterface $realTimeManager
   *   The real-time manager.
   * @param \Drupal\flowdrop_runtime\Service\ParameterResolverInterface $parameterResolver
   *   The parameter resolver.
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   The entity type manager.
   */
  public function __construct(
    private readonly ExecutionLoggerInterface $executionLogger,
    private readonly EventDispatcherInterface $eventDispatcher,
    private readonly FlowDropNodeProcessorPluginManager $nodeProcessorManager,
    private readonly RealTimeManagerInterface $realTimeManager,
    private readonly ParameterResolverInterface $parameterResolver,
    private readonly EntityTypeManagerInterface $entityTypeManager,
  ) {}

  /**
   * Execute a single node with real-time updates.
   *
   * This method uses the Unified Parameter System to resolve parameters
   * from runtime inputs, workflow values, and defaults before execution.
   *
   * @param string $executionId
   *   The execution ID.
   * @param string $nodeId
   *   The node instance ID.
   * @param string $nodeType
   *   The node type ID (FlowDropNodeType entity ID).
   * @param array<string, mixed> $runtimeInputs
   *   Runtime inputs from upstream connections.
   * @param array<string, mixed> $workflowValues
   *   Workflow-level configuration values.
   * @param \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext $context
   *   The execution context.
   * @param \Drupal\flowdrop\DTO\Tool\ToolCollectionInterface|null $tools
   *   Tools wired to this node, injected when the processor is ToolsAware.
   * @param callable|null $configureProcessor
   *   Optional hook called with the processor instance after the built-in
   *   capability injection, before parameter resolution. Lets orchestrators
   *   inject orchestrator-specific capabilities without the runtime
   *   depending on those modules.
   *
   * @return \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionResult
   *   The execution result.
   *
   * @throws \Drupal\flowdrop_runtime\Exception\RuntimeException
   *   When node execution fails.
   */
  public function executeNode(
    string $executionId,
    string $nodeId,
    string $nodeType,
    array $runtimeInputs,
    array $workflowValues,
    NodeExecutionContext $context,
    ?ToolCollectionInterface $tools = NULL,
    ?callable $configureProcessor = NULL,
  ): NodeExecutionResult {
    // Log node start via centralized logger.
    $this->executionLogger->logNodeStart($executionId, $nodeId, $nodeType);

    // Update node status to running.
    $this->realTimeManager->updateNodeStatus($executionId, $nodeId, ExecutionStatus::RUNNING, [
      "node_type" => $nodeType,
      "start_time" => time(),
    ]);

    $startTime = microtime(TRUE);

    try {
      // Load the node type entity.
      $nodeTypeEntity = $this->loadNodeTypeEntity($nodeType);
      $executorPlugin = $nodeTypeEntity->getExecutorPlugin();

      // Get the node processor plugin.
      $processor = $this->nodeProcessorManager->createInstance($executorPlugin);

      // Inject execution context for processors that need it.
      // This provides type-safe access to workflow data, IDs, and metadata.
      if ($processor instanceof ExecutionContextAwareInterface) {
        $executionContext = $this->buildExecutionContext($executionId, $nodeId, $context);
        $processor->setExecutionContext($executionContext);
      }

      // Inject connected tools for processors that consume them.
      if ($tools !== NULL && $processor instanceof ToolsAwareInterface) {
        $processor->setTools($tools);
      }

      // Let the caller inject orchestrator-specific capabilities (e.g.
      // StateGraph state) the runtime cannot know about.
      if ($configureProcessor !== NULL) {
        $configureProcessor($processor);
      }

      // Merge per-node initialData into runtime inputs.
      // If initialData contains an entry keyed by this node's ID, merge it
      // as a fallback — runtime inputs from edges take precedence.
      $allInitialData = $context->getInitialData();
      if (isset($allInitialData[$nodeId]) && is_array($allInitialData[$nodeId])) {
        $runtimeInputs = array_merge($allInitialData[$nodeId], $runtimeInputs);
      }

      // Resolve parameters using the Unified Parameter System.
      $parameterBag = $this->resolveParameters(
        $processor,
        $nodeTypeEntity,
        $runtimeInputs,
        $workflowValues
      );

      // Log parameters at appropriate verbosity level.
      $this->executionLogger->logParameters($executionId, $nodeId, $parameterBag);

      // Execute using the unified parameter execution path.
      try {
        $contextMetadata = $context->getMetadata();
        $executingJobId = isset($contextMetadata['job_id'])
          ? (int) $contextMetadata['job_id']
          : NULL;
        $resumeInterrupt = $this->resolveResumeInterrupt($processor, $parameterBag, $executingJobId);
        $result = $resumeInterrupt !== NULL
          ? $processor->resume($resumeInterrupt, $parameterBag)
          : $processor->process($parameterBag);

        // Drop the values of any output ports the author left not-exposed
        // before anything else observes them. In v2 exposure is semantic: a
        // not-exposed output port's data must not leave the node. Filtering at
        // this production point keeps it out of the unified output below, job
        // records, checkpoints, real-time updates, and any tool result handed
        // back to a model — it is never serialized or stored anywhere.
        $result = $this->filterUnexposedOutputs($result, $processor, $workflowValues);

        // Compose unified "output" port (reserved name) if configured.
        // This provides all exposed outputs as a single JSON object.
        if ($nodeTypeEntity->isUnifiedOutputExposed()) {
          $result["output"] = $this->composeUnifiedOutput($result, $nodeTypeEntity);
        }

        // Reject output json_encode cannot faithfully persist before it is
        // wrapped. Every engine stores node output through json_encode (job
        // records, StateGraph checkpoints), which silently emits {} for a
        // value it cannot represent — an entity, Url, Markup, closure or
        // resource — so the loss would otherwise surface as empty data at a
        // downstream node, or not at all on the in-memory synchronous path.
        // The thrown NonSerializableOutputException is a \RuntimeException, so
        // the inner catch below converts it to a non-retryable error Output:
        // it is not RetryableExceptionInterface (retrying a deterministic
        // output bug is pointless), so it takes the node's normal failure path
        // — error-edge routing if the author declared one, else failing the
        // run. Runs after filterUnexposedOutputs so a dropped, never-persisted
        // value is not policed.
        $this->assertSerializableOutput($result, $nodeId, $executorPlugin);

        $output = new Output();
        // Convert the result to output format first: fromArray() replaces the
        // internal data wholesale, so the status must be stamped afterwards.
        $output->fromArray($result);
        $output->setSuccess();
      }
      catch (\RuntimeException $exception) {
        // A node processor failure surfaces as an error-status Output rather
        // than propagating; the orchestrator turns that into a failed job.
        // The exception type itself does not survive past this point, so
        // preserve whether the failure is retryable (the Phase 1 marker) as
        // output metadata for the orchestrator's retry policy to act on.
        $output = new Output();
        $output->setError($exception->getMessage());
        if ($exception instanceof RetryableExceptionInterface) {
          $output->addMetadata('error_retryable', TRUE);
        }
      }

      $executionTime = microtime(TRUE) - $startTime;

      // Create execution result.
      $result = new NodeExecutionResult(
        nodeId: $nodeId,
        output: $output,
        status: $output->getStatus(),
        executionTime: $executionTime,
        nodeType: $nodeType,
        timestamp: time(),
        context: $context
      );

      // Update node status to complete.
      $this->realTimeManager->updateNodeStatus($executionId, $nodeId, ExecutionStatus::COMPLETED, [
        "execution_time" => $executionTime,
        "output_size" => strlen(serialize($output->toArray())),
        "end_time" => time(),
      ]);

      // Dispatch node execution completed event.
      $event = new NodeCompletedEvent(
        result: $result,
        executionId: $executionId,
        nodeId: $nodeId,
        executionTime: $executionTime,
        timestamp: time(),
      );
      $this->eventDispatcher->dispatch($event, NodeCompletedEvent::NAME);

      // Log completion via centralized logger.
      $this->executionLogger->logNodeComplete($executionId, $result);

      return $result;
    }
    catch (InterruptExceptionInterface $e) {
      // Bridge to flowdrop_interrupt without a hard module dependency. The
      // flowdrop_interrupt subscriber persists descriptor-shape exceptions,
      // marks the job interrupted directly, and attaches the resulting
      // entity to the event. The exception itself is immutable — read the
      // entity from the event, not from the exception.
      $metadata = $context->getMetadata();
      $sessionId = isset($metadata['session_id']) ? (int) $metadata['session_id'] : NULL;
      $jobId = isset($metadata['job_id']) ? (int) $metadata['job_id'] : NULL;
      $pipelineIdRaw = $context->getPipelineId();
      $event = new JobInterruptCaughtEvent(
        exception: $e,
        workflowId: $context->getWorkflowId(),
        pipelineId: $pipelineIdRaw !== '' ? $pipelineIdRaw : NULL,
        nodeId: $nodeId,
        jobId: $jobId,
        sessionId: $sessionId,
      );
      $this->eventDispatcher->dispatch($event, JobInterruptCaughtEvent::NAME);

      // Read the (possibly newly-persisted) entity from the event for
      // real-time status updates. For legacy-shape exceptions the
      // subscriber attaches the pre-existing entity; for descriptor-shape
      // it attaches the just-persisted one.
      $interrupt = $event->getInterrupt();
      $interruptId = $interrupt?->uuid() ?? '';
      $interruptType = $interrupt?->getType()->value ?? '';
      $executionTime = microtime(TRUE) - $startTime;
      $this->realTimeManager->updateNodeStatus($executionId, $nodeId, ExecutionStatus::INTERRUPTED, [
        "interrupt_id" => $interruptId,
        "interrupt_type" => $interruptType,
        "execution_time" => $executionTime,
        "end_time" => time(),
      ]);
      throw $e;
    }
    catch (WorkflowStopException $e) {
      // A terminal node signals a clean stop, not a failure. Extends
      // \Exception (not \RuntimeException), so it skipped the inner
      // error-routing catch; re-throw it unchanged here — before the generic
      // \Exception catch below, which would otherwise re-wrap it into a
      // RuntimeException and lose the stop signal — so the orchestrator can map
      // it to a completed run. The node itself completed its intent.
      $executionTime = microtime(TRUE) - $startTime;
      $this->realTimeManager->updateNodeStatus($executionId, $nodeId, ExecutionStatus::COMPLETED, [
        "execution_time" => $executionTime,
        "end_time" => time(),
      ]);
      throw $e;
    }
    catch (\Exception $e) {
      $executionTime = microtime(TRUE) - $startTime;

      // Update node status to failed.
      $this->realTimeManager->updateNodeStatus($executionId, $nodeId, ExecutionStatus::FAILED, [
        "error" => $e->getMessage(),
        "execution_time" => $executionTime,
        "end_time" => time(),
      ]);

      // Log failure via centralized logger.
      $this->executionLogger->logNodeFailure($executionId, $nodeId, $e, $executionTime);

      throw new RuntimeException(
        "Node execution failed for {$nodeId}: " . $e->getMessage(),
        0,
        $e
      );
    }
  }

  /**
   * Asserts a node's output contains only json_encode-safe values.
   *
   * The three execution engines gather inputs differently but all converge on
   * this service and persist output through json_encode (job records,
   * StateGraph checkpoints). json_encode does not throw on an object it cannot
   * represent — it silently emits an empty object — so a stray entity, Url,
   * Markup, closure or resource in the output round-trips to nothing and the
   * loss only shows up as empty data at a downstream node, or not at all in
   * the in-memory synchronous path. Validating the composed output here, at
   * the single waist every engine shares, turns that silent corruption into a
   * loud, node-scoped failure.
   *
   * Called from within executeNode()'s processor try block, so the thrown
   * NonSerializableOutputException — a \RuntimeException — is caught by the
   * inner handler and converted to a non-retryable error Output (it is not
   * RetryableExceptionInterface). The node then follows its normal failure
   * path: error-edge routing where declared, else failing the run.
   *
   * We check against json_encode's own contract rather than the Symfony/Drupal
   * serializer: an entity normalises fine through the serializer but this
   * pipeline stores raw json_encode output, where it does not. Allowed leaves
   * are therefore scalars, null, arrays, stdClass, and \JsonSerializable — the
   * interface json_encode honours.
   *
   * @param array $output
   *   The composed node output, before it is wrapped in an Output DTO.
   * @param string $nodeId
   *   The node id, for the error message.
   * @param string $executorPlugin
   *   The executor plugin id, for the error message.
   *
   * @throws \Drupal\flowdrop_runtime\Exception\NonSerializableOutputException
   *   When a value json_encode cannot faithfully persist is found.
   */
  protected function assertSerializableOutput(array $output, string $nodeId, string $executorPlugin): void {
    $this->assertSerializableValue($output, $nodeId, $executorPlugin, 'output');
  }

  /**
   * Recursively validates a single output value for json_encode safety.
   *
   * @param mixed $value
   *   The value to validate.
   * @param string $nodeId
   *   The node id, for the error message.
   * @param string $executorPlugin
   *   The executor plugin id, for the error message.
   * @param string $path
   *   Dotted path to $value within the output, for the error message.
   *
   * @throws \Drupal\flowdrop_runtime\Exception\NonSerializableOutputException
   *   When $value (or a nested value) cannot be faithfully json_encoded.
   */
  private function assertSerializableValue(mixed $value, string $nodeId, string $executorPlugin, string $path): void {
    if ($value === NULL || is_scalar($value)) {
      return;
    }
    if (is_array($value)) {
      foreach ($value as $key => $item) {
        $this->assertSerializableValue($item, $nodeId, $executorPlugin, $path . '.' . $key);
      }
      return;
    }
    // stdClass round-trips through json_encode, but its properties may not —
    // recurse so a nested entity is still caught.
    if ($value instanceof \stdClass) {
      foreach (get_object_vars($value) as $key => $item) {
        $this->assertSerializableValue($item, $nodeId, $executorPlugin, $path . '.' . $key);
      }
      return;
    }
    // Objects that explicitly opt into json_encode are safe by contract.
    if ($value instanceof \JsonSerializable) {
      return;
    }
    $type = is_object($value) ? get_class($value) : gettype($value);
    throw new NonSerializableOutputException(sprintf(
      'Node "%s" (plugin "%s") returned a non-serializable value at "%s": %s. Node output is persisted via json_encode across all execution engines and must contain only JSON-safe values — reduce entities to a reference such as {entity_type, id} and cast Markup/Url objects to strings before returning them.',
      $nodeId,
      $executorPlugin,
      $path,
      $type,
    ));
  }

  /**
   * Resolves the interrupt entity for a resume invocation.
   *
   * Detects four conditions and returns a ResolvedInterruptInterface only
   * when all four hold:
   *
   *  1. The parameter bag carries a `__interrupt_id__` (orchestrator
   *     injected this for jobs whose metadata.interrupt_id is set).
   *  2. The processor opts into the resume hook by implementing
   *     {@see ResumableNodeProcessorInterface}.
   *  3. The pointed-to interrupt has reached a terminal state — signalled
   *     by `getResolvedAt() !== NULL`. Pending interrupts fall through to
   *     `process()` (typically a race condition during pipeline resume).
   *  4. The interrupt is OWNED by the executing job: its `getJobId()`
   *     matches $executingJobId. A resolution is the answer to one specific
   *     question asked by one specific job; job metadata can carry a foreign
   *     interrupt UUID (loop-iteration clones, retries, copies), and
   *     honoring it would replay another execution's answer — for HITL
   *     gates, a silent auto-approve. Strict on missing data: if either id
   *     is NULL the interrupt is NOT honored, because falling through to
   *     `process()` re-asks the question, which is always the safe
   *     direction for a consent gate.
   *
   * Returns NULL otherwise — the caller invokes `process()` as before.
   * The runtime intentionally does not import flowdrop_interrupt types;
   * the load goes through entity_type.manager with the entity type ID as
   * a string and the result is narrowed by the orchestration-layer
   * interface.
   */
  protected function resolveResumeInterrupt(
    object $processor,
    ParameterBagInterface $parameterBag,
    ?int $executingJobId,
  ): ?ResolvedInterruptInterface {
    if (!$processor instanceof ResumableNodeProcessorInterface) {
      return NULL;
    }
    $interruptId = $parameterBag->get('__interrupt_id__');
    if ($interruptId === NULL || $interruptId === '') {
      return NULL;
    }
    if (!$this->entityTypeManager->hasDefinition('flowdrop_interrupt')) {
      return NULL;
    }
    $entities = $this->entityTypeManager
      ->getStorage('flowdrop_interrupt')
      ->loadByProperties(['uuid' => (string) $interruptId]);
    $entity = reset($entities) ?: NULL;
    if (!$entity instanceof ResolvedInterruptInterface) {
      return NULL;
    }
    // Pending interrupts: keep the job interrupted; do NOT call resume().
    if ($entity->getResolvedAt() === NULL) {
      return NULL;
    }
    // Ownership gate: only the job that raised the interrupt may consume
    // its resolution. Foreign or unverifiable interrupts fall through to
    // process(), which re-asks — the safe direction for a consent gate.
    $ownerJobId = $entity->getJobId();
    if ($executingJobId === NULL || $ownerJobId === NULL || $ownerJobId !== $executingJobId) {
      return NULL;
    }
    return $entity;
  }

  /**
   * Resolve parameters using the Unified Parameter System.
   *
   * @param object $processor
   *   The node processor plugin instance.
   * @param \Drupal\flowdrop_node_type\FlowDropNodeTypeInterface $nodeTypeEntity
   *   The node type entity with parameter configuration.
   * @param array<string, mixed> $runtimeInputs
   *   Runtime inputs from upstream connections.
   * @param array<string, mixed> $workflowValues
   *   Workflow-level configuration values.
   *
   * @return \Drupal\flowdrop\DTO\ParameterBagInterface
   *   The resolved parameters.
   */
  protected function resolveParameters(
    object $processor,
    FlowDropNodeTypeInterface $nodeTypeEntity,
    array $runtimeInputs,
    array $workflowValues,
  ): ParameterBagInterface {
    // Get the plugin's parameter schema.
    $parameterSchema = $processor->getParameterSchema();

    // Get the entity's parameter configuration (overrides).
    $entityParameters = $nodeTypeEntity->getParameters();

    // Use ParameterResolver to resolve values from all sources.
    return $this->parameterResolver->resolve(
      $parameterSchema,
      $entityParameters,
      $workflowValues,
      $runtimeInputs
    );
  }

  /**
   * Load a node type entity.
   *
   * @param string $nodeType
   *   The node type ID.
   *
   * @return \Drupal\flowdrop_node_type\FlowDropNodeTypeInterface
   *   The node type entity.
   *
   * @throws \Drupal\flowdrop_runtime\Exception\RuntimeException
   *   When node type is not found.
   */
  protected function loadNodeTypeEntity(string $nodeType): FlowDropNodeTypeInterface {
    $entity = $this->entityTypeManager
      ->getStorage("flowdrop_node_type")
      ->load($nodeType);

    if (!$entity instanceof FlowDropNodeTypeInterface) {
      throw new RuntimeException("Node type '{$nodeType}' not found");
    }

    return $entity;
  }

  /**
   * Execute multiple nodes in sequence with real-time updates.
   *
   * @param string $executionId
   *   The execution ID.
   * @param array<array<string, mixed>> $nodes
   *   Array of node definitions with 'id', 'type', 'inputs', and 'config'.
   * @param \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext $context
   *   The execution context.
   *
   * @return array<string, \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionResult>
   *   Array of execution results keyed by node ID.
   */
  public function executeNodes(
    string $executionId,
    array $nodes,
    NodeExecutionContext $context,
  ): array {
    $results = [];

    foreach ($nodes as $node) {
      $result = $this->executeNode(
        $executionId,
        $node["id"],
        $node["type"],
        $node["inputs"] ?? [],
        $node["config"] ?? [],
        $context
      );

      $results[$node["id"]] = $result;

      // Update context with node output for next nodes.
      $context->addNodeOutput($node["id"], $result->getOutput());
    }

    return $results;
  }

  /**
   * Validate node before execution.
   *
   * @param string $executionId
   *   The execution ID for logging.
   * @param string $nodeId
   *   The node instance ID for logging.
   * @param string $nodeType
   *   The node type ID.
   * @param array<string, mixed> $params
   *   The parameters to validate.
   *
   * @return \Drupal\flowdrop\DTO\ValidationResult
   *   The validation result.
   */
  public function validateNode(
    string $executionId,
    string $nodeId,
    string $nodeType,
    array $params,
  ): ValidationResult {
    try {
      $nodeTypeEntity = $this->loadNodeTypeEntity($nodeType);
      $processor = $this->nodeProcessorManager->createInstance(
        $nodeTypeEntity->getExecutorPlugin()
      );
      $result = $processor->validateParams($params);

      // Log validation result via centralized logger.
      $this->executionLogger->logValidation($executionId, $nodeId, $result);

      return $result;
    }
    catch (\Exception $e) {
      // Return validation failure for load errors.
      return ValidationResult::error(
        "node_type",
        "Failed to load node type '{$nodeType}': " . $e->getMessage(),
        "NODE_TYPE_LOAD_ERROR"
      );
    }
  }

  /**
   * Get node processor information.
   *
   * @param string $nodeType
   *   The node type ID.
   *
   * @return array<string, mixed>
   *   Node processor information.
   */
  public function getNodeProcessorInfo(string $nodeType): array {
    try {
      $nodeTypeEntity = $this->loadNodeTypeEntity($nodeType);
      $processor = $this->nodeProcessorManager->createInstance(
        $nodeTypeEntity->getExecutorPlugin()
      );

      return [
        "id" => $processor->getPluginId(),
        "name" => $processor->getName(),
        "description" => $processor->getDescription(),
        "category" => $nodeTypeEntity->getCategory(),
        "tags" => $nodeTypeEntity->getTags(),
        "version" => $processor->getVersion(),
        "parameter_schema" => $processor->getParameterSchema(),
        "output_schema" => $processor->getOutputSchema(),
      ];
    }
    catch (\Exception $e) {
      return [];
    }
  }

  /**
   * Builds an ExecutionContextDTO from NodeExecutionContext.
   *
   * Context is now always passed via NodeExecutionContext, which is
   * injected into processors implementing ExecutionContextAwareInterface.
   *
   * @param string $executionId
   *   The execution ID.
   * @param string $nodeId
   *   The node instance ID.
   * @param \Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext $context
   *   The node execution context.
   *
   * @return \Drupal\flowdrop\DTO\ExecutionContextDTO
   *   The execution context DTO.
   */
  protected function buildExecutionContext(
    string $executionId,
    string $nodeId,
    NodeExecutionContext $context,
  ): ExecutionContextDTO {
    return new ExecutionContextDTO(
      initialData: $context->getInitialData(),
      workflowId: $context->getWorkflowId(),
      pipelineId: $context->getPipelineId(),
      executionId: $executionId,
      nodeId: $nodeId,
      metadata: $context->getMetadata(),
    );
  }

  /**
   * Get the execution logger.
   *
   * Allows external code to access the logger for workflow-level logging.
   *
   * @return \Drupal\flowdrop_runtime\Service\Logging\ExecutionLoggerInterface
   *   The execution logger.
   */
  public function getExecutionLogger(): ExecutionLoggerInterface {
    return $this->executionLogger;
  }

  /**
   * Strips the values of output ports the author left not-exposed.
   *
   * Exposure is semantic, not cosmetic: a not-exposed port is hidden, not
   * wireable, and the engine must not surface its data. For an
   * output port that means its value must not leave the node — it cannot be
   * read by a downstream edge, and for a tool-exposed node it must never be
   * handed back to the model (which would leak data the author deliberately
   * withheld, e.g. PII). This runs at the production point, before the Output
   * is built, so a not-exposed output is never serialized into the unified
   * "output" port, job records, checkpoints, real-time updates, or a tool
   * result.
   *
   * Effective exposure is read through {@see PortExposure}, mirroring the input
   * side in {@see \Drupal\flowdrop_runtime\Service\ParameterResolver}: the
   * instance's `config.ports` override (carried in $workflowValues) else the
   * port's metadata `x-exposed-by-default` from the processor's output schema.
   * Only declared output ports are considered (deny-list semantics); keys the
   * schema does not declare — dynamic outputs, reserved control ports — pass
   * through untouched. Output ports are exposed by default, so an untouched
   * node strips nothing.
   *
   * @param array<string, mixed> $result
   *   The processor's raw output (port id => value).
   * @param object $processor
   *   The node processor plugin instance.
   * @param array<string, mixed> $workflowValues
   *   The node instance config; the reserved `ports` key carries per-port
   *   exposure overrides.
   *
   * @return array<string, mixed>
   *   The output with the values of not-exposed declared ports removed.
   */
  protected function filterUnexposedOutputs(array $result, object $processor, array $workflowValues): array {
    if ($result === []) {
      return $result;
    }

    $outputSchema = $processor->getOutputSchema();
    $properties = is_array($outputSchema["properties"] ?? NULL) ? $outputSchema["properties"] : [];
    if ($properties === []) {
      return $result;
    }

    $configPorts = is_array($workflowValues[ReservedName::PORTS] ?? NULL)
      ? $workflowValues[ReservedName::PORTS]
      : [];

    foreach ($properties as $portId => $portSchema) {
      $portId = (string) $portId;
      if (!array_key_exists($portId, $result)) {
        continue;
      }
      $exposedByDefault = PortExposure::isExposedByDefault(is_array($portSchema) ? $portSchema : []);
      if (!PortExposure::isExposed($configPorts, ReservedName::PORT_OUTPUT, $portId, $exposedByDefault)) {
        unset($result[$portId]);
      }
    }

    return $result;
  }

  /**
   * Compose unified output from exposed outputs.
   *
   * Creates a single JSON object containing all output values that are
   * configured as exposed (exposed: true) in the node type entity.
   * Internal outputs (starting with _) and trigger are excluded.
   *
   * @param array<string, mixed> $outputData
   *   Raw output data from the plugin's process() method.
   * @param \Drupal\flowdrop_node_type\FlowDropNodeTypeInterface $nodeTypeEntity
   *   The node type entity with output configuration.
   *
   * @return array<string, mixed>
   *   Filtered output containing only exposed keys.
   */
  protected function composeUnifiedOutput(
    array $outputData,
    FlowDropNodeTypeInterface $nodeTypeEntity,
  ): array {
    $entityOutputs = $nodeTypeEntity->getOutputs();
    $unified = [];

    foreach ($outputData as $key => $value) {
      // Skip internal keys (starting with _ or __).
      if (str_starts_with($key, "_")) {
        continue;
      }

      // Skip trigger - it has its own port.
      if ($key === "trigger") {
        continue;
      }

      // Check if exposed (default to TRUE if not configured).
      $outputConfig = $entityOutputs[$key] ?? [];
      $exposed = $outputConfig["exposed"] ?? TRUE;

      if ($exposed) {
        $unified[$key] = $value;
      }
    }

    return $unified;
  }

}
