Task: workarena.servicenow.single-chart-min-max-retrieval
Goal: In the Catalog Item count by Classic Mobile Picture Type report, what is the maximum value in the chart? Give me both the label and the count. If there are many, pick one.
Task configuration: url: /now/nav/ui/classic/params/target/sys_report_template.do%3Fjvar_report_id=37681d24931f7a106162fb67dd03d692
question: max

Validation code:
import json
import math
import os
import re
from typing import Any, Dict, List, Optional
from urllib.parse import parse_qsl, unquote, urlparse

import requests

def _multi_unquote(s: str, rounds: int = 3) -> str:
    for _ in range(rounds):
        new = unquote(s)
        if new == s:
            break
        s = new
    return s


def _api_get(endpoint: str, **kwargs) -> requests.Response:
    return requests.get(
        os.getenv("SERVICENOW_INSTANCE_URL", "").rstrip("/") + endpoint,
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json",
            **json.loads(os.getenv("SERVICENOW_EXTRA_HTTP_HEADERS", "{}")),
        },
        **kwargs,
    )


def extract_dashboard_sys_id(url: str) -> str | None:
    raw = _multi_unquote(url)
    parsed = urlparse(raw)
    if parsed.query:
        val = dict(parse_qsl(parsed.query)).get("sysparm_dashboard")
        if val:
            return val


def get_widget(dashboard_sys_id: str, title: str) -> Optional[Dict[str, Any]]:
    # 1) dashboard -> tabs (m2m)
    m2m = _api_get(
        "/api/now/table/pa_m2m_dashboard_tabs",
        params={
            "sysparm_query": f"dashboard={dashboard_sys_id}",
            "sysparm_fields": "tab",
            "sysparm_limit": "1000",
        },
    ).json()["result"]
    tab_ids = [
        row["tab"]["value"] if isinstance(row["tab"], dict) else row["tab"]
        for row in m2m
    ]
    for tab_id in tab_ids:
        # 2) tab -> canvas_page
        tab = _api_get(
            f"/api/now/table/pa_tabs/{tab_id}",
            params={"sysparm_fields": "name,canvas_page"},
        ).json()["result"]
        canvas_id = (
            tab["canvas_page"]["value"]
            if isinstance(tab.get("canvas_page"), dict)
            else tab.get("canvas_page")
        )

        # 3) canvas_page -> panes -> portal widgets
        panes = _api_get(
            "/api/now/table/sys_grid_canvas_pane",
            params={
                "sysparm_query": f"grid_canvas={canvas_id}",
                "sysparm_fields": "portal_widget",
                "sysparm_limit": "1000",
                "sysparm_display_value": "all",
            },
        ).json()["result"]
        portal_widget_ids = [
            (
                p["portal_widget"]["value"]
                if isinstance(p["portal_widget"], dict)
                else p["portal_widget"]
            )
            for p in panes
            if p.get("portal_widget")
        ]
        if not portal_widget_ids:
            continue
        all_pw_id = ",".join(portal_widget_ids)

        prefs = _api_get(
            "/api/now/table/sys_portal_preferences",
            params={"sysparm_query": f"portal_sectionIN{all_pw_id}^name=sys_id"},
        ).json()["result"]
        prefs_sys_ids = [p["value"] for p in prefs]
        if not prefs_sys_ids:
            continue
        all_prefs_sys_id = ",".join(prefs_sys_ids)

        # Try PA widgets first
        pa = _api_get(
            "/api/now/table/pa_widgets",
            params={
                "sysparm_query": f"sys_idIN{all_prefs_sys_id}",
                "sysparm_limit": "1000",
                "sysparm_fields": "sys_id,name,indicator,visualization",
            },
        ).json()["result"]
        filtered_pa = [p for p in pa if p.get("name") == title]
        if filtered_pa:
            w = filtered_pa[0]
            return {
                "type": "pa_widget",
                "sys_id": w["sys_id"],
                "name": w.get("name"),
                "indicator": (
                    w["indicator"]["value"]
                    if isinstance(w.get("indicator"), dict)
                    else w.get("indicator")
                ),
                "visualization": w.get("visualization"),
            }

        # Try Reports
        rpt = _api_get(
            "/api/now/table/sys_report",
            params={
                "sysparm_query": f"sys_idIN{all_prefs_sys_id}",
                "sysparm_limit": "1000",
                "sysparm_fields": "sys_id,title,type,table,field,trend_field,filter,active",
                "sysparm_display_value": "true",
            },
        ).json()["result"]
        filtered_rpt = [r for r in rpt if r.get("title") == title]
        if filtered_rpt:
            r = filtered_rpt[0]
            return {
                "type": "report",
                "sys_id": r["sys_id"],
                "title": r.get("title"),
                "report_type": r.get("type"),
                "table": r.get("table"),
                "field_name": r.get("field"),
                "trend_field": r.get("trend_field"),
                "filter": r.get("filter"),
                "active": r.get("active"),
            }

    return None


