<?php

declare(strict_types=1);

namespace Drupal\Tests\flowdrop_stategraph\Integration;

use Drupal\Core\Config\ConfigFactoryInterface;
use Drupal\Core\Config\ImmutableConfig;
use Drupal\Tests\UnitTestCase;
use Drupal\flowdrop\DTO\Output;
use Drupal\flowdrop_job\FlowDropJobInterface;
use Drupal\flowdrop_orchestration\DTO\Snapshot\IteratorState;
use Drupal\flowdrop_pipeline\Entity\FlowDropPipelineInterface;
use Drupal\flowdrop_stategraph\DTO\GraphState;
use Drupal\flowdrop_pipeline\Service\JobGenerationService;
use Drupal\flowdrop_orchestration\DTO\Orchestrator\OrchestrationRequest;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionContext;
use Drupal\flowdrop_runtime\DTO\Runtime\NodeExecutionResult;
use Drupal\flowdrop_runtime\Exception\OrchestrationException;
use Drupal\flowdrop_runtime\Service\Runtime\NodeRuntimeService;
use Drupal\flowdrop_runtime\Service\Storage\JobStorageInterface;
use Drupal\flowdrop_runtime\Service\Storage\PipelineStorageInterface;
use Drupal\flowdrop_runtime\Service\Storage\PipelineStorageRegistryInterface;
use Drupal\flowdrop_stategraph\Checkpointer\MemoryCheckpointer;
use Drupal\flowdrop_stategraph\Routing\ConditionalEdgeEvaluator;
use Drupal\flowdrop_stategraph\Service\CheckpointerFactory;
use Drupal\flowdrop_stategraph\Service\StateGraphOrchestrator;
use Drupal\flowdrop_stategraph\Service\StateManager;
use Drupal\flowdrop_workflow\FlowDropWorkflowInterface;
use Prophecy\Argument;
use Prophecy\PhpUnit\ProphecyTrait;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

/**
 * Integration tests for the StateGraphOrchestrator.
 *
 * @coversDefaultClass \Drupal\flowdrop_stategraph\Service\StateGraphOrchestrator
 * @group flowdrop_stategraph
 */
class StateGraphOrchestratorTest extends UnitTestCase {

  use ProphecyTrait;

  /**
   * The orchestrator under test.
   *
   * @var \Drupal\flowdrop_stategraph\Service\StateGraphOrchestrator
   */
  private StateGraphOrchestrator $orchestrator;

  /**
   * Mock node runtime service.
   *
   * @var \Prophecy\Prophecy\ObjectProphecy<\Drupal\flowdrop_runtime\Service\Runtime\NodeRuntimeService>
   */
  private $nodeRuntime;

  /**
   * Mock pipeline storage.
   *
   * @var \Prophecy\Prophecy\ObjectProphecy<\Drupal\flowdrop_runtime\Service\Storage\PipelineStorageInterface>
   */
  private $pipelineStorage;

  /**
   * Mock job generation service.
   *
   * @var \Prophecy\Prophecy\ObjectProphecy<\Drupal\flowdrop_pipeline\Service\JobGenerationService>
   */
  private $jobGenerationService;

  /**
   * Mock event dispatcher.
   *
   * @var \Prophecy\Prophecy\ObjectProphecy<\Symfony\Component\EventDispatcher\EventDispatcherInterface>
   */
  private $eventDispatcher;

  /**
   * Memory checkpointer for testing.
   *
   * @var \Drupal\flowdrop_stategraph\Checkpointer\MemoryCheckpointer
   */
  private MemoryCheckpointer $memoryCheckpointer;

  /**
   * Tracks executed nodes.
   *
   * @var array<string>
   */
  private array $executedNodes = [];

  /**
   * {@inheritdoc}
   */
  protected function setUp(): void {
    parent::setUp();

    $this->executedNodes = [];

    // Create logger mock.
    // Note: For void methods, just stub without willReturn().
    $logger = $this->prophesize(LoggerInterface::class);
    $logger->debug(Argument::cetera());
    $logger->info(Argument::cetera());
    $logger->error(Argument::cetera());
    $logger->warning(Argument::cetera());

    // Create config mock.
    $config = $this->prophesize(ImmutableConfig::class);
    $config->get("max_iterations")->willReturn(100);
    $configFactory = $this->prophesize(ConfigFactoryInterface::class);
    $configFactory->get("flowdrop_stategraph.settings")->willReturn($config->reveal());

    // Create real services.
    $stateManager = new StateManager($logger->reveal(), $configFactory->reveal());

    $this->memoryCheckpointer = new MemoryCheckpointer($logger->reveal());
    $this->memoryCheckpointer->clear();

    $checkpointerFactory = $this->prophesize(CheckpointerFactory::class);
    $checkpointerFactory->create(Argument::any())->willReturn($this->memoryCheckpointer);

    $conditionalEdgeEvaluator = new ConditionalEdgeEvaluator($logger->reveal());

    // Create mocks for Drupal services.
    $this->nodeRuntime = $this->prophesize(NodeRuntimeService::class);
    $this->jobGenerationService = $this->prophesize(JobGenerationService::class);
    $this->eventDispatcher = $this->prophesize(EventDispatcherInterface::class);
    $this->eventDispatcher->dispatch(Argument::any(), Argument::any())->willReturn(Argument::any());

    // Create mocks for storage services.
    $this->pipelineStorage = $this->prophesize(PipelineStorageInterface::class);
    $this->pipelineStorage->save(Argument::any());
    $this->pipelineStorage->resetCache(Argument::any());
    $pipelineStorageRegistry = $this->prophesize(PipelineStorageRegistryInterface::class);
    $pipelineStorageRegistry->getStorage(Argument::any())->willReturn($this->pipelineStorage->reveal());
    $jobStorage = $this->prophesize(JobStorageInterface::class);
    $jobStorage->save(Argument::any());
    $jobStorage->resetCache(Argument::any());

    // Create orchestrator with new constructor signature.
    $this->orchestrator = new StateGraphOrchestrator(
      $pipelineStorageRegistry->reveal(),
      $jobStorage->reveal(),
      $this->nodeRuntime->reveal(),
      $this->jobGenerationService->reveal(),
      $this->eventDispatcher->reveal(),
      $logger->reveal(),
      $stateManager,
      $checkpointerFactory->reveal(),
      $conditionalEdgeEvaluator,
    );
  }

