# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Utility functions for generating reports."""
import contextlib
import datetime
import glob
import json
import os
import re
import typing
from collections.abc import Callable, Mapping, Sequence
from typing import Any
import jinja2
import pandas as pd
import yaml
from cxas_scrapi.core import tools
from cxas_scrapi.evals import runner as evals_runner
from cxas_scrapi.utils import (
base_components,
eval_utils,
gcs_utils,
report_components,
)
def _escape(text: typing.Any) -> typing.Any:
"""HTML-escape a string."""
return (
str(text)
.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
.replace('"', """)
)
def _fmt_duration(seconds: typing.Any) -> str:
"""Format duration: seconds if < 60, minutes otherwise."""
seconds_per_minute = 60
if seconds is None:
return ""
if seconds >= seconds_per_minute:
return f"{seconds / seconds_per_minute:.1f}m"
return f"{seconds:.1f}s"
def _resolve_tool_name(
raw_name: typing.Any, tools_map: typing.Any
) -> typing.Any:
"""Resolve a full resource path to a display name."""
if not raw_name:
return raw_name
# Check reverse map (resource path → display name)
if raw_name in tools_map:
return tools_map[raw_name]
# Try matching by tool ID (last segment)
tool_id = raw_name.split("/")[-1] if "/" in raw_name else raw_name
for path, display in tools_map.items():
if path.endswith(f"/{tool_id}"):
return display
# Fallback: just use last segment
return tool_id if "/" in raw_name else raw_name
def _format_trace_line(line: typing.Any, tools_map: typing.Any) -> typing.Any:
"""Format a trace line, resolving tool IDs to display names."""
if "Tool Call:" in line or "Tool Response:" in line:
# Replace resource paths with display names
for path, display in tools_map.items():
line = line.replace(path, display)
return line
def _get_html_head(ts: typing.Any) -> str:
"""Return the HTML head with CSS and JS."""
css_path = os.path.join(
os.path.dirname(__file__), "../resources/components/base/base.css"
)
js_path = os.path.join(
os.path.dirname(__file__), "../resources/components/base/interaction.js"
)
with open(css_path, encoding="utf-8") as f:
css = f.read()
with open(js_path, encoding="utf-8") as f:
js = f.read()
return f"""
Simulation Report - {ts}
"""
def _get_summary_block(
passed: typing.Any,
total: typing.Any,
errors: typing.Any,
modality: typing.Any,
model: typing.Any,
ts: typing.Any,
wall_clock_s: typing.Any,
) -> str:
"""Return the HTML summary block."""
pct = 100 * passed / total if total else 0
pass_threshold = 90
cls = "pass" if pct >= pass_threshold else "fail"
return f"""Simulation Eval Report
{pct:.1f}%
{passed}/{total} passed | {errors} errors |
{modality} | model: {model}
Generated {ts}
{f" | Runtime: {_fmt_duration(wall_clock_s)}" if wall_clock_s else ""}
"""
def _get_results_table(eval_stats: typing.Any) -> typing.Any:
"""Return the HTML results table."""
html = """
Results by Eval
| Score | Eval | Runs |
|---|
"""
for name, s in sorted(
eval_stats.items(), key=lambda x: x[1]["pass"] / max(x[1]["total"], 1)
):
score = f"{s['pass']}/{s['total']}"
cls = "pass" if s["pass"] == s["total"] else "fail"
dots = ""
for i, r in enumerate(s["runs"]):
dot_cls = "p" if r.get("passed") else ("e" if "error" in r else "f")
safe_name = name.replace("'", "\\'")
dots += (
f'"
)
html += (
f' | {score} | '
f"{_escape(name)} | {dots} |
\n"
)
html += "
\n"
return html
def _render_session_link(session_id: typing.Any, ces_base: typing.Any) -> str:
"""Render the session link."""
if not session_id:
return ""
if ces_base:
session_url = (
f"{ces_base}?panel=conversation_list&id={session_id}&source=EVAL"
)
return (
'\n"
)
return (
f'Session: {session_id}
\n'
)
def _render_session_parameters(sparams: typing.Any) -> typing.Any:
"""Render session parameters."""
if not sparams:
return ""
html = (
''
"⚙ Session Parameters
"
)
html += (
''
f"{_escape(json.dumps(sparams, indent=2))}Goal: {_escape(step["goal"])}
'
f"Criteria: {_escape(step['success_criteria'])}
"
)
badge_cls = step_cls.replace("pass", "met").replace("fail", "not-met")
html += (
f'Status: '
f"{_escape(step['status'])}
"
)
if step.get("justification"):
html += f"Justification: {_escape(step['justification'])}"
html += "
\n"
return html
def _render_expectation_details(expectation_details: typing.Any) -> typing.Any:
"""Render expectation details."""
if not expectation_details:
return ""
html = ""
for exp in expectation_details:
exp_cls = "met" if exp["status"] == "Met" else "not-met"
html += f''
html += f"{_escape(exp['status'])} {_escape(exp['expectation'])}"
if exp.get("justification"):
html += (
f'
{_escape(exp["justification"])}'
)
html += "
\n"
return html
def _parse_trace(trace: typing.Any, tools_map: typing.Any) -> typing.Any:
"""Parse trace lines into typed entries."""
parsed_lines = []
for entry in trace:
for line in entry.split("\n"):
stripped_line = line.strip()
if not stripped_line:
continue
formatted_line = _format_trace_line(stripped_line, tools_map)
if formatted_line.startswith(
"Agent Text (Diag):"
) or formatted_line.startswith("User Query:"):
continue
elif formatted_line.startswith("Agent Text:"):
parsed_lines.append(
("agent", formatted_line[len("Agent Text:") :].strip())
)
elif formatted_line.startswith("User:"):
parsed_lines.append(("user", formatted_line[5:].strip()))
elif formatted_line.startswith("Tool Call"):
parsed_lines.append(("tool_call", formatted_line))
elif formatted_line.startswith("Tool Response"):
parsed_lines.append(("tool_resp", formatted_line))
elif formatted_line.startswith("Agent Transfer:"):
parsed_lines.append(
(
"agent_transfer",
formatted_line[len("Agent Transfer:") :].strip(),
)
)
elif formatted_line.startswith("Custom Payload:"):
parsed_lines.append(
(
"custom_payload",
formatted_line[len("Custom Payload:") :].strip(),
)
)
else:
parsed_lines.append(("system", formatted_line))
return parsed_lines
def _merge_trace_lines(parsed_lines: typing.Any) -> typing.Any:
"""Merge consecutive agent lines and pair tool calls with responses."""
merged = []
for kind, text in parsed_lines:
if kind == "agent" and merged and merged[-1][0] == "agent":
merged[-1] = ("agent", merged[-1][1] + " " + text)
elif kind == "tool_resp" and merged and merged[-1][0] == "tool_call":
merged[-1] = ("tool_pair", merged[-1][1], text)
else:
merged.append((kind, text))
return merged
def _render_merged_items(merged: typing.Any) -> typing.Any:
"""Render merged trace items to HTML."""
html = ""
for item in merged:
kind = item[0]
if kind == "user":
html += f'User: {_escape(item[1])}
\n'
elif kind == "agent":
html += (
f'Agent: {_escape(item[1])}
\n'
)
elif kind in ("tool_call", "tool_pair"):
call_text = item[1]
lbl, _, args = call_text.partition(" with args ")
lbl = lbl.replace("Tool Call: ", "").replace(
"Tool Call (Output): ", ""
)
lbl = lbl.split("/")[-1] if "/" in lbl else lbl
html += (
''
f"🔧 {_escape(lbl)}
"
)
if args:
html += (
'Input:
'
f'{_escape(args)}'
)
if kind == "tool_pair":
_, _, result = item[2].partition(" with result ")
if result:
html += (
'Output:
'
f'{_escape(result)}'
)
html += "'
f"📤 {_escape(lbl)} response
"
)
if result:
html += f'{_escape(result)}'
html += "🔀 Agent Transfer:'
f" {_escape(item[1])}
\n"
)
elif kind == "custom_payload":
html += (
''
''
"📦 Custom Payload
"
f'{_escape(item[1])}'
" \n"
)
else:
html += f'{_escape(item[1])}
\n'
return html
def _render_trace(
trace: typing.Any, tools_map: typing.Any, turns: typing.Any
) -> typing.Any:
"""Render the conversation trace."""
if not trace:
return ""
html = (
f"Conversation Trace "
f"({turns} turns)
\n"
f'\n'
)
parsed_lines = _parse_trace(trace, tools_map)
merged = _merge_trace_lines(parsed_lines)
html += _render_merged_items(merged)
html += "
\n\n'
html += (
f"Run {r['run']} — "
f''
f"{'PASS' if r.get('passed') else 'FAIL'}"
)
html += (
f" | goals: {r.get('goals', '?')} | "
f"expectations: {r.get('expectations', '?')} | "
f"turns: {r.get('turns', '?')}
\n"
)
html += _render_session_link(session_id, ces_base)
sparams = r.get("session_parameters", {})
html += _render_session_parameters(sparams)
if "error" in r:
html += (
f'Error: '
f"{_escape(r['error'])}
\n"
)
else:
html += _render_step_details(r.get("step_details", []))
html += _render_expectation_details(r.get("expectation_details", []))
html += _render_trace(
r.get("detailed_trace", []), tools_map, r.get("turns", "?")
)
html += " \n"
return html
def _upload_to_gcs(
output_path: str,
content: str,
content_type: str = "text/html; charset=utf-8",
) -> str | None:
"""Uploads the report to GCS and returns the mTLS URL or None on failure."""
try:
gcs = gcs_utils.GCSUtils()
mtls_url = gcs.upload_string(
output_path, content, content_type=content_type
)
print(f"Report uploaded to GCS: {output_path}")
print(f"Authenticated URL: {mtls_url}")
return mtls_url
except Exception as e: # pylint: disable=broad-exception-caught
print(f"WARNING: GCS upload failed ({e}). Falling back to local file.")
return None
def generate_html_report(
results: list[dict[str, Any]],
output_path: str,
modality: str,
model: str,
app_name: str = "",
wall_clock_s: float | None = None,
user_agent_extension: str | None = None,
) -> None:
"""Generate an HTML report and save it locally or upload to GCS.
If output_path starts with 'gs://', the report is uploaded to GCS.
If the upload fails, it falls back to saving a local file.
Args:
results: The list of evaluation result dicts.
output_path: The local or GCS file path to write the HTML report to.
modality: The modality used for the evaluation (e.g., 'text').
model: The model name used for the evaluation.
app_name: The CX Agent Studio (CXAS) agent resource name.
wall_clock_s: Total elapsed execution time in seconds.
user_agent_extension: Optional user agent extension string.
"""
total = len(results)
passed = sum(1 for r in results if r.get("passed"))
errors = sum(1 for r in results if "error" in r)
eval_stats = {}
for r in results:
n = r["name"]
if n not in eval_stats:
eval_stats[n] = {"pass": 0, "total": 0, "runs": []}
eval_stats[n]["total"] += 1
if r.get("passed"):
eval_stats[n]["pass"] += 1
eval_stats[n]["runs"].append(r)
tools_map = {}
if app_name:
try: # noqa: SIM105
tools_map = tools.Tools(
app_name=app_name, user_agent_extension=user_agent_extension
).get_tools_map()
except Exception: # pylint: disable=broad-exception-caught
pass
parts = app_name.split("/") if app_name else []
project_id_idx = 1
location_idx = 3
app_id_idx = 5
project_id = parts[project_id_idx] if len(parts) > project_id_idx else ""
location = parts[location_idx] if len(parts) > location_idx else ""
app_id = parts[app_id_idx] if len(parts) > app_id_idx else ""
ces_base = (
f"https://ces.cloud.google.com/projects/{project_id}/locations/{location}/apps/{app_id}"
if app_id
else ""
)
ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")
html = _get_html_head(ts)
html += _get_summary_block(
passed, total, errors, modality, model, ts, wall_clock_s
)
html += _get_results_table(eval_stats)
html += "\nEval Details
\n"
for name, s in sorted(
eval_stats.items(), key=lambda x: x[1]["pass"] / max(x[1]["total"], 1)
):
score = f"{s['pass']}/{s['total']}"
cls = "pass-bg" if s["pass"] == s["total"] else "fail-bg"
html += f'\n'
html += (
f'\n"
)
html += '
\n'
for r in s["runs"]:
html += _get_run_detail(r, ces_base, tools_map)
html += "