<?php

declare(strict_types=1);

namespace Drupal\flowdrop_interrupt\EventSubscriber;

use Drupal\Component\Serialization\Json;
use Drupal\Core\Entity\EntityTypeManagerInterface;
use Drupal\flowdrop\Utility\NodeAnchor;
use Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface;
use Drupal\flowdrop_job\Constants\JobStatus;
use Drupal\flowdrop_job\FlowDropJobInterface;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface;
use Drupal\flowdrop_interrupt\Event\InterruptResolvedEvent;
use Drupal\flowdrop_orchestration\DTO\Snapshot\WorkflowSnapshot;
use Drupal\flowdrop_session\Constants\MessageRole;
use Drupal\flowdrop_session\Constants\MessageStatus;
use Drupal\flowdrop_session\Constants\SessionStatus;
use Drupal\flowdrop_session\Entity\FlowDropSessionMessageInterface;
use Drupal\flowdrop_session\Entity\FlowDropSessionInterface;
use Drupal\flowdrop_session\Service\SessionServiceInterface;
use Drupal\flowdrop_runtime\Service\Orchestrator\PipelineExecutorResolver;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;

/**
 * Event subscriber for resolved interrupts in sessions.
 *
 * This subscriber handles session-specific logic for interrupt resolution:
 * - Updates session status (running, awaiting_input, idle).
 * - Adds messages to the session conversation.
 * - Handles nested interrupts.
 *
 * It runs with higher priority than the generic subscriber and stops
 * propagation to prevent duplicate pipeline execution.
 *
 * @internal This class is not part of the public API and may change without notice.
 */
class SessionInterruptResolvedSubscriber implements EventSubscriberInterface {

  /**
   * Constructs a SessionInterruptResolvedSubscriber.
   *
   * @param \Drupal\Core\Entity\EntityTypeManagerInterface $entityTypeManager
   *   The entity type manager.
   * @param \Drupal\flowdrop_session\Service\SessionServiceInterface $sessionService
   *   The session service.
   * @param \Drupal\flowdrop_runtime\Service\Orchestrator\PipelineExecutorResolver $pipelineExecutorResolver
   *   Resolves the executor matching the pipeline's declared orchestrator.
   * @param \Psr\Log\LoggerInterface $logger
   *   The logger.
   */
  public function __construct(
    protected readonly EntityTypeManagerInterface $entityTypeManager,
    protected readonly SessionServiceInterface $sessionService,
    protected readonly PipelineExecutorResolver $pipelineExecutorResolver,
    protected readonly LoggerInterface $logger,
  ) {}

  /**
   * {@inheritdoc}
   */
  public static function getSubscribedEvents(): array {
    return [
      "flowdrop_interrupt.resolved" => ["onInterruptResolved", 10],
    ];
  }

