SimulationEvals¶

@staticmethod
def requires_local_execution(
    test_case: dict[str, Any],
) -> tuple[bool, str]:
    """Checks if a test case requires local execution (cannot be run as a
    platform Milestone eval).

    Returns:
        (requires_local: bool, reason: str)
    """
    if test_case.get("modality") == "audio":
        return True, "Audio modality requires local client simulation"
    if test_case.get("audio_expectations"):
        return True, "audio_expectations requires local client simulation"
    if test_case.get("audio_analysis"):
        return True, "audio_analysis requires local client simulation"
    if test_case.get("voice_config"):
        return True, "voice_config requires local client simulation"
    if test_case.get("naturalness_metric") or test_case.get("naturalness"):
        return True, "naturalness_metric requires local client evaluation"
    if test_case.get("barge_in_after_ms") is not None:
        return True, "barge_in_after_ms requires local audio bidi session"
    for step in test_case.get("steps", []):
        if (
            isinstance(step, dict)
            and step.get("barge_in_after_ms") is not None
        ):
            return (
                True,
                "Step barge_in_after_ms requires local bidi session",
            )
    return False, ""

def to_milestone_yaml(
    self, test_cases: list[dict[str, Any]] | dict[str, Any]
) -> str:
    """Converts simulation test case(s) into CES Platform Milestone YAML."""
    from cxas_scrapi.utils.eval_utils import (  # noqa: PLC0415
        EvalUtils,
    )

    return EvalUtils.convert_sim_evals_to_milestone_yaml(test_cases)

def push_to_platform(
    self,
    test_cases: list[dict[str, Any]] | dict[str, Any],
    conflict_strategy: str = "OVERWRITE",
) -> Any:
    """Pushes simulation test case(s) to CES as Milestone evaluations."""
    from cxas_scrapi.utils.eval_utils import (  # noqa: PLC0415
        EvalUtils,
    )

    eval_utils = EvalUtils(app_name=self.app_name, creds=self.creds)
    return eval_utils.push_milestone_evals_from_yaml(
        source=test_cases,
        app_name=self.app_name,
        conflict_strategy=conflict_strategy,
    )