def extract_report_sys_id(url: str) -> str | None:
    raw = _multi_unquote(url)
    parsed = urlparse(raw)
    if parsed.query:
        val = dict(parse_qsl(parsed.query)).get("jvar_report_id")
        if val:
            return val


def get_report(report_sys_id: str) -> Dict[str, Any]:
    r = _api_get(
        f"/api/now/table/sys_report/{report_sys_id}",
        params={
            "sysparm_fields": "sys_id,title,type,table,field,trend_field,filter,active",
            "sysparm_display_value": "true",
        },
    ).json()["result"]
    return {
        "type": "report",
        "sys_id": r["sys_id"],
        "title": r.get("title"),
        "report_type": r.get("type"),
        "table": r.get("table"),
        "field_name": r.get("field"),
        "trend_field": r.get("trend_field"),
        "filter": r.get("filter"),
        "active": r.get("active"),
    }


def get_chart_values(chart: Dict[str, Any]) -> List[Dict[str, Any]]:
    query_filter = chart.get("filter", "")
    results = _api_get(
        f"/api/now/stats/{chart['table']}",
        params={
            "sysparm_query": query_filter,
            "sysparm_group_by": chart["field_name"],
            "sysparm_count": "true",
            "sysparm_display_value": "true",
        },
    ).json()["result"]
    return results


def get_trend_values(chart: Dict[str, Any], interval: str = "month") -> List[Dict[str, Any]]:
    """Query table records and bucket by trend_field into time periods."""
    from collections import Counter
    from datetime import datetime, timedelta

    query_filter = chart.get("filter", "")
    trend_field = chart["field_name"]  # already swapped to trend_field by caller

    # Fetch all matching records, only the trend field
    offset = 0
    limit = 1000
    all_timestamps = []
    while True:
        records = _api_get(
            f"/api/now/table/{chart['table']}",
            params={
                "sysparm_query": query_filter,
                "sysparm_fields": trend_field,
                "sysparm_limit": str(limit),
                "sysparm_offset": str(offset),
            },
        ).json()["result"]
        for rec in records:
            ts = rec.get(trend_field, "")
            if ts:
                all_timestamps.append(ts)
        if len(records) < limit:
            break
        offset += limit

    # Bucket timestamps by interval
    buckets: Counter = Counter()
    for ts in all_timestamps:
        try:
            dt = datetime.strptime(ts[:19], "%Y-%m-%d %H:%M:%S")
        except ValueError:
            continue

        if interval == "month":
            label = dt.strftime("%b/%Y")
        elif interval == "week":
            monday = dt - timedelta(days=dt.weekday())
            label = monday.strftime("%Y-%m-%d")
        elif interval == "year":
            label = dt.strftime("%Y")
        elif interval == "day":
            label = dt.strftime("%Y-%m-%d")
        else:
            label = dt.strftime("%b/%Y")

        buckets[label] += 1

    total = sum(buckets.values())
    points = []
    for label, count in buckets.items():
        percent = round(count / total * 100, 4) if total > 0 else 0
        points.append({"label": label, "count": count, "percent": percent})
    return points


