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 (QA-28699), 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.

Latency is measured directly by the harness, not re-derived from Langfuse
after the fact: ChatResult.turn_wall_clock_sec is set by ChatClient around
the SSE read itself (after the connection opens, excluding any transient
retry's backoff sleep), stamped on both the successful result and any
ChatError.partial_result so a turn that dies mid-stream after KDA already
succeeded still gets a number. KdaRunResult.kda_turn_sec captures it in
_run_once at the exact point create_args gets set -- atomic with the
kda_triggered signal, not a separate step that can fail independently.
Logged as the kda_wall_clock_sec Langfuse score; combo_report.py (gdc-nas)
reads it directly, with no trace re-querying or observation-shape matching
needed on that side anymore.

Adds ChatResult.stream_ended (from the SSE response_ended event, derived --
not a raw server field, see sse_client.py's _RESPONSE_ENDED_EVENT comment).
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.
- run_agentic_kda_skill rejects k < 1 -- the single initial run happens
  unconditionally regardless of k, so a bad env-driven KDA_RUN_K value (0, a
  typo, negative) previously ran silently once instead of surfacing the bad
  config, indistinguishable from a deliberate k=1.
- Regression tests for find_traces_per_conversation's None-safety and
  _filters_match's isinstance guard.

JIRA: QA-28800

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/kda_skill.py

@@ -28,18 +28,38 @@
 _log = logging.getLogger(__name__)
 
 SSE_DATA_PREFIX = "data: "
+SSE_EVENT_PREFIX = "event: "
+# gen-ai's last event, only if at least one item was already emitted (conversations_controller.py).
+_RESPONSE_ENDED_EVENT = "response_ended"
 
 _RETRYABLE_STATUS_CODES: frozenset[int] = frozenset({429, 502, 503, 504})
 _METADATA_SYNC_MARKER = "METADATA_SYNC_IN_PROGRESS"
 
 
 class ChatError(RuntimeError):
-    """Non-retryable error reported by the chat SSE stream."""
+    """Non-retryable error reported by the chat SSE stream.
 
-    def __init__(self, message: str, *, status_code: int | None = None, detail: str | None = None) -> None:
+    ``partial_result`` carries whatever the accumulator captured before this error fired
+    (tool calls included) -- an error event ends the stream before ``_build_chat_result``
+    ever runs, so without this a caller has no way to see, e.g., that KDA's own tool calls
+    already succeeded before an unrelated later error (a failed final-summary generation)
+    killed the turn. Callers must not assume it's complete: fields normally filled in only
+    at the very end of the stream (``stream_ended``, in particular) reflect the state at
+    the moment of the error, not a genuinely finished turn.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        status_code: int | None = None,
+        detail: str | None = None,
+        partial_result: ChatResult | None = None,
+    ) -> None:
         super().__init__(message)
         self.status_code = status_code
         self.detail = detail
+        self.partial_result = partial_result
 
 
 class TransientChatError(ChatError):
@@ -109,6 +129,7 @@ class _SseAccumulator:
     reasoning_steps: list[dict[str, Any]] = field(default_factory=list)
     adhoc_viz_args: list[dict[str, Any]] = field(default_factory=list)
     response_id: str | None = None
+    stream_ended: bool = False
 
 
 def _handle_text(content: dict[str, Any], acc: _SseAccumulator) -> None:
@@ -187,22 +208,46 @@ 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 failure from iterating `lines` itself (e.g. httpx.RemoteProtocolError/
+            # ReadError from a connection drop mid-stream) is rescued here. A bug in the
+            # processing below must propagate as-is, loudly -- catching it the same way
+            # would blend a real parser bug into the same "error" bucket as a network blip,
+            # with no statusCode payload to tell them apart later.
+            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()
+            continue
+        if not line.startswith(SSE_DATA_PREFIX):
+            continue
+        if current_event == _RESPONSE_ENDED_EVENT:
+            acc.stream_ended = True
             continue
         data_str = line[len(SSE_DATA_PREFIX) :]
         if _METADATA_SYNC_MARKER in data_str:
             raise TransientChatError(
                 f"SSE transient error: {_METADATA_SYNC_MARKER}",
                 status_code=None,
                 detail=None,
+                partial_result=_build_chat_result(acc),
             )
         try:
             event_data = json.loads(data_str)
@@ -213,8 +258,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 +340,21 @@ def send_message(self, conversation_id: str, question: str) -> ChatResult:
             body["options"] = {"reasoningEffort": self._reasoning_effort}
 
         def _do() -> ChatResult:
+            # t0 here, not around send_message(): includes the request/connection/server
+            # setup time a caller actually waits through, but still excludes
+            # _retry_transient's backoff sleep between attempts (harness overhead, not
+            # gen-ai's time), since each retry calls _do() -- and this timer -- fresh.
+            t0 = time.monotonic()
             with self._client.stream("POST", url, json=body, headers=headers) as resp:
                 resp.raise_for_status()
-                return parse_sse_lines(resp.iter_lines())
+                try:
+                    result = parse_sse_lines(resp.iter_lines())
+                except ChatError as exc:
+                    if exc.partial_result is not None:
+                        exc.partial_result.turn_wall_clock_sec = time.monotonic() - t0
+                    raise
+                result.turn_wall_clock_sec = time.monotonic() - t0
+                return result
 
         return _retry_transient(_do, is_retryable=_is_retryable_exc)
 

packages/gooddata-eval/src/gooddata_eval/core/chat/sse_client.py

@@ -100,6 +100,11 @@ class ChatResult(BaseModel):
     reasoning_step_count: int = Field(default=0, alias="reasoningStepCount")
     conversation_id: str | None = Field(default=None, alias="conversationId")
     response_id: str | None = Field(default=None, alias="responseId")
+    # Derived, not a raw server field -- see sse_client.py's _RESPONSE_ENDED_EVENT.
+    stream_ended: bool = False
+    # Set by ChatClient, not from the payload: wall-clock time of the SSE read itself,
+    # excluding retry backoff.
+    turn_wall_clock_sec: float | None = None
 
 
 class SummaryInput(BaseModel):