chore(detectors): added async detectors execution (WIP)

src/strands_evals/_async.py

@@ -1,12 +1,12 @@
-"""Async execution utilities for detectors.
+"""Async execution utilities.
 
-Provides a helper to run async functions from sync contexts, used by
-detectors that call Model.stream() directly.
+Provides helpers to run async functions from sync contexts and to fan out
+awaitables under a concurrency cap.
 """
 
 import asyncio
 import contextvars
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Iterable
 from concurrent.futures import ThreadPoolExecutor
 from typing import TypeVar
 
@@ -36,3 +36,101 @@ def execute() -> T:
         context = contextvars.copy_context()
         future = executor.submit(context.run, execute)
         return future.result()
+
+
+async def bounded_gather(
+    coros: Iterable[Awaitable[T]],
+    max_workers: int,
+    *,
+    return_exceptions: bool = True,
+) -> list[T | BaseException]:
+    """Run awaitables concurrently with at most `max_workers` running at once.
+
+    Wraps each coroutine in a `Semaphore`-gated task and `asyncio.gather`s them.
+    `return_exceptions` defaults to `True` so a single failing task does not
+    cancel its siblings — callers filter the result.
+
+    Args:
+        coros: An iterable of awaitables to run.
+        max_workers: Maximum concurrent tasks. Must be `>= 1`.
+        return_exceptions: Forwarded to `asyncio.gather`.
+
+    Returns:
+        The list of results in the order the awaitables were supplied. When
+        `return_exceptions=True`, failed tasks appear as `BaseException`
+        instances rather than raising.
+    """
+    if max_workers < 1:
+        raise ValueError(f"max_workers must be >= 1, got {max_workers}")
+
+    semaphore = asyncio.Semaphore(max_workers)
+
+    async def _bounded(coro: Awaitable[T]) -> T:
+        async with semaphore:
+            return await coro
+
+    return await asyncio.gather(*(_bounded(c) for c in coros), return_exceptions=return_exceptions)
+
+
+async def bounded_gather_fail_fast(
+    coros: Iterable[Awaitable[T]],
+    max_workers: int,
+) -> list[T]:
+    """Run awaitables concurrently with at most `max_workers` running at once,
+    cancelling siblings on the first exception.
+
+    Wraps each coroutine in a `Semaphore`-gated task and waits with
+    `FIRST_EXCEPTION`. As soon as one task raises, every still-running
+    sibling is cancelled and drained, then the original exception is
+    re-raised. Use when sibling work has nontrivial cost (e.g. LLM calls)
+    and one failure means the rest aren't worth completing. Prefer
+    `bounded_gather` when you want sibling isolation (one bad item must not
+    kill the rest of the batch).
+
+    Args:
+        coros: Iterable of awaitables to run.
+        max_workers: Maximum concurrent tasks. Must be `>= 1`.
+
+    Returns:
+        Results in the order the awaitables were supplied.
+
+    Raises:
+        Re-raises the first exception observed; pending tasks are cancelled
+        and drained before the raise propagates.
+    """
+    if max_workers < 1:
+        raise ValueError(f"max_workers must be >= 1, got {max_workers}")
+
+    semaphore = asyncio.Semaphore(max_workers)
+
+    async def _bounded(coro: Awaitable[T]) -> T:
+        async with semaphore:
+            return await coro
+
+    tasks: list[asyncio.Task[T]] = [asyncio.create_task(_bounded(c)) for c in coros]
+    if not tasks:
+        return []
+
+    try:
+        await asyncio.wait(tasks, return_when=asyncio.FIRST_EXCEPTION)
+    except BaseException:
+        # External cancellation: roll back our spawned tasks before propagating.
+        for t in tasks:
+            t.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
+        raise
+
+    # A task raised: cancel still-running siblings, drain, then re-raise.
+    pending = [t for t in tasks if not t.done()]
+    if pending:
+        for t in pending:
+            t.cancel()
+        await asyncio.gather(*pending, return_exceptions=True)
+        for t in tasks:
+            if t.cancelled():
+                continue
+            exc = t.exception()
+            if exc is not None:
+                raise exc
+
+    return [t.result() for t in tasks]

src/strands_evals/detectors/__init__.py

@@ -17,16 +17,20 @@
     RCAOutput,
     RCAStructuredOutput,
 )
-from .diagnosis import diagnose_session
-from .failure_detector import detect_failures
-from .root_cause_analyzer import analyze_root_cause
+from .diagnosis import diagnose_session, diagnose_session_async, diagnose_sessions_async
+from .failure_detector import detect_failures, detect_failures_async
+from .root_cause_analyzer import analyze_root_cause, analyze_root_cause_async
 
 __all__ = [
     # Core detectors
     "detect_failures",
+    "detect_failures_async",
     "analyze_root_cause",
+    "analyze_root_cause_async",
     # Diagnosis
     "diagnose_session",
+    "diagnose_session_async",
+    "diagnose_sessions_async",
     "DiagnosisConfig",
     "DiagnosisResult",
     "DiagnosisTrigger",

src/strands_evals/detectors/diagnosis.py

@@ -1,11 +1,17 @@
 """Session diagnosis: detect failures and analyze root causes."""
 
+import logging
+from collections.abc import Iterable
+
 from strands.models.model import Model
 
+from .._async import bounded_gather, run_async
 from ..types.detector import ConfidenceLevel, DiagnosisResult
 from ..types.trace import Session
-from .failure_detector import detect_failures
-from .root_cause_analyzer import analyze_root_cause
+from .failure_detector import detect_failures_async
+from .root_cause_analyzer import analyze_root_cause_async
+
+logger = logging.getLogger(__name__)
 
 
 def diagnose_session(
@@ -16,28 +22,62 @@ def diagnose_session(
 ) -> DiagnosisResult:
     """Run failure detection and root cause analysis on a session.
 
