From e9c0b7e8dc33060c0c8dd5006a2d64a3970464fa Mon Sep 17 00:00:00 2001
From: poshinchen <pschen@amazon.com>
Date: Wed, 17 Jun 2026 13:46:45 -0400
Subject: [PATCH] chore(detectors): added async detectors execution

---
 src/strands_evals/_async.py                   | 106 +++++++-
 src/strands_evals/detectors/__init__.py       |  10 +-
 src/strands_evals/detectors/diagnosis.py      | 102 +++++++-
 .../detectors/failure_detector.py             | 106 +++++---
 .../detectors/root_cause_analyzer.py          | 116 +++++----
 src/strands_evals/experiment.py               |   6 +-
 src/strands_evals/types/detector.py           |   4 +
 .../strands_evals/detectors/test_diagnosis.py | 201 ++++++++++++----
 .../detectors/test_failure_detector.py        | 226 +++++++++++++++++-
 .../detectors/test_root_cause_analyzer.py     | 108 ++++++---
 tests/strands_evals/test_async.py             | 107 +++++++++
 tests/strands_evals/test_experiment.py        |  24 +-
 12 files changed, 935 insertions(+), 181 deletions(-)
 create mode 100644 tests/strands_evals/test_async.py

diff --git a/src/strands_evals/_async.py b/src/strands_evals/_async.py
index 71a9f5da..ca395be4 100644
--- a/src/strands_evals/_async.py
+++ b/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]
diff --git a/src/strands_evals/detectors/__init__.py b/src/strands_evals/detectors/__init__.py
index 39a601b5..200cf60c 100644
--- a/src/strands_evals/detectors/__init__.py
+++ b/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",
diff --git a/src/strands_evals/detectors/diagnosis.py b/src/strands_evals/detectors/diagnosis.py
index e4e8d6dd..ecd93339 100644
--- a/src/strands_evals/detectors/diagnosis.py
+++ b/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
diff --git a/src/strands_evals/detectors/failure_detector.py b/src/strands_evals/detectors/failure_detector.py
index 26c9eba5..6139895a 100644
--- a/src/strands_evals/detectors/failure_detector.py
+++ b/src/strands_evals/detectors/failure_detector.py
@@ -14,7 +14,7 @@
 from strands.models.model import Model
 from strands.types.content import ContentBlock, Message, Messages
 
-from .._async import run_async
+from .._async import bounded_gather_fail_fast, run_async
 from ..types.detector import ConfidenceLevel, FailureDetectionStructuredOutput, FailureItem, FailureOutput
 from ..types.trace import Session
 from .chunking import merge_chunk_failures, split_spans_by_tokens, would_exceed_context
@@ -32,8 +32,8 @@
 
 
 def _resolve_confidence_threshold(threshold: ConfidenceLevel) -> float:
-    """Map the categorical ``"low"|"medium"|"high"`` threshold to the numeric
-    value used for comparison (see ``CONFIDENCE_MAP``).
+    """Map the categorical `"low"|"medium"|"high"` threshold to the numeric
+    value used for comparison (see `CONFIDENCE_MAP`).
     """
     return CONFIDENCE_MAP[threshold]
 
@@ -62,12 +62,16 @@ def detect_failures(
 ) -> FailureOutput:
     """Detect semantic failures in an agent execution session.
 
