EvalUtils¶

@staticmethod
def parse_variables_input(v: Any) -> dict[str, Any]:
    """Allows YAML to accept a list, a JSON string OR a dictionary.

    See :func:`_parse_variables_input` for the accepted shapes.
    """
    return _parse_variables_input(v)

@staticmethod
def is_milestone_eval_data(data: Any) -> bool:
    """Determines if the given data represents a Milestone evaluation."""
    if isinstance(data, dict):
        if "evals" in data and isinstance(data["evals"], list):
            return True
        if "steps" in data and isinstance(data["steps"], list):
            return any(
                isinstance(s, dict)
                and ("goal" in s or "success_criteria" in s)
                for s in data["steps"]
            )
        scenario = data.get("scenario")
        if isinstance(scenario, dict) and isinstance(
            scenario.get("milestones"), list
        ):
            return True
        if isinstance(data.get("milestones"), list):
            return True
    elif isinstance(data, list):
        if data and all(
            isinstance(item, dict)
            and "steps" in item
            and isinstance(item["steps"], list)
            for item in data
        ):
            return True
    return False

@staticmethod
def sanitize_milestone_eval_dict(data: Any) -> dict[str, Any]:
    """Normalizes and sanitizes simulation evals data into strict Milestone
    YAML format compatible with CES platform MilestoneYamlTranslator.

    Removes non-whitelisted keys (e.g. client-only audio_expectations,
    naturalness_metric, barge_in_after_ms) and ensures inject_variables
    only exists on step 0 (hoisting to session_parameters or step 0 if found
    on later steps).
    """
    if not data:
        return {"evals": []}

    if isinstance(data, dict) and (
        (
            isinstance(data.get("scenario"), dict)
            and isinstance(data["scenario"].get("milestones"), list)
        )
        or isinstance(data.get("milestones"), list)
    ):
        data = Evaluations.eval_dict_to_yaml(data)

    root_dict: dict[str, Any] = {}
    evals_list: list[dict[str, Any]] = []

    if isinstance(data, dict):
        if "evals" in data and isinstance(data["evals"], list):
            common_params = data.get(
                "common_session_parameters"
            ) or data.get("commonSessionParameters")
            if isinstance(common_params, dict) and common_params:
                root_dict["common_session_parameters"] = (
                    Common.unwrap_struct(common_params)
                )
            common_exps = data.get("common_expectations") or data.get(
                "commonExpectations"
            )
            if isinstance(common_exps, list) and common_exps:
                formatted_common_exps = [
                    EvalUtils._format_milestone_expectation(e)
                    for e in common_exps
                ]
                formatted_common_exps = [
                    e for e in formatted_common_exps if e
                ]
                if formatted_common_exps:
                    root_dict["common_expectations"] = formatted_common_exps
            init_evt = data.get("initial_event") or data.get("initialEvent")
            if init_evt:
                root_dict["initial_event"] = init_evt
            evals_list = data["evals"]
        elif "steps" in data:
            evals_list = [data]
        else:
            evals_list = []
    elif isinstance(data, list):
        evals_list = [item for item in data if isinstance(item, dict)]

    sanitized_evals: list[dict[str, Any]] = []
    client_only_fields = [
        "barge_in_after_ms",
        "audio_expectations",
        "audio_analysis",
        "naturalness_metric",
        "naturalness",
        "voice_config",
    ]
    for eval_idx, item in enumerate(evals_list):
        if not isinstance(item, dict):
            continue

        name = item.get("name") or f"milestone_eval_{eval_idx + 1}"
        stripped_eval_fields = [
            f for f in client_only_fields if item.get(f) is not None
        ]
        if stripped_eval_fields:
            logger.warning(
                "Stripping unsupported client-only field(s) %s from "
                "evaluation '%s' for Milestone platform compatibility.",
                stripped_eval_fields,
                name,
            )
        sanitized_eval: dict[str, Any] = {"name": str(name)}

        tags = item.get("tags")
        if tags and isinstance(tags, list):
            sanitized_eval["tags"] = [str(t) for t in tags]

        session_params = (
            item.get("session_parameters")
            or item.get("sessionParameters")
            or {}
        )
        if session_params:
            sanitized_eval["session_parameters"] = Common.unwrap_struct(
                session_params
            )
        else:
            sanitized_eval["session_parameters"] = {}

        expectations = (
            item.get("expectations")
            or item.get("evaluation_expectations")
            or item.get("evaluationExpectations")
            or []
        )
        if expectations and isinstance(expectations, list):
            formatted_exps = [
                EvalUtils._format_milestone_expectation(e)
                for e in expectations
            ]
            formatted_exps = [e for e in formatted_exps if e]
            if formatted_exps:
                sanitized_eval["expectations"] = formatted_exps

        max_turns = (
            item.get("max_turns")
            or item.get("maxTurns")
            or item.get("max_turn_count")
            or item.get("maxTurnCount")
        )
        if max_turns:
            with contextlib.suppress(ValueError, TypeError):
                sanitized_eval["max_turns"] = int(max_turns)

        init_event = item.get("initial_event") or item.get("initialEvent")
        if init_event:
            sanitized_eval["initial_event"] = init_event

        eval_meta = item.get("metadata")
        if isinstance(eval_meta, dict) and eval_meta:
            sanitized_eval["metadata"] = Common.unwrap_struct(eval_meta)

        raw_steps = item.get("steps", [])
        if not isinstance(raw_steps, list):
            raw_steps = [raw_steps]

        sanitized_steps: list[dict[str, Any]] = []
        hoisted_vars: dict[str, Any] = {}

        for step_idx, raw_step in enumerate(raw_steps):
            if not isinstance(raw_step, dict):
                continue

            stripped_step_fields = [
                f for f in client_only_fields if raw_step.get(f) is not None
            ]
            if stripped_step_fields:
                logger.warning(
                    "Stripping unsupported client-only field(s) %s from "
                    "step %d in evaluation '%s'.",
                    stripped_step_fields,
                    step_idx,
                    name,
                )

            goal = raw_step.get("goal", "")
            success_criteria = (
                raw_step.get("success_criteria")
                or raw_step.get("successCriteria")
                or ""
            )
            if not goal and not success_criteria:
                goal = f"Step {step_idx + 1}"
                success_criteria = "Goal achieved"
            elif not goal:
                goal = success_criteria
            elif not success_criteria:
                success_criteria = goal

            step_dict: dict[str, Any] = {
                "goal": str(goal),
                "success_criteria": str(success_criteria),
            }

            resp_guide = (
                raw_step.get("response_guide")
                or raw_step.get("responseGuide")
                or raw_step.get("protocol")
            )
            if resp_guide:
                step_dict["response_guide"] = str(resp_guide)

            step_max_turns = (
                raw_step.get("max_turns")
                or raw_step.get("max_turn_count")
                or raw_step.get("maxTurnCount")
            )
            if step_max_turns:
                with contextlib.suppress(ValueError, TypeError):
                    step_dict["max_turns"] = int(step_max_turns)

            static_utt = raw_step.get("static_utterance") or raw_step.get(
                "staticUtterance"
            )
            if static_utt:
                step_dict["static_utterance"] = str(static_utt)

            event_val = raw_step.get("event")
            if event_val:
                step_dict["event"] = str(event_val)

            inject_vars = raw_step.get("inject_variables") or raw_step.get(
                "injectVariables"
            )
            if inject_vars and isinstance(inject_vars, dict):
                unwrapped_inj = Common.unwrap_struct(inject_vars)
                if step_idx == 0:
                    step_dict["inject_variables"] = unwrapped_inj
                else:
                    logger.warning(
                        "Hoisting inject_variables from step %d to "
                        "session_parameters for evaluation '%s' (platform "
                        "only allows inject_variables on step 0).",
                        step_idx,
                        name,
                    )
                    hoisted_vars.update(unwrapped_inj)

            step_meta = raw_step.get("metadata")
            if isinstance(step_meta, dict) and step_meta:
                step_dict["metadata"] = Common.unwrap_struct(step_meta)

            sanitized_steps.append(step_dict)

        if hoisted_vars:
            sanitized_eval["session_parameters"].update(hoisted_vars)

        if not sanitized_eval["session_parameters"]:
            sanitized_eval.pop("session_parameters", None)

        sanitized_eval["steps"] = sanitized_steps

        sanitized_eval = {
            k: v
            for k, v in sanitized_eval.items()
            if k in MILESTONE_ALLOWED_EVAL_KEYS
        }
        sanitized_evals.append(sanitized_eval)

    root_dict["evals"] = sanitized_evals
    return {
        k: v
        for k, v in root_dict.items()
        if k in MILESTONE_ALLOWED_ROOT_KEYS
    }