def run_simulations_on_platform(
    self,
    test_cases: list[dict[str, Any]],
    runs: int = 1,
    timeout_seconds: int = 600,
    use_tool_fakes: bool | None = None,
) -> list[dict[str, Any]]:
    """Pushes and runs simulations on the CES platform via RunEvaluation,
    returning results formatted as simulation result dictionaries.
    """
    from cxas_scrapi.core.evaluations import (  # noqa: PLC0415
        Evaluations,
    )
    from cxas_scrapi.utils.eval_utils import (  # noqa: PLC0415
        EvalUtils,
    )

    eval_client = Evaluations(app_name=self.app_name, creds=self.creds)
    eval_utils = EvalUtils(app_name=self.app_name, creds=self.creds)

    import_op = self.push_to_platform(test_cases)

    eval_names: list[str] = []
    if isinstance(import_op, dict):
        imported_evals = import_op.get("response", {}).get(
            "evaluations", []
        )
        if isinstance(imported_evals, list):
            eval_names = [
                e["name"]
                for e in imported_evals
                if isinstance(e, dict) and "name" in e
            ]

    if not eval_names:
        eval_names = [
            tc["name"]
            for tc in test_cases
            if isinstance(tc, dict) and "name" in tc
        ]
    if not eval_names:
        return []

    run_eval_kwargs: dict[str, Any] = {
        "evaluations": eval_names,
        "app_name": self.app_name,
        "modality": "text",
        "run_count": runs,
    }
    if use_tool_fakes is not None:
        run_eval_kwargs["use_tool_fakes"] = use_tool_fakes

    try:
        run_op = eval_client.run_evaluation(**run_eval_kwargs)
    except Exception:
        resolved_names = []
        evals_map = None
        for e_name in eval_names:
            if e_name.startswith("projects/") and "/evaluations/" in e_name:
                resolved_names.append(e_name)
            else:
                if evals_map is None:
                    evals_map = eval_client._get_or_load_evals_map(
                        self.app_name
                    )
                r_name = evals_map.get("scenarios", {}).get(
                    e_name
                ) or evals_map.get("goldens", {}).get(e_name)
                if r_name:
                    resolved_names.append(r_name)
        rest_kwargs: dict[str, Any] = {
            "evaluations": resolved_names or eval_names,
            "app_name": self.app_name,
            "modality": "text",
            "run_count": runs,
        }
        if use_tool_fakes is not None:
            rest_kwargs["use_tool_fakes"] = use_tool_fakes
        run_op = eval_client.run_evaluation_rest(**rest_kwargs)

    meta_run = getattr(
        getattr(run_op, "metadata", None), "evaluation_run", None
    )
    run_id = (
        meta_run
        if isinstance(meta_run, str) and meta_run
        else getattr(run_op, "name", None)
    )
    if not isinstance(run_id, str) or not run_id:
        run_id = None
    if not run_id and hasattr(run_op, "operation"):
        op_name_attr = getattr(run_op.operation, "name", None)
        if isinstance(op_name_attr, str) and op_name_attr:
            run_id = op_name_attr
    if not run_id and isinstance(run_op, dict):
        run_id = (
            run_op.get("evaluationRun")
            or run_op.get("metadata", {}).get("evaluationRun")
            or run_op.get("response", {}).get("evaluationRun")
            or run_op.get("name")
        )

    if (
        run_id
        and "/evaluationRuns/" not in run_id
        and hasattr(run_op, "_refresh")
    ):
        from google.cloud.ces_v1beta.types import (  # noqa: PLC0415
            RunEvaluationOperationMetadata,
        )

        for _ in range(12):
            with contextlib.suppress(Exception):
                refreshed = run_op._refresh(None)
                meta = RunEvaluationOperationMetadata()
                meta._pb.ParseFromString(refreshed.metadata.value)
                if meta.evaluation_run:
                    run_id = meta.evaluation_run
                    break
            time.sleep(5)

    if (
        run_id
        and "/evaluationRuns/" not in run_id
        and "/operations/" in run_id
    ):
        op_name = run_id
        for _ in range(12):
            with contextlib.suppress(Exception):
                op_state = eval_client.get_operation_rest(op_name)
                candidate = op_state.get("metadata", {}).get(
                    "evaluationRun"
                ) or op_state.get("response", {}).get("evaluationRun")
                if candidate:
                    run_id = candidate
                    break
            time.sleep(5)

    raw_results = []
    if run_id and "/evaluationRuns/" in run_id:
        raw_results = eval_utils.wait_for_run_and_get_results(
            run_id, timeout_seconds=timeout_seconds
        )

    df_dict = eval_utils.evals_to_dataframe(results=raw_results)
    df_summary = df_dict.get("summary", pd.DataFrame())
    df_metadata = df_dict.get("metadata", pd.DataFrame())
    df_failures = df_dict.get("failures", pd.DataFrame())

    sim_results: list[dict[str, Any]] = []
    tc_map = {
        tc.get("name"): tc for tc in test_cases if isinstance(tc, dict)
    }

    for tc_name, tc_def in tc_map.items():
        tc_runs = (
            df_summary[df_summary["display_name"] == tc_name]
            if not df_summary.empty
            else pd.DataFrame()
        )
        if tc_runs.empty:
            sim_results.append(
                {
                    "name": tc_name,
                    "passed": False,
                    "status": "FAIL",
                    "goals": "0/0",
                    "expectations": "0/0",
                    "duration_s": 0.0,
                    "run": 1,
                    "runs": runs,
                    "platform": True,
                    "step_details": [],
                    "expectation_details": [],
                    "session_parameters": tc_def.get(
                        "session_parameters", {}
                    ),
                }
            )
            continue

        tc_steps = (
            tc_def.get("steps", [])
            if isinstance(tc_def.get("steps"), list)
            else []
        )

        for run_idx, (_, row) in enumerate(tc_runs.iterrows()):
            res_id = row.get("eval_result_id", "")
            res_meta = (
                df_metadata[df_metadata["eval_result_id"] == res_id]
                if not df_metadata.empty
                else pd.DataFrame()
            )
            res_fails = (
                df_failures[df_failures["eval_result_id"] == res_id]
                if not df_failures.empty
                else pd.DataFrame()
            )

            milestone_rows = (
                res_meta[res_meta["type"] == "Milestone"]
                if not res_meta.empty and "type" in res_meta.columns
                else pd.DataFrame()
            )
            exp_rows = (
                res_meta[res_meta["type"] == "Custom Expectation"]
                if not res_meta.empty and "type" in res_meta.columns
                else pd.DataFrame()
            )

            milestones_met = (
                sum(milestone_rows["outcome"] == "PASS")
                if not milestone_rows.empty
                else 0
            )
            total_milestones = len(milestone_rows)
            exps_met = (
                sum(exp_rows["outcome"] == "PASS")
                if not exp_rows.empty
                else 0
            )
            total_exps = len(exp_rows)

            step_details = []
            for step_idx, (_, m_row) in enumerate(
                milestone_rows.iterrows()
            ):
                raw_score = str(m_row.get("score", "")).upper()
                if (
                    raw_score == "COMPLETED"
                    or m_row.get("outcome") == "PASS"
                ):
                    step_status = "Completed"
                elif raw_score == "IN_PROGRESS":
                    step_status = "In Progress"
                else:
                    step_status = "Not Started"
                criteria = str(m_row.get("success_criteria") or "")
                if (
                    not criteria
                    and step_idx < len(tc_steps)
                    and isinstance(tc_steps[step_idx], dict)
                ):
                    criteria = str(
                        tc_steps[step_idx].get("success_criteria")
                        or tc_steps[step_idx].get("successCriteria")
                        or ""
                    )
                step_details.append(
                    {
                        "goal": str(m_row.get("expected", "")),
                        "success_criteria": criteria,
                        "status": step_status,
                        "justification": str(m_row.get("actual", "")),
                    }
                )

            expectation_details = []
            for _, e_row in exp_rows.iterrows():
                expectation_details.append(
                    {
                        "expectation": str(e_row.get("expected", "")),
                        "status": (
                            "Met"
                            if e_row.get("outcome") == "PASS"
                            else "Not Met"
                        ),
                        "justification": str(e_row.get("actual", "")),
                    }
                )

            status_val = str(row.get("evaluation_status", "")).upper()
            exec_state_val = str(row.get("execution_state", "")).upper()
            passed = (
                status_val == "PASS"
                and exec_state_val not in ("ERROR", "ERRORED", "CANCELLED")
                and res_fails.empty
            )
            conv_name = str(row.get("conversation", ""))
            sess_id = (
                conv_name.rsplit("/", 1)[-1]
                if "/" in conv_name
                else conv_name
            )

            result_entry: dict[str, Any] = {
                "name": tc_name,
                "passed": passed,
                "status": "PASS" if passed else "FAIL",
                "goals": f"{milestones_met}/{total_milestones}",
                "expectations": f"{exps_met}/{total_exps}",
                "duration_s": 0.0,
                "run": run_idx + 1,
                "runs": runs,
                "platform": True,
                "eval_result_id": res_id,
                "evaluation_run": str(
                    row.get("evaluation_run", "") or run_id or ""
                ),
                "conversation_name": conv_name,
                "session_id": sess_id,
                "step_details": step_details,
                "expectation_details": expectation_details,
                "session_parameters": tc_def.get("session_parameters", {}),
            }
            if exec_state_val in (
                "ERROR",
                "ERRORED",
            ) or status_val in ("ERROR", "ERRORED"):
                result_entry["error"] = str(
                    row.get("error_message") or "Unknown Agent Exception"
                )

            sim_results.append(result_entry)

    return sim_results

