diff --git a/py/benchmarks/benches/bench_eval_scheduling.py b/py/benchmarks/benches/bench_eval_scheduling.py new file mode 100644 index 000000000..0b4bd4358 --- /dev/null +++ b/py/benchmarks/benches/bench_eval_scheduling.py @@ -0,0 +1,58 @@ +"""Benchmarks for ordinary eval task execution and trial scheduling.""" + +import asyncio +import pathlib +import sys + +import pyperf + + +if __package__ in (None, ""): + sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2])) + +from braintrust.framework import EvalCase, Evaluator, await_or_run, run_evaluator +from braintrust.logger import BraintrustState + +from benchmarks._utils import disable_pyperf_psutil + + +async def _async_task(input_value): + return input_value + + +def _sync_task(input_value): + return input_value + + +_state = BraintrustState() +_evaluator = Evaluator( + project_name="benchmark", + eval_name="eval-scheduling", + data=[EvalCase(input=i) for i in range(25)], + task=_async_task, + scores=[], + experiment_name=None, + metadata=None, + max_concurrency=5, +) + + +async def _bench_sync_task(): + await await_or_run(asyncio.get_running_loop(), _sync_task, 1) + + +async def _bench_eval_scheduler(): + await run_evaluator(None, _evaluator, None, [], state=_state, enable_cache=False) + + +def main(runner: pyperf.Runner | None = None) -> None: + if runner is None: + disable_pyperf_psutil() + runner = pyperf.Runner() + + runner.bench_async_func("eval_scheduling.sync_task", _bench_sync_task) + runner.bench_async_func("eval_scheduling.25_async_trials", _bench_eval_scheduler) + + +if __name__ == "__main__": + main() diff --git a/py/src/braintrust/framework.py b/py/src/braintrust/framework.py index b352252e9..58a8c08ce 100644 --- a/py/src/braintrust/framework.py +++ b/py/src/braintrust/framework.py @@ -6,6 +6,7 @@ import json import re import sys +import threading import traceback import uuid import warnings @@ -161,6 +162,13 @@ class EvalHooks(abc.ABC, Generic[Expected]): An object that can be used to add metadata to an evaluation. This is passed to the `task` function. """ + @property + def cancel_event(self) -> threading.Event: + """Set when the evaluation is cancelled or times out.""" + if not hasattr(self, "_cancel_event"): + self._cancel_event = threading.Event() + return self._cancel_event + @property @abc.abstractmethod def metadata(self) -> Metadata | None: @@ -462,6 +470,9 @@ class Evaluator(Generic[Input, Output, Expected]): parameter_values: dict[str, Any] | None = None + cancel_event: threading.Event | None = None + """Optional event that cancels the evaluation when set. Also set on timeout.""" + @dataclasses.dataclass class EvalResultWithSummary(SerializableDataClass, Generic[Input, Output, Expected]): @@ -489,9 +500,18 @@ def run_f(args, kwargs, ctx): var.reset(tok) with _THREAD_POOL_SINGLETON.get() as thread_pool: - return await event_loop.run_in_executor( - thread_pool.thread_pool(), run_f, args, kwargs, contextvars.copy_context() - ) + submitted = thread_pool.thread_pool().submit(run_f, args, kwargs, contextvars.copy_context()) + future = asyncio.wrap_future(submitted, loop=event_loop) + try: + return await asyncio.shield(future) + except asyncio.CancelledError: + # Skip jobs still in the pool; running threads must finish before their eval state is disposed. + if submitted.cancel(): + raise + try: + await future + finally: + raise def _call_user_fn_args(fn, kwargs): @@ -688,6 +708,7 @@ def _EvalCommon( update: bool, reporter: ReporterDef[Input, Output, Expected, EvalReport] | None, timeout: float | None, + cancel_event: threading.Event | None, max_concurrency: int | None, project_id: str | None, base_experiment_name: str | None, @@ -733,6 +754,7 @@ def _EvalCommon( is_public=is_public, update=update, timeout=timeout, + cancel_event=cancel_event, max_concurrency=max_concurrency, project_id=project_id, base_experiment_name=base_experiment_name, @@ -845,6 +867,7 @@ async def EvalAsync( parent: str | None = None, state: BraintrustState | None = None, enable_cache: bool = True, + cancel_event: threading.Event | None = None, ) -> EvalResultWithSummary[Input, Output, Expected]: """ A function you can use to define an evaluator. This is a convenience wrapper around the `Evaluator` class. @@ -884,6 +907,8 @@ async def EvalAsync( :param reporter: (Optional) A reporter that takes an evaluator and its result and returns a report. :param timeout: (Optional) The duration, in seconds, after which to time out the evaluation. Defaults to None, in which case there is no timeout. + :param cancel_event: Optional threading.Event. Setting it stops the evaluation and cancels running async tasks. + A timeout also sets this event. Running synchronous tasks must cooperate or finish before the eval returns. :param project_id: (Optional) If specified, uses the given project ID instead of the evaluator's name to identify the project. :param base_experiment_name: An optional experiment name to use as a base. If specified, the new experiment will be summarized and compared to this experiment. @@ -922,6 +947,7 @@ async def EvalAsync( update=update, reporter=reporter, timeout=timeout, + cancel_event=cancel_event, max_concurrency=max_concurrency, project_id=project_id, base_experiment_name=base_experiment_name, @@ -974,6 +1000,7 @@ def Eval( parent: str | None = None, state: BraintrustState | None = None, enable_cache: bool = True, + cancel_event: threading.Event | None = None, ) -> EvalResultWithSummary[Input, Output, Expected]: """ A function you can use to define an evaluator. This is a convenience wrapper around the `Evaluator` class. @@ -1013,6 +1040,8 @@ def Eval( :param reporter: (Optional) A reporter that takes an evaluator and its result and returns a report. :param timeout: (Optional) The duration, in seconds, after which to time out the evaluation. Defaults to None, in which case there is no timeout. + :param cancel_event: Optional threading.Event. Setting it stops the evaluation and cancels running async tasks. + A timeout also sets this event. Running synchronous tasks must cooperate or finish before the eval returns. :param project_id: (Optional) If specified, uses the given project ID instead of the evaluator's name to identify the project. :param base_experiment_name: An optional experiment name to use as a base. If specified, the new experiment will be summarized and compared to this experiment. @@ -1052,6 +1081,7 @@ def Eval( update=update, reporter=reporter, timeout=timeout, + cancel_event=cancel_event, max_concurrency=max_concurrency, project_id=project_id, base_experiment_name=base_experiment_name, @@ -1195,6 +1225,7 @@ def __init__( tags: Sequence[str] | None = None, report_progress: Callable[[TaskProgressEvent], None] = None, parameters: ValidatedParameters | None = None, + cancel_event: threading.Event | None = None, ): if metadata is not None: self.update({"metadata": metadata}) @@ -1209,6 +1240,11 @@ def __init__( self._report_progress = report_progress self._parameters = parameters + self._cancel_event = cancel_event if cancel_event is not None else threading.Event() + + @property + def cancel_event(self) -> threading.Event: + return self._cancel_event @property def metadata(self) -> Metadata | None: @@ -1441,11 +1477,59 @@ async def run_evaluator( state: BraintrustState | None = None, enable_cache: bool = True, ) -> EvalResultWithSummary[Input, Output, Expected]: - """Wrapper on _run_evaluator_internal that times out execution after evaluator.timeout.""" - results = await asyncio.wait_for( - _run_evaluator_internal(experiment, evaluator, position, filters, stream, state, enable_cache), - evaluator.timeout, + """Run an evaluator, cancelling its work on timeout or when cancel_event is set.""" + cancel_event = evaluator.cancel_event if evaluator.cancel_event is not None else threading.Event() + if cancel_event.is_set(): + raise asyncio.CancelledError("Evaluator cancelled") + if evaluator.timeout is not None and evaluator.timeout <= 0: + cancel_event.set() + raise asyncio.TimeoutError("Evaluator timed out") + + loop = asyncio.get_running_loop() + cancelled = loop.create_future() + timed_out = False + + def on_timeout(): + nonlocal timed_out + if not cancel_event.is_set(): + timed_out = True + cancel_event.set() + if not cancelled.done(): + cancelled.set_result(None) + + async def watch_cancel_event(): + while not cancel_event.is_set(): + await asyncio.sleep(0.01) + if not cancelled.done(): + cancelled.set_result(None) + + timeout_handle = loop.call_later(evaluator.timeout, on_timeout) if evaluator.timeout is not None else None + watcher = asyncio.create_task(watch_cancel_event()) if evaluator.cancel_event is not None else None + worker = asyncio.create_task( + _run_evaluator_internal(experiment, evaluator, position, filters, stream, state, enable_cache, cancel_event) ) + try: + if watcher is not None or timeout_handle is not None: + await asyncio.wait((worker, cancelled), return_when=asyncio.FIRST_COMPLETED) + if cancel_event.is_set(): + worker.cancel() + (worker_result,) = await asyncio.gather(worker, return_exceptions=True) + worker_error = worker_result if isinstance(worker_result, Exception) else None + if timed_out: + raise asyncio.TimeoutError("Evaluator timed out") from worker_error + raise asyncio.CancelledError("Evaluator cancelled") from worker_error + results = await worker + except asyncio.CancelledError: + cancel_event.set() + worker.cancel() + await asyncio.gather(worker, return_exceptions=True) + raise + finally: + if timeout_handle is not None: + timeout_handle.cancel() + if watcher is not None: + watcher.cancel() + await asyncio.gather(watcher, return_exceptions=True) if experiment: comparison_experiment_id = evaluator.base_experiment_id @@ -1479,6 +1563,7 @@ async def _run_evaluator_internal( stream: Callable[[SSEProgressEvent], None] | None = None, state: BraintrustState | None = None, enable_cache: bool = True, + cancel_event: threading.Event | None = None, ): # Start span cache for this eval (it's disabled by default to avoid temp files outside of evals) if state is None: @@ -1489,7 +1574,9 @@ async def _run_evaluator_internal( if enable_cache: state.span_cache.start() try: - return await _run_evaluator_internal_impl(experiment, evaluator, position, filters, stream, state) + return await _run_evaluator_internal_impl( + experiment, evaluator, position, filters, stream, state, cancel_event + ) finally: # Clean up disk-based span cache after eval completes and stop caching if enable_cache: @@ -1504,8 +1591,11 @@ async def _run_evaluator_internal_impl( filters: list[Filter], stream: Callable[[SSEProgressEvent], None] | None = None, state: BraintrustState | None = None, + cancel_event: threading.Event | None = None, ): event_loop = asyncio.get_event_loop() + if cancel_event is None: + cancel_event = threading.Event() async def await_or_run_scorer(root_span, scorer, name, **kwargs): # Merge purpose into parent's propagated_event rather than replacing it @@ -1686,6 +1776,7 @@ def report_progress(event: TaskProgressEvent): tags=tags, report_progress=report_progress, parameters=resolved_evaluator_parameters, + cancel_event=cancel_event, ) # Check if the task takes a hooks argument @@ -1698,8 +1789,12 @@ def report_progress(event: TaskProgressEvent): with root_span.start_span("task", span_attributes={"type": SpanTypeAttribute.TASK}) as span: hooks.set_span(span) + if cancel_event.is_set(): + raise asyncio.CancelledError output = await await_or_run(event_loop, evaluator.task, *task_args) span.log(input=task_args[0], output=output) + if cancel_event.is_set(): + raise asyncio.CancelledError tags = hooks.tags if hooks.tags else None root_span.log(output=output, metadata=metadata, tags=tags) @@ -1781,6 +1876,8 @@ async def ensure_spans_flushed(): "id": datum.id, "tags": tags, } + if cancel_event.is_set(): + raise asyncio.CancelledError score_promises = [ asyncio.create_task(await_or_run_scorer(root_span, score, name, **scorer_kwargs)) for score, name in zip(scorers, scorer_names) @@ -1789,29 +1886,35 @@ async def ensure_spans_flushed(): asyncio.create_task(await_or_run_classifier(root_span, classifier, name, **scorer_kwargs)) for classifier, name in zip(classifiers, classifier_names) ] - - failing_scorers_and_exceptions = [] - for name, p in zip(scorer_names, score_promises): - try: - score_results = await p - for score in score_results: - scores[score.name] = score.score - except Exception as e: - failing_scorers_and_exceptions.append((name, e, traceback.format_exc())) - - failing_classifiers_and_exceptions = [] - for name, p in zip(classifier_names, classifier_promises): - try: - classifier_results = await p - if classifier_results is None: - continue - for classification in classifier_results: - item = classification.as_item() - if classification.name not in classifications: - classifications[classification.name] = [] - classifications[classification.name].append(item) - except Exception as e: - failing_classifiers_and_exceptions.append((name, e, traceback.format_exc())) + try: + failing_scorers_and_exceptions = [] + for name, p in zip(scorer_names, score_promises): + try: + score_results = await p + for score in score_results: + scores[score.name] = score.score + except Exception as e: + failing_scorers_and_exceptions.append((name, e, traceback.format_exc())) + + failing_classifiers_and_exceptions = [] + for name, p in zip(classifier_names, classifier_promises): + try: + classifier_results = await p + if classifier_results is None: + continue + for classification in classifier_results: + item = classification.as_item() + if classification.name not in classifications: + classifications[classification.name] = [] + classifications[classification.name].append(item) + except Exception as e: + failing_classifiers_and_exceptions.append((name, e, traceback.format_exc())) + finally: + # Scorers run in separate tasks and must finish before their root span closes. + for promise in (*score_promises, *classifier_promises): + if not promise.done(): + promise.cancel() + await asyncio.gather(*score_promises, *classifier_promises, return_exceptions=True) if classifications: root_span.log(classifications=classifications) @@ -1915,38 +2018,57 @@ async def filtered_iterator(it): asyncio.Semaphore(evaluator.max_concurrency) if evaluator.max_concurrency is not None else None ) - async def with_max_concurrency(coro): + async def with_max_concurrency(datum, trial_index): if max_concurrency_semaphore: async with max_concurrency_semaphore: - return await coro + if cancel_event.is_set(): + return None + return await run_evaluator_task(datum, trial_index) else: - return await coro + if cancel_event.is_set(): + return None + return await run_evaluator_task(datum, trial_index) tasks = [] - with async_tqdm( - filtered_iterator(data_iterator), - desc=f"{evaluator.eval_name} (data)", - position=position, - disable=position is None, - ) as pbar: - async for datum in pbar: - if isinstance(datum, dict): - datum_trial_count = datum.get("trial_count") - else: - datum_trial_count = getattr(datum, "trial_count", None) - trial_count = datum_trial_count if datum_trial_count is not None else evaluator.trial_count - for trial_index in range(trial_count): - tasks.append(asyncio.create_task(with_max_concurrency(run_evaluator_task(datum, trial_index)))) - - if not tasks: - eprint( - f"{bcolors.WARNING}Warning: no data rows found for evaluator '{evaluator.eval_name}'. The experiment will be empty.{bcolors.ENDC}" - ) + try: + with async_tqdm( + filtered_iterator(data_iterator), + desc=f"{evaluator.eval_name} (data)", + position=position, + disable=position is None, + ) as pbar: + async for datum in pbar: + if cancel_event.is_set(): + break + if isinstance(datum, dict): + datum_trial_count = datum.get("trial_count") + else: + datum_trial_count = getattr(datum, "trial_count", None) + trial_count = datum_trial_count if datum_trial_count is not None else evaluator.trial_count + for trial_index in range(trial_count): + if cancel_event.is_set(): + break + tasks.append(asyncio.create_task(with_max_concurrency(datum, trial_index))) + + if not tasks and not cancel_event.is_set(): + eprint( + f"{bcolors.WARNING}Warning: no data rows found for evaluator '{evaluator.eval_name}'. The experiment will be empty.{bcolors.ENDC}" + ) - results = [] - for task in std_tqdm(tasks, desc=f"{evaluator.eval_name} (tasks)", position=position, disable=position is None): - results.append(await task) - return results + results = [] + for task in std_tqdm( + tasks, desc=f"{evaluator.eval_name} (tasks)", position=position, disable=position is None + ): + result = await task + if result is not None: + results.append(result) + return results + finally: + for task in tasks: + if not task.done(): + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) def build_local_summary( diff --git a/py/src/braintrust/test_framework.py b/py/src/braintrust/test_framework.py index 3e9dfe70b..c3284e537 100644 --- a/py/src/braintrust/test_framework.py +++ b/py/src/braintrust/test_framework.py @@ -1,6 +1,9 @@ +import asyncio import importlib.util import re import sys +import threading +from concurrent.futures import ThreadPoolExecutor from unittest.mock import MagicMock, patch import pytest @@ -9,16 +12,19 @@ from .framework import ( Eval, + EvalAsync, EvalCase, EvalHooks, EvalResultWithSummary, Evaluator, Filter, _call_user_fn_args, + await_or_run, evaluate_filter, parse_filters, run_evaluator, ) +from .resource_manager import ResourceManager from .score import Classification, Score, Scorer from .test_helpers import init_test_exp, with_memory_logger, with_simulate_login # noqa: F401 @@ -279,6 +285,254 @@ def exact_match(input_value, output, expected): assert result.summary.scores["exact_match"].score == 1.0 +@pytest.mark.asyncio +@pytest.mark.parametrize("trigger", ["event", "caller"]) +async def test_eval_async_cancellation_stops_queued_trials_and_drains_running_task(trigger): + cancel_event = threading.Event() + started = asyncio.Event() + cleaned_up = asyncio.Event() + started_inputs = [] + + async def task(input_value, hooks): + started_inputs.append(input_value) + assert hooks.cancel_event is cancel_event + started.set() + try: + await asyncio.Event().wait() + finally: + await asyncio.sleep(0.01) + cleaned_up.set() + + evaluation = asyncio.create_task( + EvalAsync( + "test-cancel-event", + data=[EvalCase(input=1), EvalCase(input=2)], + task=task, + scores=[], + max_concurrency=1, + cancel_event=cancel_event, + no_send_logs=True, + ) + ) + try: + await asyncio.wait_for(started.wait(), 2) + if trigger == "event": + cancel_event.set() + else: + evaluation.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(evaluation, 2) + assert cancel_event.is_set() + assert cleaned_up.is_set() + assert started_inputs == [1] + finally: + cancel_event.set() + evaluation.cancel() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("already_set", "timeout", "error"), + [(True, None, asyncio.CancelledError), (False, 0, asyncio.TimeoutError)], + ids=["pre-set-event", "zero-timeout"], +) +async def test_eval_async_cancelled_before_tasks_start(already_set, timeout, error): + cancel_event = threading.Event() + if already_set: + cancel_event.set() + started = [] + + with pytest.raises(error): + await EvalAsync( + "test-cancelled-before-start", + data=[EvalCase(input=1)], + task=lambda input_value: started.append(input_value), + scores=[], + timeout=timeout, + cancel_event=cancel_event, + no_send_logs=True, + ) + assert cancel_event.is_set() + assert started == [] + + +@pytest.mark.asyncio +async def test_eval_sync_cancel_event_reaches_running_thread_and_skips_queue(): + cancel_event = threading.Event() + started = threading.Event() + release = threading.Event() + finished = threading.Event() + started_inputs = [] + scored_inputs = [] + + def task(input_value, hooks): + started_inputs.append(input_value) + assert hooks.cancel_event is cancel_event + started.set() + assert cancel_event.wait(2) + assert release.wait(2) + finished.set() + return input_value + + evaluation = asyncio.create_task( + asyncio.to_thread( + Eval, + "test-sync-cancel", + data=[EvalCase(input=1), EvalCase(input=2)], + task=task, + scores=[lambda input_value, output, expected: scored_inputs.append(input_value)], + max_concurrency=1, + cancel_event=cancel_event, + no_send_logs=True, + ) + ) + try: + assert await asyncio.to_thread(started.wait, 2) + cancel_event.set() + await asyncio.sleep(0.05) + assert not evaluation.done() + release.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(evaluation, 2) + assert finished.is_set() + assert started_inputs == [1] + assert scored_inputs == [] + finally: + release.set() + cancel_event.set() + + +@pytest.mark.asyncio +async def test_eval_async_timeout_waits_for_scorer_cleanup(): + cancel_event = threading.Event() + scorer_started = asyncio.Event() + scorer_cleaned_up = asyncio.Event() + + async def scorer(input_value, output, expected): + scorer_started.set() + try: + await asyncio.Event().wait() + finally: + await asyncio.sleep(0.01) + scorer_cleaned_up.set() + + evaluation = asyncio.create_task( + EvalAsync( + "test-timeout-scorer", + data=[EvalCase(input=1)], + task=lambda input_value: input_value, + scores=[scorer], + timeout=0.5, + cancel_event=cancel_event, + no_send_logs=True, + ) + ) + try: + await asyncio.wait_for(scorer_started.wait(), 2) + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(evaluation, 2) + assert cancel_event.is_set() + assert scorer_cleaned_up.is_set() + finally: + evaluation.cancel() + + +@pytest.mark.asyncio +async def test_eval_async_cancel_event_closes_pending_data_iterator(): + cancel_event = threading.Event() + reading_next_row = asyncio.Event() + iterator_closed = asyncio.Event() + + async def data(): + try: + yield EvalCase(input=1) + reading_next_row.set() + await asyncio.Event().wait() + yield EvalCase(input=2) + finally: + iterator_closed.set() + + evaluation = asyncio.create_task( + EvalAsync( + "test-cancel-data-iterator", + data=data(), + task=lambda input_value: input_value, + scores=[], + cancel_event=cancel_event, + no_send_logs=True, + ) + ) + try: + await asyncio.wait_for(reading_next_row.wait(), 2) + cancel_event.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(evaluation, 2) + assert iterator_closed.is_set() + finally: + cancel_event.set() + evaluation.cancel() + + +@pytest.mark.asyncio +async def test_eval_async_cancel_event_preserves_data_iterator_error(): + cancel_event = threading.Event() + + async def data(): + cancel_event.set() + raise ValueError("iterator failed during cancellation") + yield EvalCase(input=1) + + with pytest.raises(asyncio.CancelledError) as exc_info: + await EvalAsync( + "test-cancel-iterator-error", + data=data(), + task=lambda input_value: input_value, + scores=[], + cancel_event=cancel_event, + no_send_logs=True, + ) + assert isinstance(exc_info.value.__cause__, ValueError) + + +@pytest.mark.asyncio +async def test_cancelled_sync_task_does_not_start_while_queued_in_thread_pool(): + started = threading.Event() + release = threading.Event() + queued_started = threading.Event() + + class Pool: + def __init__(self, executor): + self.executor = executor + + def thread_pool(self): + return self.executor + + def blocking(): + started.set() + assert release.wait(2) + + def queued(): + queued_started.set() + + with ThreadPoolExecutor(max_workers=1) as executor: + with patch("braintrust.framework._THREAD_POOL_SINGLETON", ResourceManager(Pool(executor))): + loop = asyncio.get_running_loop() + first = asyncio.create_task(await_or_run(loop, blocking)) + try: + assert await asyncio.to_thread(started.wait, 2) + second = asyncio.create_task(await_or_run(loop, queued)) + await asyncio.sleep(0) + second.cancel() + await asyncio.sleep(0) + release.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(second, 2) + await asyncio.wait_for(first, 2) + assert not queued_started.is_set() + finally: + release.set() + + @pytest.mark.asyncio async def test_eval_case_id_and_tags_are_passed_to_scorers(): scorer_args = None diff --git a/py/src/braintrust/type_tests/test_eval_generics.py b/py/src/braintrust/type_tests/test_eval_generics.py index 9d655e918..3329a90bd 100644 --- a/py/src/braintrust/type_tests/test_eval_generics.py +++ b/py/src/braintrust/type_tests/test_eval_generics.py @@ -10,10 +10,11 @@ pytest src/braintrust/type_tests/test_eval_generics.py """ +import threading from typing import TypedDict import pytest -from braintrust.framework import EvalAsync, EvalCase, EvalResultWithSummary +from braintrust.framework import EvalAsync, EvalCase, EvalHooks, EvalResultWithSummary from braintrust.generated_types import ObjectReference from braintrust.score import Score from braintrust.types._eval import EvalCaseDict, EvalCaseDictNoOutput @@ -105,6 +106,26 @@ async def test_eval_same_type_output_and_expected(): assert result.results[0].expected == "golden answer" +@pytest.mark.asyncio +async def test_eval_cancel_event_types(): + cancel_event = threading.Event() + + async def task(input: str, hooks: EvalHooks[str]) -> str: + assert hooks.cancel_event is cancel_event + return input + + result = await EvalAsync( + "test-cancel-event-types", + data=[EvalCase(input="hello", expected="hello")], + task=task, + scores=[], + cancel_event=cancel_event, + no_send_logs=True, + ) + assert result.results[0].output == "hello" + assert not cancel_event.is_set() + + @pytest.mark.asyncio async def test_eval_divergent_output_and_expected(): """Output and Expected differ — the pattern reported in #240."""