<?php

declare(strict_types=1);

namespace Drupal\Tests\flowdrop_stategraph\Kernel;

use Drupal\Tests\flowdrop\Kernel\WorkflowExecutionTestBase;
use Drupal\flowdrop_interrupt\Enum\InterruptType;
use Drupal\flowdrop_interrupt\Service\InterruptManagerInterface;
use Drupal\flowdrop_job\Constants\JobStatus;
use Drupal\flowdrop_pipeline\Constants\PausedReason;
use Drupal\flowdrop_pipeline\Constants\PipelineStatus;
use Drupal\flowdrop_stategraph\Service\StateGraphOrchestrator;
use Drupal\flowdrop_workflow\WorkflowDefinitionInterface;

/**
 * Pins the budget-pause → system Pause signal → resume chain end-to-end.
 *
 * Reproduces field report #7: pipelines paused by the time budget or
 * max-iterations used to be operator-terminal — no interrupt entity
 * existed, so neither resume subscriber could fire and `POST .../resume`
 * 409'd. Now the orchestrator stamps a `paused_reason` on the execution
 * context, BudgetPauseSignalSubscriber converts it into a pending system
 * Pause signal (initiator NULL), and resolving that signal re-executes the
 * pipeline through the standard InterruptResolvedSubscriber chain.
 *
 * No LLM required; budgets are breached deterministically (a sub-ms time
 * budget that the loop's first iteration always exceeds, and a 1-iteration
 * budget on a two-wave workflow).
 *
 * @group flowdrop_stategraph
 */
class StateGraphBudgetPauseResumeTest extends WorkflowExecutionTestBase {

  /**
   * A time-budget pause raises a resumable system Pause signal.
   *
   * The sub-ms `timeout` is breached before the loop's first iteration
   * (pipeline creation and job generation alone exceed it), so the run
   * pauses with every job still ready. The operator raises the budget,
   * resolves the signal, and the run completes — reason cleared, no
   * re-pause.
   */
  public function testTimeBudgetPauseIsResumable(): void {
    $workflow = $this->createChainWorkflow('sg_time_budget', [
      'timeout' => 0.0001,
    ]);

    $result = $this->launchAndWait($workflow);
    $this->assertSame(PipelineStatus::PAUSED, $result->status);

    $pipeline = $this->loadPipeline($result->pipelineId);
    $this->assertSame(
      PausedReason::TIME_BUDGET,
      $pipeline->getExecutionContext()[PausedReason::CONTEXT_KEY] ?? NULL,
      'A time-budget pause must stamp its reason on the execution context.',
    );

    // The pause must surface as a pending system Pause signal.
    $signal = $this->interruptManager()->findPendingInwardSignal($result->pipelineId);
    $this->assertNotNull($signal, 'A budget pause must raise a pending inward Pause signal.');
    $this->assertSame(InterruptType::Pause, $signal->getType());
    $this->assertSame(PausedReason::TIME_BUDGET, $signal->getReason());
    $this->assertNull($signal->getInitiatorUid(), 'The signal must be system-raised (NULL initiator).');

    // Operator raises the budget before resuming — with the original
    // sub-ms budget every resume window would re-pause immediately.
    $context = $pipeline->getExecutionContext();
    $context[StateGraphOrchestrator::CONTEXT_STATEGRAPH_CONFIG]['maxExecutionTime'] = NULL;
    $pipeline->setExecutionContext($context);
    $pipeline->save();

    // Resolving the signal resumes the pipeline through the standard
    // InterruptResolvedSubscriber chain.
    $resolved = $this->interruptManager()->resolveInterrupt(
      (string) $signal->uuid(),
      ['resumed_at' => 1],
    );
    $this->assertNotNull($resolved);

    $pipeline = $this->loadPipeline($result->pipelineId);
    $this->assertSame(PipelineStatus::COMPLETED, $pipeline->getStatus(), $this->jobTrail($pipeline));
    $this->assertArrayNotHasKey(
      PausedReason::CONTEXT_KEY,
      $pipeline->getExecutionContext(),
      'The paused reason must be cleared once the run completes.',
    );
    $this->assertNull(
      $this->interruptManager()->findPendingInwardSignal($result->pipelineId),
      'A completed run must not re-pause or leave a dangling signal.',
    );
  }

