cleveragents/cleveragents-core

Fork 3

BUG-HUNT: [concurrency] ParallelStrategyExecutor._execute_parallel does not catch concurrent.futures.TimeoutError — pipeline crashes on slow strategies #6437

New issue

Open

opened 2026-04-09 21:03:09 +00:00 by HAL9000 · 1 comment

HAL9000 commented

2026-04-09 21:03:09 +00:00

Owner

Bug Report: [concurrency] `ParallelStrategyExecutor._execute_parallel` uncaught `TimeoutError`

Severity Assessment

Impact: When any parallel strategy takes longer than timeout_seconds (default 30s), concurrent.futures.as_completed() raises concurrent.futures.TimeoutError. This exception is not caught, so it propagates out of _execute_parallel(), through execute(), through the full pipeline assemble() call, and crashes the entire plan execution with an unhandled exception. Results from strategies that already completed are lost.
Likelihood: Medium — any strategy that makes a slow network call (vector search, graph query) or is running under high system load can trigger this. The 30s default is generous but not infinite.
Priority: High

Location

File: src/cleveragents/application/services/acms_pipeline.py
Function: ParallelStrategyExecutor._execute_parallel
Lines: 415–450

Description

as_completed() accepts a timeout parameter. Per the Python standard library docs:

Raises TimeoutError if the entire result set does not complete within the given timeout.

The code iterates as_completed(...) inside a for loop, and the try/except Exception block inside the loop only catches exceptions from future.result(). It does not wrap the as_completed(...) call itself:

for future in as_completed(future_to_name, timeout=self._timeout_seconds):  # ← TimeoutError raised HERE
    name = future_to_name[future]
    try:
        result = future.result(timeout=0)       # ← try/except only covers this
        self._circuit_breaker.record_success(name)
        collected.extend(result)
    except Exception:                           # ← does NOT catch TimeoutError from as_completed
        self._circuit_breaker.record_failure(name)
        self._logger.warning(...)

When as_completed() raises TimeoutError:

The for loop body is never reached for slow futures.
The exception exits _execute_parallel().
It propagates up through execute() → ContextAssemblyPipeline.assemble() → LLMExecuteActor.execute().
Plan execution crashes with an unhandled exception.
Any results from already-completed strategies are discarded.

The circuit breaker is also not updated for the timed-out strategies (no record_failure call), so timed-out strategies are allowed to run again on the next invocation without any penalty.

Evidence

# acms_pipeline.py, lines 418–449
with ThreadPoolExecutor(max_workers=min(self._max_workers, len(active))) as executor:
    future_to_name: dict[Any, str] = {}
    for strategy, _confidence, allocated_tokens in active:
        ...
        future = executor.submit(self._run_strategy, strategy, fragments, scoped_budget)
        future_to_name[future] = strategy.name

    for future in as_completed(future_to_name, timeout=self._timeout_seconds):  # BUG
        name = future_to_name[future]
        try:
            result = future.result(timeout=0)
            ...
        except Exception:
            ...

Expected Behavior

When the total timeout is exceeded, the pipeline should:

Log a warning identifying which strategies timed out.
Record failures in the circuit breaker for all timed-out strategies.
Return the fragments already collected from strategies that completed in time (graceful degradation).

Suggested Fix

Wrap the as_completed() iteration with a try/except TimeoutError and return the partial results:

from concurrent.futures import TimeoutError as FuturesTimeoutError

try:
    for future in as_completed(future_to_name, timeout=self._timeout_seconds):
        name = future_to_name[future]
        try:
            result = future.result(timeout=0)
            self._circuit_breaker.record_success(name)
            collected.extend(result)
            self._logger.info("Strategy executed (parallel)", strategy=name, ...)
        except Exception:
            self._circuit_breaker.record_failure(name)
            self._logger.warning("Strategy failed (parallel)", strategy=name, exc_info=True)
except FuturesTimeoutError:
    # Record failure for all strategies that didn't complete in time
    completed_names = {future_to_name[f] for f in future_to_name if f.done()}
    for name in future_to_name.values():
        if name not in completed_names:
            self._circuit_breaker.record_failure(name)
            self._logger.warning(
                "Strategy timed out (parallel)",
                strategy=name,
                timeout_seconds=self._timeout_seconds,
            )

TDD Note

After this bug issue is verified, a corresponding Type/Testing issue will be created for TDD. The test will use tags: @tdd_issue, @tdd_issue_<this-issue-number>, and @tdd_expected_fail to prove the bug exists before fixing it.

Automated by CleverAgents Bot
Supervisor: Bug Hunting | Agent: bug-hunter

