WUM-2: bound auto_verified optimize runtime

[wuman001]

Summary

• short-circuit failed local candidate preflight gates before replay/evaluation
• add explicit CLI/runtime knobs for replay concurrency, judge concurrency, judge failure retries, wall-clock budget, and fast verified mode
• bound replay repetition concurrency, apply default evaluator timeouts, and surface replay/judge cost estimates in progress/report output

Fixes #931

Verification

• PYTHONPATH=.:aworld-cli/src pytest tests/self_evolve/test_runner.py tests/self_evolve/test_cli_trajectory_case_script.py tests/self_evolve/test_optimizer_contract.py
• PYTHONPATH=.:aworld-cli/src pytest tests/self_evolve/test_runner.py -k "replay or auto_verified or evaluator or population"
• PYTHONPATH=.:aworld-cli/src pytest tests/self_evolve

[gemini-code-assist]

Code Review

This pull request introduces bounded concurrency controls, failure retries, and wall-clock time budgeting to the self-evolve optimization process, along with preflight gate rejection optimizations and corresponding tests. The review feedback correctly identifies that the worst-case wall-clock time estimation formula in estimate_replay_wall_clock_seconds significantly underestimates execution times by assuming parallel execution for processes that actually run sequentially. The reviewer provides a corrected estimation formula along with the necessary updates to the runner call and test assertions.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

The current wall-clock time estimation logic in estimate_replay_wall_clock_seconds is highly inaccurate and significantly underestimates the worst-case execution time. This is because:

1. Sequential Candidate Evaluation: Candidates in the population are evaluated sequentially in a for loop in SelfEvolveRunner.run_explicit_target, not in parallel. Therefore, replays and judge calls cannot be parallelized across different candidates.
2. Sequential Baseline and Candidate Replays: Within a single candidate's replay, the baseline replays and candidate replays are executed sequentially (first baseline, then candidate), not in parallel.
3. Sequential Judge Repetitions: Within AWorldTrajectoryEvaluatorBackend, the judge repetitions (attempts) are executed sequentially in a for loop, not in parallel.
4. Batch Case Evaluation: AWorldTrajectoryEvaluatorBackend evaluates all cases in a single runner call, so the number of sequential evaluator calls does not scale with case_count in the way estimate.judge_call_count (which multiplies by case_count) suggests.

We should update the estimation formula to accurately reflect these sequential/parallel execution boundaries.

def estimate_replay_wall_clock_seconds(
    *,
    replayed_candidate_count: int,
    baseline_repetitions: int,
    candidate_repetitions: int,
    judge_repetitions: int,
    judge_failure_retries: int,
    replay_timeout_seconds: float | None,
    judge_timeout_seconds: float | None,
    replay_concurrency: int = 1,
    judge_concurrency: int = 1,
) -> float | None:
    if judge_failure_retries < 0:
        raise ValueError("judge_failure_retries must be non-negative")
    if replay_concurrency <= 0:
        raise ValueError("replay_concurrency must be positive")
    if judge_concurrency <= 0:
        raise ValueError("judge_concurrency must be positive")
    if replayed_candidate_count < 0:
        raise ValueError("replayed_candidate_count must be non-negative")

    replay_seconds = None
    if replay_timeout_seconds is not None:
        baseline_batches = (baseline_repetitions + replay_concurrency - 1) // replay_concurrency
        candidate_batches = (candidate_repetitions + replay_concurrency - 1) // replay_concurrency
        total_batches = baseline_batches + replayed_candidate_count * candidate_batches
        replay_seconds = total_batches * replay_timeout_seconds

    judge_seconds = None
    if judge_timeout_seconds is not None:
        eval_batches = 1 if judge_concurrency >= 2 else 2
        judge_attempts = judge_repetitions + judge_failure_retries
        judge_seconds = replayed_candidate_count * eval_batches * judge_attempts * judge_timeout_seconds

    if replay_seconds is None and judge_seconds is None:
        return None
    return (replay_seconds or 0.0) + (judge_seconds or 0.0)

[gemini-code-assist]

Update the call to estimate_replay_wall_clock_seconds to pass the correct parameters reflecting the sequential/parallel execution boundaries.

        replayed_candidate_count = (
            min(candidate_count, self.replay_candidate_limit)
            if self.replay_candidate_limit is not None
            else candidate_count
        )
        wall_clock_seconds = estimate_replay_wall_clock_seconds(
            replayed_candidate_count=replayed_candidate_count,
            baseline_repetitions=self.baseline_replay_repetitions,
            candidate_repetitions=self.candidate_replay_repetitions,
            judge_repetitions=self.judge_repetitions,
            judge_failure_retries=self.judge_failure_retries,
            replay_timeout_seconds=self.replay_timeout_seconds,
            judge_timeout_seconds=self.judge_timeout_seconds,
            replay_concurrency=self.replay_concurrency,
            judge_concurrency=self.judge_concurrency,
        )

[gemini-code-assist]

Update the expected worst-case wall-clock seconds assertion to match the corrected estimation formula.

Suggested change

	assert report["replay_cost"]["estimated_worst_case_wall_clock_seconds"] == 2700
	assert report["replay_cost"]["estimated_worst_case_wall_clock_seconds"] == 4200