@staticmethod
def convert_sim_evals_to_milestone_yaml(source: Any) -> str:
    """Converts local simulation evals (file path, directory path, or
    dict/list) into a CES Platform-compatible Milestone YAML string.
    """
    raw_cases: list[dict[str, Any]] = []
    root_data: dict[str, Any] = {}

    if isinstance(source, str):
        if os.path.isdir(source):
            yaml_files = sorted(
                glob.glob(os.path.join(source, "*.yaml"))
                + glob.glob(os.path.join(source, "*.yml"))
                + glob.glob(os.path.join(source, "*.json"))
            )
            for fpath in yaml_files:
                with open(fpath, encoding="utf-8") as f:
                    content = (
                        yaml.safe_load(f)
                        if fpath.endswith((".yaml", ".yml"))
                        else json.load(f)
                    )
                    if isinstance(content, dict):
                        if "evals" in content and isinstance(
                            content["evals"], list
                        ):
                            file_common_params = (
                                content.get("common_session_parameters")
                                or content.get("commonSessionParameters")
                                or {}
                            )
                            file_common_exps = (
                                content.get("common_expectations")
                                or content.get("commonExpectations")
                                or []
                            )
                            file_init_evt = content.get(
                                "initial_event"
                            ) or content.get("initialEvent")
                            for c in content["evals"]:
                                if not isinstance(c, dict):
                                    continue
                                case_copy = dict(c)
                                if (
                                    isinstance(file_common_params, dict)
                                    and file_common_params
                                ):
                                    existing_params = (
                                        case_copy.get("session_parameters")
                                        or case_copy.get(
                                            "sessionParameters"
                                        )
                                        or {}
                                    )
                                    case_copy["session_parameters"] = {
                                        **Common.unwrap_struct(
                                            file_common_params
                                        ),
                                        **Common.unwrap_struct(
                                            existing_params
                                        ),
                                    }
                                if (
                                    isinstance(file_common_exps, list)
                                    and file_common_exps
                                ):
                                    existing_exps = (
                                        case_copy.get("expectations")
                                        or case_copy.get(
                                            "evaluation_expectations"
                                        )
                                        or case_copy.get(
                                            "evaluationExpectations"
                                        )
                                        or []
                                    )
                                    case_copy["expectations"] = list(
                                        file_common_exps
                                    ) + list(existing_exps)
                                if (
                                    file_init_evt
                                    and "initial_event" not in case_copy
                                    and "initialEvent" not in case_copy
                                ):
                                    case_copy["initial_event"] = (
                                        file_init_evt
                                    )
                                raw_cases.append(case_copy)
                        elif "steps" in content:
                            raw_cases.append(content)
                    elif isinstance(content, list):
                        raw_cases.extend(
                            [c for c in content if isinstance(c, dict)]
                        )
            root_data = {"evals": raw_cases}
        elif os.path.isfile(source):
            with open(source, encoding="utf-8") as f:
                content = (
                    yaml.safe_load(f)
                    if source.endswith((".yaml", ".yml"))
                    else json.load(f)
                )
                root_data = content
        else:
            try:
                root_data = yaml.safe_load(source)
            except Exception:
                root_data = json.loads(source)
    elif isinstance(source, (dict, list)):
        root_data = source
    else:
        raise ValueError(f"Unsupported source type: {type(source)}")

    sanitized = EvalUtils.sanitize_milestone_eval_dict(root_data)
    return yaml.dump(sanitized, sort_keys=False, allow_unicode=True)

