From 4123c681f901b78a21fe679c8c41a2bc29f5862a Mon Sep 17 00:00:00 2001 From: Frank Huynh Date: Fri, 7 Aug 2026 20:28:05 +0700 Subject: [PATCH] feat(gooddata-eval): add KDA-skill agentic evaluator Adds kda_skill.py to gooddata-eval, evaluating the chatbot's create_key_driver_analysis/execute_key_driver_analysis tool calls against the agent_kda_skill Langfuse dataset. Scope is completion (kda_triggered -> executed -> success -> turn_completed), not per-field correctness -- per-field checks are computed and logged as kda_-prefixed informational scores for a follow-up ticket, and are None (not False) both when expected_output has no key for a field AND when that field's own precondition (kda_triggered, or executed+success for Summary) isn't met. Adds ChatResult.stream_ended (from the SSE response_ended event). turn_completed requires both stream_ended AND a non-empty text_response -- a stream that ends cleanly but delivers nothing to the user isn't a completed turn either, and a turn cut off mid-answer can still emit partial, non-empty text before dying. Fixes from review: - kda_ prefix pass_at_k/pass_power_k Langfuse scores -- unprefixed, "pass_at_2" at k=2 collides with visualization.py's own score name that gdc-nas's combo_report.py.verdict() checks first, silently misfiling every KDA record as visualization once KDA_RUN_K=2 is ever set. - Relative (not absolute) tolerance for Summary's revenue-scale values by default; change is checked against reference_value's scale, not against itself. Also recognizes an explicit absolute_tolerance key -- every real agent_kda_skill dataset item uses it, which this module never read before, silently running ~3000x looser than the dataset author intended. Warns on any other *tolerance* key so a future typo surfaces instead of repeating. - Filters compared as canonicalized sets, not order-sensitive lists. - except Exception (not except ChatError) around send_message -- a stream cut off mid-turn raises a raw httpx transport error, which a narrower catch would let escape uncaught, skipping Langfuse scoring entirely for that run. - ChatError/TransientChatError carry partial_result so tool calls that already succeeded before a later, unrelated stream error aren't discarded and misreported as "the agent never called KDA at all". - turn_completed resets to False on an exception, so a crash on a later disambiguation iteration can't leave a stale True from an earlier iteration. - kda_disambiguated logged and immediately nulls the six *_correct informational fields: the simulated user reply names the acceptable candidate(s) drawn from expected_output itself, so those fields aren't an independent signal once a run went through disambiguation. - KDA-specific _is_asking_clarification instead of a heuristic shared with metric_skill.py/conversation.py, which had silently drifted apart; strips a leading "to clarify, " discourse marker before its substring checks, since that phrase means "in other words" in a final answer, not a request for one. - log_quality_and_value_scores renormalizes over known components instead of scoring unresolved latency/cost as the worst possible outcome. This changes behavior for every agentic skill, not just KDA -- any dashboard trending value_score over time will show a discontinuity at this PR's merge point. - Regression tests for find_traces_per_conversation's None-safety and _filters_match's isinstance guard. trace.latency is a direct start-of-call-to-stream-exhausted timestamp (see Trace selection comment in kda_skill.py), not an approximation from scanning observation timestamps -- the only real gap versus a harness-side wall clock is network round-trip time, accepted as negligible against the 60s threshold. JIRA: QA-28800 --- .../gooddata_eval/core/agentic/__init__.py | 14 + .../gooddata_eval/core/agentic/_langfuse.py | 30 +- .../gooddata_eval/core/agentic/kda_skill.py | 700 +++++++++++ .../src/gooddata_eval/core/chat/sse_client.py | 49 +- .../src/gooddata_eval/core/models.py | 6 + .../tests/test_agentic_kda_skill.py | 1025 +++++++++++++++++ .../tests/test_agentic_langfuse_trace.py | 79 ++ .../gooddata-eval/tests/test_sse_client.py | 72 ++ 8 files changed, 1964 insertions(+), 11 deletions(-) create mode 100644 packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py create mode 100644 packages/gooddata-eval/tests/test_agentic_kda_skill.py create mode 100644 packages/gooddata-eval/tests/test_agentic_langfuse_trace.py 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/_langfuse.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py index 67630ce2f..5a69918cf 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py @@ -31,7 +31,12 @@ def __init__(self, raw: dict) -> None: self.id: str = raw.get("id", "") self.metadata: dict = raw.get("metadata") or {} self.session_id: str | None = raw.get("sessionId") or raw.get("session_id") - self.latency: float = float(raw.get("latency") or 0.0) + # None (missing/null) is preserved, not coerced to 0.0 -- a trace that hasn't + # finished ingesting has UNKNOWN latency, not zero latency, and callers (e.g. + # log_quality_and_value_scores below, and every skill's own `pt.latency if pt + # else None` gating) rely on that distinction to not treat "unknown" as the best + # possible outcome. + self.latency: float | None = float(raw["latency"]) if raw.get("latency") is not None else None self.total_cost: float = float(raw.get("totalCost") or raw.get("total_cost") or 0.0) @@ -358,11 +363,24 @@ def log_quality_and_value_scores( data_type="NUMERIC", comment=f"{passed}/{total} strict checks passed", ) - speed = 0.0 if latency_sec is None else max(0.0, 1.0 - latency_sec / _MAX_LATENCY_SEC) - cost_factor = 0.0 if cost_usd is None else max(0.0, 1.0 - cost_usd / _MAX_COST_USD) - value = _QUALITY_WEIGHT * quality + _SPEED_WEIGHT * speed + _COST_WEIGHT * cost_factor + # An unresolved latency/cost (trace not yet settled, price not available) is UNKNOWN, + # not the best (1.0) or worst (0.0) possible outcome -- substituting either would + # silently pull value_score toward one extreme. Drop that weighted term instead and + # renormalize over whichever components do have a real value, so value_score always + # reflects only the signals actually measured for this run. + components = [(_QUALITY_WEIGHT, quality)] + speed = None if latency_sec is None else max(0.0, 1.0 - latency_sec / _MAX_LATENCY_SEC) + if speed is not None: + components.append((_SPEED_WEIGHT, speed)) + cost_factor = None if cost_usd is None else max(0.0, 1.0 - cost_usd / _MAX_COST_USD) + if cost_factor is not None: + components.append((_COST_WEIGHT, cost_factor)) + weight_total = sum(w for w, _ in components) + value = sum(w * v for w, v in components) / weight_total latency_str = "unknown" if latency_sec is None else f"{latency_sec:.2f}s" cost_str = "unknown" if cost_usd is None else f"${cost_usd:.4f}" + speed_str = "n/a" if speed is None else f"{speed:.2f}" + cost_factor_str = "n/a" if cost_factor is None else f"{cost_factor:.2f}" score_safe( langfuse, trace_id, @@ -371,8 +389,8 @@ def log_quality_and_value_scores( data_type="NUMERIC", comment=( f"{_QUALITY_WEIGHT}*quality({quality:.2f}) + " - f"{_SPEED_WEIGHT}*speed({speed:.2f}) + " - f"{_COST_WEIGHT}*cost({cost_factor:.2f}); " + f"{_SPEED_WEIGHT}*speed({speed_str}) + " + f"{_COST_WEIGHT}*cost({cost_factor_str}), renormalized /{weight_total:.1f}; " f"latency={latency_str}; cost={cost_str}" ), ) 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..89845d4fb --- /dev/null +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py @@ -0,0 +1,700 @@ +# (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 (see the +# handoff's known-collision cases) or a metric-vs-fact form choice -- not a general +# multi-turn budget. +_DEFAULT_MAX_ITERATIONS = 2 +# 1% relative tolerance on Summary's revenue-scale values (reference_value/analyzed_value/ +# change) -- a fixed absolute band would be either meaninglessly loose or an effectively +# exact-match bar that fails near-constantly, depending on the metric's scale. +_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_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 + + +# Trace selection: a conversation can have more than one Langfuse trace sharing the same +# session_id (a conversation-title-generation trace, a disambiguation turn, the actual KDA +# turn), and picking the real one requires checking observations, which are ingested +# asynchronously just like latency -- so this module doesn't try to pick one at all. It +# links whichever trace the default selector finds, exactly like every other skill; +# combo_report.py re-resolves the real KDA trace itself, well after the run, when +# ingestion has settled. +# +# trace.latency is not approximated from scanning observation timestamps. gen-ai's +# @observe(name="reply_to_question") (observable_chatbot_decorator.py) wraps the whole +# turn's async generator with end_on_exit=False specifically so Langfuse only calls .end() +# once the generator is exhausted (langfuse's own observe.py, +# _ContextPreservedAsyncGeneratorWrapper) -- i.e. trace.latency is a direct +# start-of-call-to-generator-exhaustion timestamp, equivalent to a monotonic timer wrapped +# around the same code path. The remaining gap versus a harness-side wall clock is network +# round-trip time plus sub-millisecond post-`.end()` bookkeeping in gen-ai -- both additive +# to gen-ai's own number, not time missing from it, and small next to the 60s threshold +# this feature cares about. + + +@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 + + +@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. + """ + 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_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 -- safety net, not the assertion; end only this run + # Broad on purpose, not `except ChatError`: a stream cut off mid-turn + # (a recurring failure mode in this suite) raises a raw httpx transport + # error (RemoteProtocolError/ReadError from inside resp.iter_lines()), + # not ChatError -- narrower would miss exactly the case this exists to + # catch. An uncaught error here would propagate out of + # run_agentic_kda_skill entirely, before evaluate_agentic_kda_skill's + # Langfuse-logging loop ever runs for this conversation -- + # combo_report.py falls back to JUnit to bucket the case as "error", but + # it's indistinguishable there from every other kind of crash. Treating + # it as a normal, unsuccessful run instead lets it flow through the usual + # scoring path, so kda_triggered/executed/success/kda_turn_completed + # still get logged and this failure mode is diagnosable in Langfuse too. + _log.warning("KDA send_message failed for conversation %s: %s", conv_id, exc) + # ChatError/TransientChatError carry whatever the SSE accumulator captured + # before the error fired (see ChatError.partial_result) -- an error after + # KDA's own create/execute calls 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" just because the error + # arrived before parse_sse_lines could return a normal ChatResult. + 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 + # 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 + 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_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_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, + ) + + 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: which trace is the real KDA turn is re-checked later by + # combo_report.py instead (see "Trace selection" above), so this is the same + # default (max-latency) selection every other skill uses. + 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, + } + # pt can be any trace of the conversation, not necessarily the KDA turn + # (see "Trace selection" above) -- only trust it as a rough approximation + # (this feeds value_score, not the daily report) when KDA triggered at all. + kda_latency_sec = pt.latency if pt and ev.kda_triggered else None + _log.info("[kda-report] %s: strict_pass=%s latency_sec=%s", run_name, ev.strict_pass, kda_latency_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" + ) + log_quality_and_value_scores( + langfuse, + tid, + strict_checks=strict_checks, + latency_sec=kda_latency_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..f93379361 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,42 @@ _log = logging.getLogger(__name__) SSE_DATA_PREFIX = "data: " +SSE_EVENT_PREFIX = "event: " +# gen-ai's conversations_controller.py yields this as the LAST event of a turn's stream, +# in a `finally` block, whether the turn ended normally or was already handled as an error. +# Its absence (the connection dropped mid-stream, or the process was killed) is what +# distinguishes a genuinely completed turn from a truncated one -- text_response alone +# cannot: a turn cut off mid-answer can still have emitted a partial, non-empty response. +_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 +133,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,15 +212,26 @@ 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() + current_event = "message" # SSE default in the absence of an explicit "event: " line 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): + 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: @@ -203,6 +239,7 @@ def parse_sse_lines(lines: Iterable[str]) -> ChatResult: 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) @@ -213,8 +250,10 @@ def parse_sse_lines(lines: Iterable[str]) -> ChatResult: 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) + 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") diff --git a/packages/gooddata-eval/src/gooddata_eval/core/models.py b/packages/gooddata-eval/src/gooddata_eval/core/models.py index 336c313b9..15446c520 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/models.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/models.py @@ -100,6 +100,12 @@ 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") + # True only if the SSE stream carried gen-ai's own "response_ended" event -- set by + # sse_client.py after parsing, not from the payload dict (there is no such server + # field; it's derived from which named SSE events were seen). Distinguishes a turn that + # genuinely finished from one cut off mid-stream, which can still have emitted a + # partial, non-empty text_response. + stream_ended: bool = False 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..f11bdf40c --- /dev/null +++ b/packages/gooddata-eval/tests/test_agentic_kda_skill.py @@ -0,0 +1,1025 @@ +# (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_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 +) -> 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, + } + ) + + +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_clarification_true(text): + assert _is_asking_clarification(text) is True + + +def test_is_asking_clarification_false_on_plain_statement(): + assert _is_asking_clarification("Here is the key driver analysis result.") is False + + +def test_is_asking_clarification_false_on_empty(): + assert _is_asking_clarification("") is False + + +def test_is_asking_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_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_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_clarification(text) is False + + +def test_is_asking_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_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 + + +# --------------------------------------------------------------------------- # +# 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(): + # When KDA did trigger, whatever trace find_traces_per_conversation returns is + # reported as-is -- same as every other skill. Picking the *right* trace among a + # conversation's several is combo_report.py's job now, not this module's (see + # kda_skill.py's "Trace selection" comment) -- there is no "matched" concept here. + 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", latency=76.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"), + 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 + + +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", latency=10.0, 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..5089867ca --- /dev/null +++ b/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py @@ -0,0 +1,79 @@ +# (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 _TraceObj, find_traces_per_conversation, log_quality_and_value_scores + + +def test_trace_obj_latency_none_when_missing(): + assert _TraceObj({"id": "t1"}).latency is None + + +def test_trace_obj_latency_none_when_explicitly_null(): + assert _TraceObj({"id": "t1", "latency": None}).latency is None + + +def test_trace_obj_latency_preserves_real_zero(): + # A real 0.0 (start == end) must stay 0.0, not be confused with "unknown". + assert _TraceObj({"id": "t1", "latency": 0.0}).latency == 0.0 + + +def test_trace_obj_latency_preserves_real_value(): + assert _TraceObj({"id": "t1", "latency": 45.3}).latency == 45.3 + + +def _value_score(langfuse: MagicMock) -> float: + for call in langfuse.create_score.call_args_list: + if call.kwargs.get("name") == "value_score": + return call.kwargs["value"] + raise AssertionError("value_score was never logged") + + +def test_value_score_uses_all_three_components_when_all_known(): + langfuse = MagicMock() + log_quality_and_value_scores(langfuse, "t1", strict_checks={"a": True, "b": True}, latency_sec=10.0, cost_usd=0.0) + # speed = 1 - 10/60 ≈ 0.833, cost_factor = 1.0 (cost_usd=0.0) -- same value as before this + # fix (all three known -> weights already summed to 1.0, nothing to renormalize). + assert abs(_value_score(langfuse) - (0.6 * 1.0 + 0.2 * (1 - 10 / 60) + 0.2 * 1.0)) < 0.001 + + +def test_value_score_renormalizes_instead_of_treating_unknown_latency_as_worst(): + langfuse = MagicMock() + log_quality_and_value_scores(langfuse, "t1", strict_checks={"a": True}, latency_sec=None, cost_usd=0.0) + # Unresolved latency drops the 0.2-weighted speed term entirely instead of scoring it 0.0 + # (which would silently drag value_score down for a run that may well have been fast). + # Renormalized over quality(0.6) + cost(0.2) = 0.8. + assert abs(_value_score(langfuse) - (0.6 * 1.0 + 0.2 * 1.0) / 0.8) < 0.001 + + +def test_value_score_renormalizes_instead_of_treating_unknown_cost_as_worst(): + langfuse = MagicMock() + log_quality_and_value_scores(langfuse, "t1", strict_checks={"a": True}, latency_sec=0.0, cost_usd=None) + assert abs(_value_score(langfuse) - (0.6 * 1.0 + 0.2 * 1.0) / 0.8) < 0.001 + + +def test_value_score_is_quality_only_when_latency_and_cost_both_unknown(): + langfuse = MagicMock() + log_quality_and_value_scores(langfuse, "t1", strict_checks={"a": True, "b": False}, latency_sec=None, cost_usd=None) + assert abs(_value_score(langfuse) - 0.5) < 0.001 + + +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..888582875 100644 --- a/packages/gooddata-eval/tests/test_sse_client.py +++ b/packages/gooddata-eval/tests/test_sse_client.py @@ -23,12 +23,84 @@ 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_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 = {