  /**
   * Handles the interrupt resolved event for sessions.
   *
   * @param \Drupal\flowdrop_interrupt\Event\InterruptResolvedEvent $event
   *   The event containing the resolved interrupt.
   */
  public function onInterruptResolved(InterruptResolvedEvent $event): void {
    $interrupt = $event->interrupt;

    if (!$interrupt instanceof FlowDropInterruptInterface) {
      return;
    }

    $this->logger->info("SessionInterruptResolvedSubscriber: Received interrupt @id (type: @type, node: @node)", [
      "@id" => $interrupt->uuid(),
      "@type" => $interrupt->getType()->value,
      "@node" => $interrupt->getNodeId(),
    ]);

    $sessionId = $interrupt->getSessionId();
    if ($sessionId === NULL) {
      $this->logger->debug("Interrupt @id resolved but has no session ID", [
        "@id" => $interrupt->uuid(),
      ]);
      return;
    }

    $sessionStorage = $this->entityTypeManager->getStorage("flowdrop_session");
    $session = $sessionStorage->load($sessionId);

    if (!$session instanceof FlowDropSessionInterface) {
      $this->logger->warning("Interrupt @id has invalid session @session", [
        "@id" => $interrupt->uuid(),
        "@session" => $sessionId,
      ]);
      return;
    }

    // Pipeline carries the pending-interrupt slot now. Look it up via the
    // interrupt's pipeline id rather than session metadata, so concurrent
    // sub-pipelines paused in the same session don't collide on a single
    // session-level slot.
    $pendingPipelineId = $interrupt->getEffectivePipelineId();
    $pendingPipeline = $pendingPipelineId !== NULL ? $this->loadPipeline($pendingPipelineId) : NULL;
    $pendingInterruptId = $pendingPipeline?->getPendingInterruptId();
    $this->logger->info("SessionInterruptResolvedSubscriber: Session @session status=@status, pipeline=@pipeline, pipeline.pending=@pending", [
      "@session" => $sessionId,
      "@status" => $session->getStatus(),
      "@pipeline" => $pendingPipelineId ?? "(null)",
      "@pending" => $pendingInterruptId ?? "(null)",
    ]);

    if ($session->getStatus() !== SessionStatus::AWAITING_INPUT) {
      $this->logger->debug("Session @session is not awaiting input (status: @status)", [
        "@session" => $sessionId,
        "@status" => $session->getStatus(),
      ]);
      return;
    }

    if ($pendingInterruptId !== $interrupt->uuid()) {
      $this->logger->warning("Interrupt @id does not match pending @pending on pipeline @pipeline (session @session)", [
        "@id" => $interrupt->uuid(),
        "@pending" => $pendingInterruptId,
        "@pipeline" => $pendingPipelineId,
        "@session" => $sessionId,
      ]);
      return;
    }

    $event->stopPropagation();

    try {
      $this->resumeWorkflowExecution($session, $interrupt);
    }
    catch (\Throwable $e) {
      $this->logger->error("Failed to resume workflow for session @session: @message", [
        "@session" => $sessionId,
        "@message" => $e->getMessage(),
      ]);

      $this->sessionService->addSystemMessage($session, 'Execution failed', [
        "level" => "error",
        "type" => "resume_error",
        "interrupt_id" => $interrupt->uuid(),
      ]);

      $session->setStatus(SessionStatus::IDLE);
      $session->save();
    }
  }

