ShadowEvals¶

def preflight(
    self,
    test_case: ShadowTestCase | dict[str, Any],
    modality: str = "audio",
) -> ShadowPreflightResult:
    """Checks that a ShadowEval can faithfully replay its past conversation
    before any session is opened.

    Errors (the case should not run):
      - the past conversation trace cannot be loaded;
      - it has no caller turns;
      - in audio modality, none of the caller turns has a recording (the
        replay would be pure TTS).

    Warnings / info:
      - some caller turns have no recording;
      - the target app does not record audio (new calls will not be in
        GCS; use `artifacts_dir` to keep local copies);
      - no `session_parameters` / `use_tool_fakes` are set (replays often
        diverge when the original session relied on seeded state).
    """
    tc = (
        ShadowTestCase(**test_case)
        if isinstance(test_case, dict)
        else test_case
    )
    result = ShadowPreflightResult(
        name=tc.name, conversation_id=tc.conversation_id
    )
    source_app_name = self._resolve_source_app_name(tc)
    try:
        traces_client, normalized, bucket = self._open_past_conversation(tc)
    except Exception as exc:
        result.add(
            "error",
            f"Could not load past conversation '{tc.conversation_id}' "
            f"from {source_app_name}: {exc}. Check `project_id`, "
            "`location`, `app_id` and the conversation ID.",
        )
        return result

    spoken = self._spoken_user_turn_numbers(normalized)
    replayable = self._all_replayable_user_turn_numbers(normalized)
    result.past_user_turns = len(spoken) or len(replayable)
    if not replayable:
        result.add(
            "error",
            f"Past conversation '{tc.conversation_id}' has no caller turns "
            "to replay.",
        )
        return result

    if modality == "audio" and spoken:
        uris = self._list_turn_audio_uris(
            traces_client, tc.conversation_id, normalized, "user"
        )
        result.past_user_turns_with_audio = sum(
            1 for n in spoken if n in uris
        )
        bucket_label = bucket or "resolved from the source app"
        recording_hint = (
            "Set `gcs_bucket` to the bucket the original call was "
            f"recorded to (currently {bucket_label}) "
            "and make sure the source app had audio recording "
            "(loggingSettings.audioRecordingConfig) enabled at the time."
        )
        if result.past_user_turns_with_audio == 0:
            result.add(
                "error",
                "No recorded caller audio was found for any of the "
                f"{len(spoken)} caller turns, so the replay would be pure "
                f"TTS. {recording_hint}",
            )
        elif result.past_user_turns_with_audio < len(spoken):
            result.add(
                "warning",
                f"Only {result.past_user_turns_with_audio}/{len(spoken)} "
                "caller turns have recorded audio; the rest will be "
                f"synthesized via TTS. {recording_hint}",
            )
    if modality == "audio" and not self._target_recording_bucket():
        result.add(
            "info",
            f"Audio recording is not enabled on {self.app_name}, so "
            "new calls will not be recorded to GCS. Pass "
            "`artifacts_dir` to keep local copies of the replayed "
            "audio and a side-by-side HTML report.",
        )

    if not tc.session_parameters and tc.use_tool_fakes is None:
        result.add(
            "info",
            "No `session_parameters` or `use_tool_fakes` set. If the "
            "original session relied on seeded session variables or tool "
            "fakes (e.g. a mocked caller lookup), set them in the case or "
            "the `config:` block, or the replay may diverge.",
        )
    return result

