diff --git a/src/coder_eval/criteria/agent_judge.py b/src/coder_eval/criteria/agent_judge.py index 88b0b412..13ef97db 100644 --- a/src/coder_eval/criteria/agent_judge.py +++ b/src/coder_eval/criteria/agent_judge.py @@ -95,6 +95,11 @@ class AgentJudgeChecker(BaseCriterion[AgentJudgeCriterion]): """Checker for AgentJudgeCriterion — spawns a Claude Code SDK agent as the judge.""" criterion_type = "agent_judge" + # The async path awaits the sub-agent lifecycle directly on the event loop + # (SubAgentRunner.run_async) instead of pinning a thread for its whole + # duration — see ``_check_impl_async`` — so SuccessChecker gathers it + # concurrently with other judge-type criteria instead of serializing them. + supports_native_async = True def _check_impl( self, @@ -105,122 +110,52 @@ def _check_impl( turn_records: list[TurnRecord] | None = None, context: CheckContext | None = None, ) -> CriterionResult: - ctx = context or CheckContext() - route = ctx.route - reference_dir = ctx.reference_dir - - # Master enablement gate. Skipped criteria don't spawn the sub-agent and - # don't affect cost; weighted score includes them as 1.0 so they don't penalize. - # The route precondition below is intentionally NOT checked when skipped — - # disabled criteria should be free to declare in tasks where the route isn't set. - if not criterion.enabled: - return JudgeCriterionResult( - criterion_type=criterion.type, - description=criterion.description, - score=1.0, - details="(skipped: enabled=false)", - ) + setup = _prepare(criterion, sandbox, reference_code, turn_records, context) + if isinstance(setup, JudgeCriterionResult): + return setup + runner, user_msg, judge_ctx, scrub_secrets, capture = setup - assert sandbox.sandbox_dir is not None, "sandbox not initialized" - if route is None: - # Explicit raise (not assert) so `python -O` cannot strip this guard; - # `test_agent_judge_with_no_route_raises` locks in the contract message. - msg = ( - "agent_judge requires a route from SuccessChecker; the orchestrator " - + "must construct one (DirectRoute / BedrockRoute) and pass " - + "it via SuccessChecker(..., route=...)" + try: + turn = runner.run( + user_msg, + max_turns=criterion.max_turns, + turn_timeout=float(criterion.turn_timeout), ) - raise ValueError(msg) - - # ``criterion.files`` is optional: when set, the named paths are pre-attached - # to the prompt envelope (fast verdict, narrow tool surface); when empty, the - # judge inspects the sandbox copy via its tools. Both modes compose — the judge - # can ``Read`` anything else even when files are pre-attached. - judge_ctx = JudgeContextBuilder( - files=criterion.files, - include_reference=criterion.include_reference, - include_agent_output=criterion.include_agent_output, - include_tool_calls=criterion.include_tool_calls, - include_dialog=criterion.include_dialog, - max_dialog_chars=criterion.max_dialog_chars, - max_file_chars=criterion.max_file_chars, - ).build(sandbox, reference_code, turn_records) - - # Mount the reference directory only when the criterion opted into seeing it. - # When include_reference=False the judge MUST NOT see the grading material, so - # zero out reference_dir before passing to the runner regardless of what came in. - ref_dir_for_runner = reference_dir if criterion.include_reference else None - user_msg = _render_user_message( - criterion.prompt, + except (TurnTimeoutError, AgentCrashError) as e: + return _crash_result(criterion, e) + + return _build_result( + criterion, + turn, judge_ctx, - reference_dir_mounted=ref_dir_for_runner is not None, - ) - agent_config = _build_agent_config(criterion, system_prompt=_SYSTEM_PROMPT) - - capture = VerdictCapture() - server_name, server = build_submit_verdict_mcp_server(capture) - - runner = SubAgentRunner( - sandbox=sandbox, - agent_config=agent_config, - # Use the floor-enforced patterns from the built config, not the - # user's raw ``criterion.agent.ignore_patterns`` — _build_agent_config - # injects the required security entries (.claude / .mcp.json / - # _reference) unconditionally. - ignore_patterns=agent_config.ignore_patterns, - route=route, - reference_dir=ref_dir_for_runner, - extra_mcp_servers={server_name: server}, + scrub_secrets=scrub_secrets, + system_prompt=_SYSTEM_PROMPT, + user_msg=user_msg, capture=capture, ) + async def _check_impl_async( + self, + criterion: AgentJudgeCriterion, + sandbox: Sandbox, + reference_code: str | None = None, + *, + turn_records: list[TurnRecord] | None = None, + context: CheckContext | None = None, + ) -> CriterionResult: + setup = _prepare(criterion, sandbox, reference_code, turn_records, context) + if isinstance(setup, JudgeCriterionResult): + return setup + runner, user_msg, judge_ctx, scrub_secrets, capture = setup + try: - turn = runner.run( + turn = await runner.run_async( user_msg, max_turns=criterion.max_turns, turn_timeout=float(criterion.turn_timeout), ) except (TurnTimeoutError, AgentCrashError) as e: - # Two pre-output failure modes share one return path: - # - TurnTimeoutError: sub-agent's per-turn watchdog fired. - # - AgentCrashError: SDK/CLI emitted an is_error ResultMessage - # (e.g. Bedrock's ``error_during_execution / end_turn`` flake - # when the model returns an empty assistant turn). - # No turn was produced, so no token usage is attributable to the judge. - is_timeout = isinstance(e, TurnTimeoutError) - if is_timeout: - logger.warning("agent_judge: turn timeout after %ds", criterion.turn_timeout) - details = f"Judge agent timed out after {criterion.turn_timeout}s" - else: - logger.warning("agent_judge: sub-agent crashed: %s", str(e)[:200]) - details = f"Judge agent crashed before producing a verdict: {str(e)[:200]}" - - # No turn was produced, so there's no transcript to capture — but - # return a JudgeCriterionResult anyway so renderers / aggregators - # that switch on ``isinstance(cr, JudgeCriterionResult)`` see the - # uniform shape (findings=[], transcript=None) instead of having to - # special-case a base CriterionResult with criterion_type='agent_judge'. - return JudgeCriterionResult( - criterion_type=criterion.type, - description=criterion.description, - score=0.0, - details=details, - error=f"{e.__class__.__name__}: {e}", - findings=[], - transcript=None, - token_usage=None, - ) - - # Build the scrub set: reference_code (if any) plus every file's content - # under reference_dir (if any). Either is honored only when the criterion - # opted into seeing the reference; otherwise no scrub key is needed because - # the judge never saw the material. - scrub_secrets: list[str] = [] - if criterion.include_reference: - if reference_code: - scrub_secrets.append(reference_code) - if reference_dir is not None: - scrub_secrets.extend(collect_reference_secrets(reference_dir)) + return _crash_result(criterion, e) return _build_result( criterion, @@ -233,6 +168,137 @@ def _check_impl( ) +def _crash_result(criterion: AgentJudgeCriterion, e: TurnTimeoutError | AgentCrashError) -> JudgeCriterionResult: + """Shared pre-output failure mapping for both the sync and async check paths. + + Two pre-output failure modes share one return path: + - TurnTimeoutError: sub-agent's per-turn watchdog fired. + - AgentCrashError: SDK/CLI emitted an is_error ResultMessage + (e.g. Bedrock's ``error_during_execution / end_turn`` flake + when the model returns an empty assistant turn). + No turn was produced, so no token usage is attributable to the judge. + """ + is_timeout = isinstance(e, TurnTimeoutError) + if is_timeout: + logger.warning("agent_judge: turn timeout after %ds", criterion.turn_timeout) + details = f"Judge agent timed out after {criterion.turn_timeout}s" + else: + logger.warning("agent_judge: sub-agent crashed: %s", str(e)[:200]) + details = f"Judge agent crashed before producing a verdict: {str(e)[:200]}" + + # No turn was produced, so there's no transcript to capture — but + # return a JudgeCriterionResult anyway so renderers / aggregators + # that switch on ``isinstance(cr, JudgeCriterionResult)`` see the + # uniform shape (findings=[], transcript=None) instead of having to + # special-case a base CriterionResult with criterion_type='agent_judge'. + return JudgeCriterionResult( + criterion_type=criterion.type, + description=criterion.description, + score=0.0, + details=details, + error=f"{e.__class__.__name__}: {e}", + findings=[], + transcript=None, + token_usage=None, + ) + + +def _prepare( + criterion: AgentJudgeCriterion, + sandbox: Sandbox, + reference_code: str | None, + turn_records: list[TurnRecord] | None, + context: CheckContext | None, +) -> tuple[SubAgentRunner, str, JudgeContext, list[str], VerdictCapture] | JudgeCriterionResult: + """Shared pre-dispatch setup for both the sync and async check paths. + + Returns either ``(runner, user_msg, judge_ctx, scrub_secrets, capture)`` to + dispatch with, or a terminal ``JudgeCriterionResult`` when the criterion is + disabled (short-circuits before any sub-agent spawn). + """ + ctx = context or CheckContext() + route = ctx.route + reference_dir = ctx.reference_dir + + # Master enablement gate. Skipped criteria don't spawn the sub-agent and + # don't affect cost; weighted score includes them as 1.0 so they don't penalize. + # The route precondition below is intentionally NOT checked when skipped — + # disabled criteria should be free to declare in tasks where the route isn't set. + if not criterion.enabled: + return JudgeCriterionResult( + criterion_type=criterion.type, + description=criterion.description, + score=1.0, + details="(skipped: enabled=false)", + ) + + assert sandbox.sandbox_dir is not None, "sandbox not initialized" + if route is None: + # Explicit raise (not assert) so `python -O` cannot strip this guard; + # `test_agent_judge_with_no_route_raises` locks in the contract message. + msg = ( + "agent_judge requires a route from SuccessChecker; the orchestrator " + + "must construct one (DirectRoute / BedrockRoute) and pass " + + "it via SuccessChecker(..., route=...)" + ) + raise ValueError(msg) + + # ``criterion.files`` is optional: when set, the named paths are pre-attached + # to the prompt envelope (fast verdict, narrow tool surface); when empty, the + # judge inspects the sandbox copy via its tools. Both modes compose — the judge + # can ``Read`` anything else even when files are pre-attached. + judge_ctx = JudgeContextBuilder( + files=criterion.files, + include_reference=criterion.include_reference, + include_agent_output=criterion.include_agent_output, + include_tool_calls=criterion.include_tool_calls, + include_dialog=criterion.include_dialog, + max_dialog_chars=criterion.max_dialog_chars, + max_file_chars=criterion.max_file_chars, + ).build(sandbox, reference_code, turn_records) + + # Mount the reference directory only when the criterion opted into seeing it. + # When include_reference=False the judge MUST NOT see the grading material, so + # zero out reference_dir before passing to the runner regardless of what came in. + ref_dir_for_runner = reference_dir if criterion.include_reference else None + user_msg = _render_user_message( + criterion.prompt, + judge_ctx, + reference_dir_mounted=ref_dir_for_runner is not None, + ) + agent_config = _build_agent_config(criterion, system_prompt=_SYSTEM_PROMPT) + + capture = VerdictCapture() + server_name, server = build_submit_verdict_mcp_server(capture) + + runner = SubAgentRunner( + sandbox=sandbox, + agent_config=agent_config, + # Use the floor-enforced patterns from the built config, not the + # user's raw ``criterion.agent.ignore_patterns`` — _build_agent_config + # injects the required security entries (.claude / .mcp.json / + # _reference) unconditionally. + ignore_patterns=agent_config.ignore_patterns, + route=route, + reference_dir=ref_dir_for_runner, + extra_mcp_servers={server_name: server}, + capture=capture, + ) + + # Build the scrub set: reference_code (if any) plus every file's content + # under reference_dir (if any). Either is honored only when the criterion + # opted into seeing the reference; otherwise no scrub key is needed because + # the judge never saw the material. + scrub_secrets: list[str] = [] + if criterion.include_reference: + if reference_code: + scrub_secrets.append(reference_code) + if reference_dir is not None: + scrub_secrets.extend(collect_reference_secrets(reference_dir)) + + return runner, user_msg, judge_ctx, scrub_secrets, capture + + def _build_agent_config( criterion: AgentJudgeCriterion, *, diff --git a/src/coder_eval/criteria/base.py b/src/coder_eval/criteria/base.py index 6d6753c2..c40d503d 100644 --- a/src/coder_eval/criteria/base.py +++ b/src/coder_eval/criteria/base.py @@ -1,11 +1,12 @@ """Base criterion checker interface with error handling.""" +import asyncio import logging import os import statistics import traceback from abc import ABC, abstractmethod -from collections.abc import Callable +from collections.abc import Awaitable, Callable from dataclasses import dataclass from functools import wraps from typing import TYPE_CHECKING, Any, ClassVar, Literal @@ -98,6 +99,57 @@ def wrapper( return wrapper +def handle_criterion_errors_async( + func: Callable[..., Awaitable[CriterionResult]], +) -> Callable[..., Awaitable[CriterionResult]]: + """Async twin of :func:`handle_criterion_errors` for ``check_async``. + + Same contract: infra failures escalate, everything else is captured into a + failed ``CriterionResult`` instead of propagating. + """ + + @wraps(func) + async def wrapper( + self: Any, + criterion: BaseSuccessCriterion, + sandbox: "Sandbox", + reference_code: str | None = None, + turn_records: list["TurnRecord"] | None = None, + context: "CheckContext | None" = None, + ) -> CriterionResult: + try: + return await func( + self, + criterion, + sandbox, + reference_code, + turn_records=turn_records, + context=context, + ) + except JudgeInfrastructureError: + raise + except Exception as e: + exc_info = f"{e.__class__.__name__}: {e}" + tb = "" + if os.getenv("CODER_EVAL_DEBUG") == "1": + tb = "\n" + "".join(traceback.format_exc(limit=5)) + + criterion_type = criterion.type + logger.error( + f"Error in {self.__class__.__name__}.check_async() for criterion type '{criterion_type}': {exc_info}", + exc_info=True, + ) + return CriterionResult( + criterion_type=criterion_type, + description=criterion.description, + score=0.0, + details=f"Error during check: {exc_info}{tb}", + error=exc_info, + ) + + return wrapper + + class BaseCriterion[C: BaseSuccessCriterion](ABC): """Abstract base class for all criterion checkers. @@ -135,6 +187,15 @@ def _check_impl( # live_verdict; CE025 enforces that the two stay consistent. live_stop_polarities: ClassVar[frozenset[str]] = frozenset() + # Whether this checker's ``_check_impl_async`` does genuine async I/O + # (awaits a non-blocking client) instead of the base default, which just + # offloads the sync ``_check_impl`` to a worker thread. ``SuccessChecker`` + # uses this to decide batching: native-async criteria (llm_judge, + # agent_judge) are gathered concurrently on the event loop so their + # network waits overlap instead of serializing or pinning extra threads; + # everything else still runs together in a single ``to_thread`` slot. + supports_native_async: ClassVar[bool] = False + @handle_criterion_errors def check( self, @@ -202,6 +263,58 @@ def _check_impl( """ pass + @handle_criterion_errors_async + async def check_async( + self, + criterion: C, + sandbox: "Sandbox", + reference_code: str | None = None, + turn_records: list["TurnRecord"] | None = None, + context: "CheckContext | None" = None, + ) -> CriterionResult: + """Async entry point. This method is FINAL - subclasses must NOT override it. + + Implement ``_check_impl_async`` instead. Mirrors ``check()``/``_check_impl`` + exactly, so behavior and error handling are identical between the sync and + async paths — the only difference is what runs on the event loop. + """ + return await self._check_impl_async( + criterion, + sandbox, + reference_code, + turn_records=turn_records, + context=context, + ) + + async def _check_impl_async( + self, + criterion: C, + sandbox: "Sandbox", + reference_code: str | None = None, + *, + turn_records: list["TurnRecord"] | None = None, + context: "CheckContext | None" = None, + ) -> CriterionResult: + """Async counterpart of ``_check_impl``. + + Base default offloads the sync ``_check_impl`` to a worker thread via + ``asyncio.to_thread`` — correct for the CPU/file-bound checkers, which + have no reason to change. Criteria that make a genuine network call + (``llm_judge``, ``agent_judge``) override this with real async I/O + (an async HTTP client) so the call yields the event loop instead of + pinning a thread-pool thread for the wait, and set + ``supports_native_async = True`` so ``SuccessChecker`` gathers them + concurrently with other judge-type criteria instead of serializing them. + """ + return await asyncio.to_thread( + self._check_impl, + criterion, + sandbox, + reference_code, + turn_records=turn_records, + context=context, + ) + def live_verdict( self, criterion: C, diff --git a/src/coder_eval/criteria/llm_judge.py b/src/coder_eval/criteria/llm_judge.py index 73db9f4a..92bc74c4 100644 --- a/src/coder_eval/criteria/llm_judge.py +++ b/src/coder_eval/criteria/llm_judge.py @@ -4,8 +4,8 @@ from typing import TYPE_CHECKING from coder_eval.criteria.base import BaseCriterion, CheckContext, register_criterion -from coder_eval.evaluation.judge_anthropic import invoke_anthropic_judge -from coder_eval.evaluation.judge_bedrock import invoke_bedrock_judge +from coder_eval.evaluation.judge_anthropic import invoke_anthropic_judge, invoke_anthropic_judge_async +from coder_eval.evaluation.judge_bedrock import invoke_bedrock_judge, invoke_bedrock_judge_async from coder_eval.evaluation.judge_context import ( DIALOG_HEADER, JudgeContext, @@ -56,6 +56,10 @@ class LLMJudgeChecker(BaseCriterion[LLMJudgeCriterion]): """Checker for LLMJudgeCriterion — grades the task via an LLM rubric.""" criterion_type = "llm_judge" + # The async path makes a real (non-blocking) network call via AsyncAnthropic / + # httpx.AsyncClient — see ``_check_impl_async`` — so SuccessChecker gathers it + # concurrently with other judge-type criteria instead of serializing them. + supports_native_async = True def _check_impl( self, @@ -66,100 +70,156 @@ def _check_impl( turn_records: "list[TurnRecord] | None" = None, context: CheckContext | None = None, ) -> CriterionResult: - ctx = context or CheckContext() - route = ctx.route - - # Master enablement gate. Skipped criteria don't make an LLM call and don't - # affect cost; weighted score includes them as 1.0 so they don't penalize. - # Authors who want them excluded from weighted score should remove the - # criterion from the YAML or use experiment variants to override. - if not criterion.enabled: - return JudgeCriterionResult( - criterion_type=criterion.type, - description=criterion.description, - score=1.0, - details="(skipped: enabled=false)", - ) + setup = _prepare(criterion, sandbox, reference_code, turn_records, context) + if isinstance(setup, JudgeCriterionResult): + return setup + judge_ctx, user_msg, scrub_key, route = setup - judge_ctx = JudgeContextBuilder( - files=criterion.files, - include_reference=criterion.include_reference, - include_agent_output=criterion.include_agent_output, - include_tool_calls=criterion.include_tool_calls, - include_dialog=criterion.include_dialog, - max_dialog_chars=criterion.max_dialog_chars, - max_file_chars=criterion.max_file_chars, - ).build(sandbox, reference_code, turn_records) - - user_msg = _render_user_message(criterion.prompt, judge_ctx) - - # Transport-unconfigured arm needs to short-circuit BEFORE backend dispatch. - # Hit when the run uses the Direct backend with no ANTHROPIC_API_KEY (or no - # route at all). The Bedrock backend always has a usable judge transport. - if route is None or (isinstance(route, DirectRoute) and route.judge_transport is None): - logger.error("llm_judge unreachable: no usable judge transport for the current backend") - return JudgeCriterionResult( - criterion_type=criterion.type, - description=criterion.description, - score=0.0, - details="(judge transport unconfigured)", - error=( - "llm_judge needs the run to use a backend that can reach a judge model:\n" - " - Bedrock (--backend bedrock), or\n" - " - Anthropic direct with ANTHROPIC_API_KEY set.\n" - "Set one of the above, or remove/disable the llm_judge criterion." - ), - ) + verdict, parse_error, raw_verdict_text, judge_usage = _invoke_tool_channel( + criterion=criterion, + route=route, + system_msg=_SYSTEM_PROMPT, + user_msg=user_msg, + ) + return _build_result( + criterion, judge_ctx, user_msg, scrub_key, verdict, parse_error, raw_verdict_text, judge_usage + ) - scrub_key = reference_code if criterion.include_reference else None + async def _check_impl_async( + self, + criterion: LLMJudgeCriterion, + sandbox: "Sandbox", + reference_code: str | None = None, + *, + turn_records: "list[TurnRecord] | None" = None, + context: CheckContext | None = None, + ) -> CriterionResult: + setup = _prepare(criterion, sandbox, reference_code, turn_records, context) + if isinstance(setup, JudgeCriterionResult): + return setup + judge_ctx, user_msg, scrub_key, route = setup - # Attribute the judge's API call to ``JudgeCriterionResult.token_usage`` - # from the usage the backend reported in its response. - verdict, parse_error, raw_verdict_text, response_usage = _invoke_tool_channel( + verdict, parse_error, raw_verdict_text, judge_usage = await _invoke_tool_channel_async( criterion=criterion, route=route, system_msg=_SYSTEM_PROMPT, user_msg=user_msg, ) - judge_usage = response_usage - - # Sanitize any raw model text we persist to CriterionResult.details. A misbehaving - # model could echo the reference back in an unparseable response, so we scrub it. - scrubbed = scrub_reference(raw_verdict_text, scrub_key) - - def _maybe_transcript() -> JudgeTranscript | None: - if not criterion.capture_transcript: - return None - return build_judge_transcript( - raw_verdict=raw_verdict_text, - max_chars=criterion.max_transcript_chars, - judge_system_prompt=_SYSTEM_PROMPT, - judge_prompt=user_msg, - scrub_key=scrub_key, - ) + return _build_result( + criterion, judge_ctx, user_msg, scrub_key, verdict, parse_error, raw_verdict_text, judge_usage + ) - if parse_error is not None: - return JudgeCriterionResult( - criterion_type=criterion.type, - description=criterion.description, - score=0.0, - details=scrubbed[:500], - error=scrub_reference(parse_error, scrub_key), - transcript=_maybe_transcript(), - token_usage=judge_usage, - ) - assert verdict is not None # parser contract: verdict is set iff parse_error is None - details = format_details(verdict.score, verdict.rationale, judge_ctx.missing_files, judge_ctx.degraded_notes) +def _prepare( + criterion: LLMJudgeCriterion, + sandbox: "Sandbox", + reference_code: str | None, + turn_records: "list[TurnRecord] | None", + context: CheckContext | None, +) -> tuple[JudgeContext, str, str | None, "ApiRoute | None"] | JudgeCriterionResult: + """Shared pre-dispatch setup for both the sync and async check paths. + + Returns either the tuple to dispatch with, or a terminal + ``JudgeCriterionResult`` when the criterion is disabled or the route has no + usable judge transport (both short-circuit before any LLM call). + """ + ctx = context or CheckContext() + route = ctx.route + + # Master enablement gate. Skipped criteria don't make an LLM call and don't + # affect cost; weighted score includes them as 1.0 so they don't penalize. + # Authors who want them excluded from weighted score should remove the + # criterion from the YAML or use experiment variants to override. + if not criterion.enabled: + return JudgeCriterionResult( + criterion_type=criterion.type, + description=criterion.description, + score=1.0, + details="(skipped: enabled=false)", + ) + + judge_ctx = JudgeContextBuilder( + files=criterion.files, + include_reference=criterion.include_reference, + include_agent_output=criterion.include_agent_output, + include_tool_calls=criterion.include_tool_calls, + include_dialog=criterion.include_dialog, + max_dialog_chars=criterion.max_dialog_chars, + max_file_chars=criterion.max_file_chars, + ).build(sandbox, reference_code, turn_records) + + user_msg = _render_user_message(criterion.prompt, judge_ctx) + + # Transport-unconfigured arm needs to short-circuit BEFORE backend dispatch. + # Hit when the run uses the Direct backend with no ANTHROPIC_API_KEY (or no + # route at all). The Bedrock backend always has a usable judge transport. + if route is None or (isinstance(route, DirectRoute) and route.judge_transport is None): + logger.error("llm_judge unreachable: no usable judge transport for the current backend") return JudgeCriterionResult( criterion_type=criterion.type, description=criterion.description, - score=verdict.score, - details=scrub_reference(details, scrub_key), - findings=[scrub_reference(f, scrub_key) for f in verdict.findings], + score=0.0, + details="(judge transport unconfigured)", + error=( + "llm_judge needs the run to use a backend that can reach a judge model:\n" + " - Bedrock (--backend bedrock), or\n" + " - Anthropic direct with ANTHROPIC_API_KEY set.\n" + "Set one of the above, or remove/disable the llm_judge criterion." + ), + ) + + scrub_key = reference_code if criterion.include_reference else None + return judge_ctx, user_msg, scrub_key, route + + +def _build_result( + criterion: LLMJudgeCriterion, + judge_ctx: JudgeContext, + user_msg: str, + scrub_key: str | None, + verdict: JudgeVerdict | None, + parse_error: str | None, + raw_verdict_text: str, + judge_usage: TokenUsage | None, +) -> JudgeCriterionResult: + """Shared post-dispatch result assembly for both the sync and async check paths.""" + # Sanitize any raw model text we persist to CriterionResult.details. A misbehaving + # model could echo the reference back in an unparseable response, so we scrub it. + scrubbed = scrub_reference(raw_verdict_text, scrub_key) + + def _maybe_transcript() -> JudgeTranscript | None: + if not criterion.capture_transcript: + return None + return build_judge_transcript( + raw_verdict=raw_verdict_text, + max_chars=criterion.max_transcript_chars, + judge_system_prompt=_SYSTEM_PROMPT, + judge_prompt=user_msg, + scrub_key=scrub_key, + ) + + if parse_error is not None: + return JudgeCriterionResult( + criterion_type=criterion.type, + description=criterion.description, + score=0.0, + details=scrubbed[:500], + error=scrub_reference(parse_error, scrub_key), transcript=_maybe_transcript(), token_usage=judge_usage, ) + assert verdict is not None # parser contract: verdict is set iff parse_error is None + + details = format_details(verdict.score, verdict.rationale, judge_ctx.missing_files, judge_ctx.degraded_notes) + return JudgeCriterionResult( + criterion_type=criterion.type, + description=criterion.description, + score=verdict.score, + details=scrub_reference(details, scrub_key), + findings=[scrub_reference(f, scrub_key) for f in verdict.findings], + transcript=_maybe_transcript(), + token_usage=judge_usage, + ) def _invoke_tool_channel( @@ -213,6 +273,49 @@ def _invoke_tool_channel( return None, err, f"(no verdict — {err})", response_usage +async def _invoke_tool_channel_async( + *, + criterion: LLMJudgeCriterion, + route: "ApiRoute | None", + system_msg: str, + user_msg: str, +) -> tuple[JudgeVerdict | None, str | None, str, TokenUsage | None]: + """Async twin of ``_invoke_tool_channel`` — dispatches to the non-blocking invokers.""" + response_usage: TokenUsage | None + match route: + case BedrockRoute(): + response = await invoke_bedrock_judge_async( + route=route, + model=criterion.model, + system=system_msg, + user=user_msg, + temperature=criterion.temperature, + max_tokens=criterion.max_tokens, + tool_spec=SUBMIT_VERDICT_ANTHROPIC_TOOL, + ) + verdict, err = extract_verdict_from_anthropic_response(response) + response_usage = token_usage_from_anthropic_dict(response) + case DirectRoute(): + anthropic_response = await invoke_anthropic_judge_async( + model=criterion.model, + system=system_msg, + user=user_msg, + temperature=criterion.temperature, + max_tokens=criterion.max_tokens, + tool_spec=SUBMIT_VERDICT_ANTHROPIC_TOOL, + ) + verdict, err = extract_verdict_from_anthropic_response(anthropic_response) + response_usage = token_usage_from_anthropic_dict(anthropic_response) + case _: + # route is None or an unexpected type — the unconfigured-arm guard in + # _prepare handles None before dispatch, so this is defensive only. + return None, "llm_judge: no usable API route", "(no route)", None + + if verdict is not None: + return verdict, None, verdict.model_dump_json(), response_usage + return None, err, f"(no verdict — {err})", response_usage + + def _render_user_message(prompt: str, context: JudgeContext) -> str: """Render the user-facing prompt envelope for the LLM judge.""" reference_block = "" diff --git a/src/coder_eval/evaluation/checker.py b/src/coder_eval/evaluation/checker.py index 60515ca0..aabf928c 100644 --- a/src/coder_eval/evaluation/checker.py +++ b/src/coder_eval/evaluation/checker.py @@ -5,6 +5,7 @@ the criteria registry. """ +import asyncio import logging from pathlib import Path from typing import TYPE_CHECKING, Any @@ -161,11 +162,77 @@ def check_all( ref_code = reference_code if reference_code is not None else self._reference_code ref_dir = reference_dir if reference_dir is not None else self._reference_dir records = turn_records if turn_records is not None else self._turn_records - results = [] - for criterion in criteria: - result = self._check_single(criterion, ref_code, records, ref_dir) - results.append(result) - return results + return self._check_all_sync(criteria, ref_code, records, ref_dir) + + async def check_all_async( + self, + criteria: SuccessCriteria, + reference_code: str | None = None, + turn_records: TurnRecords | None = None, + reference_dir: Path | None = None, + ) -> CriteriaResults: + """Async twin of ``check_all`` — the orchestrator's entry point. + + Non-judge criteria (file/command/trajectory checks — CPU/file-bound, + no I/O wait) still run together in a single ``asyncio.to_thread`` slot, + same as before. Criteria whose checker declares + ``supports_native_async = True`` (``llm_judge``, ``agent_judge``) are + instead awaited directly on the event loop via ``check_async``, and all + of them run concurrently with each other AND with the sync batch — so a + task stacking two judge criteria fires both LLM calls at once instead of + serializing them, and neither pins a thread for the network wait. + + Results are returned in the same order as ``criteria``, regardless of + which path each one took. + """ + if reference_code is not None: + self._reference_code = reference_code + if reference_dir is not None: + self._reference_dir = reference_dir + if turn_records is not None: + self._turn_records = turn_records + ref_code = reference_code if reference_code is not None else self._reference_code + ref_dir = reference_dir if reference_dir is not None else self._reference_dir + records = turn_records if turn_records is not None else self._turn_records + + if not criteria: + return [] + + def _is_native_async(criterion_type: str) -> bool: + try: + return self._get_checker_instance(criterion_type).supports_native_async + except KeyError: + # Unregistered type — let the shared sync path's KeyError handling + # in ``_check_single`` produce the usual failed CriterionResult. + return False + + native_async_indices = [i for i, c in enumerate(criteria) if _is_native_async(c.type)] + sync_indices = [i for i in range(len(criteria)) if i not in native_async_indices] + + results: list[CriterionResult | None] = [None] * len(criteria) + + async def run_sync_batch() -> None: + if not sync_indices: + return + sync_criteria = [criteria[i] for i in sync_indices] + sync_results = await asyncio.to_thread(self._check_all_sync, sync_criteria, ref_code, records, ref_dir) + for idx, result in zip(sync_indices, sync_results, strict=True): + results[idx] = result + + async def run_async_one(i: int) -> None: + results[i] = await self._check_single_async(criteria[i], ref_code, records, ref_dir) + + await asyncio.gather(run_sync_batch(), *(run_async_one(i) for i in native_async_indices)) + return results # type: ignore[return-value] # every slot filled by the gather above + + def _check_all_sync( + self, + criteria: SuccessCriteria, + reference_code: str | None, + turn_records: TurnRecords | None, + reference_dir: Path | None, + ) -> CriteriaResults: + return [self._check_single(criterion, reference_code, turn_records, reference_dir) for criterion in criteria] def _get_checker_instance(self, criterion_type: str) -> BaseCriterion[Any]: """Get or create a checker instance (V3: cached). @@ -265,3 +332,74 @@ def _check_single( _short_failure_reason(failed), ) return failed + + async def _check_single_async( + self, + criterion: SuccessCriterion, + reference_code: str | None, + turn_records: TurnRecords | None = None, + reference_dir: Path | None = None, + ) -> CriterionResult: + """Async twin of ``_check_single`` — used only for ``supports_native_async`` + checkers, dispatched via ``check_async`` instead of ``check``. Logging and + error-shape parity with ``_check_single`` is intentional. + """ + criterion_type = criterion.type + + try: + checker = self._get_checker_instance(criterion_type) + context = CheckContext(route=self.route, reference_dir=reference_dir) + result = await checker.check_async( + criterion, + self.sandbox, + reference_code, + turn_records=turn_records, + context=context, + ) + result.pass_threshold = criterion.pass_threshold + result.gating = criterion.is_gating + + logger.debug(f"Criterion '{criterion_type}' score: {result.score:.2f}") + if result.score < criterion.pass_threshold: + logger.info( + "Criterion '%s' %s (score=%.2f, threshold=%.2f): %s", + criterion_type, + "FAILED" if criterion.is_gating else "below threshold (informational)", + result.score, + criterion.pass_threshold, + _short_failure_reason(result), + ) + return result + + except KeyError: + logger.error(f"No checker found for criterion type '{criterion_type}'") + return CriterionResult( + criterion_type=criterion_type, + description=criterion.description, + score=0.0, + details=f"No checker registered for criterion type '{criterion_type}'", + error=f"Unsupported criterion type: '{criterion_type}'", + pass_threshold=criterion.pass_threshold, + gating=criterion.is_gating, + ) + except JudgeInfrastructureError: + raise + except Exception as e: + logger.exception(f"Checker failure for criterion '{criterion_type}': {e}") + failed = CriterionResult( + criterion_type=criterion_type, + description=criterion.description, + score=0.0, + details="Error running checker", + error=str(e), + pass_threshold=criterion.pass_threshold, + gating=criterion.is_gating, + ) + logger.info( + "Criterion '%s' %s (score=0.00, threshold=%.2f): %s", + criterion_type, + "FAILED" if criterion.is_gating else "errored (informational)", + criterion.pass_threshold, + _short_failure_reason(failed), + ) + return failed diff --git a/src/coder_eval/evaluation/judge_anthropic.py b/src/coder_eval/evaluation/judge_anthropic.py index 287e9dfe..a60ef348 100644 --- a/src/coder_eval/evaluation/judge_anthropic.py +++ b/src/coder_eval/evaluation/judge_anthropic.py @@ -12,7 +12,7 @@ from typing import Any -from anthropic import Anthropic, APIError +from anthropic import Anthropic, APIError, AsyncAnthropic from coder_eval.errors import JudgeInfrastructureError from coder_eval.evaluation.judge_models import to_anthropic_alias @@ -53,3 +53,38 @@ def invoke_anthropic_judge( # by default) — do not add another retry loop here. raise JudgeInfrastructureError(f"Anthropic judge API error: {e}") from e return response.model_dump() + + +async def invoke_anthropic_judge_async( + *, + model: str, + system: str, + user: str, + temperature: float, + max_tokens: int, + tool_spec: dict[str, Any], + timeout_seconds: float = 120.0, +) -> dict[str, Any]: + """Async twin of :func:`invoke_anthropic_judge`. + + Uses ``AsyncAnthropic`` so the call yields the event loop instead of + blocking a thread-pool thread for the network wait — lets + ``SuccessChecker.check_all_async`` run several judge criteria concurrently + without pinning a thread per judge. + """ + alias = to_anthropic_alias(model) + client = AsyncAnthropic(timeout=timeout_seconds) + async with client: + try: + response = await client.messages.create( + model=alias, + system=system, + messages=[{"role": "user", "content": user}], + max_tokens=max_tokens, + temperature=temperature, + tools=[tool_spec], # type: ignore[arg-type] + tool_choice={"type": "tool", "name": tool_spec["name"]}, + ) + except APIError as e: + raise JudgeInfrastructureError(f"Anthropic judge API error: {e}") from e + return response.model_dump() diff --git a/src/coder_eval/evaluation/judge_bedrock.py b/src/coder_eval/evaluation/judge_bedrock.py index 6627534b..8c2f6001 100644 --- a/src/coder_eval/evaluation/judge_bedrock.py +++ b/src/coder_eval/evaluation/judge_bedrock.py @@ -16,6 +16,7 @@ from __future__ import annotations +import asyncio import logging import time from typing import Any @@ -112,3 +113,69 @@ def invoke_bedrock_judge( raise JudgeInfrastructureError(f"Bedrock response is not a JSON object: {str(data)[:500]}") return data raise JudgeInfrastructureError(f"{last_failure} (after {attempts} attempts)") from last_exc + + +async def invoke_bedrock_judge_async( + *, + route: BedrockRoute, + model: str, + system: str, + user: str, + temperature: float, + max_tokens: int, + tool_spec: dict[str, Any], + timeout_seconds: float = 120.0, +) -> dict[str, Any]: + """Async twin of :func:`invoke_bedrock_judge`. + + Uses ``httpx.AsyncClient`` + ``asyncio.sleep`` between retries so the call + yields the event loop instead of blocking a thread-pool thread for the + network wait — lets ``SuccessChecker.check_all_async`` run several judge + criteria concurrently without pinning a thread per judge. + """ + qualified = to_bedrock_model(model, route.region) + url = f"https://bedrock-runtime.{route.region}.amazonaws.com/model/{qualified}/invoke" + body = { + "anthropic_version": "bedrock-2023-05-31", + "max_tokens": max_tokens, + "temperature": temperature, + "system": system, + "messages": [{"role": "user", "content": user}], + "tools": [tool_spec], + "tool_choice": {"type": "tool", "name": tool_spec["name"]}, + } + headers = { + "Authorization": f"Bearer {route.bearer_token}", + "Content-Type": "application/json", + "Accept": "application/json", + } + + attempts = _JUDGE_RETRY.max_retries + 1 + last_failure = "" + last_exc: Exception | None = None + async with httpx.AsyncClient() as client: + for attempt in range(attempts): + if attempt: + await asyncio.sleep(compute_backoff(_JUDGE_RETRY, attempt - 1)) + try: + response = await client.post(url, headers=headers, json=body, timeout=timeout_seconds) + except httpx.HTTPError as e: + last_failure = f"Bedrock invoke transport error: {e}" + last_exc = e + logger.warning("Bedrock judge attempt %d/%d failed: %s", attempt + 1, attempts, last_failure) + continue + if _is_retryable_status(response.status_code): + last_failure = f"Bedrock invoke failed: {response.status_code} {response.text[:500]}" + last_exc = None + logger.warning("Bedrock judge attempt %d/%d failed: %s", attempt + 1, attempts, last_failure) + continue + if response.status_code >= 300: + raise JudgeInfrastructureError(f"Bedrock invoke failed: {response.status_code} {response.text[:500]}") + try: + data = response.json() + except ValueError as e: + raise JudgeInfrastructureError(f"Bedrock response is not valid JSON: {e}") from e + if not isinstance(data, dict): + raise JudgeInfrastructureError(f"Bedrock response is not a JSON object: {str(data)[:500]}") + return data + raise JudgeInfrastructureError(f"{last_failure} (after {attempts} attempts)") from last_exc diff --git a/src/coder_eval/evaluation/sub_agent.py b/src/coder_eval/evaluation/sub_agent.py index f90721ef..4f29e0d1 100644 --- a/src/coder_eval/evaluation/sub_agent.py +++ b/src/coder_eval/evaluation/sub_agent.py @@ -204,6 +204,71 @@ def run(self, user_msg: str, *, max_turns: int | None, turn_timeout: float) -> T finally: shutil.rmtree(judge_dir, ignore_errors=True) + async def run_async(self, user_msg: str, *, max_turns: int | None, turn_timeout: float) -> TurnRecord: + """Async twin of :meth:`run` — for callers that already own the event loop. + + Same lifecycle (copy sandbox → start agent → communicate → stop, kill on + any exception), but awaits ``_run_agent`` directly instead of bridging + through a fresh ``asyncio.run()``, and pushes the blocking filesystem + work (``copytree``/``rmtree``) to a worker thread so neither blocks the + loop. Lets a native-async criterion (``agent_judge``) run concurrently + with other judge-type criteria instead of pinning a thread for the + duration of the sub-agent's run. + """ + src_dir = self._sandbox.sandbox_dir + assert src_dir is not None, "sandbox not initialized" + + judge_dir = Path(await asyncio.to_thread(tempfile.mkdtemp, prefix="sub_agent_")) + try: + await asyncio.to_thread( + shutil.copytree, + src_dir, + judge_dir, + symlinks=True, + ignore=_ignore_patterns_and_symlinks(self._ignore_patterns), + dirs_exist_ok=True, + ) + + if self._reference_dir is not None: + ref_dest = judge_dir / "_reference" + if ref_dest.exists(): + await asyncio.to_thread(shutil.rmtree, ref_dest, ignore_errors=True) + await asyncio.to_thread( + shutil.copytree, + self._reference_dir, + ref_dest, + symlinks=True, + ignore=_ignore_patterns_and_symlinks(self._reference_ignore_patterns), + ) + + agent = ClaudeCodeAgent( + self._agent_config, + route=self._route, + extra_mcp_servers=self._extra_mcp_servers, + ) + logger.info( + "sub_agent: starting (model=%s, max_turns=%s, allowed_tools=%s)", + self._agent_config.model, + max_turns, + self._agent_config.allowed_tools, + ) + turn = await self._run_agent( + agent, + judge_dir, + user_msg, + max_turns, + turn_timeout, + plugin_tools_dir=self._sandbox.plugin_tools_dir, + ) + logger.info( + "sub_agent: finished (duration=%.1fs, tokens=%s)", + turn.duration_seconds, + turn.token_usage, + ) + return turn + finally: + await asyncio.to_thread(shutil.rmtree, judge_dir, ignore_errors=True) + @staticmethod async def _run_agent( agent: ClaudeCodeAgent, diff --git a/src/coder_eval/orchestrator.py b/src/coder_eval/orchestrator.py index b1581470..a6369e75 100644 --- a/src/coder_eval/orchestrator.py +++ b/src/coder_eval/orchestrator.py @@ -1346,8 +1346,7 @@ async def _evaluation_loop(self) -> bool: task_file=self.task_file, cached_reference=self._reference_code, ) - criteria_results = await asyncio.to_thread( - self.success_checker.check_all, + criteria_results = await self.success_checker.check_all_async( self.task.success_criteria, reference_code=reference_code, reference_dir=reference_dir, @@ -1405,8 +1404,7 @@ async def _evaluation_loop(self) -> bool: task_file=self.task_file, cached_reference=self._reference_code, ) - criteria_results = await asyncio.to_thread( - self.success_checker.check_all, + criteria_results = await self.success_checker.check_all_async( self.task.success_criteria, reference_code=reference_code, reference_dir=reference_dir, @@ -1498,8 +1496,7 @@ async def _run_dialog_criteria_check( task_file=self.task_file, cached_reference=self._reference_code, ) - criteria_results = await asyncio.to_thread( - self.success_checker.check_all, + criteria_results = await self.success_checker.check_all_async( self.task.success_criteria, reference_code=reference_code, reference_dir=reference_dir, diff --git a/tests/test_check_all_async.py b/tests/test_check_all_async.py new file mode 100644 index 00000000..be0221c2 --- /dev/null +++ b/tests/test_check_all_async.py @@ -0,0 +1,176 @@ +"""Tests for SuccessChecker.check_all_async (GH #55). + +Judge-type criteria (llm_judge / agent_judge) declare +``supports_native_async = True`` and make a genuine non-blocking network call +from ``_check_impl_async``. ``check_all_async`` must: + - run all native-async criteria concurrently with each other (not serially), + - run the remaining sync criteria together in a single ``to_thread`` slot, + - overlap the sync batch with the async batch, + - preserve result ordering regardless of which path each criterion took, + - preserve the sync path's error-handling contract (KeyError / generic + Exception captured into a failed CriterionResult; JudgeInfrastructureError + escalates). +""" + +import asyncio +import time + +import pytest + +from coder_eval.criteria.base import BaseCriterion +from coder_eval.errors import JudgeInfrastructureError +from coder_eval.evaluation.checker import SuccessChecker +from coder_eval.models import CriterionResult, FileExistsCriterion, LLMJudgeCriterion, SandboxConfig +from coder_eval.sandbox import Sandbox + + +SLEEP_SECONDS = 0.2 + + +@pytest.fixture +def sandbox(tmp_path): + config = SandboxConfig(driver="tempdir") + sb = Sandbox(config, task_id="check_all_async_test") + try: + sb.setup() + yield sb + finally: + sb.cleanup() + + +@pytest.fixture +def checker(sandbox): + return SuccessChecker(sandbox, init_registry=True, validate_registry=False) + + +class _SleepyAsyncChecker(BaseCriterion[LLMJudgeCriterion]): + """Stand-in for llm_judge/agent_judge: sleeps on the event loop, not a thread.""" + + criterion_type = "llm_judge" + supports_native_async = True + + def __init__(self, sleep_seconds: float = SLEEP_SECONDS): + self.sleep_seconds = sleep_seconds + self.calls = 0 + + def _check_impl(self, criterion, sandbox, reference_code=None, *, turn_records=None, context=None): + raise AssertionError("sync _check_impl should not be invoked by check_all_async") + + async def _check_impl_async(self, criterion, sandbox, reference_code=None, *, turn_records=None, context=None): + self.calls += 1 + await asyncio.sleep(self.sleep_seconds) + return CriterionResult(criterion_type=self.criterion_type, description=criterion.description, score=1.0) + + +class _SleepyThreadChecker(BaseCriterion[FileExistsCriterion]): + """Stand-in for a CPU/file-bound checker: blocks a thread, not async-native.""" + + criterion_type = "file_exists" + + def __init__(self, sleep_seconds: float = SLEEP_SECONDS): + self.sleep_seconds = sleep_seconds + + def _check_impl(self, criterion, sandbox, reference_code=None, *, turn_records=None, context=None): + time.sleep(self.sleep_seconds) + return CriterionResult(criterion_type=self.criterion_type, description=criterion.description, score=1.0) + + +class _RaisingAsyncChecker(BaseCriterion[LLMJudgeCriterion]): + criterion_type = "llm_judge" + supports_native_async = True + + def __init__(self, exc: Exception): + self.exc = exc + + def _check_impl(self, criterion, sandbox, reference_code=None, *, turn_records=None, context=None): + raise self.exc + + async def _check_impl_async(self, criterion, sandbox, reference_code=None, *, turn_records=None, context=None): + raise self.exc + + +def _llm_criterion(description: str) -> LLMJudgeCriterion: + return LLMJudgeCriterion(description=description, prompt="grade it") + + +def _file_criterion(description: str, path: str = "x.txt") -> FileExistsCriterion: + return FileExistsCriterion(description=description, path=path) + + +class TestNativeAsyncConcurrency: + @pytest.mark.asyncio + async def test_two_judge_criteria_run_concurrently(self, checker): + """Two llm_judge-type criteria must overlap, not serialize (issue #55 point 1).""" + fake = _SleepyAsyncChecker() + checker._checker_instances["llm_judge"] = fake + + criteria = [_llm_criterion("a"), _llm_criterion("b")] + start = time.monotonic() + results = await checker.check_all_async(criteria) + elapsed = time.monotonic() - start + + assert fake.calls == 2 + assert elapsed < SLEEP_SECONDS * 1.5, f"judge criteria serialized: took {elapsed:.3f}s" + assert [r.score for r in results] == [1.0, 1.0] + + @pytest.mark.asyncio + async def test_async_batch_overlaps_sync_batch(self, checker): + """The native-async batch and the sync-criteria to_thread batch must overlap.""" + async_fake = _SleepyAsyncChecker() + sync_fake = _SleepyThreadChecker() + checker._checker_instances["llm_judge"] = async_fake + checker._checker_instances["file_exists"] = sync_fake + + criteria = [_llm_criterion("judge"), _file_criterion("file")] + start = time.monotonic() + results = await checker.check_all_async(criteria) + elapsed = time.monotonic() - start + + assert elapsed < SLEEP_SECONDS * 1.5, f"sync and async batches serialized: took {elapsed:.3f}s" + assert [r.criterion_type for r in results] == ["llm_judge", "file_exists"] + assert [r.score for r in results] == [1.0, 1.0] + + @pytest.mark.asyncio + async def test_result_order_preserved_regardless_of_dispatch_path(self, checker): + """Interleaved sync/async criteria must come back in declaration order.""" + checker._checker_instances["llm_judge"] = _SleepyAsyncChecker(sleep_seconds=0.01) + checker._checker_instances["file_exists"] = _SleepyThreadChecker(sleep_seconds=0.01) + + criteria = [ + _file_criterion("f1"), + _llm_criterion("j1"), + _file_criterion("f2"), + _llm_criterion("j2"), + ] + results = await checker.check_all_async(criteria) + assert [r.description for r in results] == ["f1", "j1", "f2", "j2"] + assert [r.criterion_type for r in results] == ["file_exists", "llm_judge", "file_exists", "llm_judge"] + + +class TestNativeAsyncErrorHandling: + @pytest.mark.asyncio + async def test_generic_exception_captured_as_failed_result(self, checker): + checker._checker_instances["llm_judge"] = _RaisingAsyncChecker(ValueError("boom")) + results = await checker.check_all_async([_llm_criterion("j")]) + assert len(results) == 1 + assert results[0].score == 0.0 + assert "ValueError: boom" in (results[0].error or "") + + @pytest.mark.asyncio + async def test_judge_infrastructure_error_propagates(self, checker): + checker._checker_instances["llm_judge"] = _RaisingAsyncChecker(JudgeInfrastructureError("down")) + with pytest.raises(JudgeInfrastructureError, match="down"): + await checker.check_all_async([_llm_criterion("j")]) + + @pytest.mark.asyncio + async def test_empty_criteria_returns_empty(self, checker): + assert await checker.check_all_async([]) == [] + + +class TestCheckAllAsyncMatchesSyncBehavior: + @pytest.mark.asyncio + async def test_all_sync_criteria_still_produce_same_results_as_check_all(self, checker): + criteria = [_file_criterion("f1", path="missing.txt")] + sync_results = checker.check_all(criteria) + async_results = await checker.check_all_async(criteria) + assert [r.score for r in sync_results] == [r.score for r in async_results] diff --git a/tests/test_early_stop.py b/tests/test_early_stop.py index 71cb8c7d..09fe6024 100644 --- a/tests/test_early_stop.py +++ b/tests/test_early_stop.py @@ -24,7 +24,7 @@ from datetime import datetime from pathlib import Path from typing import Any -from unittest.mock import MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest import typer @@ -1379,7 +1379,9 @@ async def _run_wiring( orch.sandbox = sandbox checker = MagicMock() - checker.check_all = MagicMock(return_value=[_crit_result(c.type, s) for c, s in zip(criteria, scores, strict=True)]) + checker.check_all_async = AsyncMock( + return_value=[_crit_result(c.type, s) for c, s in zip(criteria, scores, strict=True)] + ) orch.success_checker = checker if stop_early: diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py index 6b3f25e8..692d8a2a 100644 --- a/tests/test_orchestrator.py +++ b/tests/test_orchestrator.py @@ -1623,7 +1623,7 @@ async def test_evaluation_loop_breaks_on_max_turns_exhausted(tmp_path): # Mock success checker that always fails mock_checker = MagicMock() - mock_checker.check_all = MagicMock( + mock_checker.check_all_async = AsyncMock( return_value=[CriterionResult(criterion_type="file_exists", description="test", score=0.0)] ) orchestrator.success_checker = mock_checker @@ -1740,7 +1740,7 @@ async def crash_then_succeed_impl(_prompt, **kwargs): orchestrator.sandbox = mock_sandbox mock_checker = MagicMock() - mock_checker.check_all = MagicMock( + mock_checker.check_all_async = AsyncMock( return_value=[CriterionResult(criterion_type="file_exists", description="x", score=1.0)] ) orchestrator.success_checker = mock_checker @@ -1843,7 +1843,7 @@ async def timeout_impl(_prompt, **kwargs): orchestrator.sandbox = mock_sandbox mock_checker = MagicMock() - mock_checker.check_all = MagicMock( + mock_checker.check_all_async = AsyncMock( return_value=[CriterionResult(criterion_type="file_exists", description="x", score=1.0)] ) orchestrator.success_checker = mock_checker @@ -2075,7 +2075,7 @@ async def test_evaluation_loop_evaluate_only_loads_reference(tmp_path): "include_reference=True but reference not set" in the judge_context log. """ from datetime import datetime - from unittest.mock import MagicMock + from unittest.mock import AsyncMock, MagicMock from coder_eval.models import ( CriterionResult, @@ -2135,15 +2135,15 @@ async def test_evaluation_loop_evaluate_only_loads_reference(tmp_path): assert orchestrator.agent is None mock_checker = MagicMock() - mock_checker.check_all = MagicMock( + mock_checker.check_all_async = AsyncMock( return_value=[CriterionResult(criterion_type="file_exists", description="x", score=1.0)] ) orchestrator.success_checker = mock_checker await orchestrator._evaluation_loop() - mock_checker.check_all.assert_called_once() - kwargs = mock_checker.check_all.call_args.kwargs + mock_checker.check_all_async.assert_called_once() + kwargs = mock_checker.check_all_async.call_args.kwargs assert kwargs["reference_code"] == "REFERENCE_CONTENT" # turn_records is empty in evaluate-only mode but the kwarg should still be wired. assert kwargs["turn_records"] == [] diff --git a/tests/test_run_limits_orchestrator.py b/tests/test_run_limits_orchestrator.py index 77ca2394..b586e17e 100644 --- a/tests/test_run_limits_orchestrator.py +++ b/tests/test_run_limits_orchestrator.py @@ -225,7 +225,7 @@ async def _run_eval_loop_with_turn(self, task: TaskDefinition, tmp_path, turn: T orch.agent = mock_agent mock_checker = MagicMock() - mock_checker.check_all = MagicMock( + mock_checker.check_all_async = AsyncMock( return_value=[CriterionResult(criterion_type="file_exists", description="x", score=1.0)] ) orch.success_checker = mock_checker @@ -361,7 +361,7 @@ async def test_dialog_aborts_with_run_limit_stop_reason(self, tmp_path): orch.agent = mock_agent mock_checker = MagicMock() - mock_checker.check_all = MagicMock( + mock_checker.check_all_async = AsyncMock( return_value=[CriterionResult(criterion_type="file_exists", description="x", score=0.0)] ) orch.success_checker = mock_checker @@ -475,7 +475,7 @@ async def test_warning_does_not_abort_run(self, tmp_path, caplog): mock_agent.communicate = AsyncMock(return_value=turn) orch.agent = mock_agent mock_checker = MagicMock() - mock_checker.check_all = MagicMock( + mock_checker.check_all_async = AsyncMock( return_value=[CriterionResult(criterion_type="file_exists", description="x", score=1.0)] ) orch.success_checker = mock_checker @@ -521,7 +521,7 @@ async def test_warning_fires_in_simulation_and_does_not_abort(self, tmp_path, ca orch.agent = mock_agent mock_checker = MagicMock() - mock_checker.check_all = MagicMock( + mock_checker.check_all_async = AsyncMock( return_value=[CriterionResult(criterion_type="file_exists", description="x", score=1.0)] ) orch.success_checker = mock_checker @@ -574,7 +574,7 @@ async def test_warning_fires_when_single_simulation_turn_exceeds(self, tmp_path, orch.agent = mock_agent mock_checker = MagicMock() - mock_checker.check_all = MagicMock( + mock_checker.check_all_async = AsyncMock( return_value=[CriterionResult(criterion_type="file_exists", description="x", score=1.0)] ) orch.success_checker = mock_checker diff --git a/tests/test_timeout_orchestrator.py b/tests/test_timeout_orchestrator.py index cac28b1b..395ffdba 100644 --- a/tests/test_timeout_orchestrator.py +++ b/tests/test_timeout_orchestrator.py @@ -189,7 +189,7 @@ async def test_no_timeout_when_none(tmp_path) -> None: mock_agent.communicate = AsyncMock(return_value=_make_turn_record()) orchestrator.agent = mock_agent - orchestrator.success_checker.check_all = MagicMock( # type: ignore[union-attr] + orchestrator.success_checker.check_all_async = AsyncMock( # type: ignore[union-attr] return_value=[CriterionResult(criterion_type="file_exists", description="test", score=1.0)] ) @@ -325,7 +325,7 @@ async def flaky_communicate(_prompt, **kwargs): orchestrator.sandbox = mock_sandbox mock_checker = MagicMock() - mock_checker.check_all = MagicMock( + mock_checker.check_all_async = AsyncMock( return_value=[CriterionResult(criterion_type="file_exists", description="x", score=1.0)] ) orchestrator.success_checker = mock_checker