  /**
   * {@inheritdoc}
   */
  protected function tearDown(): void {
    $this->memoryCheckpointer->clear();
    parent::tearDown();
  }

  /**
   * Tests getType returns "flowdrop_stategraph:stategraph".
   *
   * @covers ::getType
   */
  public function testGetType(): void {
    $this->assertEquals("flowdrop_stategraph:stategraph", $this->orchestrator->getType());
  }

  /**
   * Tests getCapabilities returns expected flags.
   *
   * @covers ::getCapabilities
   */
  public function testGetCapabilities(): void {
    $capabilities = $this->orchestrator->getCapabilities();

    $this->assertTrue($capabilities["synchronous_execution"]);
    $this->assertFalse($capabilities["parallel_execution"]);
    $this->assertFalse($capabilities["async_execution"]);
    $this->assertTrue($capabilities["retry_support"]);
    $this->assertTrue($capabilities["error_recovery"]);
    $this->assertTrue($capabilities["pipeline_based"]);
    $this->assertTrue($capabilities["job_tracking"]);
    $this->assertTrue($capabilities["stateful"]);
    $this->assertTrue($capabilities["checkpointing"]);
    $this->assertTrue($capabilities["conditional_routing"]);
    $this->assertTrue($capabilities["iterator_support"]);
    $this->assertTrue($capabilities["message_history"]);
  }

  /**
   * Tests supportsWorkflow returns true for any workflow.
   *
   * @covers ::supportsWorkflow
   */
  public function testSupportsWorkflow(): void {
    $this->assertTrue($this->orchestrator->supportsWorkflow(["id" => "any_workflow"]));
    $this->assertTrue($this->orchestrator->supportsWorkflow([]));
    $this->assertTrue($this->orchestrator->supportsWorkflow(["complex" => ["nested" => "data"]]));
  }

  /**
   * Tests linear workflow execution A -> B -> C.
   *
   * @covers ::orchestrate
   */
  public function testLinearWorkflowExecution(): void {
    $workflow = [
      "id" => "linear_workflow",
      "label" => "Linear Test Workflow",
      "nodes" => [
        ["id" => "node_a", "type" => "processor_a", "config" => []],
        ["id" => "node_b", "type" => "processor_b", "config" => []],
        ["id" => "node_c", "type" => "processor_c", "config" => []],
      ],
      "edges" => [
        ["source" => "node_a", "target" => "node_b"],
        ["source" => "node_b", "target" => "node_c"],
      ],
    ];

    $this->setupMocksForLinearWorkflow($workflow);

    $request = new OrchestrationRequest(
      workflowId: "linear_workflow",
      pipelineId: "pipeline_1",
      workflow: $workflow,
      initialData: ["input" => "test_data"],
      options: [
        "stategraph" => [
          "checkpointer_type" => "memory",
          "create_checkpoints" => TRUE,
        ],
      ],
    );

    $response = $this->orchestrator->orchestrate($request);

    $this->assertEquals("completed", $response->getStatus());
    $this->assertNotEmpty($response->getExecutionId());
    $this->assertGreaterThan(0, $response->getExecutionTime());

    // Verify execution order.
    $this->assertEquals(["node_a", "node_b", "node_c"], $this->executedNodes);
  }

  /**
   * Tests orchestration response contains correct metadata.
   *
   * @covers ::orchestrate
   */
  public function testOrchestrationResponseMetadata(): void {
    $workflow = [
      "id" => "metadata_workflow",
      "label" => "Metadata Test",
      "nodes" => [
        ["id" => "node_1", "type" => "processor", "config" => []],
      ],
      "edges" => [],
    ];

    $this->setupMocksForSingleNodeWorkflow($workflow);

    $request = new OrchestrationRequest(
      workflowId: "metadata_workflow",
      pipelineId: "pipeline_metadata",
      workflow: $workflow,
      initialData: [],
      options: [
        "stategraph" => [
          "thread_id" => "test_thread_123",
        ],
      ],
    );

    $response = $this->orchestrator->orchestrate($request);

    $metadata = $response->getMetadata();

    $this->assertArrayHasKey("pipeline_id", $metadata);
    $this->assertArrayHasKey("executed_jobs", $metadata);
    $this->assertArrayHasKey("iterations", $metadata);
    $this->assertArrayHasKey("total_jobs", $metadata);
    $this->assertArrayHasKey("thread_id", $metadata);
    $this->assertArrayHasKey("final_state", $metadata);
  }

  /**
   * Tests max iterations exceeded returns correct status.
   *
   * @covers ::orchestrate
   */
  public function testMaxIterationsExceeded(): void {
    $workflow = [
      "id" => "loop_workflow",
      "label" => "Loop Test",
      "nodes" => [
        ["id" => "loop_node", "type" => "processor", "config" => []],
      ],
      "edges" => [],
    ];

    // Setup mocks that create an infinite loop scenario.
    $this->setupMocksForLoopingWorkflow($workflow);

    $request = new OrchestrationRequest(
      workflowId: "loop_workflow",
      pipelineId: "pipeline_loop",
      workflow: $workflow,
      initialData: [],
      options: [
        "stategraph" => [
          "max_iterations" => 3,
        ],
      ],
    );

    $response = $this->orchestrator->orchestrate($request);

    // Should complete normally since we limit iterations.
    // The status depends on whether jobs are exhausted or max reached.
    $this->assertContains($response->getStatus(), ["completed", "paused", "max_iterations_exceeded"]);
  }

