From a049f4cbd3b3b42d14c568d3e5632fbb1dad0d51 Mon Sep 17 00:00:00 2001
From: Nicolas Joubert <njoubert@clever-age.com>
Date: Thu, 22 Oct 2026 17:10:36 +0200
Subject: [PATCH] #331 - ProcessManager: track the processed iterable and
 blocking tasks by task configuration instead of task code, and forget them
 when their process ends: a task of a sub-process with the same code as a task
 of the parent no longer makes the parent lose the output of its blocking task
 or skip the flush of its iteration. Add tests.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 CHANGELOG.md                                  |   3 +
 src/Manager/ProcessManager.php                |  22 +-
 .../Manager/ProcessManagerSubProcessTest.php  | 195 ++++++++++++++++++
 3 files changed, 214 insertions(+), 6 deletions(-)
 create mode 100644 tests/Manager/ProcessManagerSubProcessTest.php

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9b33c604..86dcd994 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,9 @@
 Latest
 ------
 
+## Fixes
+* [#331](https://github.com/cleverage/process-bundle/issues/331) `ProcessManager`: the iterable tasks that produced items and the blocking tasks waiting for `proceed()` are tracked by task configuration instead of task code, and forgotten when their process ends. A task of a sub-process (`process_executor`) with the same code as a task of the parent no longer makes the parent lose the output of its blocking task (without error) or skip the flush at the end of its iteration (merged batches), and a failed sub-process no longer leaves its tasks to its next execution. Add tests.
+
 v5.4.1
 -----
 
diff --git a/src/Manager/ProcessManager.php b/src/Manager/ProcessManager.php
index bf5e7d96..5fa96bf3 100644
--- a/src/Manager/ProcessManager.php
+++ b/src/Manager/ProcessManager.php
@@ -53,11 +53,15 @@ class ProcessManager implements ProcessManagerInterface, ResetInterface
     protected const EXECUTE_FLUSH = 4;
 
     /**
+     * Indexed by task configuration (spl_object_id()): a sub-process may have a task with the same code.
+     *
      * @var TaskConfiguration[]
      */
     protected array $processedIterables = [];
 
     /**
+     * Indexed by task configuration (spl_object_id()): a sub-process may have a task with the same code.
+     *
      * @var TaskConfiguration[]
      */
     protected array $processedBlockings = [];
@@ -220,6 +224,12 @@ protected function doExecute(string $processCode, mixed $input = null, array $co
             $this->endProcess($processHistory);
 
             throw $error;
+        } finally {
+            // A failed or stopped process may leave tasks in the registries, seen by its next execution
+            foreach ($processConfiguration->getTaskConfigurations() as $taskConfiguration) {
+                $this->removeProcessedIterable($taskConfiguration);
+                $this->removeProcessedBlocking($taskConfiguration);
+            }
         }
 
         $this->processHistory = $parentProcessHistory;
@@ -646,7 +656,7 @@ protected function checkProcess(ProcessConfiguration $processConfiguration): voi
      */
     protected function addProcessedIterable(TaskConfiguration $taskConfiguration): void
     {
-        $this->processedIterables[$taskConfiguration->getCode()] = $taskConfiguration;
+        $this->processedIterables[spl_object_id($taskConfiguration)] = $taskConfiguration;
     }
 
     /**
@@ -654,7 +664,7 @@ protected function addProcessedIterable(TaskConfiguration $taskConfiguration): v
      */
     protected function hasProcessedIterable(TaskConfiguration $taskConfiguration): bool
     {
-        return \array_key_exists($taskConfiguration->getCode(), $this->processedIterables);
+        return \array_key_exists(spl_object_id($taskConfiguration), $this->processedIterables);
     }
 
     /**
@@ -662,7 +672,7 @@ protected function hasProcessedIterable(TaskConfiguration $taskConfiguration): b
      */
     protected function removeProcessedIterable(TaskConfiguration $taskConfiguration): void
     {
-        unset($this->processedIterables[$taskConfiguration->getCode()]);
+        unset($this->processedIterables[spl_object_id($taskConfiguration)]);
     }
 
     /**
@@ -670,7 +680,7 @@ protected function removeProcessedIterable(TaskConfiguration $taskConfiguration)
      */
     protected function addProcessedBlocking(TaskConfiguration $taskConfiguration): void
     {
-        $this->processedBlockings[$taskConfiguration->getCode()] = $taskConfiguration;
+        $this->processedBlockings[spl_object_id($taskConfiguration)] = $taskConfiguration;
     }
 
     /**
@@ -678,7 +688,7 @@ protected function addProcessedBlocking(TaskConfiguration $taskConfiguration): v
      */
     protected function hasProcessedBlocking(TaskConfiguration $taskConfiguration): bool
     {
-        return \array_key_exists($taskConfiguration->getCode(), $this->processedBlockings);
+        return \array_key_exists(spl_object_id($taskConfiguration), $this->processedBlockings);
     }
 
     /**
@@ -686,6 +696,6 @@ protected function hasProcessedBlocking(TaskConfiguration $taskConfiguration): b
      */
     protected function removeProcessedBlocking(TaskConfiguration $taskConfiguration): void
     {
-        unset($this->processedBlockings[$taskConfiguration->getCode()]);
+        unset($this->processedBlockings[spl_object_id($taskConfiguration)]);
     }
 }
