diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py index 639bee5b7..89e93dde8 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py @@ -30,6 +30,14 @@ evaluate_agentic_guardrail, run_agentic_guardrail, ) +from gooddata_eval.core.agentic.kda_skill import ( + AgenticKdaSummary, + KdaEvaluation, + KdaRunResult, + KdaSkillAssertionError, + evaluate_agentic_kda_skill, + run_agentic_kda_skill, +) from gooddata_eval.core.agentic.metric_skill import ( AgenticMetricSummary, MetricRunResult, @@ -56,6 +64,7 @@ "AgenticAlertSummary", "AgenticGeneralQuestionSummary", "AgenticGuardrailSummary", + "AgenticKdaSummary", "AgenticMetricSummary", "AgenticSearchSummary", "AgenticRunSummary", @@ -69,6 +78,9 @@ "GeneralQuestionResult", "GuardrailAssertionError", "GuardrailResult", + "KdaEvaluation", + "KdaRunResult", + "KdaSkillAssertionError", "MetricRunResult", "MetricSkillAssertionError", "RunResult", @@ -81,6 +93,7 @@ "evaluate_agentic_conversation", "evaluate_agentic_general_question", "evaluate_agentic_guardrail", + "evaluate_agentic_kda_skill", "evaluate_agentic_metric_skill", "evaluate_agentic_search_tool", "evaluate_agentic_visualization", @@ -88,6 +101,7 @@ "run_agentic_conversation", "run_agentic_general_question", "run_agentic_guardrail", + "run_agentic_kda_skill", "run_agentic_metric_skill", "run_agentic_search_tool", "run_agentic_visualization", diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py new file mode 100644 index 000000000..643efe669 --- /dev/null +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py @@ -0,0 +1,680 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +"""Agentic KDA (Key Driver Analysis)-skill evaluation runner.""" + +from __future__ import annotations + +import json +import logging +import os +import re +from dataclasses import dataclass +from typing import Any + +from gooddata_eval.core.chat.sse_client import ChatClient +from gooddata_eval.core.config import ReasoningEffort +from gooddata_eval.core.models import ToolCallEvent + +_log = logging.getLogger(__name__) + +# A single run per case; callers that want pass_at_k/pass_power_k variance pass k +# explicitly (e.g. gdc-nas's KDA_RUN_K for the daily cron). +_DEFAULT_K = 1 +# KDA cases are designed to resolve in one turn (unlike alert/metric skills), so this is +# only a safety net for the rare disambiguation turn -- a title collision, or a +# metric-vs-fact form choice -- not a general multi-turn budget. +_DEFAULT_MAX_ITERATIONS = 2 +# Fallback only -- every real agent_kda_skill item sets its own absolute_tolerance. +_DEFAULT_SUMMARY_REL_TOLERANCE = 0.01 + + +def _to_number(value: object) -> float | int | None: + """Convert string/number to int or float, None on failure. Mirrors alert_skill._to_number + -- the API is contractually numeric here, but this guards against a malformed response + raising a raw ValueError instead of failing the check cleanly.""" + if value is None: + return None + try: + f = float(str(value)) + return int(f) if f == int(f) else f + except (ValueError, TypeError): + return None + + +def _normalize_measure(m: dict) -> tuple[Any, Any, Any]: + return (m.get("type"), m.get("id"), m.get("aggregation")) + + +def _measure_matches(actual: object, expected: dict | list[dict] | None) -> bool: + """expected may be a single candidate dict or a list of candidate dicts (mirrors + metric_skill's expected_output: dict | list -- e.g. case 1 accepts either the + catalog metric id or the mathematically equivalent ad-hoc fact+SUM). + + ``actual`` is typed ``object``, not ``dict``, and checked with ``isinstance`` (mirroring + alert_skill._deep_subset) because it comes from a tool call the LLM constructed -- + a malformed call could put a non-dict value there. + """ + if not isinstance(actual, dict) or expected is None: + return False + candidates = expected if isinstance(expected, list) else [expected] + actual_norm = _normalize_measure(actual) + return any(actual_norm == _normalize_measure(c) for c in candidates if isinstance(c, dict)) + + +def _filters_match(actual: object, expected: list) -> bool: + """Set-equal on filters, not list-equal: the LLM applying the same filters in a + different order is not a mismatch. ``sort_keys=True`` alone only orders keys *within* + each dict -- it does nothing for the outer list's element order, which is exactly what + ``_measure_matches``'s normalized-tuple comparison doesn't have to worry about. Each + filter is canonicalized to its sorted-keys JSON string, then the two *sets* of those + strings are compared, so order differences no longer produce a false negative. + """ + # isinstance, not truthiness (mirrors _measure_matches's own guard): actual comes from + # a tool call the LLM constructed, so a malformed call could put a non-list value there. + actual_list = actual if isinstance(actual, list) else [] + try: + canon_actual = sorted(json.dumps(f, sort_keys=True) for f in actual_list) + canon_expected = sorted(json.dumps(f, sort_keys=True) for f in expected) + return canon_actual == canon_expected + except TypeError: + return False + + +def _within_relative_tolerance(actual: object, expected: object, rel_tolerance: float, *, base: object = None) -> bool: + """abs(actual - expected) / abs(base) <= rel_tolerance -- NOT absolute difference. + + Summary values are revenue-scale (can be in the millions), where a fixed absolute band + is either meaninglessly loose or effectively an exact-match bar that fails + near-constantly, depending on the metric's scale. Falls back to an absolute band only + when the base is exactly 0 (a relative comparison is undefined there). + + ``base`` defaults to ``expected`` (the usual case: reference_value/analyzed_value are + each their own scale anchor), but callers checking ``change`` must pass an explicit + ``base`` (reference_value, not ``change`` itself) -- ``change`` is a DIFFERENCE, often + orders of magnitude smaller than reference/analyzed_value, so tolerance relative to + ``change`` itself is far tighter than the same nominal percent applied to the other two + fields (e.g. 1% of a 10,000 change is 100, while 1% of the 1,000,000 reference_value it + was computed from is 10,000 -- a 100x tighter absolute band for no intentional reason). + """ + a, e = _to_number(actual), _to_number(expected) + b = e if base is None else _to_number(base) + if a is None or e is None or b is None: + return False + if b == 0: + return abs(a - e) <= rel_tolerance + return abs(a - e) / abs(b) <= rel_tolerance + + +def _within_absolute_tolerance(actual: object, expected: object, abs_tolerance: float) -> bool: + """abs(actual - expected) <= abs_tolerance -- for dataset items that explicitly opt into + an absolute band via ``absolute_tolerance`` (see _resolve_summary_tolerance) instead of + the module default of relative tolerance. + """ + a, e = _to_number(actual), _to_number(expected) + if a is None or e is None: + return False + return abs(a - e) <= abs_tolerance + + +_SUMMARY_TOLERANCE_KEYS = frozenset({"absolute_tolerance", "relative_tolerance"}) + + +def _resolve_summary_tolerance(expected_summary: dict) -> tuple[str, float]: + """("absolute" | "relative", tolerance value) for a Summary block. + + ``absolute_tolerance`` wins if present -- it's the tolerance vocabulary already + established elsewhere in this repo (td_config/main_config.py, ext_comparators.py), and + the one every current agent_kda_skill dataset item actually uses; ``relative_tolerance`` + (this module's own default) is checked second for backward compatibility. Any OTHER + key containing "tolerance" is almost certainly a typo of one of the two real names -- + warn instead of silently falling back to the default, since a dataset author who wrote + the wrong key would otherwise never find out their intended tolerance was ignored (this + exact silent failure is why agent_kda_skill's own absolute_tolerance: 0.01 never took + effect until this function existed). + """ + unknown = sorted(k for k in expected_summary if "tolerance" in k.lower() and k not in _SUMMARY_TOLERANCE_KEYS) + if unknown: + _log.warning( + "KDA dataset item's Summary has unrecognized tolerance key(s) %s -- ignored, " + "falling back to the default relative tolerance. Recognized keys: %s", + unknown, + sorted(_SUMMARY_TOLERANCE_KEYS), + ) + if "absolute_tolerance" in expected_summary: + return "absolute", expected_summary["absolute_tolerance"] + return "relative", expected_summary.get("relative_tolerance", _DEFAULT_SUMMARY_REL_TOLERANCE) + + +def _is_asking_kda_clarification(text: str) -> bool: + """True if ``text`` reads as the agent asking the user for input, not a final answer. + + KDA-specific, not shared with metric_skill.py/conversation.py: each skill's disambiguation + turns have independently drifted in shape, so a shared heuristic silently changes behavior + for skills it wasn't tuned against. Only the bare ``"?"``-anywhere check is tightened here, + to require the message actually end on a question -- a "?" anywhere in the text also + matches a final answer that merely quotes or rhetorically references a question, which + would wrongly keep KDA's single-turn cases going into a simulated-reply retry and could + mask a real turn-1 failure behind an artificial turn-2 pass. The other phrase checks + mirror metric_skill.py's own heuristic, since KDA's disambiguation scope (a title + collision, a metric-vs-fact form choice) is the same shape of question. + """ + if not text: + return False + t = text.strip().lower() + if t.endswith("?"): + return True + # "to clarify, ..." (optionally "just to clarify, ...") is a discourse marker meaning + # "in other words" -- it introduces a restated FINAL answer, not a request for one. + # Stripping it before the substring checks below keeps "clarif" able to catch genuine + # requests ("Could you clarify...", "I need clarification on...") without matching a + # final answer that merely opens with this phrase (e.g. "To clarify, revenue rose 12%"). + t = re.sub(r"^(just )?to clarify,?\s*", "", t) + return "could you" in t or "please provide" in t or "clarif" in t + + +def generate_simulated_kda_response(agent_message: str, measure_candidates: dict | list[dict] | None) -> str: + """Generate a user reply to keep the KDA-skill conversation going (gpt-4o-mini). + + Used only when the agent asks a clarifying question instead of triggering KDA + directly (e.g. a title collision between two metrics). Picks *any* candidate from + ``measure_candidates`` -- not necessarily the one an eventual correctness ticket + would require -- because the current scope only needs KDA to trigger, not the + resulting measure to be exactly right (see KdaEvaluation docstring). + + Always uses OpenAI regardless of which provider the combo under test runs -- this is + test-harness plumbing to keep a disambiguation turn moving, not the system under test, + and CI always has ``OPENAI_API_KEY`` from Vault for every combo (see + rw_e2e_test_tavern.yml) independent of the combo's own provider/model. + """ + try: + from openai import OpenAI # noqa: PLC0415 + except ImportError as exc: + raise RuntimeError("openai package is required for generate_simulated_kda_response") from exc + + api_key = os.environ.get("OPENAI_API_KEY") + if not api_key: + raise OSError("OPENAI_API_KEY environment variable is not set") + + client = OpenAI(api_key=api_key) + candidates = measure_candidates if isinstance(measure_candidates, list) else [measure_candidates or {}] + candidate_desc = "; or ".join( + f"{c.get('type')} '{c.get('id')}'" + (f" (aggregation {c['aggregation']})" if c.get("aggregation") else "") + for c in candidates + ) + prompt = ( + f"You are simulating a user in a conversation with a BI assistant that runs key driver " + f"analysis. The assistant said: '{agent_message}'. " + f"The user is happy to proceed with any of the following: {candidate_desc}. " + f"Reply briefly as the user, picking whichever of those the assistant offered." + ) + response = client.chat.completions.create( + model="gpt-4o-mini", + messages=[{"role": "user", "content": prompt}], + max_tokens=150, + ) + return response.choices[0].message.content or "Please proceed with either option." + + +def _extract_kda_calls(tool_call_events: list[ToolCallEvent]) -> tuple[dict | None, dict | None]: + """Return (create_args, execute_result) for the LAST create/execute *pair* -- not the + last create and last execute picked independently. Taking the last pair (not the + first) matches the observed retry-loop behaviour (kda_1 fails, kda_2 retries): the + last attempt is what actually determined the answer the chatbot gave. A new create + call clears any earlier execute_result, since that result belongs to the create it + followed, not to this one -- without that reset, `create_1 -> execute_1(success) -> + create_2 (never executed)` would wrongly pair create_2's args with execute_1's result. + """ + create_args: dict | None = None + execute_result: dict | None = None + for tc in tool_call_events: + if tc.function_name == "create_key_driver_analysis": + create_args = tc.parsed_arguments() + execute_result = None + elif tc.function_name == "execute_key_driver_analysis" and tc.result: + execute_result = tc.parsed_result() + return create_args, execute_result + + +@dataclass +class KdaEvaluation: + """Evaluation scores for a single KDA-skill run. + + Scope: this suite currently asserts only that the KDA process runs to completion -- + the tool chain triggers, executes successfully, and the chat turn ends cleanly with a + non-empty response (``turn_completed`` requires both gen-ai's stream-ended signal and + a non-empty ``text_response`` -- a stream that ends cleanly but delivers nothing to + the user isn't a completed turn either). Per-field correctness (Measure/Date + Attribute/Periods/Filters/Summary matching the expected values) is computed and + logged for visibility but intentionally excluded from ``strict_pass`` -- that + verification is scoped to a follow-up ticket, not this one. + """ + + # Core: gates strict_pass. + kda_triggered: bool + executed: bool + success: bool + turn_completed: bool + + # Informational only: computed and logged, but not required for strict_pass. None (not + # bool) when the dataset item's expected_output doesn't have this key at all -- see + # _evaluate_run's own comment for why that must not be conflated with "checked, wrong". + measure_correct: bool | None + date_attribute_correct: bool | None + analyzed_period_correct: bool | None + reference_period_correct: bool | None + filters_correct: bool | None + summary_correct: bool | None + + # Diagnostic only, never None (unlike the fields above): whether a simulated user + # reply was needed to get past a clarifying question. The simulated reply names the + # acceptable candidate(s) drawn from expected_output itself (see + # generate_simulated_kda_response), so measure_correct and friends would not be an + # independent signal for a disambiguated run -- the agent was told the answer, not + # left to infer it. _evaluate_run nulls all six of them the moment this is True, + # rather than leaving a visibly-True-but-not-trustworthy value for a future + # correctness ticket to remember to exclude. + disambiguated: bool = False + + @property + def strict_pass(self) -> bool: + return all([self.kda_triggered, self.executed, self.success, self.turn_completed]) + + +@dataclass +class KdaRunResult: + """Outcome of one run (one conversation, one message) for a KDA case.""" + + conversation_id: str + eval: KdaEvaluation + actual_create_args: dict | None + actual_execute_result: dict | None + # Wall-clock time of the specific send_message() call that triggered KDA (None if it + # never did) -- see run_agentic_kda_skill's _run_once. + turn_wall_clock_sec: float | None = None + + +@dataclass +class AgenticKdaSummary: + """Aggregated outcome of K runs for a KDA case.""" + + run_results: list[KdaRunResult] + pass_at_k: bool + pass_power_k: bool + best: KdaRunResult + + +def _evaluate_run( + create_args: dict | None, + execute_result: dict | None, + turn_completed: bool, + expected: dict, + disambiguated: bool = False, +) -> KdaEvaluation: + kda_triggered = create_args is not None + executed = execute_result is not None + # Checked against the tool's own result, not compared to expected_output -- this + # scope only cares whether KDA itself reported success, not input/output correctness. + success = executed and execute_result.get("success") is True + + # Informational only (see KdaEvaluation docstring) -- still computed so a follow-up + # ticket can promote these to strict_pass without redoing the extraction logic. + # + # Each check is None, not False, both when `expected` doesn't have the corresponding + # key at all, AND when its own precondition isn't met (kda_triggered False for the five + # below, executed+success False for summary_correct) -- a dataset item that simply + # doesn't specify an expectation for a field (or has it under a misspelled/differently- + # cased key -- these are hand-authored, title-case, space-separated keys with no schema + # validation upstream), or where KDA never got far enough to have a value to check, must + # not silently score as "wrong": that would be indistinguishable on a dashboard from KDA + # genuinely getting the field wrong. Specifically for Filters, "no expectation given" is + # also not the same claim as "expected no filters at all" -- defaulting a missing key to + # `[]` would conflate the two. + measure_correct = None + if "Measure" in expected and kda_triggered: + measure_correct = _measure_matches(create_args.get("measure"), expected["Measure"]) + date_attribute_correct = None + if "Date Attribute" in expected and kda_triggered: + date_attribute_correct = create_args.get("date_attribute_id") == expected["Date Attribute"] + analyzed_period_correct = None + if "Analyzed Period" in expected and kda_triggered: + analyzed_period_correct = create_args.get("analyzed_period") == expected["Analyzed Period"] + reference_period_correct = None + if "Reference Period" in expected and kda_triggered: + reference_period_correct = create_args.get("reference_period") == expected["Reference Period"] + filters_correct = None + if "Filters" in expected and kda_triggered: + filters_correct = _filters_match(create_args.get("filters"), expected["Filters"]) + + summary_correct = None + if "Summary" in expected and executed and success: + data = execute_result.get("data") or {} + actual_summary = data.get("summary") or {} + expected_summary = expected["Summary"] or {} + mode, tolerance = _resolve_summary_tolerance(expected_summary) + if mode == "absolute": + summary_correct = ( + _within_absolute_tolerance( + actual_summary.get("reference_value"), expected_summary.get("reference_value"), tolerance + ) + and _within_absolute_tolerance( + actual_summary.get("analyzed_value"), expected_summary.get("analyzed_value"), tolerance + ) + and _within_absolute_tolerance(actual_summary.get("change"), expected_summary.get("change"), tolerance) + ) + else: + summary_correct = ( + _within_relative_tolerance( + actual_summary.get("reference_value"), expected_summary.get("reference_value"), tolerance + ) + and _within_relative_tolerance( + actual_summary.get("analyzed_value"), expected_summary.get("analyzed_value"), tolerance + ) + and _within_relative_tolerance( + actual_summary.get("change"), + expected_summary.get("change"), + tolerance, + base=expected_summary.get("reference_value"), + ) + ) + + if disambiguated: + # The simulated user reply names the acceptable candidate(s) drawn straight from + # expected_output (see KdaEvaluation.disambiguated) -- once that happened, these + # fields would be testing whether the agent copied what it was just told, not + # what it inferred on its own. Null them immediately rather than leaving a + # visibly-True-but-not-trustworthy value for a future ticket to remember to + # exclude. + measure_correct = date_attribute_correct = analyzed_period_correct = None + reference_period_correct = filters_correct = summary_correct = None + + return KdaEvaluation( + kda_triggered=kda_triggered, + executed=executed, + success=success, + turn_completed=turn_completed, + measure_correct=measure_correct, + date_attribute_correct=date_attribute_correct, + analyzed_period_correct=analyzed_period_correct, + reference_period_correct=reference_period_correct, + filters_correct=filters_correct, + summary_correct=summary_correct, + disambiguated=disambiguated, + ) + + +def run_agentic_kda_skill( + host: str, + token: str, + workspace_id: str, + question: str, + expected_output: dict, + k: int = _DEFAULT_K, + max_iterations: int = _DEFAULT_MAX_ITERATIONS, + initial_conversation_id: str | None = None, + reasoning_effort: ReasoningEffort | None = None, +) -> AgenticKdaSummary: + """Run the KDA-skill agentic evaluation K times and return a summary. + + Each run is normally a single message in a single turn -- the agent_kda_skill + dataset is designed so every question resolves unambiguously -- but if the agent + asks a clarifying question instead of triggering KDA (a title collision, or a + metric-vs-fact form choice), a simulated user reply nudges it forward for up to + ``max_iterations`` turns, so a disambiguation turn doesn't block measuring whether + KDA itself triggers and completes. + """ + if k < 1: + # k is env-driven at the call site (KDA_RUN_K) -- a bad value ("0", a typo, a + # negative) must fail loudly here rather than being silently ignored. The single + # unconditional run below (conv_id_0) happens regardless of k, and the loop that + # adds the remaining k-1 runs is a no-op for any k <= 1 -- so k=0 or a negative + # value doesn't skip testing, it silently runs exactly once, indistinguishable + # from a deliberate k=1, instead of surfacing the bad config value. + raise ValueError(f"k must be >= 1, got {k}") + run_results: list[KdaRunResult] = [] + client = ChatClient(host=host, token=token, workspace_id=workspace_id, reasoning_effort=reasoning_effort) + + def _run_once(conv_id: str) -> KdaRunResult: + create_args: dict | None = None + execute_result: dict | None = None + turn_wall_clock_sec: float | None = None + turn_completed = False + disambiguated = False + current_question = question + + for iteration in range(max_iterations): + try: + chat_result = client.send_message(conv_id, current_question) + except Exception as exc: # noqa: BLE001 -- end this run, not the whole assertion + # Scored as a normal unsuccessful run instead of an uncaught crash, so it's + # still diagnosable in Langfuse. partial_result rescues any KDA tool calls + # that already streamed before a later, unrelated error killed the turn. + _log.warning("KDA send_message failed for conversation %s: %s", conv_id, exc) + partial = getattr(exc, "partial_result", None) + if partial is not None: + c_args, e_result = _extract_kda_calls(partial.tool_call_events or []) + if c_args is not None: + create_args, execute_result = c_args, e_result + turn_wall_clock_sec = partial.turn_wall_clock_sec + # The turn did not complete regardless of what iteration N-1 left behind -- + # without this, a crash on iteration 1 after a clean iteration 0 would keep + # logging kda_turn_completed=1 for a run that never finished. + turn_completed = False + break + c_args, e_result = _extract_kda_calls(chat_result.tool_call_events or []) + response_text = (chat_result.text_response or "").strip() + # gen-ai's own "response_ended" signal AND a non-empty answer -- a turn cut + # off mid-stream can still have emitted a partial, non-empty response before + # dying (response_text alone isn't enough), and a stream that ends cleanly + # but with nothing to show the user isn't "delivers a final answer" either + # (see KdaEvaluation docstring's scope statement -- stream_ended alone was + # weaker than that). + turn_completed = chat_result.stream_ended and bool(response_text) + if c_args is not None: + create_args, execute_result = c_args, e_result + turn_wall_clock_sec = chat_result.turn_wall_clock_sec + break + # max_iterations=2 (the default) means exactly ONE simulated-reply retry, not + # two: iteration 0 asks, iteration 1 is the retry, and this check (short- + # circuiting before _is_asking_kda_clarification) stops us from generating a + # simulated reply on the last iteration that the loop has no further iteration + # left to send -- that call costs a real OpenAI request for a reply nothing + # would ever use. + if iteration >= max_iterations - 1 or not _is_asking_kda_clarification(response_text): + break + try: + current_question = generate_simulated_kda_response(response_text, expected_output.get("Measure")) + disambiguated = True + except Exception as exc: # noqa: BLE001 -- safety net, not the assertion; end only this run + _log.warning("Simulated KDA user reply failed for conversation %s: %s", conv_id, exc) + break + + ev = _evaluate_run(create_args, execute_result, turn_completed, expected_output, disambiguated) + return KdaRunResult( + conversation_id=conv_id, + eval=ev, + actual_create_args=create_args, + actual_execute_result=execute_result, + turn_wall_clock_sec=turn_wall_clock_sec, + ) + + try: + conv_id_0 = initial_conversation_id if initial_conversation_id is not None else client.create_conversation() + try: + run_results.append(_run_once(conv_id_0)) + finally: + if initial_conversation_id is None: # only delete conversations we created + client.delete_conversation(conv_id_0) + + for _ in range(1, k): + conv_id = client.create_conversation() + try: + run_results.append(_run_once(conv_id)) + finally: + client.delete_conversation(conv_id) + finally: + client.close() + + pass_at_k = any(r.eval.strict_pass for r in run_results) + pass_power_k = all(r.eval.strict_pass for r in run_results) + best = max( + run_results, + key=lambda r: sum([r.eval.kda_triggered, r.eval.executed, r.eval.success, r.eval.turn_completed]), + ) + return AgenticKdaSummary( + run_results=run_results, + pass_at_k=pass_at_k, + pass_power_k=pass_power_k, + best=best, + ) + + +class KdaSkillAssertionError(AssertionError): + """Raised when a KDA-skill evaluation fails.""" + + __tracebackhide__ = True + + +def evaluate_agentic_kda_skill( + host: str, + token: str, + workspace_id: str, + question: str, + expected_output: dict, + k: int = _DEFAULT_K, + max_iterations: int = _DEFAULT_MAX_ITERATIONS, + initial_conversation_id: str | None = None, + langfuse: object | None = None, + dataset_item_id: str = "", + dataset_name: str = "agent_kda_skill", + run_timestamp: str | None = None, + model_version_override: str | None = None, + run_metadata_extra: dict | None = None, + reasoning_effort: ReasoningEffort | None = None, +) -> None: + """Run KDA-skill evaluation, log to Langfuse, and raise KdaSkillAssertionError on failure.""" + from datetime import datetime as _dt # noqa: PLC0415 + from datetime import timezone as _tz # noqa: PLC0415 + + from gooddata_eval.core.agentic._langfuse import try_make_langfuse_client # noqa: PLC0415 + + if langfuse is None: + langfuse = try_make_langfuse_client() + window_start = _dt.now(_tz.utc) + summary = run_agentic_kda_skill( + host=host, + token=token, + workspace_id=workspace_id, + question=question, + expected_output=expected_output, + k=k, + max_iterations=max_iterations, + initial_conversation_id=initial_conversation_id, + reasoning_effort=reasoning_effort, + ) + + if langfuse is not None and dataset_item_id: + from gooddata_eval.core.agentic._langfuse import ( # noqa: PLC0415 + build_run_context, + find_traces_per_conversation, + log_quality_and_value_scores, + observe, + score_safe, + ) + + run_name_base, run_metadata = build_run_context( + host, + token, + workspace_id, + dataset_name, + run_timestamp, + model_version_override, + run_metadata_extra, + reasoning_effort, + ) + # No custom selector -- same default (max-latency) as every other skill; harmless + # here since latency comes from run.turn_wall_clock_sec below, not this trace. + traces_by_conv = find_traces_per_conversation( + langfuse, + [r.conversation_id for r in summary.run_results], + window_start, + ) + suffix_needed = len(summary.run_results) > 1 + for run_idx, run in enumerate(summary.run_results): + pt = traces_by_conv.get(run.conversation_id) + run_name = f"{run_name_base}_run{run_idx}" if suffix_needed else run_name_base + ev = run.eval + # Gates strict_pass -- current scope is completion only (see KdaEvaluation docstring). + strict_checks = { + "kda_triggered": ev.kda_triggered, + "kda_executed": ev.executed, + "kda_success": ev.success, + "kda_turn_completed": ev.turn_completed, + } + # Informational only -- logged for visibility / a future correctness ticket, + # NOT part of strict_checks/strict_pass. See KdaEvaluation docstring. All + # kda_-prefixed, like the strict scores above: a bare "filters_correct" once + # collided with alert_skill's own filters_correct in gdc-nas's combo_report.py, + # which reads score dicts across skills by name -- prefixing every KDA score + # name (not just the one that happened to collide) closes that off for good. + informational_checks = { + "kda_measure_correct": ev.measure_correct, + "kda_date_attribute_correct": ev.date_attribute_correct, + "kda_analyzed_period_correct": ev.analyzed_period_correct, + "kda_reference_period_correct": ev.reference_period_correct, + "kda_filters_correct": ev.filters_correct, + "kda_summary_correct": ev.summary_correct, + # Unlike the fields above, never None -- always log it (see + # KdaEvaluation.disambiguated's own comment for why the *_correct fields + # above aren't a trustworthy signal on their own when this is True). + "kda_disambiguated": ev.disambiguated, + } + # run.turn_wall_clock_sec, not pt.latency: pt can be any trace of the conversation, + # not necessarily the KDA turn. + turn_wall_clock_sec = run.turn_wall_clock_sec + _log.info("[kda-report] %s: strict_pass=%s latency_sec=%s", run_name, ev.strict_pass, turn_wall_clock_sec) + with observe(langfuse, pt.id if pt else None, dataset_item_id, run_name, run_metadata) as tid: + for score_name, value in {**strict_checks, **informational_checks}.items(): + # informational_checks values are None, not bool, when the dataset item's + # expected_output has no key for that field at all -- "not asserted", not + # "checked, wrong". Skip logging entirely rather than coercing to a score. + if value is None: + continue + score_safe(langfuse, tid, name=score_name, value=float(value), data_type="BOOLEAN") + # kda_-prefixed like every other score above (see informational_checks + # comment): an unprefixed f"pass_at_{k}" is literally "pass_at_2" once + # k=2, colliding with visualization.py's own pass_at_2/pass_power_2 -- + # gdc-nas's combo_report.py dispatches skill classification by checking + # for that exact score name, so the collision would silently misfile + # every KDA record as a visualization record. + score_safe(langfuse, tid, name=f"kda_pass_at_{k}", value=float(summary.pass_at_k), data_type="BOOLEAN") + score_safe( + langfuse, tid, name=f"kda_pass_power_{k}", value=float(summary.pass_power_k), data_type="BOOLEAN" + ) + if turn_wall_clock_sec is not None: + # combo_report.py reads this score directly -- no trace re-resolution needed. + score_safe( + langfuse, tid, name="kda_turn_wall_clock_sec", value=turn_wall_clock_sec, data_type="NUMERIC" + ) + log_quality_and_value_scores( + langfuse, + tid, + strict_checks=strict_checks, + latency_sec=turn_wall_clock_sec, + cost_usd=pt.total_cost if pt and ev.kda_triggered else None, + ) + + if not summary.pass_at_k: + best = summary.best + ev = best.eval + message = ( + f"KDA skill assertion failed. strict_pass={ev.strict_pass} " + f"(kda_triggered={ev.kda_triggered}, executed={ev.executed}, " + f"success={ev.success}, turn_completed={ev.turn_completed}). " + f"Informational only, not part of strict_pass: " + f"measure_correct={ev.measure_correct}, date_attribute_correct={ev.date_attribute_correct}, " + f"analyzed_period_correct={ev.analyzed_period_correct}, " + f"reference_period_correct={ev.reference_period_correct}, " + f"filters_correct={ev.filters_correct}, summary_correct={ev.summary_correct}. " + f"Actual create args: {best.actual_create_args}. " + f"Actual execute result: {best.actual_execute_result}." + ) + raise KdaSkillAssertionError(message) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/chat/sse_client.py b/packages/gooddata-eval/src/gooddata_eval/core/chat/sse_client.py index 2db50d5a2..52a079203 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/chat/sse_client.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/chat/sse_client.py @@ -28,18 +28,38 @@ _log = logging.getLogger(__name__) SSE_DATA_PREFIX = "data: " +SSE_EVENT_PREFIX = "event: " +# gen-ai's last event, only if at least one item was already emitted (conversations_controller.py). +_RESPONSE_ENDED_EVENT = "response_ended" _RETRYABLE_STATUS_CODES: frozenset[int] = frozenset({429, 502, 503, 504}) _METADATA_SYNC_MARKER = "METADATA_SYNC_IN_PROGRESS" class ChatError(RuntimeError): - """Non-retryable error reported by the chat SSE stream.""" + """Non-retryable error reported by the chat SSE stream. - def __init__(self, message: str, *, status_code: int | None = None, detail: str | None = None) -> None: + ``partial_result`` carries whatever the accumulator captured before this error fired + (tool calls included) -- an error event ends the stream before ``_build_chat_result`` + ever runs, so without this a caller has no way to see, e.g., that KDA's own tool calls + already succeeded before an unrelated later error (a failed final-summary generation) + killed the turn. Callers must not assume it's complete: fields normally filled in only + at the very end of the stream (``stream_ended``, in particular) reflect the state at + the moment of the error, not a genuinely finished turn. + """ + + def __init__( + self, + message: str, + *, + status_code: int | None = None, + detail: str | None = None, + partial_result: ChatResult | None = None, + ) -> None: super().__init__(message) self.status_code = status_code self.detail = detail + self.partial_result = partial_result class TransientChatError(ChatError): @@ -109,6 +129,7 @@ class _SseAccumulator: reasoning_steps: list[dict[str, Any]] = field(default_factory=list) adhoc_viz_args: list[dict[str, Any]] = field(default_factory=list) response_id: str | None = None + stream_ended: bool = False def _handle_text(content: dict[str, Any], acc: _SseAccumulator) -> None: @@ -187,55 +208,77 @@ def _build_chat_result(acc: _SseAccumulator) -> ChatResult: } result = ChatResult.model_validate(payload) result.response_id = acc.response_id + result.stream_ended = acc.stream_ended return result def parse_sse_lines(lines: Iterable[str]) -> ChatResult: """Parse an SSE stream (iterable of decoded lines) into a ChatResult.""" acc = _SseAccumulator() - for raw_line in lines: - line = raw_line.decode("utf-8") if isinstance(raw_line, bytes) else raw_line - if not line or line.startswith("event: ") or not line.startswith(SSE_DATA_PREFIX): - continue - data_str = line[len(SSE_DATA_PREFIX) :] - if _METADATA_SYNC_MARKER in data_str: - raise TransientChatError( - f"SSE transient error: {_METADATA_SYNC_MARKER}", - status_code=None, - detail=None, - ) - try: - event_data = json.loads(data_str) - except json.JSONDecodeError: - continue - if "statusCode" in event_data: - code = event_data.get("statusCode") - detail = event_data.get("detail") - message = f"SSE error {code}: {detail}" - if code in _RETRYABLE_STATUS_CODES: - raise TransientChatError(message, status_code=code, detail=detail) - raise ChatError(message, status_code=code, detail=detail) - if event_data.get("responseId") and not acc.response_id: - acc.response_id = event_data["responseId"] - item = event_data.get("item") - if not item: - continue - if item.get("responseId") and not acc.response_id: - acc.response_id = item["responseId"] - role = item.get("role") - content: dict[str, Any] = item.get("content") or {} - ctype = content.get("type") - if role == "assistant": - if ctype == "text": - _handle_text(content, acc) - elif ctype == "multipart": - _handle_multipart(content, acc) - elif ctype == "reasoning": - _handle_reasoning(content, acc) - elif ctype == "toolCall": - _handle_tool_call(content, acc) - elif role == "tool" and ctype == "toolResult": - _handle_tool_result(content, acc) + current_event = "message" # SSE default in the absence of an explicit "event: " line + try: + for raw_line in lines: + line = raw_line.decode("utf-8") if isinstance(raw_line, bytes) else raw_line + if not line: + current_event = "message" # blank line ends one event block per the SSE spec + continue + if line.startswith(SSE_EVENT_PREFIX): + current_event = line[len(SSE_EVENT_PREFIX) :].strip() + continue + if not line.startswith(SSE_DATA_PREFIX): + continue + if current_event == _RESPONSE_ENDED_EVENT: + acc.stream_ended = True + continue + data_str = line[len(SSE_DATA_PREFIX) :] + if _METADATA_SYNC_MARKER in data_str: + raise TransientChatError( + f"SSE transient error: {_METADATA_SYNC_MARKER}", + status_code=None, + detail=None, + partial_result=_build_chat_result(acc), + ) + try: + event_data = json.loads(data_str) + except json.JSONDecodeError: + continue + if "statusCode" in event_data: + code = event_data.get("statusCode") + detail = event_data.get("detail") + message = f"SSE error {code}: {detail}" + if code in _RETRYABLE_STATUS_CODES: + raise TransientChatError( + message, status_code=code, detail=detail, partial_result=_build_chat_result(acc) + ) + raise ChatError(message, status_code=code, detail=detail, partial_result=_build_chat_result(acc)) + if event_data.get("responseId") and not acc.response_id: + acc.response_id = event_data["responseId"] + item = event_data.get("item") + if not item: + continue + if item.get("responseId") and not acc.response_id: + acc.response_id = item["responseId"] + role = item.get("role") + content: dict[str, Any] = item.get("content") or {} + ctype = content.get("type") + if role == "assistant": + if ctype == "text": + _handle_text(content, acc) + elif ctype == "multipart": + _handle_multipart(content, acc) + elif ctype == "reasoning": + _handle_reasoning(content, acc) + elif ctype == "toolCall": + _handle_tool_call(content, acc) + elif role == "tool" and ctype == "toolResult": + _handle_tool_result(content, acc) + except ChatError: + raise + except Exception as exc: + # A connection drop mid-stream (httpx.RemoteProtocolError/ReadError from inside + # `lines` itself) has no statusCode payload to raise a ChatError from directly -- + # without this, any tool calls already streamed before the drop would be lost. + raise ChatError(f"SSE stream error: {exc}", partial_result=_build_chat_result(acc)) from exc return _build_chat_result(acc) @@ -293,9 +336,21 @@ def send_message(self, conversation_id: str, question: str) -> ChatResult: body["options"] = {"reasoningEffort": self._reasoning_effort} def _do() -> ChatResult: + # t0 here, not around send_message(): includes the request/connection/server + # setup time a caller actually waits through, but still excludes + # _retry_transient's backoff sleep between attempts (harness overhead, not + # gen-ai's time), since each retry calls _do() -- and this timer -- fresh. + t0 = time.monotonic() with self._client.stream("POST", url, json=body, headers=headers) as resp: resp.raise_for_status() - return parse_sse_lines(resp.iter_lines()) + try: + result = parse_sse_lines(resp.iter_lines()) + except ChatError as exc: + if exc.partial_result is not None: + exc.partial_result.turn_wall_clock_sec = time.monotonic() - t0 + raise + result.turn_wall_clock_sec = time.monotonic() - t0 + return result return _retry_transient(_do, is_retryable=_is_retryable_exc) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/models.py b/packages/gooddata-eval/src/gooddata_eval/core/models.py index 336c313b9..72088e717 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/models.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/models.py @@ -100,6 +100,11 @@ class ChatResult(BaseModel): reasoning_step_count: int = Field(default=0, alias="reasoningStepCount") conversation_id: str | None = Field(default=None, alias="conversationId") response_id: str | None = Field(default=None, alias="responseId") + # Derived, not a raw server field -- see sse_client.py's _RESPONSE_ENDED_EVENT. + stream_ended: bool = False + # Set by ChatClient, not from the payload: wall-clock time of the SSE read itself, + # excluding retry backoff. + turn_wall_clock_sec: float | None = None class SummaryInput(BaseModel): diff --git a/packages/gooddata-eval/tests/test_agentic_kda_skill.py b/packages/gooddata-eval/tests/test_agentic_kda_skill.py new file mode 100644 index 000000000..6993475c0 --- /dev/null +++ b/packages/gooddata-eval/tests/test_agentic_kda_skill.py @@ -0,0 +1,1046 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +# SPDX-License-Identifier: LicenseRef-GoodData-Enterprise +import json +import logging +from unittest.mock import MagicMock, patch + +import httpx +import pytest +from gooddata_eval.core.agentic.kda_skill import ( + _DEFAULT_SUMMARY_REL_TOLERANCE, + KdaEvaluation, + KdaSkillAssertionError, + _evaluate_run, + _extract_kda_calls, + _filters_match, + _is_asking_kda_clarification, + _measure_matches, + _normalize_measure, + _resolve_summary_tolerance, + _to_number, + _within_relative_tolerance, + evaluate_agentic_kda_skill, + run_agentic_kda_skill, +) +from gooddata_eval.core.chat.sse_client import ChatError, TransientChatError +from gooddata_eval.core.models import ChatResult + +_EXPECTED = {"Measure": {"type": "metric", "id": "revenue"}} + + +def _tool_call(name: str, result: dict | None = None, arguments: dict | None = None): + return { + "functionName": name, + "functionArguments": "{}" if arguments is None else json.dumps(arguments), + "result": None if result is None else json.dumps(result), + } + + +def _kda_chat_result( + *, + success: bool = True, + text: str = "Here is the analysis.", + stream_ended: bool = True, + turn_wall_clock_sec: float | None = None, +) -> ChatResult: + return ChatResult.model_validate( + { + "textResponse": text, + "toolCallEvents": [ + _tool_call("create_key_driver_analysis", arguments={"measure": {"type": "metric", "id": "revenue"}}), + _tool_call("execute_key_driver_analysis", result={"success": success, "data": {"summary": {}}}), + ], + "reasoningStepCount": 1, + "stream_ended": stream_ended, + "turn_wall_clock_sec": turn_wall_clock_sec, + } + ) + + +def _no_kda_chat_result(text: str = "I could not find that metric.", *, stream_ended: bool = True) -> ChatResult: + return ChatResult.model_validate( + {"textResponse": text, "toolCallEvents": [], "reasoningStepCount": 1, "stream_ended": stream_ended} + ) + + +# --------------------------------------------------------------------------- # +# Pure helpers +# --------------------------------------------------------------------------- # +def test_to_number_int(): + assert _to_number("42") == 42 + + +def test_to_number_float(): + assert _to_number("4.5") == 4.5 + + +def test_to_number_none_on_garbage(): + assert _to_number("not-a-number") is None + assert _to_number(None) is None + + +def test_normalize_measure(): + assert _normalize_measure({"type": "metric", "id": "revenue", "aggregation": "SUM"}) == ( + "metric", + "revenue", + "SUM", + ) + + +def test_measure_matches_single_candidate(): + assert _measure_matches({"type": "metric", "id": "revenue"}, {"type": "metric", "id": "revenue"}) is True + + +def test_measure_matches_list_of_candidates(): + actual = {"type": "fact", "id": "order_value", "aggregation": "SUM"} + expected = [{"type": "metric", "id": "revenue"}, {"type": "fact", "id": "order_value", "aggregation": "SUM"}] + assert _measure_matches(actual, expected) is True + + +def test_measure_matches_false_when_actual_not_a_dict(): + assert _measure_matches("revenue", {"type": "metric", "id": "revenue"}) is False + + +def test_measure_matches_false_when_expected_none(): + assert _measure_matches({"type": "metric", "id": "revenue"}, None) is False + + +def test_filters_match_equal_ignores_key_order(): + assert _filters_match([{"b": 2, "a": 1}], [{"a": 1, "b": 2}]) is True + + +def test_filters_match_ignores_list_element_order(): + # Regression guard: sort_keys=True only orders keys within each dict, not the outer + # list's element order -- the LLM applying the same two filters in a different order + # must not read as a mismatch. + actual = [{"field": "region", "value": "EU"}, {"field": "year", "value": 2026}] + expected = [{"field": "year", "value": 2026}, {"field": "region", "value": "EU"}] + assert _filters_match(actual, expected) is True + + +def test_filters_match_false_on_mismatch(): + assert _filters_match([{"a": 1}], [{"a": 2}]) is False + + +def test_filters_match_treats_none_actual_as_empty_list(): + assert _filters_match(None, []) is True + + +def test_filters_match_false_on_non_serializable_value(): + assert _filters_match([{"a", "not json serializable"}], []) is False + + +def test_within_relative_tolerance_true(): + # 100 vs 100.5 is 0.5% off -- within a 1% relative band. + assert _within_relative_tolerance(100.0, 100.5, 0.01) is True + + +def test_within_relative_tolerance_false_when_exceeds(): + # 100 vs 105 is 5% off -- exceeds a 1% relative band. + assert _within_relative_tolerance(100.0, 105.0, 0.01) is False + + +def test_within_relative_tolerance_scales_with_magnitude(): + # A fixed absolute band would fail this (500 off), but 500/1_000_000 is 0.05% -- + # comfortably within a 1% relative band. This is the whole point of the fix: revenue- + # scale values need a tolerance that scales with the value, not a constant. + assert _within_relative_tolerance(1_000_000.0, 1_000_500.0, 0.01) is True + + +def test_within_relative_tolerance_false_on_non_numeric(): + assert _within_relative_tolerance("n/a", 100.0, 0.01) is False + + +def test_within_relative_tolerance_falls_back_to_absolute_band_when_expected_is_zero(): + assert _within_relative_tolerance(0.005, 0.0, 0.01) is True + assert _within_relative_tolerance(0.5, 0.0, 0.01) is False + + +def test_within_relative_tolerance_base_overrides_the_denominator(): + # Regression guard: `change` (a DIFFERENCE, e.g. 10,000 on a 1,000,000 reference_value) + # must be checked relative to a stable scale anchor, not relative to itself -- 1% of + # change=10,000 is only 100, a ~100x tighter absolute band than the 1% of 1,000,000 + # (10,000) that reference_value/analyzed_value get from the same nominal tolerance. + # actual=10,050 is 0.5% off the 10,000 reference_value base -- within a 1% band. + assert _within_relative_tolerance(10_050.0, 10_000.0, 0.01, base=1_000_000.0) is True + # Without the fix (tolerance relative to change itself): 50/10_000 = 0.5% would also + # pass here, so this alone doesn't distinguish the bug -- the next case does. + # actual=10,700 is 7% off the 10,000 change itself (would fail relative-to-itself), but + # only 0.07% off the 1,000,000 base -- correctly passes once base is the anchor. + assert _within_relative_tolerance(10_700.0, 10_000.0, 0.01, base=1_000_000.0) is True + + +def test_within_relative_tolerance_base_zero_falls_back_to_absolute_band(): + assert _within_relative_tolerance(0.005, 0.0, 0.01, base=0.0) is True + assert _within_relative_tolerance(0.5, 0.0, 0.01, base=0.0) is False + + +@pytest.mark.parametrize( + "text", + ["Could you clarify which metric?", "Please provide the date range.", "Did you mean revenue?"], +) +def test_is_asking_kda_clarification_true(text): + assert _is_asking_kda_clarification(text) is True + + +def test_is_asking_kda_clarification_false_on_plain_statement(): + assert _is_asking_kda_clarification("Here is the key driver analysis result.") is False + + +def test_is_asking_kda_clarification_false_on_empty(): + assert _is_asking_kda_clarification("") is False + + +def test_is_asking_kda_clarification_false_when_question_mark_is_not_the_final_answer(): + # Regression guard for the original bug: a final answer that merely quotes or + # rhetorically references a question must not be mistaken for a clarifying question. + text = 'The user asked "what changed?" so here is the key driver breakdown they requested.' + assert _is_asking_kda_clarification(text) is False + + +@pytest.mark.parametrize( + "text", + [ + "To clarify, revenue rose 12% quarter over quarter.", + "Just to clarify, the increase was driven by the South region.", + ], +) +def test_is_asking_kda_clarification_false_on_to_clarify_discourse_marker(text): + # Regression guard: "to clarify, ..." is a discourse marker ("in other words") that + # introduces a restated FINAL answer, not a request for one -- the bare "clarif" in t + # substring check would otherwise mistake this for a clarifying question and burn a + # simulated-reply turn on an answer that was already complete. + assert _is_asking_kda_clarification(text) is False + + +def test_is_asking_kda_clarification_true_for_genuine_clarify_request_despite_marker_strip(): + # The discourse-marker strip must not eat a genuine request that happens to start the + # same way it's phrased in practice. No trailing "?" here specifically so this exercises + # the "could you" substring check post-strip, not the separate endswith("?") check. + assert _is_asking_kda_clarification("To clarify, could you tell me which region you mean") is True + + +# --------------------------------------------------------------------------- # +# _evaluate_run -- informational fields must be None (not False) when the dataset item's +# expected_output has no key for that field at all, per _evaluate_run's own comment. +# --------------------------------------------------------------------------- # +_CREATE_ARGS = { + "measure": {"type": "metric", "id": "revenue"}, + "date_attribute_id": "date.month", + "analyzed_period": "2026-07", + "reference_period": "2026-06", + "filters": [{"field": "region", "value": "EU"}], +} +_EXECUTE_RESULT = { + "success": True, + "data": {"summary": {"reference_value": 100.0, "analyzed_value": 105.0, "change": 5.0}}, +} + + +def test_evaluate_run_all_informational_fields_are_none_when_expected_output_is_bare(): + ev = _evaluate_run(_CREATE_ARGS, _EXECUTE_RESULT, turn_completed=True, expected={}) + assert ev.measure_correct is None + assert ev.date_attribute_correct is None + assert ev.analyzed_period_correct is None + assert ev.reference_period_correct is None + assert ev.filters_correct is None + assert ev.summary_correct is None + # strict_pass is unaffected -- these fields are informational only. + assert ev.strict_pass is True + + +def test_evaluate_run_measure_correct_is_computed_when_key_present(): + ev = _evaluate_run(_CREATE_ARGS, _EXECUTE_RESULT, turn_completed=True, expected=_EXPECTED) + assert ev.measure_correct is True + assert ev.date_attribute_correct is None # still absent -- not asserted + + +def test_evaluate_run_nulls_all_informational_fields_when_disambiguated(): + # The simulated user reply names the acceptable candidate(s) straight from + # expected_output, so a disambiguated run's *_correct fields aren't testing what the + # agent inferred -- they must be nulled, not left as a misleadingly-real-looking True. + ev = _evaluate_run(_CREATE_ARGS, _EXECUTE_RESULT, turn_completed=True, expected=_EXPECTED, disambiguated=True) + assert ev.measure_correct is None + assert ev.date_attribute_correct is None + assert ev.analyzed_period_correct is None + assert ev.reference_period_correct is None + assert ev.filters_correct is None + assert ev.summary_correct is None + # Core fields are unaffected -- only the informational ones are nulled. + assert ev.kda_triggered is True + assert ev.disambiguated is True + + +def test_evaluate_run_filters_correct_none_vs_expected_empty_are_different_claims(): + # Regression guard: expected.get("Filters", []) used to conflate "no expectation + # given" with "expected no filters at all" -- a dataset item that never mentions + # Filters must not silently fail just because the LLM applied a legitimate one. + no_expectation = _evaluate_run(_CREATE_ARGS, _EXECUTE_RESULT, turn_completed=True, expected={}) + assert no_expectation.filters_correct is None + + expects_none = _evaluate_run(_CREATE_ARGS, _EXECUTE_RESULT, turn_completed=True, expected={"Filters": []}) + assert expects_none.filters_correct is False # real assertion, real mismatch + + expects_match = _evaluate_run( + _CREATE_ARGS, _EXECUTE_RESULT, turn_completed=True, expected={"Filters": _CREATE_ARGS["filters"]} + ) + assert expects_match.filters_correct is True + + +def test_evaluate_run_summary_correct_none_when_summary_key_absent(): + ev = _evaluate_run(_CREATE_ARGS, _EXECUTE_RESULT, turn_completed=True, expected={}) + assert ev.summary_correct is None + + +def test_evaluate_run_summary_correct_uses_relative_tolerance(): + expected = {"Summary": {"reference_value": 1_000_000.0, "analyzed_value": 1_000_500.0, "change": 500.0}} + execute_result = { + "success": True, + "data": {"summary": {"reference_value": 1_000_000.0, "analyzed_value": 1_000_500.0, "change": 500.0}}, + } + ev = _evaluate_run(_CREATE_ARGS, execute_result, turn_completed=True, expected=expected) + assert ev.summary_correct is True + + +def test_evaluate_run_summary_correct_checks_change_against_reference_value_not_itself(): + # change is a DIFFERENCE (10,000 here), often far smaller than reference_value + # (1,000,000) it was computed from -- checking it relative to itself would make the + # tolerance on change ~100x tighter than the same nominal 1% gives reference_value/ + # analyzed_value. actual_change=10,700 is 7% off the *change* value itself (would fail + # a change-relative-to-itself check) but only 0.07% off the real 1,000,000 scale -- + # correctly within tolerance once change is checked against reference_value. + expected = {"Summary": {"reference_value": 1_000_000.0, "analyzed_value": 1_010_000.0, "change": 10_000.0}} + execute_result = { + "success": True, + "data": { + "summary": {"reference_value": 1_000_000.0, "analyzed_value": 1_010_700.0, "change": 10_700.0}, + }, + } + ev = _evaluate_run(_CREATE_ARGS, execute_result, turn_completed=True, expected=expected) + assert ev.summary_correct is True + + +def test_evaluate_run_measure_correct_false_not_none_when_key_present_but_wrong(): + ev = _evaluate_run( + _CREATE_ARGS, _EXECUTE_RESULT, turn_completed=True, expected={"Measure": {"type": "metric", "id": "other"}} + ) + assert ev.measure_correct is False + + +def test_evaluate_run_informational_fields_are_none_not_false_when_kda_never_triggered(): + # Regression guard: measure_correct and friends used to come out False (not None) when + # kda_triggered was False, indistinguishable on a dashboard from "checked, wrong" even + # though there was nothing to check at all -- summary_correct already got this right + # (gated on executed+success), the other five didn't. + ev = _evaluate_run( + None, + None, + turn_completed=False, + expected={"Measure": {"type": "metric", "id": "revenue"}, "Filters": []}, + ) + assert ev.kda_triggered is False + assert ev.measure_correct is None + assert ev.date_attribute_correct is None + assert ev.analyzed_period_correct is None + assert ev.reference_period_correct is None + assert ev.filters_correct is None + + +# --------------------------------------------------------------------------- # +# _resolve_summary_tolerance / absolute_tolerance -- the real agent_kda_skill dataset +# (verified against a live-generated copy of all 10 items) uses "absolute_tolerance" on +# every single item, never "relative_tolerance" -- this dataset-vs-code mismatch meant the +# dataset author's intended tolerance was silently ignored on every case, always falling +# back to the 1% relative default instead. +# --------------------------------------------------------------------------- # +def test_resolve_summary_tolerance_prefers_absolute_when_present(): + assert _resolve_summary_tolerance({"absolute_tolerance": 0.01}) == ("absolute", 0.01) + + +def test_resolve_summary_tolerance_falls_back_to_relative_default(): + assert _resolve_summary_tolerance({}) == ("relative", _DEFAULT_SUMMARY_REL_TOLERANCE) + + +def test_resolve_summary_tolerance_warns_on_unrecognized_tolerance_key(caplog): + with caplog.at_level(logging.WARNING): + mode, tolerance = _resolve_summary_tolerance({"tolerance": 0.01}) + assert (mode, tolerance) == ("relative", _DEFAULT_SUMMARY_REL_TOLERANCE) + assert "unrecognized tolerance key" in caplog.text + + +def test_evaluate_run_summary_correct_uses_absolute_tolerance_from_real_dataset_shape(): + # Exact shape of a real agent_kda_skill dataset item (reference_value=3188.9, + # absolute_tolerance=0.01): before this fix, "absolute_tolerance" was never read, so + # this case silently ran under a 1% RELATIVE band (~31.89) instead of the dataset + # author's intended near-exact absolute band -- about 3000x looser than intended. + expected = { + "Summary": {"reference_value": 3188.9, "analyzed_value": 4003.31, "change": 814.41, "absolute_tolerance": 0.01} + } + execute_result = { + "success": True, + "data": {"summary": {"reference_value": 3188.9, "analyzed_value": 4003.31, "change": 814.41}}, + } + ev = _evaluate_run(_CREATE_ARGS, execute_result, turn_completed=True, expected=expected) + assert ev.summary_correct is True + + # A value that's comfortably within the old (wrongly-applied) 1% relative band but + # outside the dataset's real intended absolute band of 0.01 must now correctly fail. + execute_result_off = { + "success": True, + "data": {"summary": {"reference_value": 3188.9, "analyzed_value": 4003.31, "change": 815.00}}, + } + ev_off = _evaluate_run(_CREATE_ARGS, execute_result_off, turn_completed=True, expected=expected) + assert ev_off.summary_correct is False + + +def test_extract_kda_calls_takes_last_execute_on_retry(): + events = ( + _kda_chat_result(success=False).tool_call_events + + ChatResult.model_validate( + { + "toolCallEvents": [ + _tool_call("execute_key_driver_analysis", result={"success": True, "data": {"summary": {}}}), + ], + } + ).tool_call_events + ) + create_args, execute_result = _extract_kda_calls(events) + assert create_args == {"measure": {"type": "metric", "id": "revenue"}} + assert execute_result == {"success": True, "data": {"summary": {}}} + + +def test_extract_kda_calls_does_not_pair_a_new_create_with_an_earlier_execute(): + # create_1 -> execute_1(success) -> create_2 (never executed): create_2's args must + # not get paired with execute_1's stale result -- that would wrongly report the run + # as executed/succeeded when the actual last attempt never ran. + events = ChatResult.model_validate( + { + "toolCallEvents": [ + _tool_call("create_key_driver_analysis", arguments={"measure": {"type": "metric", "id": "a"}}), + _tool_call("execute_key_driver_analysis", result={"success": True, "data": {"summary": {}}}), + _tool_call("create_key_driver_analysis", arguments={"measure": {"type": "metric", "id": "b"}}), + ] + } + ).tool_call_events + create_args, execute_result = _extract_kda_calls(events) + assert create_args == {"measure": {"type": "metric", "id": "b"}} + assert execute_result is None + + +def test_extract_kda_calls_none_when_no_tool_calls(): + create_args, execute_result = _extract_kda_calls([]) + assert create_args is None + assert execute_result is None + + +def test_extract_kda_calls_ignores_execute_call_with_no_result(): + events = ChatResult.model_validate( + {"toolCallEvents": [_tool_call("execute_key_driver_analysis", result=None)]} + ).tool_call_events + _, execute_result = _extract_kda_calls(events) + assert execute_result is None + + +# --------------------------------------------------------------------------- # +# KdaEvaluation.strict_pass +# --------------------------------------------------------------------------- # +def _evaluation(**overrides) -> KdaEvaluation: + fields = { + "kda_triggered": True, + "executed": True, + "success": True, + "turn_completed": True, + "measure_correct": True, + "date_attribute_correct": True, + "analyzed_period_correct": True, + "reference_period_correct": True, + "filters_correct": True, + "summary_correct": True, + } + fields.update(overrides) + return KdaEvaluation(**fields) + + +def test_strict_pass_true_when_all_core_checks_pass(): + assert _evaluation().strict_pass is True + + +def test_strict_pass_false_when_any_core_check_fails(): + assert _evaluation(success=False).strict_pass is False + + +# --------------------------------------------------------------------------- # +# run_agentic_kda_skill +# --------------------------------------------------------------------------- # +def test_run_agentic_kda_skill_triggers_and_succeeds(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True) + + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + ) + + assert summary.pass_at_k is True + assert summary.best.eval.kda_triggered is True + assert summary.best.eval.executed is True + assert summary.best.eval.success is True + mock_client.close.assert_called_once() + + +def test_run_agentic_kda_skill_fails_on_sse_cutoff_despite_nonempty_text(): + # Regression guard: an SSE stream cut off mid-answer (a recurring failure mode in this + # suite) can still have emitted a partial, non-empty text_response before dying. Using + # "text_response is non-empty" as the completion signal would wrongly call this turn + # completed; only gen-ai's own response_ended event may. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True, stream_ended=False) + + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + ) + + assert summary.pass_at_k is False + assert summary.best.eval.turn_completed is False + # The KDA call itself still triggered/executed/succeeded -- only completion is in doubt. + assert summary.best.eval.kda_triggered is True + assert summary.best.eval.success is True + + +def test_run_agentic_kda_skill_survives_send_message_error(): + # A ChatError/TransientChatError raised mid-turn must not propagate out of + # run_agentic_kda_skill: an uncaught raise here would skip evaluate_agentic_kda_skill's + # Langfuse-logging loop entirely for this run, leaving nothing but a bare JUnit + # failure to diagnose from. It must instead surface as a normal (failed) run result, + # so kda_triggered/executed/success/turn_completed all still get scored as False. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.side_effect = TransientChatError("gen-ai returned 503") + + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + ) + + assert summary.pass_at_k is False + assert summary.best.eval.turn_completed is False + assert summary.best.eval.kda_triggered is False + mock_client.close.assert_called_once() + + +def test_run_agentic_kda_skill_survives_a_raw_httpx_transport_error(): + # The actual failure mode this guards against, not just ChatError: a stream cut off + # mid-turn raises httpx.RemoteProtocolError/ReadError from inside resp.iter_lines(), + # which _is_retryable_exc does not recognize as retryable and re-raises as-is -- a + # narrower `except ChatError` (an earlier version of this fix) would NOT catch this + # and would still propagate out of run_agentic_kda_skill uncaught. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.side_effect = httpx.RemoteProtocolError("peer closed connection") + + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + ) + + assert summary.pass_at_k is False + assert summary.best.eval.turn_completed is False + mock_client.close.assert_called_once() + + +def test_run_agentic_kda_skill_recovers_kda_calls_from_a_chat_errors_partial_result(): + # ChatError/TransientChatError raised after KDA's own create/execute already streamed + # through (e.g. a later, unrelated final-summary generation failing with a 500) must + # not misreport as "the agent never called KDA at all" -- the partial_result attached + # to the exception is exactly the tool_call_events already seen. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.side_effect = ChatError( + "SSE error 500: boom", status_code=500, partial_result=_kda_chat_result(success=True) + ) + + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + ) + + assert summary.best.eval.kda_triggered is True + assert summary.best.eval.executed is True + assert summary.best.eval.success is True + # The error still means the turn itself didn't complete, regardless of what KDA did. + assert summary.best.eval.turn_completed is False + + +def test_run_agentic_kda_skill_resets_turn_completed_when_a_later_iteration_crashes(): + # iteration 0 asks a clarifying question and ends cleanly (turn_completed=True for + # THAT iteration); iteration 1 then crashes. Without resetting, the stale True from + # iteration 0 would still be logged for a run that never actually finished. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.side_effect = [ + _no_kda_chat_result("Could you clarify which measure?", stream_ended=True), + httpx.RemoteProtocolError("peer closed connection"), + ] + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch( + "gooddata_eval.core.agentic.kda_skill.generate_simulated_kda_response", + return_value="Use the revenue metric.", + ), + ): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=2, + ) + + assert summary.best.eval.turn_completed is False + assert summary.best.eval.kda_triggered is False + + +def test_run_agentic_kda_skill_turn_not_completed_when_stream_ends_with_empty_text(): + # stream_ended alone is not enough: a turn that ends cleanly but delivers nothing to + # the user hasn't "delivered a final answer" either (see KdaEvaluation docstring). + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True, text=" ", stream_ended=True) + + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + ) + + assert summary.best.eval.turn_completed is False + # The KDA call itself still triggered/executed/succeeded -- only completion is in doubt. + assert summary.best.eval.kda_triggered is True + assert summary.best.eval.success is True + + +def test_run_agentic_kda_skill_marks_disambiguated_after_a_simulated_reply(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.side_effect = [ + _no_kda_chat_result("Could you clarify which measure?"), + _kda_chat_result(success=True), + ] + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch( + "gooddata_eval.core.agentic.kda_skill.generate_simulated_kda_response", + return_value="Use the revenue metric.", + ), + ): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=2, + ) + + assert summary.best.eval.disambiguated is True + assert summary.best.eval.kda_triggered is True + + +def test_run_agentic_kda_skill_not_disambiguated_when_kda_triggers_immediately(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True) + + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + ) + + assert summary.best.eval.disambiguated is False + + +def test_run_agentic_kda_skill_no_tool_call(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _no_kda_chat_result() + + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + ) + + assert summary.pass_at_k is False + assert summary.best.eval.kda_triggered is False + + +def test_run_agentic_kda_skill_resolves_after_clarification_turn(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.side_effect = [ + _no_kda_chat_result("Could you clarify which revenue measure you mean?"), + _kda_chat_result(success=True), + ] + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch( + "gooddata_eval.core.agentic.kda_skill.generate_simulated_kda_response", + return_value="The revenue metric is fine.", + ) as mock_simulate, + ): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=2, + ) + + mock_simulate.assert_called_once() + assert summary.pass_at_k is True + assert mock_client.send_message.call_count == 2 + + +def test_run_agentic_kda_skill_gives_up_after_max_iterations_of_clarification(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _no_kda_chat_result("Could you clarify which measure?") + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch( + "gooddata_eval.core.agentic.kda_skill.generate_simulated_kda_response", + return_value="Please use revenue.", + ), + ): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=2, + ) + + assert summary.pass_at_k is False + assert mock_client.send_message.call_count == 2 + + +def test_run_agentic_kda_skill_survives_simulated_reply_failure(): + # The simulated-user helper is a safety net, not the assertion under test -- if it + # raises, only the current run ends early; earlier completed runs are preserved. + mock_client = MagicMock() + mock_client.create_conversation.side_effect = ["conv-1", "conv-2"] + mock_client.send_message.side_effect = [ + _kda_chat_result(success=True), # run 0: triggers KDA immediately + _no_kda_chat_result("Could you clarify which measure?"), # run 1: asks, then helper blows up + ] + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch( + "gooddata_eval.core.agentic.kda_skill.generate_simulated_kda_response", + side_effect=RuntimeError("openai down"), + ), + ): + summary = run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=2, + max_iterations=2, + ) + + assert len(summary.run_results) == 2 + assert summary.run_results[0].eval.kda_triggered is True + assert summary.run_results[1].eval.kda_triggered is False + assert summary.pass_at_k is True # run 0 still counts + + +def test_run_agentic_kda_skill_uses_initial_conversation_for_run_0(): + mock_client = MagicMock() + mock_client.send_message.return_value = _kda_chat_result(success=True) + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + initial_conversation_id="existing-conv", + ) + mock_client.create_conversation.assert_not_called() + mock_client.delete_conversation.assert_not_called() + + +def test_run_agentic_kda_skill_creates_fresh_conversations_for_remaining_runs(): + mock_client = MagicMock() + mock_client.create_conversation.side_effect = ["fresh-1", "fresh-2"] + mock_client.send_message.return_value = _kda_chat_result(success=True) + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=3, + max_iterations=1, + initial_conversation_id="existing-conv", + ) + assert mock_client.create_conversation.call_count == 2 + assert mock_client.delete_conversation.call_count == 2 + + +@pytest.mark.parametrize("bad_k", [0, -1, -5]) +def test_run_agentic_kda_skill_rejects_non_positive_k(bad_k): + with pytest.raises(ValueError, match="k must be >= 1"): + run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=bad_k, + ) + + +# --------------------------------------------------------------------------- # +# evaluate_agentic_kda_skill +# --------------------------------------------------------------------------- # +def test_evaluate_agentic_kda_skill_raises_on_failure(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _no_kda_chat_result() + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.try_make_langfuse_client", return_value=None), + pytest.raises(KdaSkillAssertionError), + ): + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=None, + ) + + +def test_evaluate_agentic_kda_skill_does_not_raise_on_success(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True) + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.try_make_langfuse_client", return_value=None), + ): + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=None, + ) + + +def test_evaluate_agentic_kda_skill_never_treats_fallback_trace_latency_as_kda_latency(): + # Regression test: when KDA never triggered, whatever trace find_traces_per_conversation's + # default (max-latency) selector picks is NOT a real KDA turn -- its latency/cost must not + # be logged as the KDA run's own value_score inputs. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _no_kda_chat_result() + + fallback_trace = MagicMock(id="fallback-trace", latency=999.0, total_cost=5.0) + mock_langfuse = MagicMock() + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.build_run_context", return_value=("run-base", {})), + patch( + "gooddata_eval.core.agentic._langfuse.find_traces_per_conversation", + return_value={"conv-1": fallback_trace}, + ), + patch("gooddata_eval.core.agentic._langfuse.observe") as mock_observe, + patch("gooddata_eval.core.agentic._langfuse.score_safe"), + patch("gooddata_eval.core.agentic._langfuse.log_quality_and_value_scores") as mock_log_scores, + pytest.raises(KdaSkillAssertionError), + ): + mock_observe.return_value.__enter__.return_value = "fallback-trace" + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=mock_langfuse, + dataset_item_id="item-1", + ) + + mock_log_scores.assert_called_once() + assert mock_log_scores.call_args.kwargs["latency_sec"] is None + assert mock_log_scores.call_args.kwargs["cost_usd"] is None + + +def test_evaluate_agentic_kda_skill_reports_trace_latency_when_kda_triggered(): + # Latency comes from the harness's own wall-clock measurement (ChatResult.turn_wall_clock_sec, + # set by ChatClient around its send_message() call), not from the trace find_traces_per_ + # conversation happens to return -- that trace isn't necessarily the KDA turn at all (see + # kda_skill.py's comment on `pt`). Only total_cost still comes from the trace. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True, turn_wall_clock_sec=76.0) + + found_trace = MagicMock(id="trace-1", latency=999.0, total_cost=0.02) + mock_langfuse = MagicMock() + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.build_run_context", return_value=("run-base", {})), + patch( + "gooddata_eval.core.agentic._langfuse.find_traces_per_conversation", + return_value={"conv-1": found_trace}, + ), + patch("gooddata_eval.core.agentic._langfuse.observe") as mock_observe, + patch("gooddata_eval.core.agentic._langfuse.score_safe") as mock_score_safe, + patch("gooddata_eval.core.agentic._langfuse.log_quality_and_value_scores") as mock_log_scores, + ): + mock_observe.return_value.__enter__.return_value = "trace-1" + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=mock_langfuse, + dataset_item_id="item-1", + ) + + mock_log_scores.assert_called_once() + assert mock_log_scores.call_args.kwargs["latency_sec"] == 76.0 + assert mock_log_scores.call_args.kwargs["cost_usd"] == 0.02 + wall_clock_calls = [c for c in mock_score_safe.call_args_list if c.kwargs.get("name") == "kda_turn_wall_clock_sec"] + assert len(wall_clock_calls) == 1 + assert wall_clock_calls[0].kwargs["value"] == 76.0 + + +def test_evaluate_agentic_kda_skill_logs_kda_pass_at_k_and_kda_pass_power_k(): + # pass_power_k (did EVERY one of k runs pass, not just one) was computed and never + # logged anywhere -- discarding exactly the cross-run consistency signal raising k is + # meant to produce. k=2 specifically: an unprefixed "pass_at_2"/"pass_power_2" is + # exactly visualization.py's own score name, which gdc-nas's combo_report.py uses to + # classify a trace as a visualization record -- k=2 is the value that would trigger + # that collision, so the regression test must use it, not an arbitrary k. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True) + + found_trace = MagicMock(id="trace-1", total_cost=0.01) + mock_langfuse = MagicMock() + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.build_run_context", return_value=("run-base", {})), + patch( + "gooddata_eval.core.agentic._langfuse.find_traces_per_conversation", + return_value={"conv-1": found_trace}, + ), + patch("gooddata_eval.core.agentic._langfuse.observe") as mock_observe, + patch("gooddata_eval.core.agentic._langfuse.score_safe") as mock_score_safe, + patch("gooddata_eval.core.agentic._langfuse.log_quality_and_value_scores"), + ): + mock_observe.return_value.__enter__.return_value = "trace-1" + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=2, + max_iterations=1, + langfuse=mock_langfuse, + dataset_item_id="item-1", + ) + + logged = {c.kwargs["name"]: c.kwargs["value"] for c in mock_score_safe.call_args_list} + assert logged["kda_pass_at_2"] == 1.0 + assert logged["kda_pass_power_2"] == 1.0 + # The exact collision this test guards against: gdc-nas's combo_report.py.verdict() + # checks "pass_at_2" in scores as its FIRST branch to classify a trace as + # visualization -- these names must never appear unprefixed, at any k. + assert "pass_at_2" not in logged + assert "pass_power_2" not in logged diff --git a/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py b/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py new file mode 100644 index 000000000..af64a78d7 --- /dev/null +++ b/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py @@ -0,0 +1,26 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +# SPDX-License-Identifier: LicenseRef-GoodData-Enterprise +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +from gooddata_eval.core.agentic._langfuse import find_traces_per_conversation + + +def test_find_traces_per_conversation_is_none_for_a_conversation_with_no_trace(): + # find_traces_per_conversation's return dict is seeded with dict.fromkeys(conversation_ids) + # (every value starts None) and only overwritten for ids where a trace was actually found -- + # callers (kda_skill.py and every other agentic skill) must treat a missing conversation as + # None, not assume every key maps to a real trace object. + found_trace = MagicMock(latency=12.0) + + def _fetch(langfuse, cid, window_start, window_end, pad): + return [found_trace] if cid == "conv-found" else [] + + with ( + patch("gooddata_eval.core.agentic._langfuse._fetch_traces_for_session", side_effect=_fetch), + patch("gooddata_eval.core.agentic._langfuse.time.sleep"), + ): + result = find_traces_per_conversation(MagicMock(), ["conv-found", "conv-missing"], datetime.now(timezone.utc)) + + assert result["conv-found"] is found_trace + assert result["conv-missing"] is None diff --git a/packages/gooddata-eval/tests/test_sse_client.py b/packages/gooddata-eval/tests/test_sse_client.py index 490dfd57d..fb3587e6f 100644 --- a/packages/gooddata-eval/tests/test_sse_client.py +++ b/packages/gooddata-eval/tests/test_sse_client.py @@ -23,12 +23,106 @@ def test_parse_sse_lines_raises_on_error_event(): parse_sse_lines(lines) +def test_parse_sse_lines_error_carries_partial_result_with_tool_calls_already_seen(): + # A statusCode error ends the stream before _build_chat_result ever runs -- without + # partial_result, a tool call that already succeeded (e.g. KDA's own create/execute) + # before a LATER, unrelated error killed the turn would be silently discarded, making + # the run look like the agent never called the tool at all. + lines = [ + json.dumps( + { + "item": { + "role": "assistant", + "content": {"type": "toolCall", "callId": "c1", "name": "create_key_driver_analysis"}, + } + } + ), + "", + json.dumps( + { + "item": { + "role": "tool", + "content": { + "type": "toolResult", + "callId": "c1", + "result": json.dumps({"success": True}), + }, + } + } + ), + "", + json.dumps({"statusCode": 500, "detail": "boom"}), + ] + lines = [f"data: {line}" if line else line for line in lines] + with pytest.raises(ChatError) as ei: + parse_sse_lines(lines) + partial = ei.value.partial_result + assert partial is not None + assert len(partial.tool_call_events) == 1 + assert partial.tool_call_events[0].function_name == "create_key_driver_analysis" + assert partial.tool_call_events[0].result == '{"success": true}' + + +def test_parse_sse_lines_raw_transport_error_also_carries_partial_result(): + # A connection drop mid-stream (httpx.RemoteProtocolError/ReadError) has no statusCode + # payload -- it's a raw exception from iterating `lines` itself, not one this module + # raises. Must still be rescued the same way a statusCode-shaped error is. + def _lines(): + yield ( + 'data: {"item": {"role": "assistant", "content": ' + + json.dumps({"type": "toolCall", "callId": "c1", "name": "create_key_driver_analysis"}) + + "}}" + ) + yield "" + raise RuntimeError("connection dropped") + + with pytest.raises(ChatError) as ei: + parse_sse_lines(_lines()) + assert not isinstance(ei.value, TransientChatError) # not retried -- same as before this fix + partial = ei.value.partial_result + assert partial is not None + assert len(partial.tool_call_events) == 1 + assert partial.tool_call_events[0].function_name == "create_key_driver_analysis" + + def test_parse_sse_lines_ignores_non_data_lines(): result = parse_sse_lines(["event: ping", "", ": comment"]) assert result.text_response is None assert result.created_visualizations is None +def test_parse_sse_lines_stream_ended_false_when_response_ended_never_arrives(): + # A turn cut off mid-stream (connection dropped, process killed) never gets to emit + # gen-ai's own "response_ended" event -- text_response can still be non-empty from + # whatever text arrived before the cutoff. + lines = [ + "event: item", + 'data: {"item": {"role": "assistant", "content": {"type": "text", "text": "partial answ"}}}', + "", + ] + result = parse_sse_lines(lines) + assert result.text_response == "partial answ" + assert result.stream_ended is False + + +def test_parse_sse_lines_stream_ended_true_when_response_ended_event_arrives(): + lines = [ + "event: item", + 'data: {"item": {"role": "assistant", "content": {"type": "text", "text": "full answer"}}}', + "", + "event: response_ended", + "data: {}", + "", + ] + result = parse_sse_lines(lines) + assert result.text_response == "full answer" + assert result.stream_ended is True + + +def test_parse_sse_lines_stream_ended_defaults_false_with_no_events_at_all(): + assert parse_sse_lines([]).stream_ended is False + + def test_parse_sse_lines_falls_back_to_adhoc_viz_when_multipart_viz_is_null(): """Visualization from create_adhoc_visualization args used when multipart viz is null.""" viz_def = { @@ -205,6 +299,40 @@ def handler(request): assert sleeps == [] +def test_send_message_sets_turn_wall_clock_sec_on_success(monkeypatch): + monkeypatch.setattr(sse_mod.time, "monotonic", iter([100.0, 102.5]).__next__) + client = _client_with_handler(lambda request: httpx.Response(200, content=_OK_SSE)) + result = client.send_message("conv", "q") + assert result.turn_wall_clock_sec == pytest.approx(2.5) + + +def test_send_message_wall_clock_excludes_retry_backoff(monkeypatch): + # t0 must be per-attempt, set inside _do() after the connection is already open -- + # not around the whole send_message() call -- or a transient retry's backoff sleep + # (harness/network overhead, not gen-ai's time) would inflate the reported latency. + monkeypatch.setattr(sse_mod.time, "sleep", lambda s: None) + monkeypatch.setattr(sse_mod.time, "monotonic", iter([1000.0, 1000.5, 2000.0, 2001.2]).__next__) + calls = {"n": 0} + + def handler(request): + calls["n"] += 1 + return httpx.Response(200, content=_TRANSIENT_SSE if calls["n"] < 2 else _OK_SSE) + + client = _client_with_handler(handler) + result = client.send_message("conv", "q") + assert calls["n"] == 2 + assert result.turn_wall_clock_sec == pytest.approx(1.2) # attempt 2 alone, not spanning attempt 1 + backoff + + +def test_send_message_stamps_turn_wall_clock_sec_on_partial_result_too(monkeypatch): + monkeypatch.setattr(sse_mod.time, "monotonic", iter([50.0, 51.0]).__next__) + client = _client_with_handler(lambda request: httpx.Response(200, content=_NONRETRY_SSE)) + with pytest.raises(ChatError) as ei: + client.send_message("conv", "q") + assert ei.value.partial_result is not None + assert ei.value.partial_result.turn_wall_clock_sec == pytest.approx(1.0) + + def test_create_conversation_retries_then_succeeds(monkeypatch): sleeps = [] monkeypatch.setattr(sse_mod.time, "sleep", lambda s: sleeps.append(s))