  /**
   * Tests that a node reporting error status fails the job and the run.
   *
   * Node processor failures are swallowed by NodeRuntimeService into an
   * error-status Output rather than propagating. Before first-class failure
   * handling the orchestrator marked every such job COMPLETED, silently
   * losing the failure. The orchestrator must now mark the job FAILED and
   * surface a failed run.
   *
   * @covers ::executeJobWithState
   */
  public function testErroredNodeOutputFailsTheJobAndRun(): void {
    $workflow = [
      "id" => "failing_workflow",
      "label" => "Failing Node Test",
      "nodes" => [
        ["id" => "boom", "type" => "processor", "config" => []],
      ],
      "edges" => [],
    ];

    $tracker = new \stdClass();
    $tracker->completed = FALSE;
    $tracker->failed = FALSE;
    $tracker->failureMessage = NULL;

    $this->setupMocksForFailingSingleNode($workflow, $tracker, "upstream blew up");

    $request = new OrchestrationRequest(
      workflowId: "failing_workflow",
      pipelineId: "pipeline_failing",
      workflow: $workflow,
      initialData: [],
      options: ["stategraph" => ["checkpointer_type" => "memory"]],
    );

    // An escaping job failure surfaces as a failed orchestration (the same
    // contract as a genuinely thrown error), not a silently completed run.
    $threw = FALSE;
    try {
      $this->orchestrator->orchestrate($request);
    }
    catch (OrchestrationException $e) {
      $threw = TRUE;
      $this->assertStringContainsString("upstream blew up", $e->getMessage());
    }

    $this->assertTrue($threw, "Errored node must surface as a failed orchestration.");
    $this->assertTrue($tracker->failed, "Errored node must mark its job FAILED.");
    $this->assertFalse($tracker->completed, "Errored node must NOT mark its job COMPLETED.");
    $this->assertStringContainsString("upstream blew up", (string) $tracker->failureMessage);
  }

  /**
   * Tests that a node with an error edge routes its failure instead of failing.
   *
   * When the node has an outgoing error edge the orchestrator must NOT throw:
   * it marks the job FAILED + `error_routed`, emits the error payload on the
   * `error` output port, and lets the run complete (the handled failure is not
   * an unhandled run failure).
   *
   * @covers ::executeJobWithState
   * @covers ::routeNodeError
   * @covers ::nodeHasErrorEdge
   */
  public function testErrorEdgeRoutesFailureWithoutFailingTheRun(): void {
    $workflow = [
      "id" => "routing_workflow",
      "label" => "Error Routing Test",
      "nodes" => [
        ["id" => "boom", "type" => "processor", "config" => []],
      ],
      "edges" => [],
    ];

    $tracker = new \stdClass();
    $tracker->completed = FALSE;
    $tracker->failed = FALSE;
    $tracker->routed = FALSE;
    $tracker->outputData = NULL;

    $this->setupMocksForRoutedFailingNode($workflow, $tracker, "boom blew up");

    $request = new OrchestrationRequest(
      workflowId: "routing_workflow",
      pipelineId: "pipeline_routing",
      workflow: $workflow,
      initialData: [],
      options: ["stategraph" => ["checkpointer_type" => "memory"]],
    );

    // A routed failure does not surface as a failed orchestration.
    $response = $this->orchestrator->orchestrate($request);

    $this->assertSame("completed", $response->getStatus(), "A routed failure must let the run complete.");
    $this->assertTrue($tracker->failed, "The failing node's job must be marked FAILED.");
    $this->assertTrue($tracker->routed, "The failing node's job must be stamped error_routed.");
    $this->assertFalse($tracker->completed, "A routed failure must NOT mark the job COMPLETED.");
    $this->assertIsArray($tracker->outputData);
    $this->assertArrayHasKey("error", $tracker->outputData);
    $this->assertSame("boom blew up", $tracker->outputData["error"]["message"]);
    $this->assertSame("boom", $tracker->outputData["error"]["node_id"]);
  }

  /**
   * Tests that loopback edges are followed for branch-driven (agent) loops.
   *
   * Loopback re-dispatch used to be gated solely on iterator state, so a loop
   * with no iterator (e.g. an agent reason/act loop whose continuation is
   * decided by an upstream gateway) was never re-dispatched. It must now be
   * followed up to the iteration cap, while ForEach (iterator-driven) semantics
   * are preserved exactly.
   *
   * @covers ::shouldFollowLoopbackEdge
   */
  public function testShouldFollowLoopbackEdge(): void {
    $decide = new \ReflectionMethod($this->orchestrator, "shouldFollowLoopbackEdge");

    // Branch-driven loop (no iterator): follow while under the iteration cap.
    $branchState = new GraphState();
    $this->assertTrue(
      $decide->invoke($this->orchestrator, $branchState, "reason", ["reason" => 0], 5),
      "Branch-driven loopback should be followed when under the cap.",
    );
    $this->assertTrue(
      $decide->invoke($this->orchestrator, $branchState, "reason", ["reason" => 4], 5),
      "Branch-driven loopback should be followed on the last allowed iteration.",
    );

    // Branch-driven loop at/over the cap: stop (runaway backstop).
    $this->assertFalse(
      $decide->invoke($this->orchestrator, $branchState, "reason", ["reason" => 5], 5),
      "Branch-driven loopback should stop once the iteration cap is reached.",
    );

    // Iterator-driven loop (ForEach): preserve hasMore semantics exactly.
    $hasMore = new GraphState(iterator: new IteratorState(hasMore: TRUE));
    $exhausted = new GraphState(iterator: new IteratorState(hasMore: FALSE));
    $this->assertTrue(
      $decide->invoke($this->orchestrator, $hasMore, "body", [], 5),
      "Iterator loop should be followed while the iterator has more items.",
    );
    $this->assertFalse(
      $decide->invoke($this->orchestrator, $exhausted, "body", [], 5),
      "Iterator loop should stop once the iterator is exhausted.",
    );
  }