def fetch_past_conversation_data(
    self,
    test_case: ShadowTestCase | dict[str, Any],
    session_dir: str | None = None,
) -> tuple[list[ShadowPastTurn], str, dict[str, Any]]:
    """Downloads past conversation transcript and per-turn user audio from
    GCS for a `ShadowTestCase`.

    Args:
        test_case: `ShadowTestCase` or dict specifying `conversation_id`
            (and optional `project_id`, `location`, `app_id`, `gcs_bucket`).
        session_dir: Optional local directory to store downloaded WAV files.

    Returns:
        Tuple of `(past_turns, full_past_history_text, normalized_trace)`.
    """
    tc = (
        ShadowTestCase(**test_case)
        if isinstance(test_case, dict)
        else test_case
    )
    traces_client, normalized, bucket_override = (
        self._open_past_conversation(tc)
    )
    user_audio_uris = self._list_turn_audio_uris(
        traces_client, tc.conversation_id, normalized, "user"
    )
    agent_audio_uris = self._list_turn_audio_uris(
        traces_client, tc.conversation_id, normalized, "agent"
    )

    gcs_client = GCSUtils(creds=self.creds) if user_audio_uris else None

    # Extract turns and full history from normalized conversation
    raw_conv = normalized.get("raw") or {}
    raw_turns = raw_conv.get("turns") or []

    past_turns: list[ShadowPastTurn] = []
    history_lines: list[str] = []
    last_agent_texts: list[str] = []
    user_turn_counter = 0

    if raw_turns:
        for turn_idx, p_turn in enumerate(raw_turns, start=1):
            turn_user_texts: list[str] = []
            turn_agent_texts: list[str] = []
            # Agent text spoken after the user in this turn (the reply).
            turn_agent_reply: list[str] = []

            for msg in p_turn.get("messages", []) or []:
                role = (msg.get("role") or "").strip()
                chunks = msg.get("chunks") or []
                for chunk in chunks:
                    text_val = (
                        chunk.get("text") or chunk.get("transcript") or ""
                    ).strip()
                    if text_val:
                        if role.lower() == "user":
                            turn_user_texts.append(text_val)
                            history_lines.append(
                                f"[Turn {turn_idx}] User: {text_val}"
                            )
                        else:
                            turn_agent_texts.append(text_val)
                            if turn_user_texts:
                                turn_agent_reply.append(text_val)
                            history_lines.append(
                                f"[Turn {turn_idx}] Agent ({role}): "
                                f"{text_val}"
                            )
                    if "tool_call" in chunk:
                        tc_chunk = chunk["tool_call"]
                        t_name = (
                            tc_chunk.get("display_name")
                            or tc_chunk.get("name")
                            or tc_chunk.get("tool")
                            or ""
                        )
                        t_args = tc_chunk.get("args", {})
                        history_lines.append(
                            f"[Turn {turn_idx}] Tool Call: {t_name} "
                            f"args={t_args}"
                        )
                    if "tool_response" in chunk:
                        tr_chunk = chunk["tool_response"]
                        t_name = (
                            tr_chunk.get("display_name")
                            or tr_chunk.get("name")
                            or tr_chunk.get("tool")
                            or ""
                        )
                        t_resp = tr_chunk.get("response", {})
                        history_lines.append(
                            f"[Turn {turn_idx}] Tool Response: {t_name} "
                            f"response={t_resp}"
                        )

            cleaned_user_text = _extract_user_turn_transcript(
                turn_user_texts
            )
            if cleaned_user_text:
                user_turn_counter += 1
                is_non_spoken = _is_non_spoken_turn_text(cleaned_user_text)
                # Platform recordings are numbered per 1-based
                # conversation turn (`user-turn-N` == raw turn N), so
                # prefer `turn_idx`; fall back to the user-turn counter.
                audio_uri = (
                    None
                    if is_non_spoken
                    else (
                        user_audio_uris.get(turn_idx)
                        or user_audio_uris.get(user_turn_counter)
                    )
                )
                audio_path = None
                pcm_bytes = None

                if audio_uri and gcs_client is not None:
                    try:
                        raw_wav = gcs_client.download_blob(audio_uri)
                        pcm_bytes = extract_pcm_bytes_from_wav(raw_wav)
                        if pcm_bytes and session_dir:
                            os.makedirs(session_dir, exist_ok=True)
                            audio_path = os.path.join(
                                session_dir,
                                f"past_user_turn_{user_turn_counter}.wav",
                            )
                            with open(audio_path, "wb") as f:
                                f.write(raw_wav)
                    except Exception as exc:
                        logger.warning(
                            "Failed to download user audio %s: %s",
                            audio_uri,
                            exc,
                        )

                past_turns.append(
                    ShadowPastTurn(
                        turn_index=user_turn_counter,
                        user_transcript=cleaned_user_text,
                        preceding_agent_response=" ".join(last_agent_texts),
                        audio_uri=audio_uri,
                        audio_path=audio_path,
                        has_audio=bool(pcm_bytes),
                        used=False,
                        audio_bytes=pcm_bytes or None,
                        agent_response=" ".join(turn_agent_reply),
                        agent_audio_uri=agent_audio_uris.get(turn_idx),
                    )
                )
                last_agent_texts = []

            if turn_agent_texts:
                last_agent_texts.extend(turn_agent_texts)
    else:
        # Fallback to flat `entries` if `raw.turns` is absent
        for entry in self._coalesce_user_entries(
            normalized.get("entries", []) or []
        ):
            kind = entry.get("kind")
            turn_num = int(entry.get("turn", 0)) + 1
            if kind == "user":
                u_text = (entry.get("text") or "").strip()
                history_lines.append(f"[Turn {turn_num}] User: {u_text}")
                cleaned_user_text = _extract_user_turn_transcript(u_text)
                if not cleaned_user_text:
                    continue
                user_turn_counter += 1
                is_non_spoken = _is_non_spoken_turn_text(cleaned_user_text)
                audio_uri = (
                    None
                    if is_non_spoken
                    else (
                        user_audio_uris.get(turn_num)
                        or user_audio_uris.get(user_turn_counter)
                    )
                )
                audio_path = None
                pcm_bytes = None
                if audio_uri and gcs_client is not None:
                    try:
                        raw_wav = gcs_client.download_blob(audio_uri)
                        pcm_bytes = extract_pcm_bytes_from_wav(raw_wav)
                    except Exception as exc:
                        logger.warning(
                            "Failed to download user audio %s: %s",
                            audio_uri,
                            exc,
                        )
                past_turns.append(
                    ShadowPastTurn(
                        turn_index=user_turn_counter,
                        user_transcript=cleaned_user_text,
                        preceding_agent_response=" ".join(last_agent_texts),
                        audio_uri=audio_uri,
                        audio_path=audio_path,
                        has_audio=bool(pcm_bytes),
                        used=False,
                        audio_bytes=pcm_bytes or None,
                        agent_audio_uri=agent_audio_uris.get(turn_num),
                    )
                )
                last_agent_texts = []
            elif kind == "agent":
                a_text = (entry.get("text") or "").strip()
                last_agent_texts.append(a_text)
                if past_turns and a_text:
                    past_turns[-1].agent_response = " ".join(
                        filter(
                            None, [past_turns[-1].agent_response, a_text]
                        )
                    )
                history_lines.append(f"[Turn {turn_num}] Agent: {a_text}")
            elif kind == "tool_call":
                history_lines.append(
                    f"[Turn {turn_num}] Tool Call: {entry.get('tool')} "
                    f"args={entry.get('args')}"
                )

    # Missing recordings silently degrade the replay to TTS, which defeats
    # the point of a ShadowEval, so surface it loudly.
    spoken = [
        pt
        for pt in past_turns
        if not _is_non_spoken_turn_text(pt.user_transcript)
    ]
    with_audio = sum(1 for pt in spoken if pt.has_audio)
    if spoken and with_audio < len(spoken):
        logger.warning(
            "Only %d/%d past user turns of conversation %s have recorded "
            "audio (bucket=%s); the rest will be synthesized via TTS. "
            "Check `gcs_bucket` and that the app's audio recording "
            "(loggingSettings.audioRecordingConfig) was enabled for the "
            "original call.",
            with_audio,
            len(spoken),
            tc.conversation_id,
            bucket_override or "<default>",
        )

    return past_turns, "\n".join(history_lines), normalized

