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

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/chat/sse_client.py

@@ -28,18 +28,38 @@
 _log = logging.getLogger(__name__)
 
 SSE_DATA_PREFIX = "data: "
+SSE_EVENT_PREFIX = "event: "
+# gen-ai's last event, only if at least one item was already emitted (conversations_controller.py).
+_RESPONSE_ENDED_EVENT = "response_ended"
 
 _RETRYABLE_STATUS_CODES: frozenset[int] = frozenset({429, 502, 503, 504})
 _METADATA_SYNC_MARKER = "METADATA_SYNC_IN_PROGRESS"
 
 
 class ChatError(RuntimeError):
-    """Non-retryable error reported by the chat SSE stream."""
+    """Non-retryable error reported by the chat SSE stream.
 
-    def __init__(self, message: str, *, status_code: int | None = None, detail: str | None = None) -> None:
+    ``partial_result`` carries whatever the accumulator captured before this error fired
+    (tool calls included) -- an error event ends the stream before ``_build_chat_result``
+    ever runs, so without this a caller has no way to see, e.g., that KDA's own tool calls
+    already succeeded before an unrelated later error (a failed final-summary generation)
+    killed the turn. Callers must not assume it's complete: fields normally filled in only
+    at the very end of the stream (``stream_ended``, in particular) reflect the state at
+    the moment of the error, not a genuinely finished turn.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        status_code: int | None = None,
+        detail: str | None = None,
+        partial_result: ChatResult | None = None,
+    ) -> None:
         super().__init__(message)
         self.status_code = status_code
         self.detail = detail
+        self.partial_result = partial_result
 
 
 class TransientChatError(ChatError):
@@ -109,6 +129,7 @@ class _SseAccumulator:
     reasoning_steps: list[dict[str, Any]] = field(default_factory=list)
     adhoc_viz_args: list[dict[str, Any]] = field(default_factory=list)
     response_id: str | None = None
+    stream_ended: bool = False
 
 
 def _handle_text(content: dict[str, Any], acc: _SseAccumulator) -> None:
@@ -187,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/models.py

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