  /**
   * Resumes workflow execution after an interrupt is resolved.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session.
   * @param \Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface $interrupt
   *   The resolved interrupt.
   */
  protected function resumeWorkflowExecution(
    FlowDropSessionInterface $session,
    FlowDropInterruptInterface $interrupt,
  ): void {
    $this->logger->info("Resuming workflow for session @session after interrupt @interrupt", [
      "@session" => $session->id(),
      "@interrupt" => $interrupt->uuid(),
    ]);

    $session->setStatus(SessionStatus::RUNNING);
    $session->save();

    $this->markInterruptMessageAsCompleted($session, $interrupt);

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

    // Use effective pipeline ID so inward Pause signals (which carry the
    // pipeline in target_pipeline_id, not pipeline_id) resume correctly
    // when resolved through the inbox.
    $pipelineId = $interrupt->getEffectivePipelineId();
    if ($pipelineId === NULL) {
      throw new \RuntimeException("Interrupt has no pipeline ID");
    }

    $pipeline = $this->loadPipeline($pipelineId);
    if ($pipeline === NULL) {
      throw new \RuntimeException("Pipeline {$pipelineId} not found");
    }

    $this->logger->info("ResumeWorkflow: Pipeline @id status=@status before execution", [
      "@id" => $pipelineId,
      "@status" => $pipeline->getStatus(),
    ]);

    if ($pipeline->getStatus() !== PipelineStatus::PAUSED) {
      $this->logger->warning("Pipeline @id is not paused (status: @status)", [
        "@id" => $pipelineId,
        "@status" => $pipeline->getStatus(),
      ]);
    }

    $this->resetInterruptedJobToPending($pipeline, $interrupt);

    $this->logger->info("ResumeWorkflow: Executing pipeline @id", [
      "@id" => $pipeline->id(),
    ]);
    // executionId defaults to pipeline id inside the orchestrator; pass it
    // explicitly so existing log breadcrumbs keep showing the same value.
    // Resume under the orchestrator the pipeline's workflow declared — a
    // stategraph pipeline must not resume through the plain sync loop.
    $response = $this->pipelineExecutorResolver
      ->getExecutorForPipeline($pipeline)
      ->executePipeline($pipeline, 100, NULL, (string) $pipeline->id());
    $this->logger->info("ResumeWorkflow: Pipeline execution returned status @status", [
      "@status" => $response->getStatus(),
    ]);

    $this->processResumeResults($session, $response, $workflow, $interrupt);

    if ($response->getStatus() === WorkflowSnapshot::STATUS_INTERRUPTED) {
      $this->handleInterruptedResume($session, $response, $interrupt);
      return;
    }

    // Clear the pipeline's pending-interrupt slot; the pipeline is no
    // longer paused on this interrupt. Re-read the pipeline first: the
    // orchestrator reloads and saves its own instance when stamping the
    // final status, so the reference taken before executePipeline() is
    // stale — saving it would overwrite the just-persisted terminal
    // status (completed regressing to running).
    $this->entityTypeManager->getStorage("flowdrop_pipeline")
      ->resetCache([$pipeline->id()]);
    $pipeline = $this->loadPipeline((string) $pipeline->id()) ?? $pipeline;
    $pipeline->setPendingInterruptId(NULL);
    $pipeline->setPendingInterruptType(NULL);
    $pipeline->save();

    $status = $response->getStatus();
    if ($status === PipelineStatus::COMPLETED || $status === "success") {
      $session->setStatus(SessionStatus::IDLE);
    }
    elseif ($status === PipelineStatus::FAILED || $status === "error") {
      $session->setStatus(SessionStatus::IDLE);
      $this->sessionService->addSystemMessage($session, 'Execution failed', [
        "level" => "error",
        "type" => "execution_error",
      ]);
    }
    else {
      $session->setStatus(SessionStatus::IDLE);
    }

    $session->save();

    $this->logger->info("Workflow resumed successfully for session @session, status: @status", [
      "@session" => $session->id(),
      "@status" => $status,
    ]);
  }

  /**
   * Resets the interrupted job to "pending" status.
   *
   * @param \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface $pipeline
   *   The pipeline containing the job.
   * @param \Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface $interrupt
   *   The resolved interrupt.
   */
  protected function resetInterruptedJobToPending(
    FlowDropPipelineInterface $pipeline,
    FlowDropInterruptInterface $interrupt,
  ): void {
    $nodeId = $interrupt->getNodeId();
    $interruptId = $interrupt->uuid();

    // Phase 2 fast-path: when the interrupt carries a job_id, load and
    // reset the job in O(1) without scanning the pipeline's job list.
    $jobId = $interrupt->getJobId();
    if ($jobId !== NULL) {
      $directJob = $this->entityTypeManager->getStorage('flowdrop_job')->load($jobId);
      if ($directJob instanceof FlowDropJobInterface && $directJob->getStatus() === JobStatus::INTERRUPTED) {
        $metadata = $directJob->getMetadata();
        $jobInterruptId = $metadata['interrupt_id'] ?? NULL;
        if ($jobInterruptId === $interruptId || $jobInterruptId === NULL) {
          $this->logger->info("Resetting interrupted job @job_id via O(1) job_id lookup (interrupt_id: @interrupt_id)", [
            "@job_id" => $directJob->id(),
            "@interrupt_id" => $interruptId,
          ]);
          unset($metadata['interrupt_message']);
          $directJob->setMetadata($metadata);
          $directJob->markAsPending();
          $directJob->save();
          return;
        }
      }
    }

    $this->logger->info("resetInterruptedJobToPending: Looking for job with nodeId=@nodeId, interruptId=@interruptId", [
      "@nodeId" => $nodeId,
      "@interruptId" => $interruptId,
    ]);

    $jobs = $pipeline->getJobs();

    foreach ($jobs as $job) {
      $this->logger->debug("resetInterruptedJobToPending: Job @id nodeId=@nodeId status=@status", [
        "@id" => $job->id(),
        "@nodeId" => $job->getNodeId(),
        "@status" => $job->getStatus(),
      ]);
    }

    foreach ($jobs as $job) {
      if ($job->getNodeId() === $nodeId && $job->getStatus() === JobStatus::INTERRUPTED) {
        $metadata = $job->getMetadata();
        $jobInterruptId = $metadata["interrupt_id"] ?? NULL;

        if ($jobInterruptId === $interruptId || $jobInterruptId === NULL) {
          $this->logger->info("Resetting interrupted job @job_id to pending (interrupt_id: @interrupt_id)", [
            "@job_id" => $job->id(),
            "@interrupt_id" => $interruptId,
          ]);

          unset($metadata["interrupt_message"]);
          $job->setMetadata($metadata);
          $job->markAsPending();
          $job->save();

          return;
        }
      }
    }

    $this->logger->warning("No interrupted job found for node @node_id in pipeline @pipeline_id", [
      "@node_id" => $nodeId,
      "@pipeline_id" => $pipeline->id(),
    ]);
  }