def _stats_to_chart_data(stats_results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Transform ServiceNow stats API results to chart_data format."""
    total = sum(int(r["stats"]["count"]) for r in stats_results)
    points = []
    for r in stats_results:
        count = int(r["stats"]["count"])
        gf = r.get("groupby_fields", [])
        label = gf[0].get("value", "") if gf else ""
        percent = round(count / total * 100, 4) if total > 0 else 0
        points.append({"label": label, "count": count, "percent": percent})
    return points


def _parse_trend_interval(filter_str: str) -> str:
    """Extract the trend interval from a TRENDBY directive in the filter string."""
    match = re.search(r"\^TRENDBY\w+,(\w+)", filter_str)
    if match:
        return match.group(1)
    return "month"



def analyze_chart(
    chart_data: List[Dict[str, Any]],
    question: str,
    chart_series: str,
    response: str,
) -> int:
    """Analyze chart data against the assistant's numeric answer.

    chart_data can be:
      - flat list of points: [{"label": str, "count": num, "percent": num}, ...]
      - multi-series: [{"name": str, "data": [point, ...]}, ...]
    """
    if not chart_data or not isinstance(chart_data, list):
        return 0

    if not response:
        return 0

    # Resolve multi-series to flat list of points
    if isinstance(chart_data[0], dict) and "data" in chart_data[0]:
        if len(chart_data) == 1:
            points = chart_data[0]["data"]
        else:
            if not chart_series:
                return 0
            matches = [s for s in chart_data if s.get("name") == chart_series]
            if not matches:
                return 0
            points = matches[0].get("data", [])
    else:
        points = chart_data

    if not points:
        return 0

    # Remove labels to avoid counting their numbers, then extract floats
    labels = {p.get("label", "") for p in points}
    scrubbed = str(response)
    for label in labels:
        if label:
            scrubbed = scrubbed.replace(label, "")
    # extract numbers like 1,234 or 12.5 or 12,5 (we normalize commas away)
    nums = re.findall(r"[\d]+(?:[.,]\d+)?", scrubbed.replace(",", ""))
    try:
        response_vals = sorted({float(x) for x in nums})
    except ValueError:
        response_vals = []
    if not response_vals:
        return 0

    def get_count_list():
        return [float(p.get("count", 0.0)) for p in points]

    def get_expected_value_for(label: str, fmt: str) -> float | None:
        for p in points:
            if p.get("label") == label:
                if fmt == "count":
                    return float(p.get("count", 0.0))
                return float(p.get("percent", 0.0))
        return None

    def close(a: float, b: float) -> bool:
        return math.isclose(a, b, rel_tol=1e-3, abs_tol=1e-6)

    q = question.strip().lower()

    # "value; format; label"
    if q.startswith("value"):
        parts = [s.strip() for s in question.split(";")]
        if len(parts) < 3:
            return 0
        fmt, label = parts[1], parts[2]
        expected = get_expected_value_for(label, fmt)
        if expected is None:
            return 0
        return 1 if any(close(expected, v) for v in response_vals) else 0

    # max/min
    if "max" in q or "min" in q:
        want_max = "max" == q
        counts = get_count_list()
        if not counts:
            return 0
        target = (max if want_max else min)(counts)
        resp_lower = response.lower()
        for p in points:
            if float(p.get("count", 0.0)) == target:
                # Normalize label: replace special chars with space, check each token
                label = re.sub(r"[^a-zA-Z0-9]", " ", p.get("label", "")).lower()
                tokens = [t for t in label.split() if t]
                label_match = tokens and all(t in resp_lower for t in tokens)
                if label_match and any(close(target, v) for v in response_vals):
                    return 1
        return 0

    # mean/median/mode
    if any(k in q for k in ("mean", "median", "mode")):
        counts = get_count_list()
        if not counts:
            return 0

        if "mean" in q:
            target = sum(counts) / len(counts)
        elif "median" in q:
            s = sorted(counts)
            mid = len(s) // 2
            target = s[mid] if len(s) % 2 else (s[mid - 1] + s[mid]) / 2.0
        else:  # "mode"
            freq: Dict[float, int] = {}
            for v in counts:
                freq[v] = freq.get(v, 0) + 1
            max_freq = max(freq.values())
            modes = sorted([v for v, c in freq.items() if c == max_freq])
            target = modes[0]

        if len(response_vals) != 1:
            return 0
        return 1 if close(target, response_vals[0]) else 0

    return 0


def validate(messages: List[Dict[str, str]], config: Dict[str, Any]) -> int:
    """
    Validate an assistant's numeric answer against chart data fetched from
    a ServiceNow dashboard or report.

    Expected config keys:
      - url: str           # dashboard or report URL
      - chart_title: str   # title of the chart on the dashboard
      - chart_series: str  # series name (empty string if single-series)
      - question: str      # e.g., "value; count; Label", "max", "min", "mean", "median", "mode"
    """
    if (
        "url" not in config
        or "chart_title" not in config
        or "chart_series" not in config
        or "question" not in config
    ):
        raise ValueError(
            "Target must contain 'url', 'chart_title', 'chart_series' and 'question'"
        )

    if not messages:
        raise ValueError("Messages must not be empty")

    # Extract last assistant message
    response = None
    for msg in reversed(messages):
        if msg.get("role") == "assistant":
            response = msg.get("content") or ""
            break
    if not response:
        return 0

    # Fetch chart data from API
    url = config["url"]
    chart_title = config["chart_title"]

    if "pa_dashboard.do" in url:
        dashboard_sys_id = extract_dashboard_sys_id(url)
        if not dashboard_sys_id:
            raise ValueError(f"Could not extract dashboard sys_id from URL: {url}")
        widget = get_widget(dashboard_sys_id, title=chart_title)
        if widget is None:
            raise ValueError(
                f"Chart '{chart_title}' not found on dashboard {dashboard_sys_id}"
            )
        chart_series = config["chart_series"]
        if chart_series and widget.get("trend_field"):
            widget_for_query = dict(widget)
            series_filter = f"{widget['field_name']}={chart_series}"
            existing_filter = widget.get("filter", "")
            if existing_filter and not existing_filter.startswith("^"):
                series_filter += "^"
            widget_for_query["filter"] = series_filter + existing_filter
            widget_for_query["field_name"] = widget["trend_field"]
            interval = _parse_trend_interval(widget.get("filter", ""))
            points = get_trend_values(widget_for_query, interval)
        elif not chart_series and widget.get("trend_field"):
            widget_for_query = dict(widget)
            widget_for_query["field_name"] = widget["trend_field"]
            interval = _parse_trend_interval(widget.get("filter", ""))
            points = get_trend_values(widget_for_query, interval)
        else:
            chart_values = get_chart_values(widget)
            points = _stats_to_chart_data(chart_values)
    elif "sys_report_template.do" in url:
        report_sys_id = extract_report_sys_id(url)
        if not report_sys_id:
            raise ValueError(f"Could not extract report sys_id from URL: {url}")
        report = get_report(report_sys_id)
        chart_series = config["chart_series"]
        if chart_series and report.get("trend_field"):
            # Trend chart with a specific series: filter by series, group by time
            report_for_query = dict(report)
            series_filter = f"{report['field_name']}={chart_series}"
            existing_filter = report.get("filter", "")
            if existing_filter and not existing_filter.startswith("^"):
                series_filter += "^"
            report_for_query["filter"] = series_filter + existing_filter
            report_for_query["field_name"] = report["trend_field"]
            interval = _parse_trend_interval(report.get("filter", ""))
            points = get_trend_values(report_for_query, interval)
        elif not chart_series and report.get("trend_field"):
            report_for_query = dict(report)
            report_for_query["field_name"] = report["trend_field"]
            interval = _parse_trend_interval(report.get("filter", ""))
            points = get_trend_values(report_for_query, interval)
        else:
            chart_values = get_chart_values(report)
            points = _stats_to_chart_data(chart_values)
    else:
        raise ValueError(f"Unsupported URL format: {url}")

    return analyze_chart(
        chart_data=points,
        question=config["question"],
        chart_series=config["chart_series"],
        response=response,
    )