From 8f83a9ee4f38253ed857b9b2aa993f1b490a5764 Mon Sep 17 00:00:00 2001 From: Frank Huynh Date: Tue, 11 Aug 2026 19:03:59 +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 strictly completion, not field correctness: strict_pass = triggered AND executed AND success AND turn_completed Per-field checks (Measure/Date Attribute/Periods/Filters/Summary matching expected values) are deferred entirely to a follow-up ticket (QA-28699) rather than half-computed here as scores nothing reads yet. Two things can extend a run past a single turn, each bounded to max_iterations: the agent asks a clarifying question (a simulated user reply, gpt-4o-mini, nudges it forward), or it calls create but not execute in the same turn (create/execute are tracked independently across turns, so a plain continuation nudge gives it another turn instead of scoring it as if execute never happened). Latency is measured directly by the harness (ChatClient times each send_message() call around the SSE stream), not re-derived from a Langfuse trace after the fact. Only the turn that actually completes KDA counts toward it -- not an earlier disambiguation/continuation turn, and not the simulated-reply's own OpenAI call. Logged as the kda_turn_wall_clock_sec Langfuse score; combo_report.py (gdc-nas) reads it directly. Fixes from review: - kda_ prefix on pass_at_k/pass_power_k Langfuse scores -- unprefixed, "pass_at_2" at k=2 collides with visualization.py's own score name, which gdc-nas's combo_report.py.verdict() checks first when classifying a trace. - send_message errors no longer propagate out of run_agentic_kda_skill uncaught -- they now surface as a normal failed run, so it still gets scored to Langfuse instead of only showing up as a bare JUnit failure. - ChatError/TransientChatError now carry partial_result, so tool calls that already succeeded before a later, unrelated error (e.g. a failed final summary) aren't discarded and misreported as "KDA never triggered". - turn_completed resets to False in the exception branch, so a crash on a later iteration can't leave a stale True from an earlier one. - run_agentic_kda_skill rejects k < 1 -- a bad env-driven KDA_RUN_K value (0, a typo, negative) previously ran silently once instead of surfacing the bad config. - stream_ended is now set the moment the response_ended event line is parsed, not its data line -- an event with no data payload previously left the flag unset. - t0 is set before opening the SSE stream, not after -- it was missing the connection/server-setup time a caller actually waits through. - Disambiguated-turn latency no longer double-counts every turn's time plus the simulated-reply's own OpenAI call. - The 6 per-field informational correctness scores and their support code are removed -- team confirmed this PR's scope is trigger+complete only. - KdaEvaluation.kda_triggered renamed to triggered, matching the other three core fields (none of which carry the kda_ prefix). The Langfuse score name kda_triggered is unchanged. - generate_simulated_kda_response's OpenAI call now has a 30s timeout. - kda_skill wired into the CLI's agentic dispatcher (agentic_runner.py) and AGENTIC_TEST_KINDS, matching every other agentic skill, so it can be run/debugged standalone via gd-eval run instead of only through gdc-nas's tavern-e2e harness. - dataset_name default renamed from agent_kda_skill to kda_skill, matching the no-prefix pattern every other skill uses; no functional change since gdc-nas always passes it explicitly. - A few comments/docstrings that had drifted from the multi-turn behavior corrected: KdaRunResult's docstring ("one message" -> up to max_iterations), the turn_wall_clock_sec field comment, and _extract_kda_calls' docstring (pairing is only guaranteed within one turn; merging across turns is the caller's job). - expected_output.get("Measure") now guards for non-dict shapes -- DatasetItem.expected_output on the gdc-nas side allows str/list, not just dict, and a list-shaped item previously raised AttributeError, silently swallowed by the broad except and disabling disambiguation with only a WARNING. - _DEFAULT_MAX_ITERATIONS raised from 2 to 3 -- 2 was lower than every other agentic skill (visualization=4, alert_skill=6, metric_skill=7) and left no room for a case that needs both disambiguation and a create/execute turn split. - turn_wall_clock_sec's field comment corrected again -- it accumulates from the first create call through every turn attempted after it, whether or not execute ever completes, not only "through the turn that completed execute". - kda_pass_at_k/kda_pass_power_k no longer logged to Langfuse -- matches metric_skill/alert_skill/guardrail/search_tool/general_question, which all compute the pair but never log it at their default k=1; nothing reads a kda_pass_at_1 score today, and the score name shifts if k ever changes, silently splitting any Langfuse view built on the old name. - sse_client.py's t0 comment corrected again -- being per-attempt excludes not just the sleep backoff between retries, but the entire duration of any earlier failed attempt too. - Two missing test cases added: create succeeds but execute never arrives even after the continuation nudge (previously only tested running out of iterations via repeated clarification, not via nudge); and a run combining both extension paths (disambiguation, then a create-without-execute continuation) exhausting its budget safely. - KdaRunResult.eval renamed to evaluation -- it shadowed the eval builtin and is part of the published surface (exported in core/agentic/__init__.py's __all__); fixed now, before gdc-nas starts consuming this module. - turn_wall_clock_sec reverted to counting only the turn that actually completed KDA (both create and execute resolved), matching this PR's original, already-reviewed latency definition. An earlier change in this PR summed a create-only turn's time into it on the theory that a continuation turn (create succeeds, execute lands a turn later) is real gen-ai processing time -- that theory doesn't match the agreed definition: create being tracked across turns is only what lets the harness recognize completion when execute arrives late, it was never meant to change what latency measures. - The "continuation" mechanism itself (create/execute tracked independently across turns, with a "Please proceed." nudge if create succeeded without execute in the same turn) is removed entirely, not just its latency accounting. Read gdc-nas's actual KDA skill (system prompt + orchestration loop): create and execute are always called together in one turn, no confirmation step -- this scenario doesn't happen. The only real no-execute case is the org having data-sharing with the LLM off, which removes execute_key_driver_analysis from the tool list entirely (permanent gap, not a delayed call); no chat nudge can work around that, so there was nothing for this mechanism to correctly handle in the first place. - _DEFAULT_MAX_ITERATIONS raised from 2 to 3 for a different reason than the removed continuation mechanism: metric and analyzed-period ambiguity can each need their own clarifying question, so a case ambiguous on both can legitimately take 2 rounds before create is ever called (confirmed against real test runs). Still asking after that is the model failing to resolve the object, not KDA itself. JIRA: QA-28800 --- .../src/gooddata_eval/cli/agentic_runner.py | 12 + .../gooddata_eval/core/agentic/__init__.py | 14 + .../gooddata_eval/core/agentic/kda_skill.py | 382 +++++++ .../src/gooddata_eval/core/chat/sse_client.py | 67 +- .../src/gooddata_eval/core/models.py | 4 + .../tests/test_agentic_kda_skill.py | 930 ++++++++++++++++++ .../tests/test_agentic_langfuse_trace.py | 26 + .../gooddata-eval/tests/test_sse_client.py | 150 +++ 8 files changed, 1578 insertions(+), 7 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/cli/agentic_runner.py b/packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py index f73b44679..7147af183 100644 --- a/packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py +++ b/packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py @@ -11,6 +11,7 @@ from gooddata_eval.core.agentic.conversation import ConversationFixture, evaluate_agentic_conversation from gooddata_eval.core.agentic.general_question import evaluate_agentic_general_question from gooddata_eval.core.agentic.guardrail import evaluate_agentic_guardrail +from gooddata_eval.core.agentic.kda_skill import evaluate_agentic_kda_skill from gooddata_eval.core.agentic.metric_skill import evaluate_agentic_metric_skill from gooddata_eval.core.agentic.search_tool import evaluate_agentic_search_tool from gooddata_eval.core.agentic.visualization import evaluate_agentic_visualization @@ -38,6 +39,7 @@ class _LfKw(TypedDict, total=False): "agentic_general_question", "agentic_guardrail", "agentic_conversation", + "agentic_kda_skill", } ) @@ -159,6 +161,16 @@ def _dispatch_agentic( k=k, **lf_kw, ) + elif kind == "agentic_kda_skill": + evaluate_agentic_kda_skill( + host=host, + token=token, + workspace_id=workspace_id, + question=item.question, + expected_output=eo if isinstance(eo, dict) else {}, + k=k, + **lf_kw, + ) elif kind == "agentic_conversation": fixture_data = eo.get("fixture") or eo if isinstance(eo, dict) else {} evaluate_agentic_conversation( diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py index 639bee5b7..89e93dde8 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/__init__.py @@ -30,6 +30,14 @@ evaluate_agentic_guardrail, run_agentic_guardrail, ) +from gooddata_eval.core.agentic.kda_skill import ( + AgenticKdaSummary, + KdaEvaluation, + KdaRunResult, + KdaSkillAssertionError, + evaluate_agentic_kda_skill, + run_agentic_kda_skill, +) from gooddata_eval.core.agentic.metric_skill import ( AgenticMetricSummary, MetricRunResult, @@ -56,6 +64,7 @@ "AgenticAlertSummary", "AgenticGeneralQuestionSummary", "AgenticGuardrailSummary", + "AgenticKdaSummary", "AgenticMetricSummary", "AgenticSearchSummary", "AgenticRunSummary", @@ -69,6 +78,9 @@ "GeneralQuestionResult", "GuardrailAssertionError", "GuardrailResult", + "KdaEvaluation", + "KdaRunResult", + "KdaSkillAssertionError", "MetricRunResult", "MetricSkillAssertionError", "RunResult", @@ -81,6 +93,7 @@ "evaluate_agentic_conversation", "evaluate_agentic_general_question", "evaluate_agentic_guardrail", + "evaluate_agentic_kda_skill", "evaluate_agentic_metric_skill", "evaluate_agentic_search_tool", "evaluate_agentic_visualization", @@ -88,6 +101,7 @@ "run_agentic_conversation", "run_agentic_general_question", "run_agentic_guardrail", + "run_agentic_kda_skill", "run_agentic_metric_skill", "run_agentic_search_tool", "run_agentic_visualization", diff --git a/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py new file mode 100644 index 000000000..f621afa27 --- /dev/null +++ b/packages/gooddata-eval/src/gooddata_eval/core/agentic/kda_skill.py @@ -0,0 +1,382 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +"""Agentic KDA (Key Driver Analysis)-skill evaluation runner.""" + +from __future__ import annotations + +import logging +import os +import re +from dataclasses import dataclass + +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__) + +_DEFAULT_K = 1 +# Disambiguation safety net only (create+execute always run together in the same +# turn) -- 3 covers metric and period each needing their own clarifying question. +_DEFAULT_MAX_ITERATIONS = 3 + + +def _is_asking_kda_clarification(text: str) -> bool: + """True if ``text`` reads as the agent asking for input, not a final answer. + + KDA-specific, not shared with metric_skill.py/conversation.py -- each skill's + disambiguation heuristic has already drifted independently. Requires the text to + end on "?" (a "?" anywhere also matches a final answer that merely quotes one). + """ + if not text: + return False + t = text.strip().lower() + if t.endswith("?"): + return True + # "To clarify, ..." means "in other words" (a final answer), not a request for one -- + # strip it first so "clarif" below only matches genuine clarification requests. + 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. Picks *any* candidate from ``measure_candidates`` -- scope only needs KDA + to trigger, not the resulting measure to be exactly right. Always OpenAI regardless + of the combo's own provider -- this is test-harness plumbing, not the system under test. + """ + 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, + timeout=30, + ) + 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 in this turn's + tool calls -- not the last create and last execute picked independently. A new create + call clears any earlier execute_result -- it belongs to the create it followed, not to + this one. + """ + create_args: dict | None = None + execute_result: dict | None = None + for tc in tool_call_events: + if tc.function_name == "create_key_driver_analysis": + create_args = tc.parsed_arguments() + execute_result = None + elif tc.function_name == "execute_key_driver_analysis" and tc.result: + execute_result = tc.parsed_result() + return create_args, execute_result + + +@dataclass +class KdaEvaluation: + """Evaluation scores for a single KDA-skill run. + + Scope: 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). + """ + + triggered: bool + executed: bool + success: bool + turn_completed: bool + disambiguated: bool = False + + @property + def strict_pass(self) -> bool: + return all([self.triggered, self.executed, self.success, self.turn_completed]) + + +@dataclass +class KdaRunResult: + """Outcome of one run (one conversation, up to max_iterations messages) for a KDA case.""" + + conversation_id: str + evaluation: KdaEvaluation + actual_create_args: dict | None + actual_execute_result: dict | None + # Wall-clock time of the turn that called create (None if create never happened) -- + # not any earlier disambiguation turn. See run_agentic_kda_skill's _run_once. + turn_wall_clock_sec: float | None = None + + +@dataclass +class AgenticKdaSummary: + """Aggregated outcome of K runs for a KDA case.""" + + run_results: list[KdaRunResult] + pass_at_k: bool + pass_power_k: bool + best: KdaRunResult + + +def _evaluate_run( + create_args: dict | None, + execute_result: dict | None, + turn_completed: bool, + disambiguated: bool = False, +) -> KdaEvaluation: + triggered = create_args is not None + executed = execute_result is not None + success = executed and execute_result.get("success") is True + return KdaEvaluation( + triggered=triggered, + executed=executed, + success=success, + turn_completed=turn_completed, + 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 one message, one turn -- create and execute are always called + together in the same turn (the skill's own system prompt: "NO confirmation needed"). + The only thing that can extend a run up to ``max_iterations`` turns is the agent + asking a clarifying question instead of triggering KDA directly; a simulated user + reply nudges it forward. + """ + if k < 1: + # k=0 or negative would otherwise silently run once, indistinguishable from k=1. + raise ValueError(f"k must be >= 1, got {k}") + run_results: list[KdaRunResult] = [] + client = ChatClient(host=host, token=token, workspace_id=workspace_id, reasoning_effort=reasoning_effort) + + def _run_once(conv_id: str) -> KdaRunResult: + create_args: dict | None = None + execute_result: dict | None = None + turn_wall_clock_sec: float | None = None + turn_completed = False + disambiguated = False + current_question = question + + for iteration in range(max_iterations): + try: + chat_result = client.send_message(conv_id, current_question) + except Exception as exc: # noqa: BLE001 -- end this run, not the whole assertion + _log.warning("KDA send_message failed for conversation %s: %s", conv_id, exc) + partial = getattr(exc, "partial_result", None) + if partial is not None: + create_args, execute_result = _extract_kda_calls(partial.tool_call_events or []) + if create_args is not None: + turn_wall_clock_sec = partial.turn_wall_clock_sec + turn_completed = False + break + create_args, execute_result = _extract_kda_calls(chat_result.tool_call_events or []) + response_text = (chat_result.text_response or "").strip() + turn_completed = chat_result.stream_ended and bool(response_text) + if create_args is not None: + # This turn's own time -- the turn that called create, not any earlier + # disambiguation turn or the simulated-reply generation. create and execute + # are always called together in the same turn (or not at all), so this is + # final either way -- execute_result may still be None (e.g. the skill's + # execute tool isn't available at all when data-sharing is off for the org). + turn_wall_clock_sec = chat_result.turn_wall_clock_sec + break + if iteration >= max_iterations - 1: + break + if _is_asking_kda_clarification(response_text): + measure_candidates = expected_output.get("Measure") if isinstance(expected_output, dict) else None + try: + current_question = generate_simulated_kda_response(response_text, measure_candidates) + 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 + else: + break + + ev = _evaluate_run(create_args, execute_result, turn_completed, disambiguated) + return KdaRunResult( + conversation_id=conv_id, + evaluation=ev, + actual_create_args=create_args, + actual_execute_result=execute_result, + turn_wall_clock_sec=turn_wall_clock_sec, + ) + + try: + conv_id_0 = initial_conversation_id if initial_conversation_id is not None else client.create_conversation() + try: + run_results.append(_run_once(conv_id_0)) + finally: + if initial_conversation_id is None: # only delete conversations we created + client.delete_conversation(conv_id_0) + + for _ in range(1, k): + conv_id = client.create_conversation() + try: + run_results.append(_run_once(conv_id)) + finally: + client.delete_conversation(conv_id) + finally: + client.close() + + pass_at_k = any(r.evaluation.strict_pass for r in run_results) + pass_power_k = all(r.evaluation.strict_pass for r in run_results) + best = max( + run_results, + key=lambda r: sum( + [r.evaluation.triggered, r.evaluation.executed, r.evaluation.success, r.evaluation.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 = "kda_skill", + run_timestamp: str | None = None, + model_version_override: str | None = None, + run_metadata_extra: dict | None = None, + reasoning_effort: ReasoningEffort | None = None, +) -> None: + """Run KDA-skill evaluation, log to Langfuse, and raise KdaSkillAssertionError on failure.""" + from datetime import datetime as _dt # noqa: PLC0415 + from datetime import timezone as _tz # noqa: PLC0415 + + from gooddata_eval.core.agentic._langfuse import try_make_langfuse_client # noqa: PLC0415 + + if langfuse is None: + langfuse = try_make_langfuse_client() + window_start = _dt.now(_tz.utc) + summary = run_agentic_kda_skill( + host=host, + token=token, + workspace_id=workspace_id, + question=question, + expected_output=expected_output, + k=k, + max_iterations=max_iterations, + initial_conversation_id=initial_conversation_id, + reasoning_effort=reasoning_effort, + ) + + if langfuse is not None and dataset_item_id: + from gooddata_eval.core.agentic._langfuse import ( # noqa: PLC0415 + build_run_context, + find_traces_per_conversation, + log_quality_and_value_scores, + observe, + score_safe, + ) + + run_name_base, run_metadata = build_run_context( + host, + token, + workspace_id, + dataset_name, + run_timestamp, + model_version_override, + run_metadata_extra, + reasoning_effort, + ) + # No custom selector -- same default (max-latency) as every other skill; harmless + # here since latency comes from run.turn_wall_clock_sec below, not this trace. + traces_by_conv = find_traces_per_conversation( + langfuse, + [r.conversation_id for r in summary.run_results], + window_start, + ) + suffix_needed = len(summary.run_results) > 1 + for run_idx, run in enumerate(summary.run_results): + pt = traces_by_conv.get(run.conversation_id) + run_name = f"{run_name_base}_run{run_idx}" if suffix_needed else run_name_base + ev = run.evaluation + # Gates strict_pass -- current scope is completion only (see KdaEvaluation docstring). + strict_checks = { + "kda_triggered": ev.triggered, + "kda_executed": ev.executed, + "kda_success": ev.success, + "kda_turn_completed": ev.turn_completed, + } + # Not pt.latency: pt can be any trace of the conversation, not necessarily the KDA turn. + turn_wall_clock_sec = run.turn_wall_clock_sec + _log.info("[kda-report] %s: strict_pass=%s latency_sec=%s", run_name, ev.strict_pass, turn_wall_clock_sec) + with observe(langfuse, pt.id if pt else None, dataset_item_id, run_name, run_metadata) as tid: + for score_name, value in strict_checks.items(): + score_safe(langfuse, tid, name=score_name, value=float(value), data_type="BOOLEAN") + score_safe(langfuse, tid, name="kda_disambiguated", value=float(ev.disambiguated), data_type="BOOLEAN") + if turn_wall_clock_sec is not None: + # combo_report.py reads this score directly -- no trace re-resolution needed. + score_safe( + langfuse, tid, name="kda_turn_wall_clock_sec", value=turn_wall_clock_sec, data_type="NUMERIC" + ) + log_quality_and_value_scores( + langfuse, + tid, + strict_checks=strict_checks, + latency_sec=turn_wall_clock_sec, + cost_usd=pt.total_cost if pt and ev.triggered else None, + ) + + if not summary.pass_at_k: + best = summary.best + ev = best.evaluation + message = ( + f"KDA skill assertion failed. strict_pass={ev.strict_pass} " + f"(triggered={ev.triggered}, executed={ev.executed}, " + f"success={ev.success}, turn_completed={ev.turn_completed}). " + 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..562b466b7 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,34 @@ _log = logging.getLogger(__name__) SSE_DATA_PREFIX = "data: " +SSE_EVENT_PREFIX = "event: " +# gen-ai's last event, only if at least one item was already emitted (conversations_controller.py). +_RESPONSE_ENDED_EVENT = "response_ended" _RETRYABLE_STATUS_CODES: frozenset[int] = frozenset({429, 502, 503, 504}) _METADATA_SYNC_MARKER = "METADATA_SYNC_IN_PROGRESS" class ChatError(RuntimeError): - """Non-retryable error reported by the chat SSE stream.""" + """Non-retryable error reported by the chat SSE stream. - def __init__(self, message: str, *, status_code: int | None = None, detail: str | None = None) -> None: + ``partial_result`` carries whatever the accumulator captured before the error fired + (tool calls included). Callers must not assume it's complete -- fields like + ``stream_ended`` reflect the state at the moment of the error, not a 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 +125,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 +204,37 @@ def _build_chat_result(acc: _SseAccumulator) -> ChatResult: } result = ChatResult.model_validate(payload) result.response_id = acc.response_id + result.stream_ended = acc.stream_ended return result def parse_sse_lines(lines: Iterable[str]) -> ChatResult: """Parse an SSE stream (iterable of decoded lines) into a ChatResult.""" acc = _SseAccumulator() - for raw_line in lines: + current_event = "message" # SSE default in the absence of an explicit "event: " line + it = iter(lines) + while True: + try: + raw_line = next(it) + except StopIteration: + break + except Exception as exc: + # Only a transport-level failure (e.g. connection drop mid-stream) is rescued + # here -- a bug in the processing below must propagate uncaught, not get + # mislabeled as a network error. + raise ChatError(f"SSE stream error: {exc}", partial_result=_build_chat_result(acc)) from exc 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() + if current_event == _RESPONSE_ENDED_EVENT: + acc.stream_ended = True + continue + if not line.startswith(SSE_DATA_PREFIX): + continue + if current_event == _RESPONSE_ENDED_EVENT: continue data_str = line[len(SSE_DATA_PREFIX) :] if _METADATA_SYNC_MARKER in data_str: @@ -203,6 +242,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 +253,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") @@ -293,9 +335,20 @@ def send_message(self, conversation_id: str, question: str) -> ChatResult: body["options"] = {"reasoningEffort": self._reasoning_effort} def _do() -> ChatResult: + # Set fresh on every retry attempt (before opening this attempt's stream, so its + # own connection setup time counts) -- excludes not just the sleep backoff between + # attempts, but the entire duration of any earlier failed attempt. + t0 = time.monotonic() with self._client.stream("POST", url, json=body, headers=headers) as resp: resp.raise_for_status() - return parse_sse_lines(resp.iter_lines()) + try: + result = parse_sse_lines(resp.iter_lines()) + except ChatError as exc: + if exc.partial_result is not None: + exc.partial_result.turn_wall_clock_sec = time.monotonic() - t0 + raise + result.turn_wall_clock_sec = time.monotonic() - t0 + return result return _retry_transient(_do, is_retryable=_is_retryable_exc) diff --git a/packages/gooddata-eval/src/gooddata_eval/core/models.py b/packages/gooddata-eval/src/gooddata_eval/core/models.py index 336c313b9..ee58dcc80 100644 --- a/packages/gooddata-eval/src/gooddata_eval/core/models.py +++ b/packages/gooddata-eval/src/gooddata_eval/core/models.py @@ -100,6 +100,10 @@ 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 once gen-ai's response_ended event arrived. + stream_ended: bool = False + # Wall-clock seconds for the whole chat turn, timed by the client. + turn_wall_clock_sec: float | None = None class SummaryInput(BaseModel): diff --git a/packages/gooddata-eval/tests/test_agentic_kda_skill.py b/packages/gooddata-eval/tests/test_agentic_kda_skill.py new file mode 100644 index 000000000..d8fa2cdfb --- /dev/null +++ b/packages/gooddata-eval/tests/test_agentic_kda_skill.py @@ -0,0 +1,930 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +# SPDX-License-Identifier: LicenseRef-GoodData-Enterprise +import json +from unittest.mock import MagicMock, patch + +import httpx +import pytest +from gooddata_eval.core.agentic.kda_skill import ( + KdaEvaluation, + KdaSkillAssertionError, + _evaluate_run, + _extract_kda_calls, + _is_asking_kda_clarification, + evaluate_agentic_kda_skill, + run_agentic_kda_skill, +) +from gooddata_eval.core.chat.sse_client import ChatError, TransientChatError +from gooddata_eval.core.models import ChatResult + +_EXPECTED = {"Measure": {"type": "metric", "id": "revenue"}} + + +def _tool_call(name: str, result: dict | None = None, arguments: dict | None = None): + return { + "functionName": name, + "functionArguments": "{}" if arguments is None else json.dumps(arguments), + "result": None if result is None else json.dumps(result), + } + + +def _kda_chat_result( + *, + success: bool = True, + text: str = "Here is the analysis.", + stream_ended: bool = True, + turn_wall_clock_sec: float | None = None, +) -> ChatResult: + return ChatResult.model_validate( + { + "textResponse": text, + "toolCallEvents": [ + _tool_call("create_key_driver_analysis", arguments={"measure": {"type": "metric", "id": "revenue"}}), + _tool_call("execute_key_driver_analysis", result={"success": success, "data": {"summary": {}}}), + ], + "reasoningStepCount": 1, + "stream_ended": stream_ended, + "turn_wall_clock_sec": turn_wall_clock_sec, + } + ) + + +def _no_kda_chat_result( + text: str = "I could not find that metric.", + *, + stream_ended: bool = True, + turn_wall_clock_sec: float | None = None, +) -> ChatResult: + return ChatResult.model_validate( + { + "textResponse": text, + "toolCallEvents": [], + "reasoningStepCount": 1, + "stream_ended": stream_ended, + "turn_wall_clock_sec": turn_wall_clock_sec, + } + ) + + +# --------------------------------------------------------------------------- # +# _is_asking_kda_clarification +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize( + "text", + ["Could you clarify which metric?", "Please provide the date range.", "Did you mean revenue?"], +) +def test_is_asking_kda_clarification_true(text): + assert _is_asking_kda_clarification(text) is True + + +def test_is_asking_kda_clarification_false_on_plain_statement(): + assert _is_asking_kda_clarification("Here is the key driver analysis result.") is False + + +def test_is_asking_kda_clarification_false_on_empty(): + assert _is_asking_kda_clarification("") is False + + +def test_is_asking_kda_clarification_false_when_question_mark_is_not_the_final_answer(): + # Regression guard for the original bug: a final answer that merely quotes or + # rhetorically references a question must not be mistaken for a clarifying question. + text = 'The user asked "what changed?" so here is the key driver breakdown they requested.' + assert _is_asking_kda_clarification(text) is False + + +@pytest.mark.parametrize( + "text", + [ + "To clarify, revenue rose 12% quarter over quarter.", + "Just to clarify, the increase was driven by the South region.", + ], +) +def test_is_asking_kda_clarification_false_on_to_clarify_discourse_marker(text): + # Regression guard: "to clarify, ..." is a discourse marker ("in other words") that + # introduces a restated FINAL answer, not a request for one -- the bare "clarif" in t + # substring check would otherwise mistake this for a clarifying question and burn a + # simulated-reply turn on an answer that was already complete. + assert _is_asking_kda_clarification(text) is False + + +def test_is_asking_kda_clarification_true_for_genuine_clarify_request_despite_marker_strip(): + # The discourse-marker strip must not eat a genuine request that happens to start the + # same way it's phrased in practice. No trailing "?" here specifically so this exercises + # the "could you" substring check post-strip, not the separate endswith("?") check. + assert _is_asking_kda_clarification("To clarify, could you tell me which region you mean") is True + + +# --------------------------------------------------------------------------- # +# _evaluate_run +# --------------------------------------------------------------------------- # +def test_evaluate_run_computes_core_fields_from_create_and_execute_args(): + ev = _evaluate_run({"measure": {"type": "metric", "id": "revenue"}}, {"success": True}, turn_completed=True) + assert (ev.triggered, ev.executed, ev.success, ev.turn_completed) == (True, True, True, True) + + +def test_evaluate_run_false_when_kda_never_triggered(): + ev = _evaluate_run(None, None, turn_completed=False) + assert (ev.triggered, ev.executed, ev.success) == (False, False, False) + + +def test_evaluate_run_success_false_when_execute_result_says_so(): + ev = _evaluate_run({"measure": {}}, {"success": False}, turn_completed=True) + assert (ev.triggered, ev.executed, ev.success) == (True, True, False) + + +def test_evaluate_run_passes_through_disambiguated(): + ev = _evaluate_run({"measure": {}}, {"success": True}, turn_completed=True, disambiguated=True) + assert ev.disambiguated is True + + +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 = { + "triggered": True, + "executed": True, + "success": True, + "turn_completed": 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.evaluation.triggered is True + assert summary.best.evaluation.executed is True + assert summary.best.evaluation.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.evaluation.turn_completed is False + # The KDA call itself still triggered/executed/succeeded -- only completion is in doubt. + assert summary.best.evaluation.triggered is True + assert summary.best.evaluation.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 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.evaluation.turn_completed is False + assert summary.best.evaluation.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.evaluation.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.evaluation.triggered is True + assert summary.best.evaluation.executed is True + assert summary.best.evaluation.success is True + # The error still means the turn itself didn't complete, regardless of what KDA did. + assert summary.best.evaluation.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.evaluation.turn_completed is False + assert summary.best.evaluation.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.evaluation.turn_completed is False + # The KDA call itself still triggered/executed/succeeded -- only completion is in doubt. + assert summary.best.evaluation.triggered is True + assert summary.best.evaluation.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.evaluation.disambiguated is True + assert summary.best.evaluation.triggered is True + + +def test_run_agentic_kda_skill_disambiguates_when_expected_output_is_not_a_dict(): + # DatasetItem.expected_output on the gdc-nas side allows str/list, not just dict. + # expected_output.get("Measure") would raise AttributeError on those shapes, silently + # swallowed by the broad except around generate_simulated_kda_response and disabling + # disambiguation with only a WARNING. Guard so the call still happens, with None + # candidates, instead of crashing. + 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.", + ) as mock_generate, + ): + 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=["not", "a", "dict"], + k=1, + max_iterations=2, + ) + + mock_generate.assert_called_once_with("Could you clarify which measure?", None) + assert summary.best.evaluation.disambiguated is True + assert summary.best.evaluation.triggered is True + + +def test_run_agentic_kda_skill_latency_is_only_the_turn_that_completed_kda(): + # The disambiguation turn's own time, and the simulated-reply generation between + # turns, must NOT be counted -- only the turn where KDA actually completed reflects + # gen-ai's own latency; the rest is test-harness overhead (an unrelated OpenAI call). + 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?", turn_wall_clock_sec=5.0), + _kda_chat_result(success=True, turn_wall_clock_sec=8.0), + ] + + 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.turn_wall_clock_sec == 8.0 + + +def test_run_agentic_kda_skill_triggered_but_not_executed_when_execute_tool_is_unavailable(): + # create and execute are always called together in the same turn, or not at all -- + # e.g. when the org has data-sharing with the LLM off, execute_key_driver_analysis + # isn't registered as a tool at all, so create can succeed alone within a single turn + # with no execute_result. Must be scored as triggered but not executed immediately, + # not treated as "execute is coming in a later turn". + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + create_only = ChatResult.model_validate( + { + "textResponse": "The analysis is ready. Open it above to review the results.", + "toolCallEvents": [ + _tool_call("create_key_driver_analysis", arguments={"measure": {"type": "metric", "id": "revenue"}}) + ], + "stream_ended": True, + "turn_wall_clock_sec": 3.0, + } + ) + mock_client.send_message.return_value = create_only + + 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=2, + ) + + assert summary.best.evaluation.triggered is True + assert summary.best.evaluation.executed is False + assert summary.best.evaluation.disambiguated is False + assert summary.best.turn_wall_clock_sec == 3.0 + assert mock_client.send_message.call_count == 1 + + +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.evaluation.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.evaluation.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_disambiguation_then_create_without_execute(): + # Combines both remaining paths in one run -- a disambiguation turn, then a turn where + # create succeeds but execute_result is None (e.g. execute is unavailable for this + # org). Confirms this still scores correctly (disambiguated + triggered, not executed) + # instead of being mistaken for "still waiting on more turns". + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + create_only = ChatResult.model_validate( + { + "textResponse": "The analysis is ready. Open it above to review the results.", + "toolCallEvents": [ + _tool_call("create_key_driver_analysis", arguments={"measure": {"type": "metric", "id": "revenue"}}) + ], + "stream_ended": True, + } + ) + mock_client.send_message.side_effect = [ + _no_kda_chat_result("Could you clarify which measure?"), + create_only, + ] + + 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 mock_client.send_message.call_count == 2 + assert summary.best.evaluation.disambiguated is True + assert summary.best.evaluation.triggered is True + assert summary.best.evaluation.executed is False + assert summary.pass_at_k is False + + +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].evaluation.triggered is True + assert summary.run_results[1].evaluation.triggered is False + assert summary.pass_at_k is True # run 0 still counts + + +def test_run_agentic_kda_skill_uses_initial_conversation_for_run_0(): + mock_client = MagicMock() + mock_client.send_message.return_value = _kda_chat_result(success=True) + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + initial_conversation_id="existing-conv", + ) + mock_client.create_conversation.assert_not_called() + mock_client.delete_conversation.assert_not_called() + + +def test_run_agentic_kda_skill_creates_fresh_conversations_for_remaining_runs(): + mock_client = MagicMock() + mock_client.create_conversation.side_effect = ["fresh-1", "fresh-2"] + mock_client.send_message.return_value = _kda_chat_result(success=True) + with patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client): + run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=3, + max_iterations=1, + initial_conversation_id="existing-conv", + ) + assert mock_client.create_conversation.call_count == 2 + assert mock_client.delete_conversation.call_count == 2 + + +@pytest.mark.parametrize("bad_k", [0, -1, -5]) +def test_run_agentic_kda_skill_rejects_non_positive_k(bad_k): + with pytest.raises(ValueError, match="k must be >= 1"): + run_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=bad_k, + ) + + +# --------------------------------------------------------------------------- # +# evaluate_agentic_kda_skill +# --------------------------------------------------------------------------- # +def test_evaluate_agentic_kda_skill_raises_on_failure(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _no_kda_chat_result() + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.try_make_langfuse_client", return_value=None), + pytest.raises(KdaSkillAssertionError), + ): + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=None, + ) + + +def test_evaluate_agentic_kda_skill_does_not_raise_on_success(): + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True) + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.try_make_langfuse_client", return_value=None), + ): + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=None, + ) + + +def test_evaluate_agentic_kda_skill_never_treats_fallback_trace_latency_as_kda_latency(): + # Regression test: when KDA never triggered, whatever trace find_traces_per_conversation's + # default (max-latency) selector picks is NOT a real KDA turn -- its latency/cost must not + # be logged as the KDA run's own value_score inputs. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _no_kda_chat_result() + + fallback_trace = MagicMock(id="fallback-trace", latency=999.0, total_cost=5.0) + mock_langfuse = MagicMock() + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.build_run_context", return_value=("run-base", {})), + patch( + "gooddata_eval.core.agentic._langfuse.find_traces_per_conversation", + return_value={"conv-1": fallback_trace}, + ), + patch("gooddata_eval.core.agentic._langfuse.observe") as mock_observe, + patch("gooddata_eval.core.agentic._langfuse.score_safe"), + patch("gooddata_eval.core.agentic._langfuse.log_quality_and_value_scores") as mock_log_scores, + pytest.raises(KdaSkillAssertionError), + ): + mock_observe.return_value.__enter__.return_value = "fallback-trace" + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=mock_langfuse, + dataset_item_id="item-1", + ) + + mock_log_scores.assert_called_once() + assert mock_log_scores.call_args.kwargs["latency_sec"] is None + assert mock_log_scores.call_args.kwargs["cost_usd"] is None + + +def test_evaluate_agentic_kda_skill_reports_trace_latency_when_kda_triggered(): + # Latency comes from the harness's own wall-clock measurement (ChatResult.turn_wall_clock_sec, + # set by ChatClient around its send_message() call), not from the trace find_traces_per_ + # conversation happens to return -- that trace isn't necessarily the KDA turn at all (see + # kda_skill.py's comment on `pt`). Only total_cost still comes from the trace. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True, turn_wall_clock_sec=76.0) + + found_trace = MagicMock(id="trace-1", latency=999.0, total_cost=0.02) + mock_langfuse = MagicMock() + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.build_run_context", return_value=("run-base", {})), + patch( + "gooddata_eval.core.agentic._langfuse.find_traces_per_conversation", + return_value={"conv-1": found_trace}, + ), + patch("gooddata_eval.core.agentic._langfuse.observe") as mock_observe, + patch("gooddata_eval.core.agentic._langfuse.score_safe") as mock_score_safe, + patch("gooddata_eval.core.agentic._langfuse.log_quality_and_value_scores") as mock_log_scores, + ): + mock_observe.return_value.__enter__.return_value = "trace-1" + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=1, + max_iterations=1, + langfuse=mock_langfuse, + dataset_item_id="item-1", + ) + + mock_log_scores.assert_called_once() + assert mock_log_scores.call_args.kwargs["latency_sec"] == 76.0 + assert mock_log_scores.call_args.kwargs["cost_usd"] == 0.02 + wall_clock_calls = [c for c in mock_score_safe.call_args_list if c.kwargs.get("name") == "kda_turn_wall_clock_sec"] + assert len(wall_clock_calls) == 1 + assert wall_clock_calls[0].kwargs["value"] == 76.0 + + +def test_evaluate_agentic_kda_skill_does_not_log_pass_at_k_or_pass_power_k(): + # Matches metric_skill/alert_skill/guardrail/search_tool/general_question, which all + # compute pass_at_k/pass_power_k but never log them to Langfuse at their default k=1 -- + # nothing reads a kda_pass_at_1 score, and the score name shifts if k ever changes, + # silently splitting any Langfuse view built on the old name. Only visualization.py + # logs this pair, with a real consumer at k=2 (combo_report.py's viz_flaky) that + # justifies it. + mock_client = MagicMock() + mock_client.create_conversation.return_value = "conv-1" + mock_client.send_message.return_value = _kda_chat_result(success=True) + + found_trace = MagicMock(id="trace-1", total_cost=0.01) + mock_langfuse = MagicMock() + + with ( + patch("gooddata_eval.core.agentic.kda_skill.ChatClient", return_value=mock_client), + patch("gooddata_eval.core.agentic._langfuse.build_run_context", return_value=("run-base", {})), + patch( + "gooddata_eval.core.agentic._langfuse.find_traces_per_conversation", + return_value={"conv-1": found_trace}, + ), + patch("gooddata_eval.core.agentic._langfuse.observe") as mock_observe, + patch("gooddata_eval.core.agentic._langfuse.score_safe") as mock_score_safe, + patch("gooddata_eval.core.agentic._langfuse.log_quality_and_value_scores"), + ): + mock_observe.return_value.__enter__.return_value = "trace-1" + evaluate_agentic_kda_skill( + host="http://host/api/v1/actions/workspaces/ws1/ai", + token="tok", + workspace_id="ws1", + question="What drove revenue change?", + expected_output=_EXPECTED, + k=2, + max_iterations=1, + langfuse=mock_langfuse, + dataset_item_id="item-1", + ) + + logged = {c.kwargs["name"] for c in mock_score_safe.call_args_list} + assert "kda_pass_at_2" not in logged + assert "kda_pass_power_2" not in logged + assert "pass_at_2" not in logged + assert "pass_power_2" not in logged diff --git a/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py b/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py new file mode 100644 index 000000000..af64a78d7 --- /dev/null +++ b/packages/gooddata-eval/tests/test_agentic_langfuse_trace.py @@ -0,0 +1,26 @@ +# (C) 2026 GoodData Corporation. All rights reserved. +# SPDX-License-Identifier: LicenseRef-GoodData-Enterprise +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +from gooddata_eval.core.agentic._langfuse import find_traces_per_conversation + + +def test_find_traces_per_conversation_is_none_for_a_conversation_with_no_trace(): + # find_traces_per_conversation's return dict is seeded with dict.fromkeys(conversation_ids) + # (every value starts None) and only overwritten for ids where a trace was actually found -- + # callers (kda_skill.py and every other agentic skill) must treat a missing conversation as + # None, not assume every key maps to a real trace object. + found_trace = MagicMock(latency=12.0) + + def _fetch(langfuse, cid, window_start, window_end, pad): + return [found_trace] if cid == "conv-found" else [] + + with ( + patch("gooddata_eval.core.agentic._langfuse._fetch_traces_for_session", side_effect=_fetch), + patch("gooddata_eval.core.agentic._langfuse.time.sleep"), + ): + result = find_traces_per_conversation(MagicMock(), ["conv-found", "conv-missing"], datetime.now(timezone.utc)) + + assert result["conv-found"] is found_trace + assert result["conv-missing"] is None diff --git a/packages/gooddata-eval/tests/test_sse_client.py b/packages/gooddata-eval/tests/test_sse_client.py index 490dfd57d..88be58abe 100644 --- a/packages/gooddata-eval/tests/test_sse_client.py +++ b/packages/gooddata-eval/tests/test_sse_client.py @@ -23,12 +23,127 @@ def test_parse_sse_lines_raises_on_error_event(): parse_sse_lines(lines) +def test_parse_sse_lines_error_carries_partial_result_with_tool_calls_already_seen(): + # A statusCode error ends the stream before _build_chat_result ever runs -- without + # partial_result, a tool call that already succeeded (e.g. KDA's own create/execute) + # before a LATER, unrelated error killed the turn would be silently discarded, making + # the run look like the agent never called the tool at all. + lines = [ + json.dumps( + { + "item": { + "role": "assistant", + "content": {"type": "toolCall", "callId": "c1", "name": "create_key_driver_analysis"}, + } + } + ), + "", + json.dumps( + { + "item": { + "role": "tool", + "content": { + "type": "toolResult", + "callId": "c1", + "result": json.dumps({"success": True}), + }, + } + } + ), + "", + json.dumps({"statusCode": 500, "detail": "boom"}), + ] + lines = [f"data: {line}" if line else line for line in lines] + with pytest.raises(ChatError) as ei: + parse_sse_lines(lines) + partial = ei.value.partial_result + assert partial is not None + assert len(partial.tool_call_events) == 1 + assert partial.tool_call_events[0].function_name == "create_key_driver_analysis" + assert partial.tool_call_events[0].result == '{"success": true}' + + +def test_parse_sse_lines_raw_transport_error_also_carries_partial_result(): + # A connection drop mid-stream (httpx.RemoteProtocolError/ReadError) has no statusCode + # payload -- it's a raw exception from iterating `lines` itself, not one this module + # raises. Must still be rescued the same way a statusCode-shaped error is. + def _lines(): + yield ( + 'data: {"item": {"role": "assistant", "content": ' + + json.dumps({"type": "toolCall", "callId": "c1", "name": "create_key_driver_analysis"}) + + "}}" + ) + yield "" + raise RuntimeError("connection dropped") + + with pytest.raises(ChatError) as ei: + parse_sse_lines(_lines()) + assert not isinstance(ei.value, TransientChatError) # not retried -- same as before this fix + partial = ei.value.partial_result + assert partial is not None + assert len(partial.tool_call_events) == 1 + assert partial.tool_call_events[0].function_name == "create_key_driver_analysis" + + +def test_parse_sse_lines_a_real_parsing_bug_propagates_uncaught_not_as_a_chat_error(): + # A malformed payload (here: "item" is a string, not a dict) crashes the processing + # code itself with a plain AttributeError -- must surface loudly as that bug, not get + # silently relabeled as a ChatError/"SSE stream error" indistinguishable from a + # genuine network blip. Only a failure from iterating `lines` itself is rescued. + lines = ['data: {"item": "not-a-dict"}'] + with pytest.raises(AttributeError): + parse_sse_lines(lines) + + 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_stream_ended_true_when_response_ended_has_no_data_line(): + lines = [ + "event: item", + 'data: {"item": {"role": "assistant", "content": {"type": "text", "text": "x"}}}', + "", + "event: response_ended", + "", + ] + assert parse_sse_lines(lines).stream_ended is True + + def test_parse_sse_lines_falls_back_to_adhoc_viz_when_multipart_viz_is_null(): """Visualization from create_adhoc_visualization args used when multipart viz is null.""" viz_def = { @@ -205,6 +320,41 @@ def handler(request): assert sleeps == [] +def test_send_message_sets_turn_wall_clock_sec_on_success(monkeypatch): + monkeypatch.setattr(sse_mod.time, "monotonic", iter([100.0, 102.5]).__next__) + client = _client_with_handler(lambda request: httpx.Response(200, content=_OK_SSE)) + result = client.send_message("conv", "q") + assert result.turn_wall_clock_sec == pytest.approx(2.5) + + +def test_send_message_wall_clock_excludes_retry_backoff(monkeypatch): + # t0 must be per-attempt, set inside _do() before the connection opens -- not around + # the whole send_message() call -- or a failed attempt's time plus the backoff sleep + # between attempts (harness/network overhead, not gen-ai's time) would inflate the + # reported latency. + monkeypatch.setattr(sse_mod.time, "sleep", lambda s: None) + monkeypatch.setattr(sse_mod.time, "monotonic", iter([1000.0, 1000.5, 2000.0, 2001.2]).__next__) + calls = {"n": 0} + + def handler(request): + calls["n"] += 1 + return httpx.Response(200, content=_TRANSIENT_SSE if calls["n"] < 2 else _OK_SSE) + + client = _client_with_handler(handler) + result = client.send_message("conv", "q") + assert calls["n"] == 2 + assert result.turn_wall_clock_sec == pytest.approx(1.2) # attempt 2 alone, not spanning attempt 1 + backoff + + +def test_send_message_stamps_turn_wall_clock_sec_on_partial_result_too(monkeypatch): + monkeypatch.setattr(sse_mod.time, "monotonic", iter([50.0, 51.0]).__next__) + client = _client_with_handler(lambda request: httpx.Response(200, content=_NONRETRY_SSE)) + with pytest.raises(ChatError) as ei: + client.send_message("conv", "q") + assert ei.value.partial_result is not None + assert ei.value.partial_result.turn_wall_clock_sec == pytest.approx(1.0) + + def test_create_conversation_retries_then_succeeds(monkeypatch): sleeps = [] monkeypatch.setattr(sse_mod.time, "sleep", lambda s: sleeps.append(s))