  /**
   * Loads a pipeline by ID.
   *
   * @param string $pipeline_id
   *   The pipeline ID (can be entity ID or execution ID).
   *
   * @return \Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface|null
   *   The pipeline entity, or NULL if not found.
   */
  protected function loadPipeline(string $pipeline_id): ?FlowDropPipelineInterface {
    $pipelineStorage = $this->entityTypeManager->getStorage("flowdrop_pipeline");

    $pipeline = $pipelineStorage->load($pipeline_id);
    if ($pipeline instanceof FlowDropPipelineInterface) {
      return $pipeline;
    }

    $entities = $pipelineStorage->loadByProperties(["execution_id" => $pipeline_id]);
    $pipeline = reset($entities);

    return $pipeline instanceof FlowDropPipelineInterface ? $pipeline : NULL;
  }

  /**
   * Handles another interrupt during resume (nested interrupts).
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session.
   * @param mixed $response
   *   The orchestration response.
   * @param \Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface $previousInterrupt
   *   The previously resolved interrupt.
   */
  protected function handleInterruptedResume(
    FlowDropSessionInterface $session,
    mixed $response,
    FlowDropInterruptInterface $previousInterrupt,
  ): 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;

    $this->logger->info("handleInterruptedResume: New interrupt id=@id, type=@type, node=@node, linked=@linked", [
      "@id" => $interruptId ?? "(null)",
      "@type" => $interruptType,
      "@node" => $interrupt["nodeId"] ?? "(null)",
      "@linked" => $linkedInterruptId ?? "(none)",
    ]);

    // Linked interrupts (e.g. the parent's WorkflowExecution placeholder
    // gated on a sub-pipeline sentinel) don't merit their own user-facing
    // bubble — the sub-workflow's own messages already populate the
    // stream and the user resolves the inner interrupt directly. Stamp
    // pipeline pending state and return without emitting noise.
    if (is_string($linkedInterruptId) && $linkedInterruptId !== '') {
      $executionId = $response->getExecutionId();
      if ($executionId !== NULL && $executionId !== '') {
        $pipeline = $this->loadPipeline($executionId);
        if ($pipeline !== NULL) {
          $pipeline->setPendingInterruptId($interruptId);
          $pipeline->setPendingInterruptType($interruptType);
          $pipeline->save();
        }
      }
      $session->setStatus(SessionStatus::AWAITING_INPUT);
      $session->save();
      return;
    }

    // Build message metadata with flat keys for the frontend library.
    $messageMetadata = [
      "type" => "interrupt_request",
      "interrupt_id" => $interruptId,
      "interrupt_type" => $interruptType,
      "schema" => $interruptSchema,
      "options" => $interruptOptions,
      "default_value" => $interruptDefaultValue,
      "node_id" => $interrupt["nodeId"] ?? NULL,
      // Surface workflow attribution for the UI bubble footer.
      "workflow_id" => $interrupt["workflowId"] ?? NULL,
      "execution_id" => $response->getExecutionId(),
      "allow_cancel" => TRUE,
    ];