def save_shadow_artifacts(
    self,
    shadow_conv: ShadowUserConversation,
    artifacts_dir: str,
    test_case: ShadowTestCase,
    session_id: str,
    use_tool_fakes: bool | None = None,
    download_past_agent_audio: bool = True,
) -> str:
    """Persists a replay's audio and a side-by-side HTML report.

    Writes into `artifacts_dir`:
      - `audio/past_user_turn_<N>.wav`: recorded caller audio per past
        turn (exactly what was streamed for `use_past_audio` turns);
      - `audio/past_agent_turn_<N>.wav`: the original agent reply (when
        the source app recorded it);
      - `audio/new_agent_turn_<N>.wav`: the new agent reply (requires
        `capture_agent_audio`);
      - `shadow_result.json` and `shadow_report.html`.

    Returns:
        Path to the HTML report.
    """
    audio_dir = os.path.join(artifacts_dir, "audio")
    os.makedirs(audio_dir, exist_ok=True)

    def _rel(path: str) -> str:
        return os.path.relpath(path, artifacts_dir)

    gcs_client = None
    past_payload: dict[int, dict[str, Any]] = {}
    for pt in shadow_conv.past_turns:
        user_audio = None
        if pt.audio_bytes or (
            pt.audio_path and os.path.exists(pt.audio_path)
        ):
            user_audio = os.path.join(
                audio_dir, f"past_user_turn_{pt.turn_index}.wav"
            )
            if pt.audio_path and os.path.exists(pt.audio_path):
                shutil.copyfile(pt.audio_path, user_audio)
            else:
                with open(user_audio, "wb") as f:
                    f.write(pcm_to_wav_bytes(pt.audio_bytes or b""))
        agent_audio = None
        if download_past_agent_audio and pt.agent_audio_uri:
            try:
                gcs_client = gcs_client or GCSUtils(creds=self.creds)
                data = gcs_client.download_blob(pt.agent_audio_uri)
                agent_audio = os.path.join(
                    audio_dir, f"past_agent_turn_{pt.turn_index}.wav"
                )
                with open(agent_audio, "wb") as f:
                    f.write(data)
            except Exception as exc:
                logger.warning(
                    "Could not download past agent audio %s: %s",
                    pt.agent_audio_uri,
                    exc,
                )
                agent_audio = None
        past_payload[pt.turn_index] = {
            "turn_index": pt.turn_index,
            "user_text": pt.user_transcript,
            "user_audio": _rel(user_audio) if user_audio else None,
            "agent_text": pt.agent_response,
            "agent_audio": _rel(agent_audio) if agent_audio else None,
        }

    replayed: set[int] = set()
    turns: list[dict[str, Any]] = []
    for log in shadow_conv.turn_logs:
        past_idx = log.selected_past_turn_index
        if log.decision == "event" and shadow_conv.past_turns:
            first = shadow_conv.past_turns[0]
            if (first.user_transcript or "").lstrip().startswith("<event"):
                past_idx = first.turn_index
        past = past_payload.get(past_idx) if past_idx is not None else None
        if past_idx is not None:
            replayed.add(past_idx)
        new_agent_audio = None
        if log.agent_audio_path and os.path.exists(log.agent_audio_path):
            new_agent_audio = os.path.join(
                audio_dir, f"new_agent_turn_{log.sim_turn}.wav"
            )
            shutil.copyfile(log.agent_audio_path, new_agent_audio)
        is_past_audio = (
            log.decision == ShadowDecisionType.USE_PAST_AUDIO.value
        )
        turns.append(
            {
                "sim_turn": log.sim_turn,
                "decision": log.decision,
                "justification": log.decision_justification,
                "user_text": log.user_utterance,
                "user_audio": (
                    past.get("user_audio")
                    if past and is_past_audio
                    else None
                ),
                "agent_text": log.agent_response,
                "agent_audio": (
                    _rel(new_agent_audio) if new_agent_audio else None
                ),
                "past": past,
            }
        )

    results = shadow_conv.expectation_results or []
    met = sum(1 for r in results if r.status == ExpectationStatus.MET)
    data = {
        "name": test_case.name,
        "conversation_id": test_case.conversation_id,
        "source_app": self._resolve_source_app_name(test_case),
        "target_app": self.app_name,
        "session_id": session_id,
        "replay_mode": test_case.replay_mode,
        "session_parameters": test_case.session_parameters,
        "use_tool_fakes": use_tool_fakes,
        "summary": {
            "passed": bool(results) and met == len(results),
            "expectations": f"{met}/{len(results)}",
            "past_audio_turns": sum(
                1
                for t in shadow_conv.turn_logs
                if t.decision == ShadowDecisionType.USE_PAST_AUDIO.value
            ),
            "tts_turns": sum(
                1
                for t in shadow_conv.turn_logs
                if t.decision == ShadowDecisionType.GENERATE_TTS.value
            ),
        },
        "expectations": [
            {
                "expectation": r.expectation,
                "status": r.status.value,
                "justification": r.justification,
            }
            for r in results
        ],
        "turns": turns,
        "unreplayed_past_turns": [
            p
            for idx, p in past_payload.items()
            if idx not in replayed
            and _extract_user_turn_transcript(p["user_text"])
            and not _extract_user_turn_transcript(
                p["user_text"]
            ).startswith(("<event", "<must_reply"))
        ],
    }
    with open(
        os.path.join(artifacts_dir, "shadow_result.json"),
        "w",
        encoding="utf-8",
    ) as f:
        json.dump(data, f, indent=2, default=str)
    report_path = os.path.join(artifacts_dir, "shadow_report.html")
    with open(report_path, "w", encoding="utf-8") as f:
        f.write(render_shadow_html_report(data))
    return report_path