def push_milestone_evals_from_yaml(
    self,
    source: Any,
    app_name: str | None = None,
    conflict_strategy: str | int = "OVERWRITE",
) -> Any:
    """Converts and pushes local simulation evaluations to the platform as
    Milestone evaluations via the REST importEvaluations endpoint.
    """
    app_name = app_name or self.app_name
    yaml_content = self.convert_sim_evals_to_milestone_yaml(source)
    return self.eval_client.import_evaluations(
        app_name=app_name,
        yaml_content=yaml_content,
        conflict_strategy=conflict_strategy,
    )

@staticmethod
def score_result_audio(result: typing.Any) -> bool:
    """Score a single result using audio-correct method.
    In audio mode, taskCompleted is broken (always False).
    Use goalScore AND allExpectationsSatisfied instead.
    """
    res_dict = (
        type(result).to_dict(result)
        if not isinstance(result, dict)
        else result
    )
    sr = res_dict.get("scenario_result", {})
    goal = sr.get("user_goal_satisfaction_result", {}).get("score", 0)
    all_exp = sr.get("all_expectations_satisfied", False)
    return (goal == 1) and all_exp

def evals_to_dataframe(
    self,
    results: list[Any] | str | None = None,
    eval_names: list[str] | str | None = None,
) -> dict[str, Any]:
    """Provides three simplified views of the evaluation data.

    Returns:
        A dict with 'summary', 'failures', and 'trace' DataFrames.
    """
    run_summaries, expectations, turns = self._parse_eval_results(
        results, eval_names
    )

    # Summary
    df_summary = pd.DataFrame(run_summaries)
    if not df_summary.empty:
        if "create_time" in df_summary.columns:
            df_summary["create_time"] = pd.to_datetime(
                df_summary["create_time"]
            )
        if "update_time" in df_summary.columns:
            df_summary["update_time"] = pd.to_datetime(
                df_summary["update_time"]
            )
        if "update_time" in df_summary.columns:
            df_summary = df_summary.sort_values(
                by="update_time", ascending=False
            ).reset_index(drop=True)

    # Failures
    failures = []
    for run_sum in run_summaries:
        if run_sum.get("execution_state") in (
            "ERROR",
            "ERRORED",
        ) or run_sum.get("evaluation_status") in ("ERROR", "ERRORED"):
            raw_err = run_sum.get(
                "error_message", "Unknown Agent Exception"
            )
            failures.append(
                {
                    "display_name": run_sum.get("display_name", "Unknown"),
                    "eval_result_id": run_sum.get("eval_result_id", ""),
                    "turn_index": None,
                    "failure_type": "System Engine Error",
                    "expected": "Run evaluation to completion",
                    "actual": f"Error: {raw_err}",
                    "score": None,
                }
            )

    # Metadata
    all_metadata = []

    for exp in expectations:
        if exp.get("record_type") == "milestone":
            if exp.get("not_met_count", 0) > 0:
                status_lbl = exp.get("status", "NOT_STARTED")
                explanation = exp.get("explanation", "")
                actual_text = (
                    f"({status_lbl}) {explanation}"
                    if explanation
                    else f"({status_lbl})"
                )
                failures.append(
                    {
                        "display_name": exp["display_name"],
                        "eval_result_id": exp["eval_result_id"],
                        "turn_index": None,
                        "failure_type": "Milestone",
                        "expected": str(exp.get("expectation", "")),
                        "actual": actual_text,
                        "score": None,
                    }
                )
            status_val = exp.get("status", "NOT_STARTED")
            outcome_str = "PASS" if status_val == "COMPLETED" else "FAIL"
            all_metadata.append(
                {
                    "display_name": exp["display_name"],
                    "evaluation_run": exp.get("evaluation_run"),
                    "eval_result_id": exp["eval_result_id"],
                    "evaluation_status": exp.get("evaluation_status"),
                    "turn_index": None,
                    "type": "Milestone",
                    "expected": exp["expectation"],
                    "actual": exp["explanation"],
                    "outcome": outcome_str,
                    "score": status_val,
                    "success_criteria": exp.get("success_criteria", ""),
                }
            )
            continue

        if exp.get("not_met_count", 0) > 0:
            explanation = exp.get("explanation", "")
            actual_text = (
                f"(Not Met) {explanation}" if explanation else "(Not Met)"
            )
            failures.append(
                {
                    "display_name": exp["display_name"],
                    "eval_result_id": exp["eval_result_id"],
                    "turn_index": None,
                    "failure_type": "Expectation",
                    "expected": str(exp.get("expectation", "")),
                    "actual": actual_text,
                    "score": None,
                }
            )
        if exp.get("record_type") == "summary_expectation":
            not_met = exp.get("not_met_count", 0)
            met = exp.get("met_count", 0)
            outcome_str = "PASS" if not_met == 0 else "FAIL"

            all_metadata.append(
                {
                    "display_name": exp["display_name"],
                    "evaluation_run": exp.get("evaluation_run"),
                    "eval_result_id": exp["eval_result_id"],
                    "evaluation_status": exp.get("evaluation_status"),
                    "turn_index": None,
                    "type": "Custom Expectation",
                    "expected": exp["expectation"],
                    "actual": exp["explanation"],
                    "outcome": outcome_str,
                    "score": f"{met} / {met + not_met}",
                }
            )

    for turn in turns:
        outcomes_str = turn.get("expectation_outcomes", "[]")
        outcomes = []
        with contextlib.suppress(json.JSONDecodeError, TypeError):
            outcomes = json.loads(outcomes_str)

        sem_handled = False
        tool_handled = False

        # Process individual explicit expectation failures
        for outcome_obj in outcomes:
            e, a, f = self._get_exp_act(outcome_obj)
            if f == "Tool Response":
                continue
            raw_outcome = outcome_obj.get("outcome")
            outcome_str = EvalUtils._map_outcome(raw_outcome)

            score_val = None
            if f == "Semantic Similarity":
                raw_score = turn.get("semantic_score")
                if isinstance(raw_score, (int, float)):
                    s_val = (
                        int(raw_score)
                        if raw_score == int(raw_score)
                        else raw_score
                    )
                    score_val = f"{s_val} / 4.0"
                else:
                    score_val = (
                        str(raw_score) if raw_score is not None else None
                    )
            elif f == "Tool Call":
                raw_score = turn.get("tool_invocation_score")
                if isinstance(raw_score, (int, float)):
                    score_val = f"{int(raw_score * 100)}%"
                else:
                    score_val = EvalUtils._map_outcome(raw_score)

            all_metadata.append(
                {
                    "display_name": turn["display_name"],
                    "eval_result_id": turn["eval_result_id"],
                    "evaluation_run": turn.get("evaluation_run"),
                    "evaluation_status": turn.get("evaluation_status"),
                    "turn_index": turn["turn_index"],
                    "type": f,
                    "expected": e,
                    "actual": a,
                    "outcome": outcome_str,
                    "score": score_val,
                }
            )

            if raw_outcome == 2 or outcome_str == "FAIL":
                if f == "Semantic Similarity":
                    sem_handled = True
                elif f == "Tool Call":
                    raw_score = turn.get("tool_invocation_score")
                    if EvalUtils._map_outcome(raw_score) == "FAIL":
                        tool_handled = True

                failures.append(
                    {
                        "display_name": turn["display_name"],
                        "eval_result_id": turn["eval_result_id"],
                        "turn_index": turn["turn_index"],
                        "failure_type": f,
                        "expected": e,
                        "actual": a,
                        "score": score_val,
                    }
                )

        # Process overall semantic failure if not caught
        sem_outcome = turn.get("semantic_outcome")
        if sem_outcome == "FAIL" and not sem_handled:
            e_text, a_text = "", ""
            for outcome_obj in outcomes:
                if "agent_response" in outcome_obj.get("expectation", {}):
                    e_text, a_text, _ = self._get_exp_act(outcome_obj)
                    break
            raw_score = turn.get("semantic_score")
            if isinstance(raw_score, (int, float)):
                s_val = (
                    int(raw_score)
                    if raw_score == int(raw_score)
                    else raw_score
                )
                score_val = f"{s_val} / 4.0"
            else:
                score_val = (
                    str(raw_score) if raw_score is not None else None
                )

            failures.append(
                {
                    "display_name": turn["display_name"],
                    "eval_result_id": turn["eval_result_id"],
                    "turn_index": turn["turn_index"],
                    "failure_type": "Semantic Similarity",
                    "expected": e_text,
                    "actual": a_text,
                    "score": score_val,
                }
            )

        # Process overall tool failure if not caught
        tool_outcome = turn.get("tool_invocation_score")
        if tool_outcome == "FAIL" and not tool_handled:
            e_text, a_text = "", ""
            for outcome_obj in outcomes:
                if "tool_call" in outcome_obj.get("expectation", {}):
                    e_text, a_text, _ = self._get_exp_act(outcome_obj)
                    break
            failures.append(
                {
                    "display_name": turn["display_name"],
                    "eval_result_id": turn["eval_result_id"],
                    "turn_index": turn["turn_index"],
                    "failure_type": "Tool Call",
                    "expected": e_text,
                    "actual": a_text,
                    "score": "FAIL",
                }
            )

    df_failures = pd.DataFrame(failures)
    if not df_failures.empty:
        df_failures = df_failures.sort_values(
            by=["eval_result_id", "turn_index"]
        ).reset_index(drop=True)

    df_traces = pd.DataFrame(turns)

    df_metadata = pd.DataFrame(all_metadata)

    return {
        "summary": df_summary,
        "failures": df_failures,
        "trace": df_traces,
        "metadata": df_metadata,
    }

