feat(gooddata-eval): add KDA-skill agentic evaluator

Add kda_skill.py: the evaluator for the KDA (Key Driver Analysis) chatbot
skill. Mirrors metric_skill/alert_skill's shape (run_agentic_*/
evaluate_agentic_*/*AssertionError), with a few KDA-specific pieces:

- Scope (QA-28800): strict_pass gates only completion (kda_triggered,
  executed, success, turn_completed), not per-field correctness or
  latency -- those are logged as informational-only for a follow-up ticket.
- A bounded (max_iterations=2) disambiguation safety net: if the agent asks
  a clarifying question instead of triggering KDA, a simulated user reply
  (gpt-4o-mini) nudges it forward. A failure in that helper ends only the
  current run (contained), not the whole evaluation.
- KDA-turn latency for the daily report is read from Langfuse, but the
  trace to read it from is selected by which one actually made the
  create_key_driver_analysis/execute_key_driver_analysis tool call
  (paginated observation lookup) -- not by picking the largest-latency
  trace in the session, which can pick the wrong turn when a case spans
  more than one (a disambiguation exchange, or a transient-retry the chat
  SDK does internally).
- classify_kda_report_bucket() classifies each run into pass (completed,
  <=60s) / failed (completed, slower) / error (didn't complete) for a
  separate daily-report reducer to read back via three Langfuse boolean
  scores -- distinct from strict_pass, which continues to gate the CI
  assertion on completion only.

_langfuse.py: add an additive Observations API (paginated) and an optional
select= override on find_traces_per_conversation (default unchanged: max
latency) so KDA's trace selection doesn't touch the default other skills use.

JIRA: QA-28800
risk: nonprod

Author: FrankHuynh

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,13 +93,15 @@
     "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",
     "run_agentic_alert_skill",
     "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",

packages/gooddata-eval/src/gooddata_eval/core/agentic/_langfuse.py

@@ -8,7 +8,7 @@
 import os
 import time
 import uuid
-from collections.abc import Iterator
+from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 from typing import Any
@@ -56,6 +56,35 @@ def _ts(v: Any) -> str:
         return _TraceListResult([_TraceObj(t) for t in resp.json().get("data", [])])
 
 
+class _ObservationObj:
+    """Duck-type wrapper around a raw Langfuse observation dict."""
+
+    def __init__(self, raw: dict) -> None:
+        self.id: str = raw.get("id", "")
+        self.name: str | None = raw.get("name")
+        self.output: Any = raw.get("output")
+
+
+class _ObservationListResult:
+    def __init__(self, data: list[_ObservationObj], total_pages: int) -> None:
+        self.data = data
+        self.total_pages = total_pages
+
+
+class _ObservationsAPI:
+    def __init__(self, client: httpx.Client) -> None:
+        self._client = client
+
+    def list(self, trace_id: str, page: int = 1, limit: int = 100) -> _ObservationListResult:
+        resp = self._client.get("/api/public/observations", params={"traceId": trace_id, "page": page, "limit": limit})
+        resp.raise_for_status()
+        body = resp.json()
+        meta = body.get("meta") or {}
+        return _ObservationListResult(
+            [_ObservationObj(o) for o in body.get("data", [])], total_pages=int(meta.get("totalPages") or 1)
+        )
+
+
 class _DatasetRunItemsAPI:
     def __init__(self, client: httpx.Client) -> None:
         self._client = client
@@ -83,6 +112,7 @@ def create(
 class _LangfuseAPI:
     def __init__(self, client: httpx.Client) -> None:
         self.trace = _TraceAPI(client)
+        self.observations = _ObservationsAPI(client)
         self.dataset_run_items = _DatasetRunItemsAPI(client)
 
 
@@ -247,11 +277,16 @@ def find_traces_per_conversation(
     langfuse: Any,
     conversation_ids: list[str],
     window_start: datetime,
+    select: Callable[[list[Any]], Any | None] | None = None,
 ) -> dict[str, Any]:
-    """Poll Langfuse until traces matching all conversation_ids are found or retries exhaust."""
+    """Check Langfuse for the trace(s) matching each conversation_id; get latency, picking
+    the right turn via ``select`` (default: largest latency) -- e.g. KDA passes a selector
+    that picks the turn that actually made the KDA tool call.
+    """
     if bool(os.environ.get(SKIP_ENV_VAR)):
         return dict.fromkeys(conversation_ids)
 
+    select = select or (lambda found: max(found, key=lambda t: getattr(t, "latency", None) or 0.0))
     by_conv: dict[str, Any] = dict.fromkeys(conversation_ids)
     window_end = datetime.now(timezone.utc)
     pad = timedelta(seconds=_WINDOW_PADDING_SEC)
@@ -269,7 +304,7 @@ def find_traces_per_conversation(
                 break
             delay *= _BACKOFF
         if found:
-            by_conv[cid] = max(found, key=lambda t: getattr(t, "latency", None) or 0.0)
+            by_conv[cid] = select(found)
         else:
             _log.warning(
                 "[langfuse] No trace found for conversation %s in window [%s, %s]", cid, window_start, window_end