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. 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)