-    Pipeline: detect_failures → analyze_root_cause (if failures found).
+    Synchronous wrapper around `diagnose_session_async`. Safe to call from
+    sync code or from inside a running event loop (Jupyter, FastAPI, async
+    tests) — the work runs on a dedicated worker thread with its own loop.
+
+    Args:
+        session: The Session object to diagnose.
+        model: Model for LLM-based detectors. None uses default.
+        confidence_threshold: Minimum confidence for failure detection.
+
+    Returns:
+        DiagnosisResult with failures and root causes.
+    """
+    return run_async(
+        lambda: diagnose_session_async(
+            session,
+            model=model,
+            confidence_threshold=confidence_threshold,
+        )
+    )
+
+
+async def diagnose_session_async(
+    session: Session,
+    *,
+    model: Model | str | None = None,
+    confidence_threshold: ConfidenceLevel = ConfidenceLevel.LOW,
+    max_workers: int = 5,
+) -> DiagnosisResult:
+    """Async variant of `diagnose_session`.
+
+    Pipeline: detect_failures_async → analyze_root_cause_async (if failures found).
 
     Args:
         session: The Session object to diagnose.
         model: Model for LLM-based detectors. None uses default.
         confidence_threshold: Minimum confidence for failure detection.
+        max_workers: Maximum concurrent LLM calls inside per-session
+            chunked detection / RCA. See `detect_failures_async`.
 
     Returns:
         DiagnosisResult with failures and root causes.
     """
-    failure_output = detect_failures(
+    failure_output = await detect_failures_async(
         session,
         confidence_threshold=confidence_threshold,
         model=model,
+        max_workers=max_workers,
     )
 
     root_causes = []
     if failure_output.failures:
-        rca_output = analyze_root_cause(
+        rca_output = await analyze_root_cause_async(
             session,
             failures=failure_output.failures,
             model=model,
+            max_workers=max_workers,
         )
         root_causes = rca_output.root_causes
 
@@ -46,3 +86,55 @@ def diagnose_session(
         failures=failure_output.failures,
         root_causes=root_causes,
     )
+
+
+async def diagnose_sessions_async(
+    sessions: Iterable[Session],
+    *,
+    model: Model | str | None = None,
+    confidence_threshold: ConfidenceLevel = ConfidenceLevel.LOW,
+    max_workers: int = 5,
+    per_session_workers: int = 5,
+) -> list[DiagnosisResult | None]:
+    """Diagnose many sessions concurrently.
+
+    Sessions fan out under a `max_workers` semaphore; per-session chunk/window
+    fan-out is bounded by `per_session_workers`. Sessions that raise are
+    logged and replaced with `None` in the result so one bad session does
+    not poison the batch.
+
+    Args:
+        sessions: Sessions to diagnose. Order is preserved in the output.
+        model: Model for LLM-based detectors.
+        confidence_threshold: Minimum confidence for failure detection.
+        max_workers: Maximum sessions diagnosed concurrently.
+        per_session_workers: Maximum concurrent LLM calls inside each session.
+
+    Returns:
+        List of `DiagnosisResult` (or `None` if that session failed),
+        in the same order as `sessions`.
+    """
+    sessions_list = list(sessions)
+
+    async def _diagnose(session: Session) -> DiagnosisResult:
+        return await diagnose_session_async(
+            session,
+            model=model,
+            confidence_threshold=confidence_threshold,
+            max_workers=per_session_workers,
+        )
+
+    gathered = await bounded_gather(
+        (_diagnose(s) for s in sessions_list),
+        max_workers,
+        return_exceptions=True,
+    )
+
+    results: list[DiagnosisResult | None] = []
+    for session, outcome in zip(sessions_list, gathered, strict=True):
+        if isinstance(outcome, BaseException):
+            logger.warning("Diagnosis failed for session %s: %s", session.session_id, outcome)
+            results.append(None)
+        else:
+            results.append(outcome)
+    return results