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,55 +208,77 @@ 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:
-        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):
-            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,
-            )
-        try:
-            event_data = json.loads(data_str)
-        except json.JSONDecodeError:
-            continue
-        if "statusCode" in event_data:
-            code = event_data.get("statusCode")
-            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)
-        if event_data.get("responseId") and not acc.response_id:
-            acc.response_id = event_data["responseId"]
-        item = event_data.get("item")
-        if not item:
-            continue
-        if item.get("responseId") and not acc.response_id:
-            acc.response_id = item["responseId"]
-        role = item.get("role")
-        content: dict[str, Any] = item.get("content") or {}
-        ctype = content.get("type")
-        if role == "assistant":
-            if ctype == "text":
-                _handle_text(content, acc)
-            elif ctype == "multipart":
-                _handle_multipart(content, acc)
-            elif ctype == "reasoning":
-                _handle_reasoning(content, acc)
-            elif ctype == "toolCall":
-                _handle_tool_call(content, acc)
-        elif role == "tool" and ctype == "toolResult":
-            _handle_tool_result(content, acc)
+    current_event = "message"  # SSE default in the absence of an explicit "event: " line
+    try:
+        for raw_line in lines:
+            line = raw_line.decode("utf-8") if isinstance(raw_line, bytes) else raw_line
+            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)
+            except json.JSONDecodeError:
+                continue
+            if "statusCode" in event_data:
+                code = event_data.get("statusCode")
+                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, 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")
+            if not item:
+                continue
+            if item.get("responseId") and not acc.response_id:
+                acc.response_id = item["responseId"]
+            role = item.get("role")
+            content: dict[str, Any] = item.get("content") or {}
+            ctype = content.get("type")
+            if role == "assistant":
+                if ctype == "text":
+                    _handle_text(content, acc)
+                elif ctype == "multipart":
+                    _handle_multipart(content, acc)
+                elif ctype == "reasoning":
+                    _handle_reasoning(content, acc)
+                elif ctype == "toolCall":
+                    _handle_tool_call(content, acc)
+            elif role == "tool" and ctype == "toolResult":
+                _handle_tool_result(content, acc)
+    except ChatError:
+        raise
+    except Exception as exc:
+        # A connection drop mid-stream (httpx.RemoteProtocolError/ReadError from inside
+        # `lines` itself) has no statusCode payload to raise a ChatError from directly --
+        # without this, any tool calls already streamed before the drop would be lost.
+        raise ChatError(f"SSE stream error: {exc}", partial_result=_build_chat_result(acc)) from exc
     return _build_chat_result(acc)
 
 
@@ -293,9 +336,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):