CHANGELOG.md

@@ -163,6 +163,16 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   `sandbox_root=.cleveragents/sandbox/`, so LLM file output (`FILE:` blocks)
   is written to disk during the execute phase. (#4222)
 
+- **SubplanExecutionService fail_fast cancellation** (#7582): Fixed a race condition where
+  already-running parallel subplans were not cancelled when `fail_fast` fired. Previously,
+  `Future.cancel()` only prevented queued futures from starting but had no effect on
+  in-flight futures that completed after `stop_flag` was set — their `COMPLETE` results
+  were incorrectly included in the merge output. The fix adds a post-completion guard that
+  overrides any non-`ERRORED`/non-`CANCELLED` result to `CANCELLED` when `stop_flag` is
+  active, and clears the associated output to prevent it from entering the merge. Also
+  replaces the O(n) linear `status` lookup in the `as_completed()` loop with an O(1)
+  `status_map` dict pre-computed before the executor block.
+
 - **Robot Framework TDD Listener Guards** (#5436): Added three guard conditions to the
   `tdd_expected_fail_listener` `end_test()` function to prevent blindly inverting ALL test
   failures to passes, which was masking infrastructure errors and causing flaky CI behavior.

features/subplan_execution.feature

@@ -258,6 +258,15 @@ Feature: Subplan Execution and Merge
     Then the first subplan should be errored
     And the remaining subplans should have CANCELLED status
 
+  @parallel @cancel_status
+  Scenario: Parallel fail_fast marks in-flight futures as CANCELLED
+    Given a parent plan with 3 subplans in parallel mode with fail_fast
+    And the subplan executor will block for 1 seconds
+    And the first subplan will fail with "ValidationError: schema mismatch"
+    When the subplans are executed
+    Then the first subplan should be errored
+    And the remaining subplans should have CANCELLED status
+
   # --- Dependency-ordered concurrent execution ---
 
   @dependency_ordered @concurrent

src/cleveragents/application/services/subplan_execution_service.py

@@ -301,6 +301,7 @@ class SubplanExecutionService:
         completion_order: list[str] = []
         stop_flag = False
         timeout = self._config.timeout_per_subplan_seconds
+        status_map = {status.subplan_id: status for status in statuses}
 
         with ThreadPoolExecutor(max_workers=max_workers) as pool:
             future_to_id: dict[Future[tuple[SubplanStatus, dict[str, str]]], str] = {}
@@ -315,20 +316,26 @@ class SubplanExecutionService:
 
             for future in as_completed(future_to_id):
                 subplan_id = future_to_id[future]
+                original_status = status_map[subplan_id]
                 try:
                     result_status, output = future.result()
                 except CancelledError:
-                    result_status = self._cancel_status(
-                        next(s for s in statuses if s.subplan_id == subplan_id)
-                    )
+                    result_status = self._cancel_status(original_status)
                     output = {}
                 except Exception as exc:  # pragma: no cover - defensive
                     result_status = self._error_status(
-                        next(s for s in statuses if s.subplan_id == subplan_id),
+                        original_status,
                         str(exc),
                     )
                     output = {}
 
+                if stop_flag and result_status.status not in (
+                    ProcessingState.ERRORED,
+                    ProcessingState.CANCELLED,
+                ):
+                    result_status = self._cancel_status(original_status)
+                    output = {}
+
                 results_map[subplan_id] = (result_status, output)
                 completion_order.append(subplan_id)