def get_latency_metrics_dfs(
    self,
    results: list[Any] | None = None,
    eval_names: list[str] | None = None,
    app_name: str | None = None,
) -> dict[str, pd.DataFrame]:
    """Generates latency metrics DataFrames from results and traces.

    Args:
        results: An optional list of Eval Result List payload chunks.
        eval_names: Alternatively, an optional list of string Display
            Names / Names of Evals.
        app_name: Optional override if retrieving Conversation traces
            dynamically.
    """
    if not results:
        if not getattr(self, "app_name", None) and not app_name:
            raise ValueError(
                "app_name must be set to look up evaluations by name."
            )
        results = []
        for name in eval_names or []:
            # Retrieve all results for the provided display name
            results.extend(self.list_evaluation_results(name))

    if not results:
        return {
            "eval_summary": pd.DataFrame(),
            "eval_details": pd.DataFrame(),
            "tool_summary": pd.DataFrame(),
            "tool_details": pd.DataFrame(),
            "callback_summary": pd.DataFrame(),
            "callback_details": pd.DataFrame(),
            "guardrail_summary": pd.DataFrame(),
            "guardrail_details": pd.DataFrame(),
        }

    conv_ids = set()
    for res_obj in results:
        res_dict = (
            type(res_obj).to_dict(res_obj)
            if not isinstance(res_obj, dict)
            else res_obj
        )
        turns = res_dict.get("golden_result", {}).get(
            "turn_replay_results", []
        )
        for t in turns:
            if t.get("conversation"):
                conv_ids.add(t.get("conversation"))

    target_app = app_name or getattr(self, "app_name", None)
    traces = {}
    if target_app and conv_ids:
        if target_app == getattr(self, "app_name", None):
            ch_getter = self.ch_client.get_conversation
        else:
            # Defer import to break circular dependency:
            # conversation_history -> latency_parser -> eval_utils
            # -> conversation_history
            from cxas_scrapi.core.conversation_history import (  # noqa: PLC0415
                ConversationHistory,
            )

            ch_client = ConversationHistory(
                app_name=target_app, creds=self.creds
            )
            ch_getter = ch_client.get_conversation
        traces = LatencyParser.fetch_conversation_traces(
            list(conv_ids), ch_getter
        )

    eval_details_rows = []
    eval_summary_agg = []
    tool_details_rows = []
    callback_details_rows = []
    guardrail_details_rows = []
    llm_details_rows = []

    for res_obj in results:
        res_dict = (
            type(res_obj).to_dict(res_obj)
            if not isinstance(res_obj, dict)
            else res_obj
        )
        result_name = res_dict.get("name", "")
        tokens = result_name.split("/")
        eval_result_id = tokens[-1] if tokens else result_name
        eval_name = "/".join(tokens[:-2]) if len(tokens) >= 2 else ""

        # Get display name
        display_name = eval_name
        evals_map = getattr(self, "_get_or_load_evals_map", lambda x: {})(
            getattr(self, "app_name", None)
        )
        if evals_map:
            for lookup_name, full_path in evals_map.get(
                "goldens", {}
            ).items():
                if full_path == eval_name:
                    display_name = lookup_name
                    break
            for lookup_name, full_path in evals_map.get(
                "scenarios", {}
            ).items():
                if full_path == eval_name:
                    display_name = lookup_name
                    break

        golden = res_dict.get("golden_result", {})
        turns = golden.get("turn_replay_results", [])

        run_total_turn_latencies = []
        run_tool_latencies = []
        run_llm_latencies = []
        run_guardrail_latencies = []
        run_callback_latencies = []

        for turn_idx, t in enumerate(turns):
            total_turn_ms = LatencyParser._parse_duration_ms(
                t.get("turn_latency", "0s")
            )

            # Turn specific items
            tool_calls = t.get("tool_call_latencies", [])
            turn_tool_ms = sum(
                LatencyParser._parse_duration_ms(
                    tc.get("execution_latency", "0s")
                )
                for tc in tool_calls
            )
            tool_names = ", ".join(
                [
                    tc.get("display_name", tc.get("tool", ""))
                    for tc in tool_calls
                ]
            )

            turn_llm_ms = 0.0
            turn_guardrail_ms = 0.0
            turn_callback_ms = 0.0

            cid = t.get("conversation")
            if cid and cid in traces:
                conv = traces[cid]
                conv_turns = conv.get("turns", [])
                # Match by index assuming evaluating linearly
                if turn_idx < len(conv_turns):
                    trace_turn = conv_turns[turn_idx]
                    root = trace_turn.get("root_span", {})
                    if root:
                        sums = LatencyParser._process_spans(
                            [root],
                            eval_result_id,
                            turn_idx + 1,
                            tool_details_rows,
                            callback_details_rows,
                            guardrail_details_rows,
                            llm_details_rows,
                            context_key="eval_result_id",
                        )
                        turn_llm_ms = sums["LLM"]
                        turn_guardrail_ms = sums["Guardrail"]
                        turn_callback_ms = sums["Callback"]

            run_total_turn_latencies.append(total_turn_ms)
            if turn_tool_ms > 0:
                run_tool_latencies.append(turn_tool_ms)
            if turn_llm_ms > 0:
                run_llm_latencies.append(turn_llm_ms)
            if turn_guardrail_ms > 0:
                run_guardrail_latencies.append(turn_guardrail_ms)
            if turn_callback_ms > 0:
                run_callback_latencies.append(turn_callback_ms)

            eval_details_rows.append(
                {
                    "display_name": display_name,
                    "eval_result_id": eval_result_id,
                    "turn_index": turn_idx + 1,
                    "Total Turn Latency (ms)": int(total_turn_ms),
                    "Tool Call Latencies (ms)": int(turn_tool_ms),
                    "LLM Latencies (ms)": int(turn_llm_ms),
                    "Guardrail Latencies (ms)": int(turn_guardrail_ms),
                    "Callback Latencies (ms)": int(turn_callback_ms),
                    "tool_names": tool_names,
                }
            )

        # Compute run summary aggregations
        t_agg = self._aggregate(run_total_turn_latencies)
        tc_agg = self._aggregate(run_tool_latencies)
        llm_agg = self._aggregate(run_llm_latencies)
        gr_agg = self._aggregate(run_guardrail_latencies)
        cb_agg = self._aggregate(run_callback_latencies)

        t_p50, t_p90, t_p99 = t_agg["p50"], t_agg["p90"], t_agg["p99"]
        llm_p50, llm_p90, llm_p99 = (
            llm_agg["p50"],
            llm_agg["p90"],
            llm_agg["p99"],
        )
        tc_p50, tc_p90, tc_p99 = tc_agg["p50"], tc_agg["p90"], tc_agg["p99"]
        gr_p50, gr_p90, gr_p99 = gr_agg["p50"], gr_agg["p90"], gr_agg["p99"]
        cb_p50, cb_p90, cb_p99 = cb_agg["p50"], cb_agg["p90"], cb_agg["p99"]

        p50_90_99_turn = f"{t_p50} ms | {t_p90} ms | {t_p99} ms"
        p50_90_99_llm = f"{llm_p50} ms | {llm_p90} ms | {llm_p99} ms"
        p50_90_99_tc = f"{tc_p50} ms | {tc_p90} ms | {tc_p99} ms"
        p50_90_99_gr = f"{gr_p50} ms | {gr_p90} ms | {gr_p99} ms"
        p50_90_99_cb = f"{cb_p50} ms | {cb_p90} ms | {cb_p99} ms"

        eval_summary_agg.append(
            {
                "display_name": display_name,
                "eval_result_id": eval_result_id,
                "evaluation_type": "Golden" if golden else "Scenario",
                "Average (Turn)": f"""{t_agg["Average"]} ms""",
                "p50 | p90 | p99 (Turn)": p50_90_99_turn,
                "Average (LLM)": f"""{llm_agg["Average"]} ms""",
                "p50 | p90 | p99 (LLM)": p50_90_99_llm,
                "Average (Tool Call)": f"""{tc_agg["Average"]} ms""",
                "p50 | p90 | p99 (Tool Call)": p50_90_99_tc,
                "Average (Guardrail)": f"""{gr_agg["Average"]} ms""",
                "p50 | p90 | p99 (Guardrail)": p50_90_99_gr,
                "Average (Callback)": f"""{cb_agg["Average"]} ms""",
                "p50 | p90 | p99 (Callback)": p50_90_99_cb,
            }
        )

    eval_details = pd.DataFrame(eval_details_rows)
    eval_summary = pd.DataFrame(eval_summary_agg)

    tool_details = pd.DataFrame(tool_details_rows)
    callback_details = pd.DataFrame(callback_details_rows)
    guardrail_details = pd.DataFrame(guardrail_details_rows)

    tool_summary = LatencyParser.build_summary_df(
        tool_details, ["tool_name"]
    )
    callback_summary = LatencyParser.build_summary_df(
        callback_details, ["agent", "stage", "description"]
    )
    guardrail_summary = LatencyParser.build_summary_df(
        guardrail_details, ["agent", "name"]
    )

    return {
        "eval_summary": eval_summary,
        "eval_details": eval_details,
        "tool_summary": tool_summary,
        "tool_details": tool_details,
        "callback_summary": callback_summary,
        "callback_details": callback_details,
        "guardrail_summary": guardrail_summary,
        "guardrail_details": guardrail_details,
    }