  /**
   * Sets up mocks for a linear workflow.
   *
   * @param array<string, mixed> $workflow
   *   The workflow definition.
   */
  private function setupMocksForLinearWorkflow(array $workflow): void {
    // Setup pipeline mock.
    $pipeline = $this->prophesize(FlowDropPipelineInterface::class);
    $workflowEntity = $this->prophesize(FlowDropWorkflowInterface::class);
    $workflowEntity->id()->willReturn($workflow["id"]);
    $workflowEntity->getGraphData()->willReturn($workflow);

    $pipeline->getWorkflow()->willReturn($workflowEntity->reveal());
    $pipeline->id()->willReturn("pipeline_123");
    $pipeline->getInputData()->willReturn(["input" => "test_data"]);
    $pipeline->isReadyToStart()->willReturn(TRUE);
    $pipeline->isPaused()->willReturn(FALSE);
    // Fluent methods returning 'static' need explicit return values.
    $pipeline->markAsStarted()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->markAsCompleted(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->hasFailedJobs()->willReturn(FALSE);
    $pipeline->pause()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->save()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->getExecutionContext()->willReturn([]);
    $pipeline->setExecutionContext(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });

    // Create jobs for each node.
    $jobs = [];
    $readyJobsCalls = [];

    foreach ($workflow["nodes"] as $index => $node) {
      $job = $this->prophesize(FlowDropJobInterface::class);
      $job->id()->willReturn("job_{$index}");
      $job->getNodeId()->willReturn($node["id"]);
      $job->getMetadata()->willReturn([
        "config" => $node["config"],
        "node_type_id" => $node["type"],
        "incoming_edges" => [],
      ]);
      // Fluent methods returning 'static' need explicit return values.
      $job->markAsStarted()->will(function () use ($job) {
        return $job->reveal();
      });
      $job->markAsCompleted(Argument::any())->will(function () use ($job) {
        return $job->reveal();
      });
      $job->setInputData(Argument::any())->will(function () use ($job) {
        return $job->reveal();
      });
      $job->setMetadata(Argument::any())->will(function () use ($job) {
        return $job->reveal();
      });
      $job->markAsSkipped()->will(function () use ($job) {
        return $job->reveal();
      });
      $job->save()->will(function () use ($job) {
        return $job->reveal();
      });

      $jobs[] = $job->reveal();
    }

    // Return one job at a time to simulate sequential execution.
    // After returning all jobs, subsequent calls return empty array.
    // Use stdClass to avoid reference issues with Prophecy closures.
    $pipeline->getJobs()->willReturn($jobs);
    $pipeline->getJobsByStatus(Argument::type("string"))->will(function ($args) use ($jobs) {
      if ($args[0] === "idle") {
        return $jobs;
      }
      return [];
    });
    $pipeline->hasInterruptedJobs()->willReturn(FALSE);
    $pipeline->markAsFailed(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });

    // Setup pipeline storage.
    $this->pipelineStorage->create(Argument::any())->willReturn($pipeline->reveal());
    $this->pipelineStorage->load(Argument::any())->willReturn($pipeline->reveal());

    // Setup job generation.
    $this->jobGenerationService->generateJobs(Argument::any(), Argument::any())->shouldBeCalled();
    $jobTracker = new \stdClass();
    $jobTracker->queue = $jobs;
    $this->jobGenerationService->getReadyJobs(Argument::any())->will(function () use ($jobTracker) {
      if (!empty($jobTracker->queue)) {
        return [array_shift($jobTracker->queue)];
      }
      return [];
    });

