From 3cfa24854a2a7927963fc38d50d6481caa523fb6 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Sun, 12 Apr 2026 04:05:18 +0000 Subject: [PATCH 1/2] fix(concurrency): fix SubplanExecutionService._execute_parallel() #7582 Ensure fail_fast cancels in-flight futures and reports them as CANCELLED. Add Behave coverage that reproduces the concurrency regression. ISSUES CLOSED: #7582 --- features/subplan_execution.feature | 9 +++++++++ .../services/subplan_execution_service.py | 15 +++++++++++---- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/features/subplan_execution.feature b/features/subplan_execution.feature index 2c0ae3ea1..1de6bf5af 100644 --- a/features/subplan_execution.feature +++ b/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 diff --git a/src/cleveragents/application/services/subplan_execution_service.py b/src/cleveragents/application/services/subplan_execution_service.py index 5f58eaab5..2032a31c5 100644 --- a/src/cleveragents/application/services/subplan_execution_service.py +++ b/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) -- 2.52.0 From c11b05b77373bca78cccb1e4852f3065c9251aad Mon Sep 17 00:00:00 2001 From: CleverThis Date: Sun, 12 Apr 2026 08:02:29 +0000 Subject: [PATCH 2/2] docs(changelog): add v3.3.0 changelog entry for #7582 fail_fast fix --- CHANGELOG.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index ade2d531a..7907767ad 100644 --- a/CHANGELOG.md +++ b/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. -- 2.52.0