+    Synchronous wrapper around `detect_failures_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 analyze.
         confidence_threshold: Minimum categorical confidence to include a
-            failure (``"low"`` | ``"medium"`` | ``"high"``). Internally mapped
-            to ``0.5 | 0.75 | 0.9`` via ``CONFIDENCE_MAP`` before filtering.
-            Defaults to ``"low"`` (include everything the LLM flagged).
+            failure (`"low"` | `"medium"` | `"high"`). Internally mapped
+            to `0.5 | 0.75 | 0.9` via `CONFIDENCE_MAP` before filtering.
+            Defaults to `"low"` (include everything the LLM flagged).
         model: A Model instance, model ID string (wrapped in BedrockModel),
             or None (uses default Haiku).
 
@@ -75,6 +79,36 @@ def detect_failures(
         FailureOutput with list of FailureItems, each with span_id, category,
         confidence (float in [0.0, 1.0]), and evidence.
     """
+    return run_async(
+        lambda: detect_failures_async(
+            session,
+            confidence_threshold=confidence_threshold,
+            model=model,
+        )
+    )
+
+
+async def detect_failures_async(
+    session: Session,
+    *,
+    confidence_threshold: ConfidenceLevel = ConfidenceLevel.LOW,
+    model: Model | str | None = None,
+    max_workers: int = 5,
+) -> FailureOutput:
+    """Async variant of `detect_failures`.
+
+    Chunked detection fans chunks out concurrently, capped at `max_workers`.
+
+    Args:
+        session: The Session object to analyze.
+        confidence_threshold: Minimum categorical confidence to include a
+            failure. See `detect_failures`.
+        model: A Model instance, model ID string, or None.
+        max_workers: Maximum concurrent LLM calls during chunked detection.
+
+    Returns:
+        FailureOutput. See `detect_failures`.
+    """
     threshold = _resolve_confidence_threshold(confidence_threshold)
     effective_model = _resolve_model(model)
     template = get_template("v0")
@@ -82,14 +116,14 @@ def detect_failures(
     user_prompt = template.build_prompt(session_json=session_json)
 
     if would_exceed_context(user_prompt):
-        raw = _detect_chunked(session, effective_model, template)
+        raw = await _detect_chunked_async(session, effective_model, template, max_workers)
     else:
         try:
-            raw = _detect_direct(user_prompt, effective_model, template)
+            raw = await _detect_direct_async(user_prompt, effective_model, template)
         except Exception as e:
             if _is_context_exceeded(e):
                 logger.warning("Context exceeded despite pre-flight check, falling back to chunking")
-                raw = _detect_chunked(session, effective_model, template)
+                raw = await _detect_chunked_async(session, effective_model, template, max_workers)
             else:
                 raise
 
@@ -108,18 +142,19 @@ def detect_failures(
     return FailureOutput(session_id=session.session_id, failures=filtered)
 
 
-def _detect_direct(user_prompt: str, model: Model, template: object) -> list[FailureItem]:
-    """Attempt direct LLM detection on the full session."""
-    text = _call_model(model, system_prompt=template.SYSTEM_PROMPT, user_prompt=user_prompt)
+async def _detect_direct_async(user_prompt: str, model: Model, template: object) -> list[FailureItem]:
+    """Direct LLM detection on the full session."""
+    text = await _call_model_async(model, system_prompt=template.SYSTEM_PROMPT, user_prompt=user_prompt)
     return _parse_text_result(text)
 
 
-def _detect_chunked(
+async def _detect_chunked_async(
     session: Session,
     model: Model,
     template: object,
+    max_workers: int,
 ) -> list[FailureItem]:
-    """Chunk session and detect failures per chunk, then merge."""
+    """Chunk session and detect failures per chunk concurrently, then merge."""
     spans = _flatten_traces_to_spans(session.traces)
 
     if len(spans) <= MIN_CHUNK_SIZE:
@@ -130,24 +165,32 @@ def _detect_chunked(
 
     logger.info("Chunked detection: %d spans -> %d chunks", len(spans), len(chunks))
 
-    chunk_results: list[list[FailureItem]] = []
-    for i, chunk_spans in enumerate(chunks):
+    async def _process_chunk(index: int, chunk_spans: list) -> list[FailureItem]:
         try:
             chunk_json = _serialize_spans(chunk_spans)
             user_prompt = template.build_prompt(session_json=chunk_json)
-            text = _call_model(model, system_prompt=template.SYSTEM_PROMPT, user_prompt=user_prompt)
-            chunk_results.append(_parse_text_result(text))
-            logger.info("Chunk %d/%d: processed %d spans", i + 1, len(chunks), len(chunk_spans))
+            text = await _call_model_async(model, system_prompt=template.SYSTEM_PROMPT, user_prompt=user_prompt)
+            result = _parse_text_result(text)
+            logger.info("Chunk %d/%d: processed %d spans", index + 1, len(chunks), len(chunk_spans))
+            return result
         except Exception as e:
             if _is_context_exceeded(e):
-                logger.warning("Chunk %d/%d still exceeds context, skipping", i + 1, len(chunks))
-            else:
-                raise
+                logger.warning("Chunk %d/%d still exceeds context, skipping", index + 1, len(chunks))
+                return []
+            raise
+
+    # `_process_chunk` already swallows context-exceeded as `[]`, so anything that
+    # reaches `bounded_gather_fail_fast` as an exception is a genuine error worth
+    # cancelling sibling LLM calls for (vs. paying for every chunk before raising).
+    chunk_results = await bounded_gather_fail_fast(
+        (_process_chunk(i, chunk_spans) for i, chunk_spans in enumerate(chunks)),
+        max_workers,
+    )
 
     return merge_chunk_failures(chunk_results)
 
 
-def _call_model(model: Model, *, system_prompt: str, user_prompt: str) -> str:
+async def _call_model_async(model: Model, *, system_prompt: str, user_prompt: str) -> str:
     """Call the model directly and return the full text response.
 
     Prompt delivery: everything goes in a single user message with no
@@ -163,16 +206,13 @@ def _call_model(model: Model, *, system_prompt: str, user_prompt: str) -> str:
     full_prompt = f"{system_prompt}\n\n{user_prompt}"
     messages: Messages = [Message(role="user", content=[ContentBlock(text=full_prompt)])]
 
-    async def _stream() -> str:
-        chunks: list[str] = []
-        async for event in model.stream(messages):
-            if "contentBlockDelta" in event:
-                delta = event["contentBlockDelta"].get("delta", {})
-                if "text" in delta:
-                    chunks.append(delta["text"])
-        return "".join(chunks)
-
-    return run_async(_stream)
+    chunks: list[str] = []
+    async for event in model.stream(messages):
+        if "contentBlockDelta" in event:
+            delta = event["contentBlockDelta"].get("delta", {})
+            if "text" in delta:
+                chunks.append(delta["text"])
+    return "".join(chunks)
 
 
 def _extract_json(text: str) -> str:
diff --git a/src/strands_evals/detectors/root_cause_analyzer.py b/src/strands_evals/detectors/root_cause_analyzer.py
index 0f460c7f..bee976e6 100644
--- a/src/strands_evals/detectors/root_cause_analyzer.py
+++ b/src/strands_evals/detectors/root_cause_analyzer.py
@@ -18,6 +18,7 @@
 from strands.types.exceptions import ContextWindowOverflowException
 from typing_extensions import cast
 
+from .._async import bounded_gather_fail_fast, run_async
 from ..types.detector import FailureItem, RCAItem, RCAOutput, RCAStructuredOutput
 from ..types.trace import Session, SpanUnion
 from .chunking import would_exceed_context
@@ -26,7 +27,7 @@
     RCA_MIN_WINDOW_SIZE,
     RCA_WINDOW_SPLIT_FACTOR,
 )
-from .failure_detector import detect_failures
+from .failure_detector import detect_failures_async
 from .prompt_templates.root_cause import get_merge_template, get_template
 from .utils import (
     _is_context_exceeded,
@@ -50,10 +51,9 @@ def analyze_root_cause(
 ) -> RCAOutput:
     """Perform root cause analysis on detected failures in a session.
 
-    Uses a 3-tier fallback strategy for handling large sessions:
-    1. Direct: analyze full session in one LLM call
-    2. Pruned: keep only spans on failure paths, retry
-    3. Chunked: split pruned session into windows, analyze each, merge
+    Synchronous wrapper around `analyze_root_cause_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 analyze.
@@ -67,8 +67,36 @@ def analyze_root_cause(
         causality, propagation_impact, root_cause_explanation, fix_type, and
         fix_recommendation.
     """
+    return run_async(lambda: analyze_root_cause_async(session, failures, model=model))
+
+
+async def analyze_root_cause_async(
+    session: Session,
+    failures: list[FailureItem] | None = None,
+    *,
+    model: Model | str | None = None,
+    max_workers: int = 5,
+) -> RCAOutput:
+    """Async variant of `analyze_root_cause`.
+
+    Chunked RCA fans windows out concurrently, capped at `max_workers`.
+
+    Uses a 3-tier fallback strategy for handling large sessions:
+    1. Direct: analyze full session in one LLM call
+    2. Pruned: keep only spans on failure paths, retry
+    3. Chunked: split pruned session into windows, analyze each, merge
+
+    Args:
+        session: The Session object to analyze.
+        failures: List of FailureItems. If None, detection runs automatically.
+        model: A Model instance, model ID string, or None.
+        max_workers: Maximum concurrent LLM calls during chunked RCA.
+
+    Returns:
+        RCAOutput. See `analyze_root_cause`.
+    """
     if failures is None:
-        failures = detect_failures(session, model=model).failures
+        failures = (await detect_failures_async(session, model=model, max_workers=max_workers)).failures
     if not failures:
         return RCAOutput()
 
@@ -85,7 +113,7 @@ def analyze_root_cause(
 
     if not would_exceed_context(system_prompt):
         try:
-            raw = _rca_direct(system_prompt, effective_model)
+            raw = await _rca_direct_async(system_prompt, effective_model)
             return RCAOutput(root_causes=raw)
         except Exception as e:
             if not _is_context_exceeded(e):
@@ -104,7 +132,7 @@ def analyze_root_cause(
 
     if not would_exceed_context(pruned_prompt):
         try:
-            raw = _rca_direct(pruned_prompt, effective_model)
+            raw = await _rca_direct_async(pruned_prompt, effective_model)
             return RCAOutput(root_causes=raw)
         except Exception as e:
             if not _is_context_exceeded(e):
@@ -112,26 +140,27 @@ def analyze_root_cause(
             logger.warning("Pruned session still exceeds context, falling back to chunking")
 
     # Tier 3: Chunked analysis with merge
-    raw = _rca_chunked(pruned_session, failures, effective_model)
+    raw = await _rca_chunked_async(pruned_session, failures, effective_model, max_workers)
     return RCAOutput(root_causes=raw)
 
 
-def _rca_direct(system_prompt: str, model: Model) -> list[RCAItem]:
+async def _rca_direct_async(system_prompt: str, model: Model) -> list[RCAItem]:
     agent = Agent(
         model=model,
         system_prompt=system_prompt,
         callback_handler=None,
     )
-    result = agent(_TEMPLATE.USER_PROMPT, structured_output_model=RCAStructuredOutput)
+    result = await agent.invoke_async(_TEMPLATE.USER_PROMPT, structured_output_model=RCAStructuredOutput)
     return _parse_structured_result(cast(RCAStructuredOutput, result.structured_output))
 
 
-def _rca_chunked(
+async def _rca_chunked_async(
     pruned_session: Session,
     failures: list[FailureItem],
     model: Model,
+    max_workers: int,
 ) -> list[RCAItem]:
-    """Split pruned session into per-trace windows, analyze each, merge results."""
+    """Split pruned session into per-trace windows, analyze each concurrently, merge."""
     total_spans = sum(len(t.spans) for t in pruned_session.traces)
 
     if total_spans <= RCA_MIN_WINDOW_SIZE:
@@ -147,40 +176,45 @@ def _rca_chunked(
         window_size,
     )
 
-    chunk_results: list[str] = []
-    window_num = 0
-
+    windows: list[list[SpanUnion]] = []
     for trace in pruned_session.traces:
         if not trace.spans:
             continue
-
         for i in range(0, len(trace.spans), window_size):
-            window_spans = trace.spans[i : i + window_size]
-            window_num += 1
+            windows.append(trace.spans[i : i + window_size])
 
-            window_span_ids = {_get_span_id(s) for s in window_spans}
-            window_failures = [f for f in failures if f.span_id in window_span_ids]
-            window_failures_json = _serialize_failures(window_failures) if window_failures else failures_json
+    async def _process_window(window_num: int, window_spans: list[SpanUnion]) -> str | None:
+        window_span_ids = {_get_span_id(s) for s in window_spans}
+        window_failures = [f for f in failures if f.span_id in window_span_ids]
+        window_failures_json = _serialize_failures(window_failures) if window_failures else failures_json
 
-            window_json = _serialize_spans(window_spans)
-            system_prompt = _TEMPLATE.build_prompt(
-                execution_json=window_json,
-                execution_failures_json=window_failures_json,
-            )
+        window_json = _serialize_spans(window_spans)
+        system_prompt = _TEMPLATE.build_prompt(
+            execution_json=window_json,
+            execution_failures_json=window_failures_json,
+        )
+
+        try:
+            agent = Agent(model=model, system_prompt=system_prompt, callback_handler=None)
+            result = await agent.invoke_async(_TEMPLATE.USER_PROMPT, structured_output_model=RCAStructuredOutput)
+            structured = cast(RCAStructuredOutput, result.structured_output)
+            logger.info("RCA chunk %d: processed %d spans", window_num, len(window_spans))
+            return structured.model_dump_json(by_alias=True, indent=2)
+        except ContextWindowOverflowException:
+            logger.warning("RCA chunk %d still too large, skipping", window_num)
+            return None
+        except Exception as e:
+            if _is_context_exceeded(e):
+                logger.warning("RCA chunk %d context exceeded (string match), skipping", window_num)
+                return None
+            raise
+
+    gathered = await bounded_gather_fail_fast(
+        (_process_window(i + 1, w) for i, w in enumerate(windows)),
+        max_workers,
+    )
 
-            try:
-                agent = Agent(model=model, system_prompt=system_prompt, callback_handler=None)
-                result = agent(_TEMPLATE.USER_PROMPT, structured_output_model=RCAStructuredOutput)
-                structured = cast(RCAStructuredOutput, result.structured_output)
-                chunk_results.append(structured.model_dump_json(by_alias=True, indent=2))
-                logger.info("RCA chunk %d: processed %d spans", window_num, len(window_spans))
-            except ContextWindowOverflowException:
-                logger.warning("RCA chunk %d still too large, skipping", window_num)
-            except Exception as e:
-                if _is_context_exceeded(e):
-                    logger.warning("RCA chunk %d context exceeded (string match), skipping", window_num)
-                else:
-                    raise
+    chunk_results: list[str] = [outcome for outcome in gathered if outcome is not None]
 
     if not chunk_results:
         return []
@@ -197,7 +231,7 @@ def _rca_chunked(
     )
 
     agent = Agent(model=model, system_prompt=merge_prompt, callback_handler=None)
-    result = agent(_MERGE_TEMPLATE.USER_PROMPT, structured_output_model=RCAStructuredOutput)
+    result = await agent.invoke_async(_MERGE_TEMPLATE.USER_PROMPT, structured_output_model=RCAStructuredOutput)
     return _parse_structured_result(cast(RCAStructuredOutput, result.structured_output))
 
 
diff --git a/src/strands_evals/experiment.py b/src/strands_evals/experiment.py
index de42b984..d024a4e8 100644
--- a/src/strands_evals/experiment.py
+++ b/src/strands_evals/experiment.py
@@ -17,7 +17,7 @@
 from typing_extensions import Any, Generic
 
 from .case import Case
-from .detectors.diagnosis import diagnose_session
+from .detectors.diagnosis import diagnose_session_async
 from .evaluation_data_store import EvaluationDataStore
 from .evaluators.coherence_evaluator import CoherenceEvaluator
 from .evaluators.conciseness_evaluator import ConcisenessEvaluator
@@ -471,11 +471,11 @@ async def _run_diagnosis(
             return None, None
 
         try:
-            result = await asyncio.to_thread(
-                diagnose_session,
+            result = await diagnose_session_async(
                 trajectory,
                 model=self._diagnosis_config.model,
                 confidence_threshold=self._diagnosis_config.confidence_threshold,
+                max_workers=self._diagnosis_config.max_workers,
             )
             recs = result.recommendations
             recs_str = "; ".join(recs) if recs else None
diff --git a/src/strands_evals/types/detector.py b/src/strands_evals/types/detector.py
index 307fac97..a4960c48 100644
--- a/src/strands_evals/types/detector.py
+++ b/src/strands_evals/types/detector.py
@@ -33,11 +33,15 @@ class DiagnosisConfig(BaseModel):
         trigger: When to run diagnosis — "on_failure" or "always".
         model: The model to use for diagnosis.
         confidence_threshold: Minimum confidence level for failure detection.
+        max_workers: Maximum concurrent LLM calls inside per-session chunked
+            detection / RCA. Defaults to 5; raise for fast targets / generous
+            TPM budgets, drop to 1 for deterministic ordering.
     """
 
     trigger: DiagnosisTrigger = DiagnosisTrigger.ON_FAILURE
     model: Model | str | None = None
     confidence_threshold: ConfidenceLevel = ConfidenceLevel.MEDIUM
+    max_workers: int = Field(default=5, ge=1)
 
     model_config = {"arbitrary_types_allowed": True}
 
diff --git a/tests/strands_evals/detectors/test_diagnosis.py b/tests/strands_evals/detectors/test_diagnosis.py
index 48834806..db8ad39a 100644
--- a/tests/strands_evals/detectors/test_diagnosis.py
+++ b/tests/strands_evals/detectors/test_diagnosis.py
@@ -1,9 +1,13 @@
 """Tests for diagnosis module."""
 
 from datetime import datetime
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
-from strands_evals.detectors.diagnosis import diagnose_session
+from strands_evals.detectors.diagnosis import (
+    diagnose_session,
+    diagnose_session_async,
+    diagnose_sessions_async,
+)
 from strands_evals.types.detector import (
     ConfidenceLevel,
     DiagnosisResult,
@@ -24,106 +28,105 @@
 )
 
 
-def _span_info(span_id: str = "span_1") -> SpanInfo:
+def _span_info(span_id: str = "span_1", session_id: str = "sess_1") -> SpanInfo:
     now = datetime.now()
-    return SpanInfo(session_id="sess_1", span_id=span_id, trace_id="trace_1", start_time=now, end_time=now)
+    return SpanInfo(session_id=session_id, span_id=span_id, trace_id="trace_1", start_time=now, end_time=now)
 
 
-def _make_session() -> Session:
+def _make_session(session_id: str = "sess_1") -> Session:
     spans = [
         AgentInvocationSpan(
-            span_info=_span_info("span_1"),
+            span_info=_span_info("span_1", session_id),
             user_prompt="Hello",
             agent_response="Hi there",
             available_tools=[ToolConfig(name="search")],
         ),
         InferenceSpan(
-            span_info=_span_info("span_2"),
+            span_info=_span_info("span_2", session_id),
             messages=[UserMessage(content=[TextContent(text="Hello")])],
         ),
     ]
     return Session(
-        session_id="sess_1",
-        traces=[Trace(trace_id="trace_1", session_id="sess_1", spans=spans)],
+        session_id=session_id,
+        traces=[Trace(trace_id="trace_1", session_id=session_id, spans=spans)],
+    )
+
+
+def _failure(span_id: str = "span_1") -> FailureItem:
+    return FailureItem(span_id=span_id, category=["hallucination"], confidence=[0.9], evidence=["made up data"])
+
+
+def _rca(span_id: str = "span_1") -> RCAItem:
+    return RCAItem(
+        failure_span_id=span_id,
+        location=span_id,
+        causality="PRIMARY_FAILURE",
+        propagation_impact=["QUALITY_DEGRADATION"],
+        failure_detection_timing="IMMEDIATELY_AT_OCCURRENCE",
+        completion_status="PARTIAL_SUCCESS",
+        root_cause_explanation="Bad tool output",
+        fix_type="TOOL_DESCRIPTION_FIX",
+        fix_recommendation="Improve tool description",
     )
 
 
 class TestDiagnoseSession:
-    @patch("strands_evals.detectors.diagnosis.detect_failures")
+    """Sync `diagnose_session` is a thin asyncio.run shim — patch the async impls."""
+
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
     def test_no_failures_skips_rca(self, mock_detect):
         mock_detect.return_value = FailureOutput(session_id="sess_1", failures=[])
-        session = _make_session()
 
-        result = diagnose_session(session)
+        result = diagnose_session(_make_session())
 
         assert isinstance(result, DiagnosisResult)
         assert result.session_id == "sess_1"
         assert result.failures == []
         assert result.root_causes == []
 
-    @patch("strands_evals.detectors.diagnosis.analyze_root_cause")
-    @patch("strands_evals.detectors.diagnosis.detect_failures")
+    @patch("strands_evals.detectors.diagnosis.analyze_root_cause_async", new_callable=AsyncMock)
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
     def test_with_failures_runs_rca(self, mock_detect, mock_rca):
-        failures = [
-            FailureItem(span_id="span_1", category=["hallucination"], confidence=[0.9], evidence=["made up data"])
-        ]
-        mock_detect.return_value = FailureOutput(session_id="sess_1", failures=failures)
-        rca_items = [
-            RCAItem(
-                failure_span_id="span_1",
-                location="span_1",
-                causality="PRIMARY_FAILURE",
-                propagation_impact=["QUALITY_DEGRADATION"],
-                failure_detection_timing="IMMEDIATELY_AT_OCCURRENCE",
-                completion_status="PARTIAL_SUCCESS",
-                root_cause_explanation="Bad tool output",
-                fix_type="TOOL_DESCRIPTION_FIX",
-                fix_recommendation="Improve tool description",
-            )
-        ]
-        mock_rca.return_value = RCAOutput(root_causes=rca_items)
-        session = _make_session()
+        mock_detect.return_value = FailureOutput(session_id="sess_1", failures=[_failure()])
+        mock_rca.return_value = RCAOutput(root_causes=[_rca()])
 
-        result = diagnose_session(session)
+        result = diagnose_session(_make_session())
 
         assert len(result.failures) == 1
         assert len(result.root_causes) == 1
         assert result.root_causes[0].fix_recommendation == "Improve tool description"
-        mock_rca.assert_called_once()
+        mock_rca.assert_awaited_once()
 
-    @patch("strands_evals.detectors.diagnosis.detect_failures")
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
     def test_passes_model_and_threshold(self, mock_detect):
         mock_detect.return_value = FailureOutput(session_id="sess_1", failures=[])
-        session = _make_session()
         mock_model = MagicMock()
 
-        diagnose_session(session, model=mock_model, confidence_threshold=ConfidenceLevel.HIGH)
+        diagnose_session(_make_session(), model=mock_model, confidence_threshold=ConfidenceLevel.HIGH)
 
-        mock_detect.assert_called_once_with(
-            session,
-            confidence_threshold=ConfidenceLevel.HIGH,
-            model=mock_model,
-        )
+        kwargs = mock_detect.call_args.kwargs
+        assert kwargs["confidence_threshold"] == ConfidenceLevel.HIGH
+        assert kwargs["model"] is mock_model
 
-    @patch("strands_evals.detectors.diagnosis.analyze_root_cause")
-    @patch("strands_evals.detectors.diagnosis.detect_failures")
+    @patch("strands_evals.detectors.diagnosis.analyze_root_cause_async", new_callable=AsyncMock)
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
     def test_passes_model_to_rca(self, mock_detect, mock_rca):
-        failures = [FailureItem(span_id="s1", category=["error"], confidence=[0.9], evidence=["e"])]
-        mock_detect.return_value = FailureOutput(session_id="sess_1", failures=failures)
+        mock_detect.return_value = FailureOutput(session_id="sess_1", failures=[_failure("s1")])
         mock_rca.return_value = RCAOutput(root_causes=[])
-        session = _make_session()
         mock_model = MagicMock()
 
-        diagnose_session(session, model=mock_model)
+        diagnose_session(_make_session(), model=mock_model)
 
-        mock_rca.assert_called_once_with(session, failures=failures, model=mock_model)
+        mock_rca.assert_awaited_once()
+        kwargs = mock_rca.call_args.kwargs
+        assert kwargs["model"] is mock_model
+        assert kwargs["failures"] == [_failure("s1")]
 
-    @patch("strands_evals.detectors.diagnosis.detect_failures")
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
     def test_serialization_round_trip(self, mock_detect):
         mock_detect.return_value = FailureOutput(session_id="sess_1", failures=[])
-        session = _make_session()
 
-        result = diagnose_session(session)
+        result = diagnose_session(_make_session())
         dumped = result.model_dump()
 
         assert dumped["session_id"] == "sess_1"
@@ -132,3 +135,95 @@ def test_serialization_round_trip(self, mock_detect):
 
         restored = DiagnosisResult.model_validate(dumped)
         assert restored == result
+
+
+class TestDiagnoseSessionAsync:
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
+    async def test_no_failures_skips_rca(self, mock_detect):
+        mock_detect.return_value = FailureOutput(session_id="sess_1", failures=[])
+
+        result = await diagnose_session_async(_make_session())
+
+        assert isinstance(result, DiagnosisResult)
+        assert result.failures == []
+        assert result.root_causes == []
+
+    @patch("strands_evals.detectors.diagnosis.analyze_root_cause_async", new_callable=AsyncMock)
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
+    async def test_with_failures_runs_rca(self, mock_detect, mock_rca):
+        mock_detect.return_value = FailureOutput(session_id="sess_1", failures=[_failure()])
+        mock_rca.return_value = RCAOutput(root_causes=[_rca()])
+
+        result = await diagnose_session_async(_make_session())
+
+        assert len(result.failures) == 1
+        assert len(result.root_causes) == 1
+
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
+    async def test_threads_max_workers(self, mock_detect):
+        mock_detect.return_value = FailureOutput(session_id="sess_1", failures=[])
+
+        await diagnose_session_async(_make_session(), max_workers=3)
+
+        assert mock_detect.call_args.kwargs["max_workers"] == 3
+
+
+class TestDiagnoseSessionsAsync:
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
+    async def test_batch_runs_in_order(self, mock_detect):
+        mock_detect.side_effect = [
+            FailureOutput(session_id="a", failures=[]),
+            FailureOutput(session_id="b", failures=[]),
+            FailureOutput(session_id="c", failures=[]),
+        ]
+        sessions = [_make_session("a"), _make_session("b"), _make_session("c")]
+
+        results = await diagnose_sessions_async(sessions)
+
+        assert len(results) == 3
+        ids = [r.session_id for r in results]
+        assert ids == ["a", "b", "c"]
+
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
+    async def test_one_session_failure_returns_none_for_that_slot(self, mock_detect):
+        mock_detect.side_effect = [
+            FailureOutput(session_id="ok1", failures=[]),
+            RuntimeError("LLM down"),
+            FailureOutput(session_id="ok2", failures=[]),
+        ]
+        sessions = [_make_session("ok1"), _make_session("bad"), _make_session("ok2")]
+
+        results = await diagnose_sessions_async(sessions)
+
+        assert len(results) == 3
+        assert results[0] is not None and results[0].session_id == "ok1"
+        assert results[1] is None
+        assert results[2] is not None and results[2].session_id == "ok2"
+
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
+    async def test_threads_per_session_workers(self, mock_detect):
+        mock_detect.return_value = FailureOutput(session_id="x", failures=[])
+
+        await diagnose_sessions_async([_make_session("x")], per_session_workers=7)
+
+        assert mock_detect.call_args.kwargs["max_workers"] == 7
+
+
+class TestSyncWrappersFromInsideEventLoop:
+    """The sync wrappers must work when called from inside a running loop
+    (Jupyter, FastAPI handlers, pytest-asyncio in `auto` mode). Because
+    `pyproject.toml` sets `asyncio_mode = "auto"`, every `async def test_...`
+    in this suite runs inside an event loop, which is exactly the regression
+    case we want to lock in.
+    """
+
+    @patch("strands_evals.detectors.diagnosis.detect_failures_async", new_callable=AsyncMock)
+    async def test_diagnose_session_works_inside_running_loop(self, mock_detect):
+        mock_detect.return_value = FailureOutput(session_id="sess_1", failures=[])
+
+        # The sync wrapper, called from inside this async test, must NOT raise
+        # `RuntimeError: asyncio.run() cannot be called from a running event loop`.
+        result = diagnose_session(_make_session())
+
+        assert isinstance(result, DiagnosisResult)
+        assert result.failures == []
diff --git a/tests/strands_evals/detectors/test_failure_detector.py b/tests/strands_evals/detectors/test_failure_detector.py
index dfe8ecae..58e24185 100644
--- a/tests/strands_evals/detectors/test_failure_detector.py
+++ b/tests/strands_evals/detectors/test_failure_detector.py
@@ -2,7 +2,7 @@
 
 import json
 from datetime import datetime
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -10,6 +10,7 @@
     _extract_json,
     _parse_text_result,
     detect_failures,
+    detect_failures_async,
 )
 from strands_evals.detectors.utils import (
     _group_spans_into_traces,
@@ -226,7 +227,7 @@ def test_serialize_spans():
     assert "span_10" in result
 
 
-@patch("strands_evals.detectors.failure_detector._call_model")
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
 def test_detect_failures_no_failures(mock_call_model):
     mock_call_model.return_value = _make_json_response([])
 
@@ -238,7 +239,7 @@ def test_detect_failures_no_failures(mock_call_model):
     assert output.failures == []
 
 
-@patch("strands_evals.detectors.failure_detector._call_model")
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
 def test_detect_failures_with_failures(mock_call_model):
     mock_call_model.return_value = _make_json_response(
         [
@@ -259,7 +260,7 @@ def test_detect_failures_with_failures(mock_call_model):
     assert output.failures[0].confidence == [0.9]
 
 
-@patch("strands_evals.detectors.failure_detector._call_model")
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
 def test_detect_failures_confidence_threshold(mock_call_model):
     mock_call_model.return_value = _make_json_response(
         [
@@ -276,7 +277,7 @@ def test_detect_failures_confidence_threshold(mock_call_model):
     assert output.failures[0].span_id == "span_2"
 
 
-@patch("strands_evals.detectors.failure_detector._call_model")
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
 def test_detect_failures_per_mode_filtering(mock_call_model):
     """Individual failure modes below threshold are pruned, not just whole spans."""
     mock_call_model.return_value = _make_json_response(
@@ -300,7 +301,7 @@ def test_detect_failures_per_mode_filtering(mock_call_model):
     assert "repetition" not in output.failures[0].category
 
 
-@patch("strands_evals.detectors.failure_detector._call_model")
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
 def test_detect_failures_context_overflow_fallback(mock_call_model):
     """When direct call raises context overflow, should fall back to chunking."""
     from strands.types.exceptions import ContextWindowOverflowException
@@ -318,7 +319,7 @@ def test_detect_failures_context_overflow_fallback(mock_call_model):
     assert isinstance(output, FailureOutput)
 
 
-@patch("strands_evals.detectors.failure_detector._call_model")
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
 def test_detect_failures_non_context_error_raises(mock_call_model):
     """Non-context errors should propagate."""
     mock_call_model.side_effect = RuntimeError("Something else broke")
@@ -329,7 +330,7 @@ def test_detect_failures_non_context_error_raises(mock_call_model):
 
 
 @patch("strands_evals.detectors.failure_detector._resolve_model")
-@patch("strands_evals.detectors.failure_detector._call_model")
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
 def test_detect_failures_passes_model(mock_call_model, mock_resolve_model):
     mock_model = MagicMock()
     mock_resolve_model.return_value = mock_model
@@ -430,3 +431,212 @@ def _make_tool_span_with_trace(span_id: str, trace_id: str) -> ToolExecutionSpan
         tool_call=ToolCall(name="t", arguments={}),
         tool_result=ToolResult(content="ok"),
     )
+
+
+# --- async tests ---
+
+
+def _make_multi_span_session(n_spans: int = 4) -> Session:
+    """A session with enough spans to clear MIN_CHUNK_SIZE guard in _detect_chunked_async."""
+    spans = [
+        ToolExecutionSpan(
+            span_info=_span_info(f"span_{i}"),
+            tool_call=ToolCall(name="t", arguments={}),
+            tool_result=ToolResult(content="ok"),
+        )
+        for i in range(n_spans)
+    ]
+    return Session(
+        session_id="sess_1",
+        traces=[Trace(trace_id="trace_1", session_id="sess_1", spans=spans)],
+    )
+
+
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
+async def test_detect_failures_async_no_failures(mock_call_model):
+    mock_call_model.return_value = _make_json_response([])
+
+    output = await detect_failures_async(_make_session())
+
+    assert isinstance(output, FailureOutput)
+    assert output.failures == []
+
+
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
+async def test_detect_failures_async_with_failures(mock_call_model):
+    mock_call_model.return_value = _make_json_response(
+        [
+            {
+                "location": "span_1",
+                "category": ["err"],
+                "confidence": ["high"],
+                "evidence": ["bad"],
+            }
+        ]
+    )
+
+    output = await detect_failures_async(_make_session())
+
+    assert len(output.failures) == 1
+    assert output.failures[0].span_id == "span_1"
+
+
+@patch("strands_evals.detectors.failure_detector._detect_chunked_async", new_callable=AsyncMock)
+@patch("strands_evals.detectors.failure_detector.would_exceed_context")
+async def test_detect_failures_async_passes_max_workers_to_chunked(mock_would_exceed, mock_chunked):
+    """When the session triggers chunked detection, max_workers is forwarded."""
+    mock_would_exceed.return_value = True
+    mock_chunked.return_value = []
+
+    await detect_failures_async(_make_session(), max_workers=4)
+
+    mock_chunked.assert_awaited_once()
+    # signature: _detect_chunked_async(session, model, template, max_workers)
+    assert mock_chunked.call_args.args[3] == 4
+
+
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
+@patch("strands_evals.detectors.failure_detector.split_spans_by_tokens")
+@patch("strands_evals.detectors.failure_detector.would_exceed_context")
+async def test_detect_failures_chunked_concurrent_calls(mock_would_exceed, mock_split, mock_call_model):
+    """Chunked path issues per-chunk LLM calls under the gather; with max_workers > 1
+    they should overlap. Asserts that >=2 chunks are in flight at once."""
+    import asyncio
+
+    mock_would_exceed.return_value = True
+    # 3 fake chunks, each containing one span
+    session = _make_multi_span_session(4)
+    spans = session.traces[0].spans
+    mock_split.return_value = [[spans[0]], [spans[1]], [spans[2]]]
+
+    in_flight = 0
+    peak = 0
+    enter = asyncio.Event()
+
+    async def fake_call(*args, **kwargs):
+        nonlocal in_flight, peak
+        in_flight += 1
+        peak = max(peak, in_flight)
+        # First chunk waits until a sibling has entered, proving overlap
+        if not enter.is_set():
+            enter.set()
+        await asyncio.sleep(0.01)
+        in_flight -= 1
+        return _make_json_response([])
+
+    mock_call_model.side_effect = fake_call
+
+    await detect_failures_async(session, max_workers=3)
+
+    assert peak >= 2, f"expected concurrent chunks; observed peak={peak}"
+
+
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
+@patch("strands_evals.detectors.failure_detector.split_spans_by_tokens")
+@patch("strands_evals.detectors.failure_detector.would_exceed_context")
+async def test_detect_failures_chunked_one_overflow_does_not_kill_batch(mock_would_exceed, mock_split, mock_call_model):
+    """One chunk raising ContextWindowOverflowException is logged and skipped;
+    sibling chunks still produce results."""
+    from strands.types.exceptions import ContextWindowOverflowException
+
+    mock_would_exceed.return_value = True
+    session = _make_multi_span_session(4)
+    spans = session.traces[0].spans
+    mock_split.return_value = [[spans[0]], [spans[1]], [spans[2]]]
+
+    mock_call_model.side_effect = [
+        _make_json_response([{"location": "span_1", "category": ["e"], "confidence": ["high"], "evidence": ["x"]}]),
+        ContextWindowOverflowException("nope"),
+        _make_json_response([{"location": "span_1", "category": ["e"], "confidence": ["high"], "evidence": ["y"]}]),
+    ]
+
+    output = await detect_failures_async(session, max_workers=1)
+
+    # Two surviving chunks, each contributing the same span — merge_chunk_failures dedupes.
+    assert len(output.failures) >= 1
+    assert output.failures[0].span_id == "span_1"
+
+
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
+@patch("strands_evals.detectors.failure_detector.split_spans_by_tokens")
+@patch("strands_evals.detectors.failure_detector.would_exceed_context")
+async def test_detect_failures_chunked_non_context_error_propagates(mock_would_exceed, mock_split, mock_call_model):
+    """A non-context exception in any chunk should raise out of the gather."""
+    mock_would_exceed.return_value = True
+    session = _make_multi_span_session(4)
+    spans = session.traces[0].spans
+    mock_split.return_value = [[spans[0]], [spans[1]]]
+
+    mock_call_model.side_effect = [
+        _make_json_response([]),
+        RuntimeError("Something else broke"),
+    ]
+
+    with pytest.raises(RuntimeError, match="Something else broke"):
+        await detect_failures_async(session, max_workers=1)
+
+
+# --- sync-wrapper-from-inside-event-loop regression ---
+
+
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
+async def test_detect_failures_sync_works_inside_running_loop(mock_call_model):
+    """`detect_failures` (the sync wrapper) must be safe to call from inside an
+    async function. `pyproject.toml` sets `asyncio_mode = "auto"`, so this test
+    body runs inside an event loop -- exactly the regression case.
+    """
+    mock_call_model.return_value = _make_json_response([])
+
+    output = detect_failures(_make_session())
+
+    assert isinstance(output, FailureOutput)
+    assert output.failures == []
+
+
+
+@patch("strands_evals.detectors.failure_detector._call_model_async", new_callable=AsyncMock)
+@patch("strands_evals.detectors.failure_detector.split_spans_by_tokens")
+@patch("strands_evals.detectors.failure_detector.would_exceed_context")
+async def test_detect_failures_chunked_fails_fast_cancels_siblings(
+    mock_would_exceed, mock_split, mock_call_model
+):
+    """When one chunk raises a non-context error, sibling chunks still in
+    flight must be cancelled before they complete their LLM call -- otherwise
+    we burn tokens on work whose result we'll throw away."""
+    import asyncio
+
+    mock_would_exceed.return_value = True
+    session = _make_multi_span_session(4)
+    spans = session.traces[0].spans
+    mock_split.return_value = [[spans[0]], [spans[1]], [spans[2]]]
+
+    sibling_finished = False
+    sibling_started = asyncio.Event()
+
+    async def fake_call(*args, **kwargs):
+        # Chunk 0 starts first and is still mid-flight when chunk 1 raises.
+        # Chunk 2 (queued behind the semaphore) is also still pending.
+        if not sibling_started.is_set():
+            sibling_started.set()
+            try:
+                await asyncio.sleep(10)
+            except asyncio.CancelledError:
+                raise
+            nonlocal sibling_finished
+            sibling_finished = True
+            return _make_json_response([])
+        # Subsequent calls: the failing chunk waits until sibling has
+        # actually started, then raises.
+        await sibling_started.wait()
+        raise RuntimeError("hard fail")
+
+    mock_call_model.side_effect = fake_call
+
+    with pytest.raises(RuntimeError, match="hard fail"):
+        await detect_failures_async(session, max_workers=3)
+
+    await asyncio.sleep(0)
+    assert sibling_finished is False, (
+        "fail-fast chunked detection must cancel the slow sibling, "
+        "not let it complete its LLM call"
+    )
diff --git a/tests/strands_evals/detectors/test_root_cause_analyzer.py b/tests/strands_evals/detectors/test_root_cause_analyzer.py
index 9ac139ec..f4e57ad9 100644
--- a/tests/strands_evals/detectors/test_root_cause_analyzer.py
+++ b/tests/strands_evals/detectors/test_root_cause_analyzer.py
@@ -1,7 +1,7 @@
 """Tests for root cause analyzer."""
 
 from datetime import datetime
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -11,6 +11,7 @@
     _prune_session_to_failure_paths,
     _trace_to_roots,
     analyze_root_cause,
+    analyze_root_cause_async,
 )
 from strands_evals.detectors.utils import _is_context_exceeded
 from strands_evals.types.detector import (
@@ -303,13 +304,18 @@ def test_analyze_root_cause_empty_failures():
     assert output.root_causes == []
 
 
+def _agent_with_result(structured_output):
+    """Build a mocked Agent whose .invoke_async returns a result with the given structured_output."""
+    agent = MagicMock()
+    result = MagicMock()
+    result.structured_output = structured_output
+    agent.invoke_async = AsyncMock(return_value=result)
+    return agent
+
+
 @patch("strands_evals.detectors.root_cause_analyzer.Agent")
 def test_analyze_root_cause_direct(mock_agent_cls):
-    mock_agent = MagicMock()
-    mock_agent_cls.return_value = mock_agent
-    mock_result = MagicMock()
-    mock_result.structured_output = _make_rca_structured_output(["span_2"])
-    mock_agent.return_value = mock_result
+    mock_agent_cls.return_value = _agent_with_result(_make_rca_structured_output(["span_2"]))
 
     session = _make_session()
     failures = _make_failures(["span_2"])
@@ -326,11 +332,7 @@ def test_analyze_root_cause_direct(mock_agent_cls):
 def test_analyze_root_cause_passes_model(mock_resolve, mock_agent_cls):
     mock_model = MagicMock()
     mock_resolve.return_value = mock_model
-    mock_agent = MagicMock()
-    mock_agent_cls.return_value = mock_agent
-    mock_result = MagicMock()
-    mock_result.structured_output = _make_rca_structured_output(["span_2"])
-    mock_agent.return_value = mock_result
+    mock_agent_cls.return_value = _agent_with_result(_make_rca_structured_output(["span_2"]))
 
     session = _make_session()
     failures = _make_failures(["span_2"])
@@ -345,15 +347,14 @@ def test_analyze_root_cause_passes_model(mock_resolve, mock_agent_cls):
 def test_analyze_root_cause_context_overflow_falls_to_pruning(mock_agent_cls):
     from strands.types.exceptions import ContextWindowOverflowException
 
-    mock_agent = MagicMock()
-    mock_agent_cls.return_value = mock_agent
-
     pruned_result = MagicMock()
     pruned_result.structured_output = _make_rca_structured_output(["span_2"])
-    mock_agent.side_effect = [
-        ContextWindowOverflowException("too big"),
-        pruned_result,
-    ]
+
+    overflow_agent = MagicMock()
+    overflow_agent.invoke_async = AsyncMock(side_effect=ContextWindowOverflowException("too big"))
+    pruned_agent = MagicMock()
+    pruned_agent.invoke_async = AsyncMock(return_value=pruned_result)
+    mock_agent_cls.side_effect = [overflow_agent, pruned_agent]
 
     session = _make_session()
     failures = _make_failures(["span_2"])
@@ -365,9 +366,9 @@ def test_analyze_root_cause_context_overflow_falls_to_pruning(mock_agent_cls):
 
 @patch("strands_evals.detectors.root_cause_analyzer.Agent")
 def test_analyze_root_cause_non_context_error_raises(mock_agent_cls):
-    mock_agent = MagicMock()
-    mock_agent_cls.return_value = mock_agent
-    mock_agent.side_effect = RuntimeError("Something else broke")
+    agent = MagicMock()
+    agent.invoke_async = AsyncMock(side_effect=RuntimeError("Something else broke"))
+    mock_agent_cls.return_value = agent
 
     session = _make_session()
     failures = _make_failures(["span_2"])
@@ -375,30 +376,26 @@ def test_analyze_root_cause_non_context_error_raises(mock_agent_cls):
         analyze_root_cause(session, failures)
 
 
-@patch("strands_evals.detectors.root_cause_analyzer.detect_failures")
+@patch("strands_evals.detectors.root_cause_analyzer.detect_failures_async", new_callable=AsyncMock)
 @patch("strands_evals.detectors.root_cause_analyzer.Agent")
 def test_analyze_root_cause_auto_detects_failures(mock_agent_cls, mock_detect):
-    """When failures=None, detect_failures is called automatically."""
+    """When failures=None, detect_failures_async is called automatically."""
     from strands_evals.types.detector import FailureOutput
 
     mock_detect.return_value = FailureOutput(
         session_id="sess_1",
         failures=_make_failures(["span_2"]),
     )
-    mock_agent = MagicMock()
-    mock_agent_cls.return_value = mock_agent
-    mock_result = MagicMock()
-    mock_result.structured_output = _make_rca_structured_output(["span_2"])
-    mock_agent.return_value = mock_result
+    mock_agent_cls.return_value = _agent_with_result(_make_rca_structured_output(["span_2"]))
 
     session = _make_session()
     output = analyze_root_cause(session, failures=None)
 
-    mock_detect.assert_called_once()
+    mock_detect.assert_awaited_once()
     assert len(output.root_causes) == 1
 
 
-@patch("strands_evals.detectors.root_cause_analyzer.detect_failures")
+@patch("strands_evals.detectors.root_cause_analyzer.detect_failures_async", new_callable=AsyncMock)
 def test_analyze_root_cause_auto_detect_no_failures(mock_detect):
     """When auto-detection finds no failures, return empty RCAOutput."""
     from strands_evals.types.detector import FailureOutput
@@ -420,3 +417,54 @@ def test_analyze_root_cause_multiple_root_causes():
     for i, sid in enumerate(["span_1", "span_2", "span_3"]):
         assert result[i].failure_span_id == sid
         assert result[i].root_cause_explanation == f"Failure at {sid} due to test issue"
+
+
+# --- async tests ---
+
+
+@patch("strands_evals.detectors.root_cause_analyzer.Agent")
+async def test_analyze_root_cause_async_direct(mock_agent_cls):
+    mock_agent_cls.return_value = _agent_with_result(_make_rca_structured_output(["span_2"]))
+
+    output = await analyze_root_cause_async(_make_session(), _make_failures(["span_2"]))
+
+    assert len(output.root_causes) == 1
+    assert output.root_causes[0].failure_span_id == "span_2"
+
+
+async def test_analyze_root_cause_async_empty_failures():
+    output = await analyze_root_cause_async(_make_session(), failures=[])
+    assert output.root_causes == []
+
+
+@patch("strands_evals.detectors.root_cause_analyzer.detect_failures_async", new_callable=AsyncMock)
+@patch("strands_evals.detectors.root_cause_analyzer.Agent")
+async def test_analyze_root_cause_async_auto_detects(mock_agent_cls, mock_detect):
+    from strands_evals.types.detector import FailureOutput
+
+    mock_detect.return_value = FailureOutput(session_id="sess_1", failures=_make_failures(["span_2"]))
+    mock_agent_cls.return_value = _agent_with_result(_make_rca_structured_output(["span_2"]))
+
+    output = await analyze_root_cause_async(_make_session(), failures=None, max_workers=2)
+
+    mock_detect.assert_awaited_once()
+    # max_workers should propagate into auto-detection
+    assert mock_detect.call_args.kwargs["max_workers"] == 2
+    assert len(output.root_causes) == 1
+
+
+# --- sync-wrapper-from-inside-event-loop regression ---
+
+
+@patch("strands_evals.detectors.root_cause_analyzer.Agent")
+async def test_analyze_root_cause_sync_works_inside_running_loop(mock_agent_cls):
+    """`analyze_root_cause` (the sync wrapper) must be safe to call from inside
+    an async function. `pyproject.toml` sets `asyncio_mode = "auto"`, so this
+    test body runs inside an event loop.
+    """
+    mock_agent_cls.return_value = _agent_with_result(_make_rca_structured_output(["span_2"]))
+
+    output = analyze_root_cause(_make_session(), _make_failures(["span_2"]))
+
+    assert isinstance(output, RCAOutput)
+    assert len(output.root_causes) == 1
diff --git a/tests/strands_evals/test_async.py b/tests/strands_evals/test_async.py
new file mode 100644
index 00000000..eb872fe6
--- /dev/null
+++ b/tests/strands_evals/test_async.py
@@ -0,0 +1,107 @@
+"""Tests for `strands_evals._async`."""
+
+import asyncio
+
+import pytest
+
+from strands_evals._async import bounded_gather, bounded_gather_fail_fast
+
+# --- bounded_gather (return_exceptions=True) ---
+
+
+async def test_bounded_gather_returns_results_in_order():
+    async def _v(x: int) -> int:
+        await asyncio.sleep(0)
+        return x
+
+    results = await bounded_gather((_v(i) for i in range(5)), max_workers=2)
+    assert results == [0, 1, 2, 3, 4]
+
+
+async def test_bounded_gather_isolates_failures():
+    async def _ok(x: int) -> int:
+        return x
+
+    async def _boom() -> int:
+        raise RuntimeError("boom")
+
+    results = await bounded_gather([_ok(1), _boom(), _ok(3)], max_workers=2)
+    assert results[0] == 1
+    assert isinstance(results[1], RuntimeError)
+    assert results[2] == 3
+
+
+async def test_bounded_gather_invalid_max_workers():
+    with pytest.raises(ValueError):
+        await bounded_gather([], 0)
+
+
+# --- bounded_gather_fail_fast ---
+
+
+async def test_bounded_gather_fail_fast_returns_results_in_order():
+    async def _v(x: int) -> int:
+        await asyncio.sleep(0)
+        return x
+
+    results = await bounded_gather_fail_fast((_v(i) for i in range(5)), max_workers=2)
+    assert results == [0, 1, 2, 3, 4]
+
+
+async def test_bounded_gather_fail_fast_empty():
+    assert await bounded_gather_fail_fast([], max_workers=3) == []
+
+
+async def test_bounded_gather_fail_fast_invalid_max_workers():
+    with pytest.raises(ValueError):
+        await bounded_gather_fail_fast([], 0)
+
+
+async def test_bounded_gather_fail_fast_cancels_siblings_on_first_error():
+    """The whole point of fail-fast: a sibling task that's still mid-flight
+    when another raises must be cancelled rather than allowed to complete.
+    """
+    sibling_completed = False
+    sibling_started = asyncio.Event()
+
+    async def _slow_sibling() -> str:
+        nonlocal sibling_completed
+        sibling_started.set()
+        try:
+            await asyncio.sleep(10)  # would never finish in a real test
+        except asyncio.CancelledError:
+            raise
+        sibling_completed = True
+        return "should-not-happen"
+
+    async def _fail_after_sibling_starts() -> str:
+        await sibling_started.wait()
+        raise RuntimeError("kaboom")
+
+    with pytest.raises(RuntimeError, match="kaboom"):
+        await bounded_gather_fail_fast(
+            [_slow_sibling(), _fail_after_sibling_starts()],
+            max_workers=2,
+        )
+
+    # Give the cancelled sibling a chance to settle.
+    await asyncio.sleep(0)
+    assert sibling_completed is False, "fail-fast must cancel the still-running sibling"
+
+
+async def test_bounded_gather_fail_fast_first_exception_wins():
+    """When two tasks fail near-simultaneously, the one observed first is the
+    one that surfaces. The contract is `wait(FIRST_EXCEPTION)`, not 'collect
+    every exception'."""
+
+    async def _fail(label: str, delay: float) -> str:
+        await asyncio.sleep(delay)
+        raise RuntimeError(label)
+
+    with pytest.raises(RuntimeError) as exc_info:
+        await bounded_gather_fail_fast(
+            [_fail("first", 0.0), _fail("second", 0.05)],
+            max_workers=2,
+        )
+
+    assert str(exc_info.value) == "first"
diff --git a/tests/strands_evals/test_experiment.py b/tests/strands_evals/test_experiment.py
index b8be325e..dbb04e18 100644
--- a/tests/strands_evals/test_experiment.py
+++ b/tests/strands_evals/test_experiment.py
@@ -2032,7 +2032,7 @@ def task_with_session(c):
         assert report.recommendations[af] == report.recommendations[m2] == "Fix the prompt"
         mock_run_diag.assert_called_once()
 
-    @patch("strands_evals.experiment.diagnose_session")
+    @patch("strands_evals.experiment.diagnose_session_async")
     def test_diagnosis_exception_returns_none(self, mock_diagnose):
         """If diagnosis throws, it should be caught and return None for both fields."""
         mock_diagnose.side_effect = RuntimeError("LLM unavailable")
@@ -2053,6 +2053,28 @@ def task_with_session(c):
         assert report.diagnoses == [None]
         assert report.recommendations == [None]
 
+    @patch("strands_evals.experiment.diagnose_session_async")
+    def test_diagnosis_threads_max_workers_from_config(self, mock_diagnose):
+        """`DiagnosisConfig.max_workers` must reach `diagnose_session_async`
+        so users can tune inner LLM concurrency without monkey-patching.
+        """
+        from strands_evals.types.detector import DiagnosisResult
+
+        mock_diagnose.return_value = DiagnosisResult(session_id="sess_1", failures=[], root_causes=[])
+        session = self._make_session()
+
+        cases = [Case(name="fail", input="foo", expected_output="bar")]
+        experiment = Experiment(
+            cases=cases,
+            evaluators=[MockEvaluator()],
+            diagnosis_config=DiagnosisConfig(max_workers=7),
+        )
+
+        experiment.run_evaluations(lambda c: {"output": c.input, "trajectory": session})
+
+        mock_diagnose.assert_called_once()
+        assert mock_diagnose.call_args.kwargs["max_workers"] == 7
+
 
 def test_run_evaluations_two_same_class_evaluators_with_distinct_names():
     """Two instances of the same class with `name=...` produce distinct rows."""