    // Setup node runtime.
    $executedNodes = &$this->executedNodes;
    $this->nodeRuntime->executeNode(
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("array"),
      Argument::type("array"),
      Argument::type(NodeExecutionContext::class),
      Argument::cetera()
    )->will(function ($args) use (&$executedNodes) {
      $nodeId = $args[1];
      $executedNodes[] = $nodeId;

      $output = new Output([
        "status" => "success",
        "result" => "output_from_{$nodeId}",
      ]);

      return new NodeExecutionResult(
        nodeId: $nodeId,
        status: "success",
        output: $output,
        executionTime: 0.01,
        metadata: [],
      );
    });
  }

  /**
   * Sets up mocks for a single node workflow.
   *
   * @param array<string, mixed> $workflow
   *   The workflow definition.
   */
  private function setupMocksForSingleNodeWorkflow(array $workflow): void {
    $node = $workflow["nodes"][0];

    // Setup pipeline mock.
    $pipeline = $this->prophesize(FlowDropPipelineInterface::class);
    $workflowEntity = $this->prophesize(FlowDropWorkflowInterface::class);
    $workflowEntity->id()->willReturn($workflow["id"]);
    $workflowEntity->getGraphData()->willReturn($workflow);

    $pipeline->getWorkflow()->willReturn($workflowEntity->reveal());
    $pipeline->id()->willReturn("pipeline_single");
    $pipeline->getInputData()->willReturn([]);
    $pipeline->isReadyToStart()->willReturn(TRUE);
    $pipeline->isPaused()->willReturn(FALSE);
    // Fluent methods returning 'static' need explicit return values.
    $pipeline->markAsStarted()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->markAsCompleted(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->hasFailedJobs()->willReturn(FALSE);
    $pipeline->pause()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->save()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->getExecutionContext()->willReturn([]);
    $pipeline->setExecutionContext(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });

    // Create single job.
    $job = $this->prophesize(FlowDropJobInterface::class);
    $job->id()->willReturn("job_0");
    $job->getNodeId()->willReturn($node["id"]);
    $job->getMetadata()->willReturn([
      "config" => $node["config"],
      "node_type_id" => $node["type"],
      "incoming_edges" => [],
    ]);
    // Fluent methods returning 'static' need explicit return values.
    $job->markAsStarted()->will(function () use ($job) {
      return $job->reveal();
    });
    $job->markAsCompleted(Argument::any())->will(function () use ($job) {
      return $job->reveal();
    });
    $job->setInputData(Argument::any())->will(function () use ($job) {
      return $job->reveal();
    });
    $job->setMetadata(Argument::any())->will(function () use ($job) {
      return $job->reveal();
    });
    $job->save()->will(function () use ($job) {
      return $job->reveal();
    });

    $jobs = [$job->reveal()];

    $pipeline->getJobs()->willReturn($jobs);
    $pipeline->getJobsByStatus(Argument::type("string"))->willReturn([]);
    $pipeline->hasInterruptedJobs()->willReturn(FALSE);
    $pipeline->markAsFailed(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });

    // Setup pipeline storage.
    $this->pipelineStorage->create(Argument::any())->willReturn($pipeline->reveal());
    $this->pipelineStorage->load(Argument::any())->willReturn($pipeline->reveal());

    $this->jobGenerationService->generateJobs(Argument::any(), Argument::any())->shouldBeCalled();
    $returned = FALSE;
    $this->jobGenerationService->getReadyJobs(Argument::any())->will(function () use ($jobs, &$returned) {
      if (!$returned) {
        $returned = TRUE;
        return $jobs;
      }
      return [];
    });

    // Setup node runtime.
    $executedNodes = &$this->executedNodes;
    $this->nodeRuntime->executeNode(
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("array"),
      Argument::type("array"),
      Argument::type(NodeExecutionContext::class),
      Argument::cetera()
    )->will(function ($args) use (&$executedNodes) {
      $nodeId = $args[1];
      $executedNodes[] = $nodeId;

      return new NodeExecutionResult(
        nodeId: $nodeId,
        status: "success",
        output: new Output(["result" => "done"]),
        executionTime: 0.01,
        metadata: [],
      );
    });
  }

  /**
   * Sets up mocks for a single node whose execution reports error status.
   *
   * Mirrors setupMocksForSingleNodeWorkflow() but the node runtime returns an
   * error-status Output, and the job's terminal-status calls are recorded on
   * the tracker so the test can assert FAILED (not COMPLETED).
   *
   * @param array<string, mixed> $workflow
   *   The workflow definition.
   * @param \stdClass $tracker
   *   Records ->completed, ->failed and ->failureMessage.
   * @param string $errorMessage
   *   The error message the node reports.
   */
  private function setupMocksForFailingSingleNode(array $workflow, \stdClass $tracker, string $errorMessage): void {
    $node = $workflow["nodes"][0];

    $pipeline = $this->prophesize(FlowDropPipelineInterface::class);
    $workflowEntity = $this->prophesize(FlowDropWorkflowInterface::class);
    $workflowEntity->id()->willReturn($workflow["id"]);
    $workflowEntity->getGraphData()->willReturn($workflow);

    $pipeline->getWorkflow()->willReturn($workflowEntity->reveal());
    $pipeline->id()->willReturn("pipeline_failing");
    $pipeline->getInputData()->willReturn([]);
    $pipeline->isReadyToStart()->willReturn(TRUE);
    $pipeline->isPaused()->willReturn(FALSE);
    $pipeline->markAsStarted()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->markAsCompleted(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->hasFailedJobs()->willReturn(FALSE);
    $pipeline->pause()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->save()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->getExecutionContext()->willReturn([]);
    $pipeline->setExecutionContext(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->markAsFailed(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });

    $job = $this->prophesize(FlowDropJobInterface::class);
    $job->id()->willReturn("job_0");
    $job->getNodeId()->willReturn($node["id"]);
    $job->getMetadata()->willReturn([
      "config" => $node["config"],
      "node_type_id" => $node["type"],
      "incoming_edges" => [],
    ]);
    $job->markAsStarted()->will(function () use ($job) {
      return $job->reveal();
    });
    $job->setInputData(Argument::any())->will(function () use ($job) {
      return $job->reveal();
    });
    $job->setMetadata(Argument::any())->will(function () use ($job) {
      return $job->reveal();
    });
    $job->save()->will(function () use ($job) {
      return $job->reveal();
    });
    $job->markAsCompleted(Argument::any())->will(function () use ($job, $tracker) {
      $tracker->completed = TRUE;
      return $job->reveal();
    });
    $job->markAsFailed(Argument::any())->will(function ($args) use ($job, $tracker) {
      $tracker->failed = TRUE;
      $tracker->failureMessage = $args[0];
      return $job->reveal();
    });

    $jobs = [$job->reveal()];
    $pipeline->getJobs()->willReturn($jobs);
    $pipeline->getJobsByStatus(Argument::type("string"))->willReturn([]);
    $pipeline->hasInterruptedJobs()->willReturn(FALSE);

    $this->pipelineStorage->create(Argument::any())->willReturn($pipeline->reveal());
    $this->pipelineStorage->load(Argument::any())->willReturn($pipeline->reveal());

    $this->jobGenerationService->generateJobs(Argument::any(), Argument::any())->shouldBeCalled();
    $returned = FALSE;
    $this->jobGenerationService->getReadyJobs(Argument::any())->will(function () use ($jobs, &$returned) {
      if (!$returned) {
        $returned = TRUE;
        return $jobs;
      }
      return [];
    });

    $this->nodeRuntime->executeNode(
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("array"),
      Argument::type("array"),
      Argument::type(NodeExecutionContext::class),
      Argument::cetera()
    )->will(function ($args) use ($errorMessage) {
      $nodeId = $args[1];
      $output = new Output();
      $output->setError($errorMessage);

      return new NodeExecutionResult(
        nodeId: $nodeId,
        status: "error",
        output: $output,
        executionTime: 0.01,
        metadata: [],
      );
    });
  }

  /**
   * Sets up a single failing node that has an outgoing error edge.
   *
   * Mirrors setupMocksForFailingSingleNode() but the job carries an outgoing
   * error edge in its metadata, so the orchestrator routes the failure
   * (FAILED + error_routed, error payload on output) instead of failing the
   * run. getReadyJobs is drained via the tracker object — a `use (&$scalar)`
   * closure does NOT persist its mutation across Prophecy promise calls, so it
   * would loop to max-iterations and the run would PAUSE, not complete.
   *
   * @param array<string, mixed> $workflow
   *   The workflow definition.
   * @param \stdClass $tracker
   *   Records ->completed, ->failed, ->routed, ->outputData, and (internal)
   *   ->served drain flag.
   * @param string $errorMessage
   *   The error message the node reports.
   */
  private function setupMocksForRoutedFailingNode(array $workflow, \stdClass $tracker, string $errorMessage): void {
    $node = $workflow["nodes"][0];
    $tracker->served = FALSE;

    $pipeline = $this->prophesize(FlowDropPipelineInterface::class);
    $workflowEntity = $this->prophesize(FlowDropWorkflowInterface::class);
    $workflowEntity->id()->willReturn($workflow["id"]);
    $workflowEntity->getGraphData()->willReturn($workflow);

    $pipeline->getWorkflow()->willReturn($workflowEntity->reveal());
    $pipeline->id()->willReturn("pipeline_routing");
    $pipeline->getInputData()->willReturn([]);
    $pipeline->isReadyToStart()->willReturn(TRUE);
    $pipeline->isPaused()->willReturn(FALSE);
    $pipeline->markAsStarted()->will(fn() => $pipeline->reveal());
    $pipeline->markAsCompleted(Argument::any())->will(fn() => $pipeline->reveal());
    $pipeline->markAsFailed(Argument::any())->will(fn() => $pipeline->reveal());
    $pipeline->hasFailedJobs()->willReturn(FALSE);
    $pipeline->hasInterruptedJobs()->willReturn(FALSE);
    $pipeline->pause()->will(fn() => $pipeline->reveal());
    $pipeline->save()->will(fn() => $pipeline->reveal());
    $pipeline->getExecutionContext()->willReturn([]);
    $pipeline->setExecutionContext(Argument::any())->will(fn() => $pipeline->reveal());

    // The error edge lives on the job's outgoing edges; nodeHasErrorEdge reads
    // it from here. The workflow graph stays edge-free so the real edge
    // evaluator schedules nothing else.
    $job = $this->prophesize(FlowDropJobInterface::class);
    $job->id()->willReturn("job_0");
    $job->getNodeId()->willReturn($node["id"]);
    $job->getMetadata()->willReturn([
      "config" => $node["config"],
      "node_type_id" => $node["type"],
      "incoming_edges" => [],
      "outgoing_edges" => [
        [
          "source" => $node["id"],
          "source_handle" => $node["id"] . "-output-error",
          "target" => "handler",
          "target_handle" => "handler-input-input",
        ],
      ],
    ]);
    $job->markAsStarted()->will(fn() => $job->reveal());
    $job->setInputData(Argument::any())->will(fn() => $job->reveal());
    $job->setOutputData(Argument::any())->will(function ($args) use ($job, $tracker) {
      $tracker->outputData = $args[0];
      return $job->reveal();
    });
    $job->setMetadata(Argument::any())->will(function ($args) use ($job, $tracker) {
      if (($args[0]["error_routed"] ?? FALSE) === TRUE) {
        $tracker->routed = TRUE;
      }
      return $job->reveal();
    });
    $job->save()->will(fn() => $job->reveal());
    $job->markAsCompleted(Argument::any())->will(function () use ($job, $tracker) {
      $tracker->completed = TRUE;
      return $job->reveal();
    });
    $job->markAsFailed(Argument::any())->will(function () use ($job, $tracker) {
      $tracker->failed = TRUE;
      return $job->reveal();
    });

    $jobs = [$job->reveal()];
    $pipeline->getJobs()->willReturn($jobs);
    $pipeline->getJobsByStatus(Argument::type("string"))->willReturn([]);

    $this->pipelineStorage->create(Argument::any())->willReturn($pipeline->reveal());
    $this->pipelineStorage->load(Argument::any())->willReturn($pipeline->reveal());

    $this->jobGenerationService->generateJobs(Argument::any(), Argument::any())->shouldBeCalled();
    $this->jobGenerationService->getReadyJobs(Argument::any())->will(function () use ($jobs, $tracker) {
      if (!$tracker->served) {
        $tracker->served = TRUE;
        return $jobs;
      }
      return [];
    });

    $this->nodeRuntime->executeNode(
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("array"),
      Argument::type("array"),
      Argument::type(NodeExecutionContext::class),
      Argument::cetera()
    )->will(function ($args) use ($errorMessage) {
      $output = new Output();
      $output->setError($errorMessage);

      return new NodeExecutionResult(
        nodeId: $args[1],
        status: "error",
        output: $output,
        executionTime: 0.01,
        metadata: [],
      );
    });
  }

  /**
   * Sets up a single node whose runtime can fail, retry, and/or succeed.
   *
   * @param array<string, mixed> $workflow
   *   The workflow definition.
   * @param \stdClass $t
   *   Tracker: ->exec (executeNode calls), ->retries, ->completed, ->failed.
   * @param int $maxRetries
   *   The job's retry budget.
   * @param int $succeedOnAttempt
   *   The 1-based attempt on which the node succeeds; use a large number to
   *   never succeed.
   * @param bool $retryable
   *   Whether the error output carries the error_retryable marker.
   */
  private function setupMocksForRetryNode(array $workflow, \stdClass $t, int $maxRetries, int $succeedOnAttempt, bool $retryable): void {
    $node = $workflow["nodes"][0];

    $pipeline = $this->prophesize(FlowDropPipelineInterface::class);
    $workflowEntity = $this->prophesize(FlowDropWorkflowInterface::class);
    $workflowEntity->id()->willReturn($workflow["id"]);
    $workflowEntity->getGraphData()->willReturn($workflow);

    $pipeline->getWorkflow()->willReturn($workflowEntity->reveal());
    $pipeline->id()->willReturn("pipeline_retry");
    $pipeline->getInputData()->willReturn([]);
    $pipeline->isReadyToStart()->willReturn(TRUE);
    $pipeline->isPaused()->willReturn(FALSE);
    $pipeline->markAsStarted()->will(fn() => $pipeline->reveal());
    $pipeline->markAsCompleted(Argument::any())->will(fn() => $pipeline->reveal());
    $pipeline->markAsFailed(Argument::any())->will(fn() => $pipeline->reveal());
    $pipeline->hasFailedJobs()->willReturn(FALSE);
    $pipeline->pause()->will(fn() => $pipeline->reveal());
    $pipeline->save()->will(fn() => $pipeline->reveal());
    $pipeline->getExecutionContext()->willReturn([]);
    $pipeline->setExecutionContext(Argument::any())->will(fn() => $pipeline->reveal());

    $job = $this->prophesize(FlowDropJobInterface::class);
    $job->id()->willReturn("job_0");
    $job->getNodeId()->willReturn($node["id"]);
    $job->getMetadata()->willReturn([
      "config" => $node["config"],
      "node_type_id" => $node["type"],
      "incoming_edges" => [],
    ]);
    $job->markAsStarted()->will(fn() => $job->reveal());
    $job->setInputData(Argument::any())->will(fn() => $job->reveal());
    $job->setMetadata(Argument::any())->will(fn() => $job->reveal());
    $job->save()->will(fn() => $job->reveal());
    $job->getMaxRetries()->willReturn($maxRetries);
    $job->getRetryCount()->will(fn() => $t->retries);
    $job->canRetry()->will(fn() => $t->retries < $maxRetries);
    $job->incrementRetryCount()->will(function () use ($job, $t) {
      $t->retries++;
      return $job->reveal();
    });
    $job->markAsCompleted(Argument::any())->will(function () use ($job, $t) {
      $t->completed = TRUE;
      return $job->reveal();
    });
    $job->markAsFailed(Argument::any())->will(function () use ($job, $t) {
      $t->failed = TRUE;
      return $job->reveal();
    });

    $jobs = [$job->reveal()];
    $pipeline->getJobs()->willReturn($jobs);
    $pipeline->getJobsByStatus(Argument::type("string"))->willReturn([]);
    $pipeline->hasInterruptedJobs()->willReturn(FALSE);

    $this->pipelineStorage->create(Argument::any())->willReturn($pipeline->reveal());
    $this->pipelineStorage->load(Argument::any())->willReturn($pipeline->reveal());

    $this->jobGenerationService->generateJobs(Argument::any(), Argument::any())->shouldBeCalled();
    // Drain via object state ($t), not a by-ref scalar: Prophecy promise
    // closures do not preserve `use (&$scalar)` mutations across calls, so a
    // by-ref flag would re-dispatch the job every iteration up to the
    // iteration cap. Object properties (captured by handle) do persist.
    $this->jobGenerationService->getReadyJobs(Argument::any())->will(function () use ($jobs, $t) {
      if (empty($t->dispatched)) {
        $t->dispatched = TRUE;
        return $jobs;
      }
      return [];
    });

    $this->nodeRuntime->executeNode(
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("array"),
      Argument::type("array"),
      Argument::type(NodeExecutionContext::class),
      Argument::cetera()
    )->will(function ($args) use ($t, $succeedOnAttempt, $retryable) {
      $t->exec++;
      $nodeId = $args[1];
      if ($t->exec >= $succeedOnAttempt) {
        return new NodeExecutionResult(
          nodeId: $nodeId,
          status: "success",
          output: new Output(["status" => "success", "result" => "ok"]),
          executionTime: 0.01,
          metadata: [],
        );
      }
      $output = new Output();
      $output->setError("transient failure");
      if ($retryable) {
        $output->addMetadata("error_retryable", TRUE);
      }
      return new NodeExecutionResult(
        nodeId: $nodeId,
        status: "error",
        output: $output,
        executionTime: 0.01,
        metadata: [],
      );
    });
  }

  /**
   * A retryable failure that clears on a later attempt completes the job.
   *
   * @covers ::executeNodeWithRetry
   */
  public function testRetryableNodeSucceedsOnRetry(): void {
    $workflow = [
      "id" => "retry_ok_workflow",
      "label" => "Retry Then Succeed",
      "nodes" => [["id" => "flaky", "type" => "processor", "config" => []]],
      "edges" => [],
    ];
    $t = (object) ["exec" => 0, "retries" => 0, "completed" => FALSE, "failed" => FALSE];
    // Budget 3 retries; succeed on the 2nd attempt (one failure, one retry).
    $this->setupMocksForRetryNode($workflow, $t, 3, 2, TRUE);

    $request = new OrchestrationRequest(
      workflowId: "retry_ok_workflow",
      pipelineId: "pipeline_retry",
      workflow: $workflow,
      initialData: [],
      options: ["stategraph" => ["checkpointer_type" => "memory"]],
    );

    $response = $this->orchestrator->orchestrate($request);

    $this->assertEquals(2, $t->exec, "Node should run twice: one failure + one retry.");
    $this->assertEquals(1, $t->retries, "Retry counter should reflect one retry.");
    $this->assertTrue($t->completed, "A successful retry must complete the job.");
    $this->assertFalse($t->failed, "Job must not be marked failed when a retry succeeds.");
    $this->assertEquals("completed", $response->getStatus());
  }

  /**
   * A retryable failure that never clears fails once the budget is exhausted.
   *
   * @covers ::executeNodeWithRetry
   */
  public function testRetryableNodeFailsAfterExhaustingBudget(): void {
    $workflow = [
      "id" => "retry_exhaust_workflow",
      "label" => "Retry Then Exhaust",
      "nodes" => [["id" => "doomed", "type" => "processor", "config" => []]],
      "edges" => [],
    ];
    $t = (object) ["exec" => 0, "retries" => 0, "completed" => FALSE, "failed" => FALSE];
    // Budget 2 retries; never succeeds → 1 initial + 2 retries = 3 runs.
    $this->setupMocksForRetryNode($workflow, $t, 2, 999, TRUE);

    $request = new OrchestrationRequest(
      workflowId: "retry_exhaust_workflow",
      pipelineId: "pipeline_retry",
      workflow: $workflow,
      initialData: [],
      options: ["stategraph" => ["checkpointer_type" => "memory"]],
    );

    $threw = FALSE;
    try {
      $this->orchestrator->orchestrate($request);
    }
    catch (OrchestrationException) {
      $threw = TRUE;
    }

    $this->assertEquals(3, $t->exec, "Node should run 1 + maxRetries(2) = 3 times.");
    $this->assertEquals(2, $t->retries, "Retry counter should reach the budget.");
    $this->assertTrue($t->failed, "Exhausted retries must fail the job.");
    $this->assertFalse($t->completed);
    $this->assertTrue($threw, "Exhausted retries must surface as a failed orchestration.");
  }

  /**
   * A non-retryable failure is never retried, even with budget available.
   *
   * @covers ::executeNodeWithRetry
   */
  public function testNonRetryableErrorIsNotRetried(): void {
    $workflow = [
      "id" => "no_retry_workflow",
      "label" => "Non-Retryable",
      "nodes" => [["id" => "hard_fail", "type" => "processor", "config" => []]],
      "edges" => [],
    ];
    $t = (object) ["exec" => 0, "retries" => 0, "completed" => FALSE, "failed" => FALSE];
    // Generous budget, but the error is NOT marked retryable.
    $this->setupMocksForRetryNode($workflow, $t, 5, 999, FALSE);

    $request = new OrchestrationRequest(
      workflowId: "no_retry_workflow",
      pipelineId: "pipeline_retry",
      workflow: $workflow,
      initialData: [],
      options: ["stategraph" => ["checkpointer_type" => "memory"]],
    );

    $threw = FALSE;
    try {
      $this->orchestrator->orchestrate($request);
    }
    catch (OrchestrationException) {
      $threw = TRUE;
    }

    $this->assertEquals(1, $t->exec, "A non-retryable error must run the node exactly once.");
    $this->assertEquals(0, $t->retries);
    $this->assertTrue($t->failed);
    $this->assertTrue($threw);
  }

  /**
   * Sets up mocks for a looping workflow.
   *
   * @param array<string, mixed> $workflow
   *   The workflow definition.
   */
  private function setupMocksForLoopingWorkflow(array $workflow): void {
    $node = $workflow["nodes"][0];

    // Setup pipeline mock.
    $pipeline = $this->prophesize(FlowDropPipelineInterface::class);
    $workflowEntity = $this->prophesize(FlowDropWorkflowInterface::class);
    $workflowEntity->id()->willReturn($workflow["id"]);
    $workflowEntity->getGraphData()->willReturn($workflow);

    $pipeline->getWorkflow()->willReturn($workflowEntity->reveal());
    $pipeline->id()->willReturn("pipeline_loop");
    $pipeline->getInputData()->willReturn([]);
    $pipeline->isReadyToStart()->willReturn(TRUE);
    $pipeline->isPaused()->willReturn(FALSE);
    // Fluent methods returning 'static' need explicit return values.
    $pipeline->markAsStarted()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->markAsCompleted(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->hasFailedJobs()->willReturn(FALSE);
    $pipeline->pause()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->save()->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });
    $pipeline->getExecutionContext()->willReturn([]);
    $pipeline->setExecutionContext(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });

    // Create job that keeps returning as ready.
    $job = $this->prophesize(FlowDropJobInterface::class);
    $job->id()->willReturn("job_loop");
    $job->getNodeId()->willReturn($node["id"]);
    $job->getMetadata()->willReturn([
      "config" => $node["config"],
      "node_type_id" => $node["type"],
      "incoming_edges" => [],
    ]);
    // Fluent methods returning 'static' need explicit return values.
    $job->markAsStarted()->will(function () use ($job) {
      return $job->reveal();
    });
    $job->markAsCompleted(Argument::any())->will(function () use ($job) {
      return $job->reveal();
    });
    $job->setInputData(Argument::any())->will(function () use ($job) {
      return $job->reveal();
    });
    $job->setMetadata(Argument::any())->will(function () use ($job) {
      return $job->reveal();
    });
    $job->save()->will(function () use ($job) {
      return $job->reveal();
    });

    $jobs = [$job->reveal()];

    $pipeline->getJobs()->willReturn($jobs);
    $pipeline->getJobsByStatus(Argument::type("string"))->willReturn([]);
    $pipeline->hasInterruptedJobs()->willReturn(FALSE);
    $pipeline->markAsFailed(Argument::any())->will(function () use ($pipeline) {
      return $pipeline->reveal();
    });

    // Setup pipeline storage.
    $this->pipelineStorage->create(Argument::any())->willReturn($pipeline->reveal());
    $this->pipelineStorage->load(Argument::any())->willReturn($pipeline->reveal());

    $this->jobGenerationService->generateJobs(Argument::any(), Argument::any())->shouldBeCalled();
    $iteration = 0;
    // Return jobs for limited iterations then stop.
    $this->jobGenerationService->getReadyJobs(Argument::any())->will(function () use ($jobs, &$iteration) {
      if ($iteration < 3) {
        $iteration++;
        return $jobs;
      }
      return [];
    });

    // Setup node runtime.
    $this->nodeRuntime->executeNode(
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("string"),
      Argument::type("array"),
      Argument::type("array"),
      Argument::type(NodeExecutionContext::class),
      Argument::cetera()
    )->will(function ($args) {
      return new NodeExecutionResult(
        nodeId: $args[1],
        status: "success",
        output: new Output(["result" => "loop_iteration"]),
        executionTime: 0.01,
        metadata: [],
      );
    });
  }

}