@cleanup_session_dir
def simulate_conversation(
    self,
    test_case: dict[str, Any],
    sim_user_model: str | None = _DEFAULT_SIM_USER_MODEL,
    eval_model: str | None = _DEFAULT_EVAL_MODEL,
    session_id: str | None = None,
    console_logging: bool = True,
    modality: str = "text",
    capture_agent_audio: bool = False,
    background_noise_file: str | None = None,
    bg_noise_snr: float | None = None,
    burst_noise_files: list[str] | None = None,
    burst_noise_snr: float | None = None,
    use_tool_fakes: bool = False,
    voice_config: dict[str, Any] | None = None,
    initial_utterance: str = _FIRST_UTTERANCE,
    skip_playback_wait: bool = False,
    single_bidi_stream: bool = False,
    max_turns: int | None = None,
    naturalness: bool | dict[str, Any] | None = None,
    audio_analysis: bool | None = None,
    **kwargs: Any,
) -> LLMUserConversation:
    """Runs the simulated conversation loop.

    Args:
        test_case: The test case dictionary defining evaluation steps.
        sim_user_model: The Gemini model used for the simulated user.
        eval_model: The Gemini model used for evaluating expectations.
        console_logging: Whether to print interaction transcript to
            the console.
        single_bidi_stream: For audio modality, keep one persistent
            bidi WebSocket open for the whole conversation instead of
            opening a new connection per turn (the default).
        max_turns: Maximum number of conversation turns. Defaults to
            the test_case's max_turns setting, or 30 if unspecified.
        naturalness: Overrides the test case's `naturalness_metric`
            block. `None` (the default) defers to the test case, so
            simulations that never declare the metric are unaffected.
        audio_analysis: Run multimodal audio checks, acoustic metrics,
            and voice drift analysis. Defaults to instance setting.
    """
    sim_user_model = sim_user_model or _DEFAULT_SIM_USER_MODEL
    eval_model = eval_model or _DEFAULT_EVAL_MODEL
    if session_id is None:
        session_id = str(uuid.uuid4())
    effective_audio_analysis = (
        audio_analysis
        if audio_analysis is not None
        else getattr(self, "audio_analysis", False)
    )
    if (
        effective_audio_analysis
        and naturalness is None
        and getattr(self, "naturalness", None) is None
    ):
        effective_naturalness = True
    else:
        effective_naturalness = (
            naturalness if naturalness is not None else self.naturalness
        )
    naturalness_config = parse_naturalness_config(
        test_case,
        effective_naturalness,
    )
    voice_config = voice_config or test_case.get("voice_config")
    eval_conv = LLMUserConversation(
        genai_client=self.genai_client,
        genai_model=sim_user_model,
        test_case=test_case,
        max_turns=max_turns,
        initial_utterance=initial_utterance,
    )

    # Initialize audio paths tracking
    eval_conv.agent_audio_paths = {}
    current_sim_turn = 0

    has_barge_in_step = any(
        isinstance(s, dict) and s.get("barge_in_after_ms") is not None
        for s in test_case.get("steps", [])
    )
    if has_barge_in_step and modality != "audio":
        logging.warning(
            "Test case has barge_in_after_ms configured, but modality is "
            f"'{modality}'. Barge-in evaluation requires audio modality."
        )
    if has_barge_in_step and modality == "audio":
        single_bidi_stream = True
        skip_playback_wait = False

    resolved_noise = resolve_noise_config(
        config_or_case=test_case,
        background_noise_file=background_noise_file,
        bg_noise_snr=bg_noise_snr,
        burst_noise_files=burst_noise_files,
        burst_noise_snr=burst_noise_snr,
    )
    effective_bg_file = resolved_noise["background_noise_file"]
    effective_bg_snr = resolved_noise["bg_noise_snr"]
    effective_burst_files = resolved_noise["burst_noise_files"]
    effective_burst_snr = resolved_noise["burst_noise_snr"]

    interactive_session = None
    if modality == "audio" and single_bidi_stream:
        client = self.sessions_client
        create_kwargs: dict[str, Any] = {
            "session_id": session_id,
            "capture_agent_audio": capture_agent_audio,
            "background_noise_file": effective_bg_file,
            "use_tool_fakes": use_tool_fakes,
            "skip_playback_wait": skip_playback_wait,
            "voice_config": voice_config,
        }
        if effective_bg_file is not None:
            create_kwargs["bg_noise_snr"] = effective_bg_snr
        if effective_burst_files is not None:
            create_kwargs["burst_noise_files"] = effective_burst_files
            create_kwargs["burst_noise_snr"] = effective_burst_snr
        interactive_session = client.create_interactive_session(
            **create_kwargs
        )
        interactive_session.start()

    try:
        if console_logging:
            print(
                f"Starting simulated conversation with session ID: "
                f"{session_id}"
            )

        # Initialize the first turn manually
        user_utterance, variables = eval_conv.next_user_utterance()
        accumulated_variables = {}
        if variables:
            accumulated_variables.update(variables)

        detailed_trace = []
        detailed_trace.append(f"User: {user_utterance}")
        turn_traces: list[dict[str, Any]] = []

        while user_utterance:
            barge_in_offset = (
                eval_conv.peek_barge_in_after_ms()
                if hasattr(eval_conv, "peek_barge_in_after_ms")
                else None
            )
            if (
                modality == "audio"
                and interactive_session
                and isinstance(barge_in_offset, int)
            ):
                barge_in_utterance, barge_in_vars = (
                    eval_conv.generate_barge_in_utterance()
                )
                barge_vars = accumulated_variables.copy()
                if barge_in_vars:
                    barge_vars.update(barge_in_vars)

                resp_n, response = (
                    interactive_session.send_turn_with_barge_in(
                        text=user_utterance,
                        barge_in_text=barge_in_utterance,
                        barge_in_after_ms=barge_in_offset,
                        variables=accumulated_variables,
                        barge_in_variables=barge_vars,
                    )
                )

                if isinstance(resp_n, dict) and resp_n.get("session_ended"):
                    if resp_n.get("connection_error"):
                        err_msg = (
                            "Interactive session WebSocket error: "
                            f"{resp_n['connection_error']}"
                        )
                        raise BidiSessionError(err_msg)
                    break

                if resp_n and getattr(resp_n, "agent_audio_paths", None):
                    audio_path_n = resp_n.agent_audio_paths.get(0)
                    if audio_path_n:
                        eval_conv.agent_audio_paths[current_sim_turn] = (
                            audio_path_n
                        )

                agent_text_n, trace_chunks_n, _, tool_calls_n = (
                    self._parse_agent_response(resp_n)
                )
                parsed_n = ParsedSessionResponse(resp_n)
                if (
                    parsed_n.barge_in_metadata
                    and parsed_n.barge_in_metadata.interrupted
                ):
                    heard = parsed_n.barge_in_metadata.estimated_heard_text
                    eval_conv._add_agent_response(
                        f"{heard}... [interrupted]"
                    )
                elif agent_text_n:
                    eval_conv._add_agent_response(agent_text_n)

                eval_conv._add_agent_tool_calls(tool_calls_n)
                detailed_trace.append("\n".join(trace_chunks_n))
                turn_root_span_n = ParsedSessionResponse.extract_root_span(
                    resp_n
                )
                if isinstance(turn_root_span_n, dict) and turn_root_span_n:
                    turn_traces.append(
                        {
                            "turn_index": current_sim_turn,
                            "user_utterance": user_utterance,
                            "agent_text": agent_text_n,
                            "root_span": turn_root_span_n,
                        }
                    )
                current_sim_turn += 1

                eval_conv.record_user_utterance(barge_in_utterance)
                detailed_trace.append(
                    f"User: [Barge-In] {barge_in_utterance}"
                )

                if isinstance(response, dict) and response.get(
                    "session_ended"
                ):
                    if response.get("connection_error"):
                        err_msg = (
                            "Interactive session WebSocket error: "
                            f"{response['connection_error']}"
                        )
                        raise BidiSessionError(err_msg)
                    break

                user_utterance = barge_in_utterance
                accumulated_variables = barge_vars
            elif modality == "audio" and interactive_session:
                response = interactive_session.send_turn(
                    user_utterance,
                    accumulated_variables,
                )
                # Check if session ended via WebSocket endSession
                if isinstance(response, dict) and response.get(
                    "session_ended"
                ):
                    if response.get("connection_error"):
                        err_msg = (
                            f"Interactive session WebSocket error: "
                            f"{response['connection_error']}"
                        )
                        raise BidiSessionError(err_msg)
                    break
            else:
                response = self._send_request_with_retry(
                    session_id=session_id,
                    user_utterance=user_utterance,
                    variables=accumulated_variables,
                    modality=modality,
                    console_logging=console_logging,
                    turn_num=current_sim_turn,
                    capture_agent_audio=capture_agent_audio,
                    background_noise_file=effective_bg_file,
                    bg_noise_snr=effective_bg_snr,
                    burst_noise_files=effective_burst_files,
                    burst_noise_snr=effective_burst_snr,
                    use_tool_fakes=use_tool_fakes,
                    voice_config=voice_config,
                )
            if not response:
                break

            # Extract and save the agent turn audio WAV if present
            # in response.
            if response and getattr(response, "agent_audio_paths", None):
                audio_path = response.agent_audio_paths.get(0)
                if audio_path:
                    paths = eval_conv.agent_audio_paths
                    paths[current_sim_turn] = 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))

            turn_root_span = ParsedSessionResponse.extract_root_span(
                response
            )
            if isinstance(turn_root_span, dict) and turn_root_span:
                turn_traces.append(
                    {
                        "turn_index": current_sim_turn,
                        "user_utterance": user_utterance,
                        "agent_text": agent_text,
                        "root_span": turn_root_span,
                    }
                )

            if session_ended:
                if agent_text:
                    eval_conv._add_agent_response(agent_text)
                eval_conv._add_agent_tool_calls(tool_calls)
                # Ensure the final agent response is evaluated
                # so that steps_progress is updated on session end.
                eval_conv._next_user_utterance()
                if console_logging:
                    print(
                        "\nSession has been closed by the Agent via "
                        "end_session tool."
                    )
                # Mark steps as completed if session ending matches
                # criteria (escalation/transfer) or if goal was primed as
                # ready for completion.
                self._reconcile_completed_steps(
                    eval_conv, session_ended=True
                )
                break

            # Get the next simulated user utterance based on the agent's
            # response
            eval_conv._add_agent_tool_calls(tool_calls)
            user_utterance, variables = eval_conv.next_user_utterance(
                agent_text
            )
            if variables:
                accumulated_variables.update(variables)
            if user_utterance:
                detailed_trace.append(f"User: {user_utterance}")

            current_sim_turn += 1

        self._reconcile_completed_steps(eval_conv)

        if console_logging:
            self._print_completion_status(eval_conv)

        self._evaluate_expectations(
            eval_conv,
            detailed_trace,
            eval_model,
            console_logging,
            capture_agent_audio=capture_agent_audio,
        )
        self._evaluate_naturalness(
            eval_conv,
            detailed_trace,
            eval_model,
            console_logging,
            naturalness_config,
            session_id=session_id,
            modality=modality,
        )
        if effective_audio_analysis:
            self._evaluate_audio_analysis(
                eval_conv=eval_conv,
                detailed_trace=detailed_trace,
                turn_traces=turn_traces,
                eval_model=eval_model,
                console_logging=console_logging,
                session_id=session_id,
                modality=modality,
            )
        eval_conv._session_id = session_id
        eval_conv.session_id = session_id
        eval_conv._detailed_trace = detailed_trace
        eval_conv.detailed_trace = detailed_trace
        eval_conv._turn_traces = turn_traces
        eval_conv.turn_traces = turn_traces
        if turn_traces and session_id:
            with contextlib.suppress(Exception):
                cache_dir = "/tmp/scrapi_traces"
                os.makedirs(cache_dir, exist_ok=True)
                cache_path = os.path.join(cache_dir, f"{session_id}.json")
                with open(cache_path, "w") as f:
                    json.dump(
                        {"session_id": session_id, "turns": turn_traces}, f
                    )
        return eval_conv
    finally:
        if interactive_session:
            interactive_session.close()