fix: dispatch adaptive optimizations in parallel path + deadlock regression test

- Parallel path now checks if a successful candidate was previously
  refined (via path_to_root ancestry). If so, dispatches adaptive
  optimization instead of batch refinement — matching sequential behavior.
- Adds regression test: 6 candidates with pool_size=2 all pass, proving
  no deadlock occurs when passes exceed available slots.

Author: KRRT7

codeflash/languages/function_optimizer.py

@@ -1602,18 +1602,34 @@ def _evaluate_candidates_parallel(
                     )
                     eval_ctx.valid_optimizations.append(best_optimization)
 
-                    batch_refiner_candidates.append(
-                        AIServiceBatchRefinerCandidate(
-                            optimization_id=candidate.optimization_id,
-                            optimized_source_code=candidate.source_code.markdown,
-                            optimized_explanation=candidate.explanation,
-                            optimized_code_runtime=candidate_result.best_test_runtime,
-                            original_code_runtime=original_code_baseline.runtime,
-                            speedup=f"{int(perf_gain * 100)}%",
-                            optimized_line_profiler_results="",
-                        )
+                    current_tree_candidates = candidate_node.path_to_root()
+                    is_candidate_refined_before = any(
+                        c.source == OptimizedCandidateSource.REFINE for c in current_tree_candidates
                     )
 
+                    if is_candidate_refined_before:
+                        future_adaptive = self.call_adaptive_optimize(
+                            trace_id=self.get_trace_id(exp_type),
+                            original_source_code=code_context.read_writable_code.markdown,
+                            prev_candidates=current_tree_candidates,
+                            eval_ctx=eval_ctx,
+                            ai_service_client=ai_service_client,
+                        )
+                        if future_adaptive:
+                            self.future_adaptive_optimizations.append(future_adaptive)
+                    else:
+                        batch_refiner_candidates.append(
+                            AIServiceBatchRefinerCandidate(
+                                optimization_id=candidate.optimization_id,
+                                optimized_source_code=candidate.source_code.markdown,
+                                optimized_explanation=candidate.explanation,
+                                optimized_code_runtime=candidate_result.best_test_runtime,
+                                original_code_runtime=original_code_baseline.runtime,
+                                speedup=f"{int(perf_gain * 100)}%",
+                                optimized_line_profiler_results="",
+                            )
+                        )
+
             # Dispatch refinement immediately so CandidateProcessor sees it
             if batch_refiner_candidates:
                 self._dispatch_refinement(

tests/test_parallel_evaluator.py

@@ -354,3 +354,53 @@ async def _run() -> list:  # type: ignore[type-arg]
         assert benchmark_call_count == 3
         for _, result in results:
             assert is_successful(result)
+
+    def test_more_candidates_than_slots_no_deadlock(self, tmp_path: Path) -> None:
+        """Regression test: more passing candidates than pool slots must not deadlock."""
+        from codeflash.optimization.parallel_evaluator import _BehavioralPass
+
+        opt = self._make_optimizer_mock(tmp_path)
+        (tmp_path / "src").mkdir(parents=True)
+        (tmp_path / "src" / "module.py").write_text("def f(): pass", encoding="utf-8")
+
+        nodes = [self._make_candidate_node(f"cand_{i}") for i in range(6)]
+        evaluator = ParallelCandidateEvaluator(opt, pool_size=2)
+
+        repo_root = Path(__file__).resolve().parents[1]
+        mock_result = MagicMock()
+        mock_result.best_test_runtime = 2000
+        mock_behavior_results = MagicMock()
+
+        async def mock_behavioral(self_eval: object, *args: object, **kwargs: object) -> Success:  # type: ignore[type-arg]
+            return Success(
+                _BehavioralPass(
+                    candidate_index=0,
+                    perf_test_files=[],
+                    test_env={},
+                    pytest_cmd_list=[],
+                    behavior_test_results=mock_behavior_results,
+                )
+            )
+
+        async def mock_benchmark(self_eval: object, *args: object, **kwargs: object) -> Success:  # type: ignore[type-arg]
+            return Success(mock_result)
+
+        async def _run() -> list:  # type: ignore[type-arg]
+            with (
+                patch("codeflash.code_utils.worktree_pool.git_root_dir", return_value=repo_root),
+                patch.object(ParallelCandidateEvaluator, "_run_behavioral", mock_behavioral),
+                patch.object(ParallelCandidateEvaluator, "_benchmark_phase", mock_benchmark),
+            ):
+                return await evaluator.evaluate_candidates(
+                    candidates=[(n, i, None) for i, n in enumerate(nodes)],
+                    code_context=MagicMock(),
+                    original_code_baseline=MagicMock(),
+                    original_helper_code={},
+                    file_path_to_helper_classes={},
+                )
+
+        # If this deadlocks, the test will timeout
+        results = anyio.run(_run)
+        assert len(results) == 6
+        for _, result in results:
+            assert is_successful(result)