    // Extract type-specific fields from schema/options into flat keys.
    if ($interruptType === "confirmation" && 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"];
          }
        }
      }
    }
    if (is_array($interruptSchema)) {
      if (isset($interruptSchema["placeholder"])) {
        $messageMetadata["placeholder"] = $interruptSchema["placeholder"];
      }
      if (isset($interruptSchema["multiline"])) {
        $messageMetadata["multiline"] = (bool) $interruptSchema["multiline"];
      }
      if (isset($interruptSchema["minLength"])) {
        $messageMetadata["min_length"] = (int) $interruptSchema["minLength"];
      }
      if (isset($interruptSchema["maxLength"])) {
        $messageMetadata["max_length"] = (int) $interruptSchema["maxLength"];
      }
    }

    $this->sessionService->addSystemMessage($session, $interruptMessage, $messageMetadata);

    // Stamp the new pending state on the executing pipeline rather than
    // session metadata. The session remains "awaiting_input" while any of
    // its pipelines carries a pending interrupt.
    $executionId = $response->getExecutionId();
    if ($executionId !== NULL && $executionId !== '') {
      $pipeline = $this->loadPipeline($executionId);
      if ($pipeline !== NULL) {
        $pipeline->setPendingInterruptId($interruptId);
        $pipeline->setPendingInterruptType($interruptType);
        $pipeline->save();
      }
      else {
        $this->logger->warning("handleInterruptedResume: Cannot stamp pending state — pipeline @id not found", [
          "@id" => $executionId,
        ]);
      }
    }

    $session->setStatus(SessionStatus::AWAITING_INPUT);
    $session->save();

    $this->logger->info("handleInterruptedResume: Session @session updated - status=awaiting_input, pipeline=@pipeline, pending_interrupt_id=@pending", [
      "@session" => $session->id(),
      "@pipeline" => $executionId ?? "(null)",
      "@pending" => $interruptId ?? "(null)",
    ]);
  }

  /**
   * Processes the results from resumed execution.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session.
   * @param mixed $response
   *   The orchestration response.
   * @param mixed $workflow
   *   The workflow entity.
   * @param \Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface $interrupt
   *   The resolved interrupt.
   */
  protected function processResumeResults(
    FlowDropSessionInterface $session,
    mixed $response,
    mixed $workflow,
    FlowDropInterruptInterface $interrupt,
  ): void {
    $this->logger->info("ProcessResumeResults: Called with response status @status", [
      "@status" => $response->getStatus(),
    ]);

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

    $outputPatterns = [
      "chat_output",
      "chatoutput",
      "text_output",
      "textoutput",
      "message_output",
      "chat_message",
      "response_output",
    ];

    $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));

      foreach ($outputPatterns as $pattern) {
        if (str_contains($nodeTypeId, $pattern)) {
          $outputNodes[$nodeId] = $nodeTypeId;
          break;
        }
      }
    }

    $this->logger->info("ProcessResumeResults: Found @count output nodes: @nodes", [
      "@count" => count($outputNodes),
      "@nodes" => implode(", ", array_keys($outputNodes)),
    ]);
    $this->logger->info("ProcessResumeResults: Results keys: @keys", [
      "@keys" => implode(", ", array_keys($results)),
    ]);

    foreach ($results as $resultKey => $output) {
      $parts = explode(":", $resultKey);
      $nodeId = $parts[0];

      if (!isset($outputNodes[$nodeId])) {
        $this->logger->debug("ProcessResumeResults: Skipping @key - not an output node", [
          "@key" => $resultKey,
        ]);
        continue;
      }

      $content = $this->extractMessageContent($output);
      $this->logger->debug("ProcessResumeResults: Extracted content for @key: @content", [
        "@key" => $resultKey,
        "@content" => $content ?? "(null)",
      ]);
      if ($content !== NULL && $content !== "") {
        $assistantMessage = $this->sessionService->addAssistantMessage(
          $session,
          $content,
          $nodeId,
          [
            "source" => "workflow_resume",
            "interrupt_id" => $interrupt->uuid(),
            "execution_id" => $response->getExecutionId(),
            "result_key" => $resultKey,
          ]
        );

        $assistantMessage->setStatus(MessageStatus::COMPLETED);
        $assistantMessage->save();

        $this->logger->info("ProcessResumeResults: Created assistant message @id for node @node", [
          "@id" => $assistantMessage->id(),
          "@node" => $nodeId,
        ]);
      }
    }
  }

  /**
   * Extracts message content from output.
   *
   * @param mixed $output
   *   The node output.
   *
   * @return string|null
   *   The message content or NULL.
   */
  protected function extractMessageContent(mixed $output): ?string {
    if (is_string($output)) {
      return $output;
    }

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

    $contentKeys = [
      "message",
      "text",
      "output",
      "content",
      "value",
      "result",
      "response",
    ];

    foreach ($contentKeys as $key) {
      if (isset($output[$key])) {
        $value = $output[$key];
        return is_string($value) ? $value : Json::encode($value);
      }
    }

    return NULL;
  }

  /**
   * Marks the interrupt request message as completed.
   *
   * @param \Drupal\flowdrop_session\Entity\FlowDropSessionInterface $session
   *   The session.
   * @param \Drupal\flowdrop_interrupt\Entity\FlowDropInterruptInterface $interrupt
   *   The resolved interrupt.
   */
  protected function markInterruptMessageAsCompleted(
    FlowDropSessionInterface $session,
    FlowDropInterruptInterface $interrupt,
  ): void {
    $interruptId = $interrupt->uuid();
    $sessionId = $session->id();

    if ($sessionId === NULL) {
      return;
    }

    $messageStorage = $this->entityTypeManager->getStorage("flowdrop_session_message");

    $query = $messageStorage->getQuery()
      // Access bypassed: internal event subscriber;
      // access enforced at event dispatch layer.
      ->accessCheck(FALSE)
      ->condition("session_id", $sessionId)
      ->condition("role", MessageRole::SYSTEM)
      ->sort("timestamp", "DESC");

    $messageIds = $query->execute();

    if (empty($messageIds)) {
      $this->logger->debug("No system messages found for session @session", [
        "@session" => $sessionId,
      ]);
      return;
    }

    $messages = $messageStorage->loadMultiple($messageIds);

    foreach ($messages as $message) {
      if (!$message instanceof FlowDropSessionMessageInterface) {
        continue;
      }

      $metadata = $message->getMessageMetadata();
      $messageInterruptId = $metadata["interrupt_id"] ?? NULL;

      if ($messageInterruptId === $interruptId) {
        $resolvedByUserId = $interrupt->getResolvedBy();
        if ($resolvedByUserId !== NULL) {
          $userStorage = $this->entityTypeManager->getStorage("user");
          $resolvedByUser = $userStorage->load($resolvedByUserId);
          if ($resolvedByUser !== NULL) {
            $metadata["resolvedByUserName"] = $resolvedByUser->getDisplayName();
            $metadata["resolvedByUserId"] = $resolvedByUserId;
          }
        }

        $responseData = $interrupt->getResponseData();
        if ($responseData !== NULL) {
          $metadata["response_value"] = $responseData;
        }

        $message->setMessageMetadata($metadata);
        $message->setStatus(MessageStatus::COMPLETED);
        $message->save();

        $this->logger->info("Marked interrupt message @message as completed for interrupt @interrupt", [
          "@message" => $message->id(),
          "@interrupt" => $interruptId,
        ]);
        return;
      }
    }

    $this->logger->debug("No message found with interrupt_id @interrupt in session @session", [
      "@interrupt" => $interruptId,
      "@session" => $sessionId,
    ]);
  }

}