@cleanup_session_dir
def run_shadow_conversation(
    self,
    test_case: ShadowTestCase | dict[str, Any],
    sim_user_model: str | None = _DEFAULT_GEMINI_MODEL,
    eval_model: str | None = _DEFAULT_GEMINI_MODEL,
    session_id: str | None = None,
    console_logging: bool = True,
    modality: str = "audio",
    capture_agent_audio: bool = False,
    background_noise_file: str | None = None,
    burst_noise_files: list[str] | None = None,
    use_tool_fakes: bool = False,
    voice_config: dict[str, Any] | None = None,
    initial_utterance: str | None = None,
    skip_playback_wait: bool = False,
    single_bidi_stream: bool = False,
    max_turns: int | None = None,
    naturalness: bool | dict[str, Any] | None = None,
    artifacts_dir: str | None = None,
    **kwargs: Any,
) -> ShadowUserConversation:
    """Replays a past conversation on the target agent over Bidi audio (or
    text) using `ShadowUserConversation` to arbitrate between past audio
    files and TTS.

    Downloaded and captured audio lives in a temporary session directory
    that is deleted when this returns. Pass `artifacts_dir` to keep it,
    together with a side-by-side HTML report (see
    `save_shadow_artifacts`); the report path is then available as
    `conversation.report_path`.
    """
    sim_user_model = sim_user_model or _DEFAULT_GEMINI_MODEL
    eval_model = eval_model or _DEFAULT_GEMINI_MODEL
    if session_id is None:
        session_id = str(uuid.uuid4())

    tc_model = (
        ShadowTestCase(**test_case)
        if isinstance(test_case, dict)
        else test_case
    )
    tc_dict = tc_model.model_dump()
    naturalness_config = parse_naturalness_config(
        tc_dict,
        naturalness if naturalness is not None else self.naturalness,
    )
    voice_config = voice_config or tc_model.voice_config
    if tc_model.use_tool_fakes is not None:
        use_tool_fakes = tc_model.use_tool_fakes
    # The report needs the new agent audio, so capture it when keeping
    # artifacts.
    capture_agent_audio = capture_agent_audio or bool(artifacts_dir)

    session_dir = f"/tmp/scrapi_evals/{session_id}"
    past_turns, full_past_history, _ = self.fetch_past_conversation_data(
        tc_model, session_dir=session_dir
    )

    shadow_conv = ShadowUserConversation(
        genai_client=self.genai_client,
        genai_model=sim_user_model,
        test_case=tc_model,
        past_turns=past_turns,
        full_past_history=full_past_history,
        max_turns=max_turns,
        initial_utterance=initial_utterance,
    )

    current_sim_turn = 0
    interactive_session = None
    if modality == "audio" and single_bidi_stream:
        client = self.sessions_client
        interactive_session = client.create_interactive_session(
            session_id=session_id,
            capture_agent_audio=capture_agent_audio,
            background_noise_file=background_noise_file,
            use_tool_fakes=use_tool_fakes,
            skip_playback_wait=skip_playback_wait,
            voice_config=voice_config,
        )
        interactive_session.start()

    try:
        if console_logging:
            print(
                f"Starting shadow conversation replay for "
                f"conversation_id={tc_model.conversation_id} "
                f"(session_id={session_id})"
            )

        user_utterance, audio_bytes, variables, turn_log = (
            shadow_conv.next_user_turn()
        )
        accumulated_variables: dict[str, Any] = {}
        if variables:
            accumulated_variables.update(variables)

        detailed_trace: list[str] = []
        if user_utterance:
            past_idx = (
                turn_log.selected_past_turn_index if turn_log else None
            )
            mode_tag = (
                f"[Past Audio Turn #{past_idx}]"
                if turn_log
                and turn_log.decision
                == ShadowDecisionType.USE_PAST_AUDIO.value
                else (
                    "[Generated TTS]"
                    if turn_log
                    and turn_log.decision
                    == ShadowDecisionType.GENERATE_TTS.value
                    else "[Event]"
                )
            )
            detailed_trace.append(f"User {mode_tag}: {user_utterance}")

        while user_utterance:
            if modality == "audio" and interactive_session:
                response = interactive_session.send_turn(
                    user_utterance,
                    accumulated_variables,
                    audio_bytes=audio_bytes,
                )
                if isinstance(response, dict) and response.get(
                    "session_ended"
                ):
                    if response.get("connection_error"):
                        raise BidiSessionError(
                            f"Interactive session WebSocket error: "
                            f"{response['connection_error']}"
                        )
                    break
            else:
                response = self._send_shadow_request_with_retry(
                    session_id=session_id,
                    user_utterance=user_utterance,
                    audio_bytes=audio_bytes,
                    variables=accumulated_variables,
                    modality=modality,
                    console_logging=console_logging,
                    turn_num=current_sim_turn,
                    capture_agent_audio=capture_agent_audio,
                    background_noise_file=background_noise_file,
                    burst_noise_files=burst_noise_files,
                    use_tool_fakes=use_tool_fakes,
                    voice_config=voice_config,
                )
            if not response:
                break

            if response and getattr(response, "agent_audio_paths", None):
                audio_path = response.agent_audio_paths.get(0)
                if audio_path:
                    shadow_conv.agent_audio_paths[current_sim_turn] = (
                        audio_path
                    )
                    if shadow_conv.turn_logs and isinstance(
                        audio_path, str
                    ):
                        shadow_conv.turn_logs[
                            -1
                        ].agent_audio_path = audio_path

            if console_logging:
                self.sessions_client.parse_result(response)

            agent_text, trace_chunks, session_ended, tool_calls = (
                self._parse_agent_response(response)
            )
            detailed_trace.append("\n".join(trace_chunks))

            if session_ended:
                if agent_text:
                    shadow_conv._add_agent_response(agent_text)
                    if shadow_conv.turn_logs:
                        shadow_conv.turn_logs[
                            -1
                        ].agent_response = agent_text
                shadow_conv._add_agent_tool_calls(tool_calls)
                for prog in shadow_conv.steps_progress:
                    if prog.status != StepStatus.COMPLETED:
                        prog.status = StepStatus.COMPLETED
                        prog.justification = (
                            "Session ended by agent; marking step complete."
                        )
                break

            shadow_conv._add_agent_tool_calls(tool_calls)
            user_utterance, audio_bytes, variables, turn_log = (
                shadow_conv.next_user_turn(agent_text)
            )
            if variables:
                accumulated_variables.update(variables)
            if user_utterance:
                past_idx = (
                    turn_log.selected_past_turn_index if turn_log else None
                )
                mode_tag = (
                    f"[Past Audio Turn #{past_idx}]"
                    if turn_log
                    and turn_log.decision
                    == ShadowDecisionType.USE_PAST_AUDIO.value
                    else "[Generated TTS]"
                )
                detailed_trace.append(f"User {mode_tag}: {user_utterance}")

            current_sim_turn += 1

        self._evaluate_expectations(
            shadow_conv,
            detailed_trace,
            eval_model,
            console_logging,
            capture_agent_audio=capture_agent_audio,
        )
        self._evaluate_naturalness(
            shadow_conv,
            detailed_trace,
            eval_model,
            console_logging,
            naturalness_config,
            modality=modality,
        )
        shadow_conv._session_id = session_id
        shadow_conv.session_id = session_id
        shadow_conv._detailed_trace = detailed_trace
        shadow_conv.detailed_trace = detailed_trace
        shadow_conv.report_path = None
        if artifacts_dir:
            try:
                shadow_conv.report_path = self.save_shadow_artifacts(
                    shadow_conv,
                    artifacts_dir,
                    tc_model,
                    session_id,
                    use_tool_fakes=use_tool_fakes,
                )
                if console_logging:
                    print(f"ShadowEval report: {shadow_conv.report_path}")
            except Exception as exc:
                logger.warning(
                    "Failed to save ShadowEval artifacts to %s: %s",
                    artifacts_dir,
                    exc,
                )
        return shadow_conv
    finally:
        if interactive_session:
            interactive_session.close()