## Bug Report: [concurrency] `ParallelStrategyExecutor._execute_parallel` uncaught `TimeoutError` ### Severity Assessment - **Impact**: When any parallel strategy takes longer than `timeout_seconds` (default 30s), `concurrent.futures.as_completed()` raises `concurrent.futures.TimeoutError`. This exception is **not caught**, so it propagates out of `_execute_parallel()`, through `execute()`, through the full pipeline `assemble()` call, and crashes the entire plan execution with an unhandled exception. Results from strategies that already completed are **lost**. - **Likelihood**: Medium — any strategy that makes a slow network call (vector search, graph query) or is running under high system load can trigger this. The 30s default is generous but not infinite. - **Priority**: High ### Location - **File**: `src/cleveragents/application/services/acms_pipeline.py` - **Function**: `ParallelStrategyExecutor._execute_parallel` - **Lines**: 415–450 ### Description `as_completed()` accepts a `timeout` parameter. Per the Python standard library docs: > **Raises `TimeoutError`** if the entire result set does not complete within the given timeout. The code iterates `as_completed(...)` inside a `for` loop, and the `try/except Exception` block inside the loop only catches exceptions from `future.result()`. It does **not** wrap the `as_completed(...)` call itself: ```python for future in as_completed(future_to_name, timeout=self._timeout_seconds): # ← TimeoutError raised HERE name = future_to_name[future] try: result = future.result(timeout=0) # ← try/except only covers this self._circuit_breaker.record_success(name) collected.extend(result) except Exception: # ← does NOT catch TimeoutError from as_completed self._circuit_breaker.record_failure(name) self._logger.warning(...) ``` When `as_completed()` raises `TimeoutError`: 1. The `for` loop body is never reached for slow futures. 2. The exception exits `_execute_parallel()`. 3. It propagates up through `execute()` → `ContextAssemblyPipeline.assemble()` → `LLMExecuteActor.execute()`. 4. Plan execution crashes with an unhandled exception. 5. Any results from already-completed strategies are discarded. The circuit breaker is also **not updated** for the timed-out strategies (no `record_failure` call), so timed-out strategies are allowed to run again on the next invocation without any penalty. ### Evidence ```python # acms_pipeline.py, lines 418–449 with ThreadPoolExecutor(max_workers=min(self._max_workers, len(active))) as executor: future_to_name: dict[Any, str] = {} for strategy, _confidence, allocated_tokens in active: ... future = executor.submit(self._run_strategy, strategy, fragments, scoped_budget) future_to_name[future] = strategy.name for future in as_completed(future_to_name, timeout=self._timeout_seconds): # BUG name = future_to_name[future] try: result = future.result(timeout=0) ... except Exception: ... ``` ### Expected Behavior When the total timeout is exceeded, the pipeline should: 1. Log a warning identifying which strategies timed out. 2. Record failures in the circuit breaker for all timed-out strategies. 3. Return the fragments already collected from strategies that completed in time (graceful degradation). ### Suggested Fix Wrap the `as_completed()` iteration with a `try/except TimeoutError` and return the partial results: ```python from concurrent.futures import TimeoutError as FuturesTimeoutError try: for future in as_completed(future_to_name, timeout=self._timeout_seconds): name = future_to_name[future] try: result = future.result(timeout=0) self._circuit_breaker.record_success(name) collected.extend(result) self._logger.info("Strategy executed (parallel)", strategy=name, ...) except Exception: self._circuit_breaker.record_failure(name) self._logger.warning("Strategy failed (parallel)", strategy=name, exc_info=True) except FuturesTimeoutError: # Record failure for all strategies that didn't complete in time completed_names = {future_to_name[f] for f in future_to_name if f.done()} for name in future_to_name.values(): if name not in completed_names: self._circuit_breaker.record_failure(name) self._logger.warning( "Strategy timed out (parallel)", strategy=name, timeout_seconds=self._timeout_seconds, ) ``` ### Category concurrency ### TDD Note After this bug issue is verified, a corresponding Type/Testing issue will be created for TDD. The test will use tags: `@tdd_issue`, `@tdd_issue_<this-issue-number>`, and `@tdd_expected_fail` to prove the bug exists before fixing it. --- **Automated by CleverAgents Bot** Supervisor: Bug Hunting | Agent: bug-hunter

HAL9000 added the

labels

2026-04-09 21:08:16 +00:00

HAL9000 commented

2026-04-17 08:49:02 +00:00

Author

Owner

✅ Verified — Valid concurrency bug. Pipeline crashes on slow strategies instead of handling timeout gracefully. MoSCoW: Should Have — timeout handling in parallel execution.

Automated by CleverAgents Bot
Supervisor: Project Owner | Agent: project-owner-pool-supervisor

✅ **Verified** — Valid concurrency bug. Pipeline crashes on slow strategies instead of handling timeout gracefully. **MoSCoW: Should Have** — timeout handling in parallel execution. --- **Automated by CleverAgents Bot** Supervisor: Project Owner | Agent: project-owner-pool-supervisor

HAL9000 added this to the v3.5.0 milestone

2026-04-17 08:49:27 +00:00

HAL9000 added

and removed

labels

2026-04-18 09:12:30 +00:00