diff --git a/tests/Manager/ProcessManagerSubProcessTest.php b/tests/Manager/ProcessManagerSubProcessTest.php
new file mode 100644
index 00000000..db97ab12
--- /dev/null
+++ b/tests/Manager/ProcessManagerSubProcessTest.php
@@ -0,0 +1,195 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the CleverAge/ProcessBundle package.
+ *
+ * Copyright (c) Clever-Age
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace CleverAge\ProcessBundle\Tests\Manager;
+
+use CleverAge\ProcessBundle\Configuration\ProcessConfiguration;
+use CleverAge\ProcessBundle\Configuration\TaskConfiguration;
+use CleverAge\ProcessBundle\Context\ContextualOptionResolver;
+use CleverAge\ProcessBundle\Event\ProcessEvent;
+use CleverAge\ProcessBundle\Logger\AbstractLogger;
+use CleverAge\ProcessBundle\Logger\ProcessLogger;
+use CleverAge\ProcessBundle\Logger\TaskLogger;
+use CleverAge\ProcessBundle\Manager\ProcessManager;
+use CleverAge\ProcessBundle\Model\AbstractConfigurableTask;
+use CleverAge\ProcessBundle\Model\ProcessHistory;
+use CleverAge\ProcessBundle\Model\ProcessState;
+use CleverAge\ProcessBundle\Model\TaskInterface;
+use CleverAge\ProcessBundle\Registry\ProcessConfigurationRegistry;
+use CleverAge\ProcessBundle\Registry\TaskRegistry;
+use CleverAge\ProcessBundle\Task\AbstractIterableOutputTask;
+use CleverAge\ProcessBundle\Task\AggregateIterableTask;
+use CleverAge\ProcessBundle\Task\ConstantIterableOutputTask;
+use CleverAge\ProcessBundle\Task\SimpleBatchTask;
+use PHPUnit\Framework\TestCase;
+use Psr\Log\NullLogger;
+use Symfony\Component\DependencyInjection\Container;
+use Symfony\Component\EventDispatcher\EventDispatcher;
+
+/**
+ * A sub-process sharing the process manager of its parent (as the process_executor task) must not change the state of
+ * the tasks of the parent, even when its tasks have the same codes.
+ */
+#[\PHPUnit\Framework\Attributes\CoversClass(ProcessManager::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessConfiguration::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(TaskRegistry::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(TaskConfiguration::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ContextualOptionResolver::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessEvent::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(AbstractLogger::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(AbstractConfigurableTask::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessHistory::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessState::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ProcessConfigurationRegistry::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(AbstractIterableOutputTask::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(ConstantIterableOutputTask::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(AggregateIterableTask::class)]
+#[\PHPUnit\Framework\Attributes\UsesClass(SimpleBatchTask::class)]
+class ProcessManagerSubProcessTest extends TestCase
+{
+    private ProcessManager $processManager;
+
+    /**
+     * @var list<mixed>
+     */
+    private array $recorded = [];
+
+    protected function setUp(): void
+    {
+        $container = new Container();
+        $container->set('test.parent_iterate', new ConstantIterableOutputTask());
+        $container->set('test.child_iterate', new ConstantIterableOutputTask());
+        $container->set('test.parent_aggregate', new AggregateIterableTask());
+        $container->set('test.child_aggregate', new AggregateIterableTask());
+        $container->set('test.batch', new SimpleBatchTask());
+        // Runs the given sub-process and forwards its input, as the process_executor task
+        $container->set('test.sub_process', new class($this) implements TaskInterface {
+            public function __construct(
+                private readonly ProcessManagerSubProcessTest $test,
+            ) {
+            }
+
+            public function execute(ProcessState $state): void
+            {
+                $this->test->executeSubProcess((string) $state->getTaskConfiguration()->getOption('process'));
+                $state->setOutput($state->getInput());
+            }
+        });
+        $container->set('test.record', new class($this) implements TaskInterface {
+            public function __construct(
+                private readonly ProcessManagerSubProcessTest $test,
+            ) {
+            }
+
+            public function execute(ProcessState $state): void
+            {
+                $this->test->record($state->getInput());
+            }
+        });
+
+        $registry = new ProcessConfigurationRegistry(
+            [
+                'test.blocking_parent' => $this->createProcessConfiguration([
+                    'iterate' => $this->createTaskConfiguration('@test.parent_iterate', ['output' => [1, 2, 3]], ['sub_process']),
+                    'sub_process' => $this->createTaskConfiguration('@test.sub_process', ['process' => 'test.blocking_child'], ['aggregate']),
+                    'aggregate' => $this->createTaskConfiguration('@test.parent_aggregate'),
+                ], 'aggregate'),
+                'test.blocking_child' => $this->createProcessConfiguration([
+                    'iterate' => $this->createTaskConfiguration('@test.child_iterate', ['output' => ['a']], ['aggregate']),
+                    'aggregate' => $this->createTaskConfiguration('@test.child_aggregate'),
+                ]),
+                'test.iterable_parent' => $this->createProcessConfiguration([
+                    'iterate' => $this->createTaskConfiguration('@test.parent_iterate', ['output' => [1, 2, 3, 4, 5]], ['sub_process']),
+                    'sub_process' => $this->createTaskConfiguration('@test.sub_process', ['process' => 'test.iterable_child'], ['batch']),
+                    'batch' => $this->createTaskConfiguration('@test.batch', ['batch_count' => 2], ['record']),
+                    'record' => $this->createTaskConfiguration('@test.record'),
+                ]),
+                'test.iterable_child' => $this->createProcessConfiguration([
+                    'iterate' => $this->createTaskConfiguration('@test.child_iterate', ['output' => ['a']]),
+                ]),
+            ],
+            'stop'
+        );
+
+        $this->processManager = new ProcessManager(
+            $container,
+            new ProcessLogger(new NullLogger()),
+            new TaskLogger(new NullLogger()),
+            $registry,
+            new ContextualOptionResolver(),
+            new EventDispatcher(),
+            taskRegistry: new TaskRegistry($container),
+        );
+    }
+
+    public function executeSubProcess(string $processCode): void
+    {
+        $this->processManager->execute($processCode);
+    }
+
+    public function record(mixed $value): void
+    {
+        $this->recorded[] = $value;
+    }
+
+    public function testABlockingTaskOfTheParentIsProceededAfterASubProcessWithATaskOfTheSameCode(): void
+    {
+        self::assertSame([1, 2, 3], $this->processManager->execute('test.blocking_parent'));
+    }
+
+    public function testAnIterableTaskOfTheParentIsFlushedAfterASubProcessWithATaskOfTheSameCode(): void
+    {
+        $this->processManager->execute('test.iterable_parent');
+
+        self::assertSame([[1, 2], [3, 4], [5]], $this->recorded);
+    }
+
+    /**
+     * @param array<string, array<string, mixed>> $tasks
+     *
+     * @return array<string, mixed>
+     */
+    private function createProcessConfiguration(array $tasks, ?string $endPoint = null): array
+    {
+        return [
+            'options' => [],
+            'entry_point' => null,
+            'end_point' => $endPoint,
+            'description' => '',
+            'help' => '',
+            'public' => true,
+            'tasks' => $tasks,
+        ];
+    }
+
+    /**
+     * @param array<string, mixed> $options
+     * @param list<string>         $outputs
+     *
+     * @return array<string, mixed>
+     */
+    private function createTaskConfiguration(string $service, array $options = [], array $outputs = []): array
+    {
+        return [
+            'service' => $service,
+            'options' => $options,
+            'description' => '',
+            'help' => '',
+            'outputs' => $outputs,
+            'errors' => [],
+            'error_outputs' => [],
+            'error_strategy' => null,
+            'log_level' => null,
+        ];
+    }
+}