def run_shadow_evals(
    self,
    test_cases: list[ShadowTestCase | dict[str, Any]],
    runs: int = 1,
    parallel: int = 1,
    sim_user_model: str | None = _DEFAULT_GEMINI_MODEL,
    eval_model: str | None = _DEFAULT_GEMINI_MODEL,
    modality: str = "audio",
    verbose: bool = False,
    capture_agent_audio: bool = False,
    background_noise_file: str | None = None,
    burst_noise_files: list[str] | None = None,
    use_tool_fakes: bool = False,
    expectations_only: bool | None = None,
    skip_playback_wait: bool = False,
    single_bidi_stream: bool = False,
    progress_callback: Callable[[int, int], None] | None = None,
    naturalness: bool | dict[str, Any] | None = None,
    artifacts_dir: str | None = None,
    preflight: bool = True,
) -> list[dict[str, Any]]:
    """Runs a batch of shadow evaluation test cases across `runs` and
    `parallel` workers.

    Args:
        artifacts_dir: If set, each run keeps its audio and writes a
            side-by-side HTML report under
            `<artifacts_dir>/<case>_run<N>/` (see `save_shadow_artifacts`).
        preflight: Check every case with `preflight` first. Cases with
            preflight errors are reported as failed without opening a
            session; warnings are printed and attached to the results as
            `preflight_issues`.
    """
    if expectations_only is not None:
        self.expectations_only = expectations_only
    if naturalness is not None:
        self.naturalness = naturalness

    sim_user_model = sim_user_model or _DEFAULT_GEMINI_MODEL
    eval_model = eval_model or _DEFAULT_GEMINI_MODEL

    validated_cases = [
        ShadowTestCase(**tc) if isinstance(tc, dict) else tc
        for tc in test_cases
    ]

    results: list[dict[str, Any]] = []
    blocked: list[dict[str, Any]] = []
    preflight_issues: dict[int, list[dict[str, str]]] = {}
    if preflight:
        runnable: list[ShadowTestCase] = []
        for tc in validated_cases:
            check = self.preflight(tc, modality=modality)
            for issue in check.issues:
                print(
                    f"  [{issue.level.upper()}] {tc.name}: {issue.message}"
                )
            issues = [i.model_dump() for i in check.issues]
            if check.ok:
                runnable.append(tc)
                if issues:
                    preflight_issues[id(tc)] = issues
                continue
            errors = "; ".join(
                i.message for i in check.issues if i.level == "error"
            )
            blocked.append(
                {
                    "name": tc.name,
                    "conversation_id": tc.conversation_id,
                    "run": 1,
                    "passed": False,
                    "error": f"Preflight failed: {errors}",
                    "preflight_issues": issues,
                }
            )
        validated_cases = runnable

    jobs = [
        (tc, run_idx) for tc in validated_cases for run_idx in range(runs)
    ]
    issues_by_name = {
        tc.name: preflight_issues[id(tc)]
        for tc in validated_cases
        if id(tc) in preflight_issues
    }

    with Progress() as progress:
        task_id = progress.add_task("Running Shadow Evals", total=len(jobs))
        if parallel <= 1:
            for tc, run_idx in jobs:
                results.append(
                    self._run_single_shadow_job(
                        tc,
                        run_idx,
                        runs,
                        sim_user_model,
                        eval_model,
                        modality,
                        verbose,
                        parallel,
                        capture_agent_audio=capture_agent_audio,
                        background_noise_file=background_noise_file,
                        burst_noise_files=burst_noise_files,
                        use_tool_fakes=use_tool_fakes,
                        skip_playback_wait=skip_playback_wait,
                        single_bidi_stream=single_bidi_stream,
                        artifacts_dir=artifacts_dir,
                    )
                )
                progress.update(task_id, advance=1)
                if progress_callback:
                    progress_callback(len(results), len(jobs))
        else:
            max_workers = min(parallel, 25)
            with ThreadPoolExecutor(max_workers=max_workers) as executor:
                futures = {
                    executor.submit(
                        self._run_single_shadow_job,
                        tc,
                        run_idx,
                        runs,
                        sim_user_model,
                        eval_model,
                        modality,
                        verbose,
                        parallel,
                        capture_agent_audio=capture_agent_audio,
                        background_noise_file=background_noise_file,
                        burst_noise_files=burst_noise_files,
                        use_tool_fakes=use_tool_fakes,
                        skip_playback_wait=skip_playback_wait,
                        single_bidi_stream=single_bidi_stream,
                        artifacts_dir=artifacts_dir,
                    ): (tc.name, run_idx)
                    for tc, run_idx in jobs
                }
                for future in as_completed(futures):
                    results.append(future.result())
                    progress.update(task_id, advance=1)
                    if progress_callback:
                        progress_callback(len(results), len(jobs))

    for res in results:
        if res.get("name") in issues_by_name:
            res.setdefault("preflight_issues", issues_by_name[res["name"]])
    return blocked + results