  /**
   * A max-iterations pause resumes with a fresh iteration window.
   *
   * With a 1-iteration budget on a two-wave chain, the first run executes
   * only the first wave; resolving the signal re-enters the loop, the
   * second wave runs, and the pipeline completes — no config change needed
   * because each resume gets a fresh iteration counter.
   */
  public function testMaxIterationsPauseIsResumable(): void {
    $workflow = $this->createChainWorkflow('sg_max_iterations', [
      'stategraph' => [
        'checkpointer_type' => 'memory',
        'create_checkpoints' => TRUE,
        'max_iterations' => 1,
      ],
    ]);

    $result = $this->launchAndWait($workflow);
    $this->assertSame(PipelineStatus::PAUSED, $result->status);

    $pipeline = $this->loadPipeline($result->pipelineId);
    $this->assertSame(
      PausedReason::MAX_ITERATIONS,
      $pipeline->getExecutionContext()[PausedReason::CONTEXT_KEY] ?? NULL,
    );
    $this->assertSame(JobStatus::COMPLETED, $this->jobStatusForNode($pipeline, 'first'));
    $this->assertNotSame(JobStatus::COMPLETED, $this->jobStatusForNode($pipeline, 'second'),
      'The second wave must not have run within a 1-iteration budget.');

    $signal = $this->interruptManager()->findPendingInwardSignal($result->pipelineId);
    $this->assertNotNull($signal);
    $this->assertSame(InterruptType::Pause, $signal->getType());
    $this->assertSame(PausedReason::MAX_ITERATIONS, $signal->getReason());

    $resolved = $this->interruptManager()->resolveInterrupt(
      (string) $signal->uuid(),
      ['resumed_at' => 1],
    );
    $this->assertNotNull($resolved);

    $pipeline = $this->loadPipeline($result->pipelineId);
    $this->assertSame(PipelineStatus::COMPLETED, $pipeline->getStatus(), $this->jobTrail($pipeline));
    $this->assertSame(JobStatus::COMPLETED, $this->jobStatusForNode($pipeline, 'second'));
    $this->assertArrayNotHasKey(PausedReason::CONTEXT_KEY, $pipeline->getExecutionContext());
    $this->assertNull($this->interruptManager()->findPendingInwardSignal($result->pipelineId));
  }

  /**
   * A HITL-style pause without a stamped reason raises no signal.
   *
   * Guard for the subscriber's reason gate: an operator Pause signal
   * arriving mid-run pauses the pipeline through the poll path, which
   * stamps no paused_reason — the subscriber must not stack a second
   * (system) signal on top of the operator's own.
   */
  public function testPauseWithoutReasonRaisesNoSignal(): void {
    $workflow = $this->createChainWorkflow('sg_no_reason', [
      'stategraph' => [
        'checkpointer_type' => 'memory',
        'create_checkpoints' => TRUE,
        'max_iterations' => 20,
      ],
    ]);

    $result = $this->launchAndWait($workflow);
    $this->assertSame(PipelineStatus::COMPLETED, $result->status);

    // A completed run must never carry a reason or a signal.
    $pipeline = $this->loadPipeline($result->pipelineId);
    $this->assertArrayNotHasKey(PausedReason::CONTEXT_KEY, $pipeline->getExecutionContext());
    $this->assertNull($this->interruptManager()->findPendingInwardSignal($result->pipelineId));
  }

  /**
   * Creates a two-wave logger chain declared for the stategraph.
   *
   * `first → second` over a plain data edge, so the run needs two ready
   * waves to complete — enough to leave work behind when a budget breaks
   * the loop early.
   *
   * @param string $id
   *   The workflow entity id.
   * @param array<string, mixed> $orchestratorSettings
   *   Extra orchestrator settings merged over the stategraph declaration.
   *
   * @return \Drupal\flowdrop_workflow\WorkflowDefinitionInterface
   *   The saved workflow.
   */
  protected function createChainWorkflow(string $id, array $orchestratorSettings): WorkflowDefinitionInterface {
    return $this->createStateGraphWorkflow($id, [
      $this->loggerNode('first', 'wave one'),
      $this->loggerNode('second', 'wave two'),
    ], [
      $this->edge('first', 'message', 'second', 'message'),
    ], $orchestratorSettings);
  }

  /**
   * Returns the interrupt manager.
   */
  protected function interruptManager(): InterruptManagerInterface {
    return $this->container->get('flowdrop_interrupt.manager');
  }

}
