feat(gooddata-eval): make the K verdict a gate, pass@K or pass^K

Author: myhoai

packages/gooddata-eval/README.md

@@ -90,7 +90,8 @@ Both provider name and provider id are accepted as the prefix.
 
 | Flag | Default | Description |
 |---|---|---|
-| `--runs K` | `2` | Independent runs per item (pass@K). An item passes if any run passes. |
+| `--runs K` | `2` | Independent runs per item. |
+| `--gate` | `any` | Which verdict decides an item: `any` = pass@K (a run passing is enough), `power` = pass^K (every run must pass, so the verdict measures stability). Identical at `--runs 1`. Agentic kinds only. |
 | `--concurrency K` | `1` | Number of items evaluated concurrently. `1` = sequential (default). Increase to load-test the agent under simultaneous requests. Progress output interleaves when K > 1. |
 | `--reasoning-effort LEVEL` | server default | `LOW`, `MEDIUM` or `HIGH`, sent as `options.reasoningEffort` on every chat message. Requires the `enableGenAiReasoningEffort` feature flag on the target organization — without it the server ignores the value. Applies to chat items only; `dashboard_summary` items go through the summary endpoint, which has no such option. |
 
@@ -241,8 +242,14 @@ Without `[llm-judge]`, those items are **skipped**.
 
 | Score | Description |
 |---|---|
-| `pass_at_k` | 1 if any of the K runs passed strict checks, else 0. |
+| `pass_at_k` | 1 if **any** of the K runs passed strict checks, else 0. |
+| `pass_power_k` | 1 only if **every** one of the K runs passed. Agentic kinds only. |
+| `gate_passed` | The verdict that decided the item: `pass_at_k` under `--gate any`, `pass_power_k` under `--gate power`. Agentic kinds only. |
 | `quality_score` | Fraction of strict check flags that are `True` (0.0–1.0). Shown in CLI as a percentage. |
 | `value_score` | Weighted blend: 0.6 × quality + 0.2 × speed (speed = max(0, 1 − latency/60s)). |
 | `latency_s` | Average per-run latency in seconds. |
 | `provider_type` | Model vendor + gateway label (e.g. `ANTHROPIC`, `BEDROCK/ANTHROPIC`, `AZURE/OPENAI`). Stored in Langfuse trace metadata and tags. |
+
+Score names carry no K; K and the gate are on the dataset-run metadata as `eval_k` and
+`eval_gate`. `agentic_visualization` also still writes its historic `pass_at_{K}` /
+`pass_power_{K}` pair. `agentic_conversation` has no K gate.

packages/gooddata-eval/src/gooddata_eval/cli/agentic_runner.py

@@ -6,6 +6,7 @@
 import time
 from typing import Any, TypedDict
 
+from gooddata_eval.core.agentic._gate import DEFAULT_GATE, EvalGate, normalize_gate
 from gooddata_eval.core.agentic._langfuse import make_langfuse_client
 from gooddata_eval.core.agentic.alert_skill import evaluate_agentic_alert_skill
 from gooddata_eval.core.agentic.conversation import ConversationFixture, evaluate_agentic_conversation
@@ -85,8 +86,12 @@ def _dispatch_agentic(
     run_ts: str,
     model_version_override: str | None,
     reasoning_effort: ReasoningEffort | None = None,
+    gate: EvalGate = DEFAULT_GATE,
 ) -> None:
-    """Call the appropriate evaluate_agentic_* function for the item's test_kind."""
+    """Call the appropriate evaluate_agentic_* function for the item's test_kind.
+
+    `gate` reaches every kind except agentic_conversation, which has no K to gate over.
+    """
     kind = item.test_kind
     eo = item.expected_output
     lf_kw: _LfKw = {
@@ -106,6 +111,7 @@ def _dispatch_agentic(
             question=item.question,
             expected_outputs=_parse_visualization_expected(eo),
             k=k,
+            gate=gate,
             **lf_kw,
         )
     elif kind == "agentic_metric_skill":
@@ -116,6 +122,7 @@ def _dispatch_agentic(
             question=item.question,
             expected_output=eo if isinstance(eo, (dict, list)) else {},
             k=k,
+            gate=gate,
             **lf_kw,
         )
     elif kind == "agentic_alert_skill":
@@ -126,6 +133,7 @@ def _dispatch_agentic(
             question=item.question,
             expected_output=eo if isinstance(eo, dict) else {},
             k=k,
+            gate=gate,
             **lf_kw,
         )
     elif kind == "agentic_search":
@@ -139,6 +147,7 @@ def _dispatch_agentic(
             question=item.question,
             expected_tool_call=expected_args,
             k=k,
+            gate=gate,
             **lf_kw,
         )
     elif kind == "agentic_general_question":
@@ -149,6 +158,7 @@ def _dispatch_agentic(
             question=item.question,
             expected_output=eo if isinstance(eo, str) else str(eo),
             k=k,
+            gate=gate,
             **lf_kw,
         )
     elif kind == "agentic_guardrail":
@@ -159,6 +169,7 @@ def _dispatch_agentic(
             question=item.question,
             expected_output=eo if isinstance(eo, str) else str(eo),
             k=k,
+            gate=gate,
             **lf_kw,
         )
     elif kind == "agentic_kda_skill":
@@ -169,6 +180,7 @@ def _dispatch_agentic(
             question=item.question,
             expected_output=eo if isinstance(eo, dict) else {},
             k=k,
+            gate=gate,
             **lf_kw,
         )
     elif kind == "agentic_conversation":
@@ -193,6 +205,7 @@ def run_agentic_items(
     k: int = 2,
     model_version: str | None = None,
     reasoning_effort: ReasoningEffort | None = None,
+    gate: EvalGate = DEFAULT_GATE,
     use_langfuse: bool = False,
     run_ts: str,
     on_item_start: Any = None,
@@ -201,7 +214,7 @@ def run_agentic_items(
     """Run agentic items through evaluate_agentic_* and return an EvalReport."""
     langfuse = make_langfuse_client() if use_langfuse else None
 
-    report = EvalReport(model=model_version)
+    report = EvalReport(model=model_version, gate=normalize_gate(gate))
     total = len(items)
 
     for index, item in enumerate(items, start=1):
@@ -219,11 +232,13 @@ def run_agentic_items(
         )
         t0 = time.perf_counter()
         try:
-            _dispatch_agentic(item, host, token, workspace_id, k, langfuse, run_ts, model_version, reasoning_effort)
-            item_report.pass_at_k = True
+            _dispatch_agentic(
+                item, host, token, workspace_id, k, langfuse, run_ts, model_version, reasoning_effort, gate
+            )
+            item_report.gate_passed = True
             item_report.runs = k
         except AssertionError as exc:
-            item_report.pass_at_k = False
+            item_report.gate_passed = False
             item_report.runs = k
             print(f"[agentic] {item.id} FAIL: {exc}", flush=True)
         except Exception as exc:

packages/gooddata-eval/src/gooddata_eval/cli/main.py

@@ -15,7 +15,7 @@
 
 from gooddata_eval.cli.agentic_runner import AGENTIC_TEST_KINDS, run_agentic_items
 from gooddata_eval.core.chat.sse_client import ChatClient
-from gooddata_eval.core.config import ReasoningEffort, RunConfig
+from gooddata_eval.core.config import DEFAULT_GATE, EvalGate, ReasoningEffort, RunConfig, normalize_gate
 from gooddata_eval.core.connection import ConnectionError_, resolve_connection
 from gooddata_eval.core.dataset.local import load_local_dataset
 from gooddata_eval.core.langfuse.sink import LangfuseSink
@@ -89,7 +89,15 @@ def _build_parser() -> argparse.ArgumentParser:
             "Default: workspace's current active model."
         ),
     )
-    run.add_argument("--runs", type=int, default=2, help="Independent runs per item (pass@K). Default 2.")
+    run.add_argument("--runs", type=int, default=2, help="Independent runs per item. Default 2.")
+    run.add_argument(
+        "--gate",
+        choices=get_args(EvalGate),
+        default=DEFAULT_GATE,
+        help="Which verdict decides an item: 'any' = pass@K (a run passing is enough, the "
+        "default and historic behaviour), 'power' = pass^K (every run must pass, so the verdict "
+        "measures stability). Identical at --runs 1. Agentic kinds only.",
+    )
     run.add_argument(
         "--concurrency",
         type=int,
@@ -178,7 +186,7 @@ def on_item_done(index: int, total: int, report: ItemReport) -> None:
             tag = "[yellow]SKIP[/yellow]"
         elif report.error:
             tag = "[red]ERR [/red]"
-        elif report.pass_at_k:
+        elif report.gate_passed:
             tag = "[green]PASS[/green]"
         else:
             tag = "[red]FAIL[/red]"
@@ -341,6 +349,7 @@ def on_langfuse_item_done(
                     token=config.token,
                     workspace_id=config.workspace_id,
                     k=config.runs,
+                    gate=config.gate,
                     model_version=resolved.model_id,
                     reasoning_effort=config.reasoning_effort,
                     use_langfuse=config.log_to_langfuse,
@@ -449,6 +458,7 @@ def main(argv: list[str] | None = None) -> int:
             kind=args.kind,
             preserve_failed=args.preserve_failed,
             reasoning_effort=args.reasoning_effort,
+            gate=normalize_gate(args.gate),
         )
         return _run(config)
     except (

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

@@ -0,0 +1,69 @@
+# (C) 2026 GoodData Corporation
+"""Which of pass@K / pass^K decides an item, and how both reach Langfuse."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from gooddata_eval.core.config import DEFAULT_GATE, EvalGate, normalize_gate
+
+__all__ = [
+    "DEFAULT_GATE",
+    "EvalGate",
+    "gate_failure_note",
+    "gate_label",
+    "gate_passed",
+    "log_gate_scores",
+    "normalize_gate",
+    "stamp_gate_metadata",
+]
+
+
+def gate_passed(gate: str | None, *, pass_at_k: bool, pass_power_k: bool) -> bool:
+    """Whether the item passes under ``gate``."""
+    return pass_power_k if normalize_gate(gate) == "power" else pass_at_k
+
+
+def gate_label(gate: str | None, k: int) -> str:
+    """Short name for the gate, e.g. ``pass^3`` or ``pass@2``."""
+    return f"pass^{k}" if normalize_gate(gate) == "power" else f"pass@{k}"
+
+
+def gate_failure_note(gate: str | None, run_passes: tuple[bool, ...] | list[bool]) -> str:
+    """Gate and how many runs met it, for the assertion message.
+
+    Needed because the message body shows the BEST run, which under pass^K can be a run that
+    passed — so the reported diff on its own looks like a pass.
+    """
+    k = len(run_passes)
+    passed = sum(1 for p in run_passes if p)
+    label = gate_label(gate, k)
+    if normalize_gate(gate) == "power" and passed:
+        return f"Gate {label} failed: {passed}/{k} runs passed — unstable, not a clean failure."
+    return f"Gate {label} failed: {passed}/{k} runs passed."
+
+
+def log_gate_scores(langfuse: Any, trace_id: str | None, *, gate: str | None, pass_at_k: bool, pass_power_k: bool):
+    """Log both candidate verdicts and the one that decided.
+
+    The names carry no K on purpose: `pass_at_2` becomes `pass_at_3` the moment K changes,
+    splitting every Langfuse view built on the old name.
+    """
+    from gooddata_eval.core.agentic._langfuse import score_safe  # noqa: PLC0415
+
+    score_safe(langfuse, trace_id, name="pass_at_k", value=pass_at_k, data_type="BOOLEAN")
+    score_safe(langfuse, trace_id, name="pass_power_k", value=pass_power_k, data_type="BOOLEAN")
+    score_safe(
+        langfuse,
+        trace_id,
+        name="gate_passed",
+        value=gate_passed(gate, pass_at_k=pass_at_k, pass_power_k=pass_power_k),
+        data_type="BOOLEAN",
+    )
+
+
+def stamp_gate_metadata(metadata: dict, *, k: int, gate: str | None) -> dict:
+    """Record K and the gate on the dataset-run metadata (mutates and returns it)."""
+    metadata["eval_k"] = k
+    metadata["eval_gate"] = normalize_gate(gate)
+    return metadata

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

@@ -12,6 +12,14 @@
 from gooddata_sdk import GoodDataSdk
 
 from gooddata_eval.core.agentic._catalog import CatalogMetricAlert
+from gooddata_eval.core.agentic._gate import (
+    DEFAULT_GATE,
+    EvalGate,
+    gate_failure_note,
+    gate_passed,
+    log_gate_scores,
+    stamp_gate_metadata,
+)
 from gooddata_eval.core.chat.sse_client import ChatClient
 from gooddata_eval.core.config import ReasoningEffort
 from gooddata_eval.core.models import ToolCallEvent
@@ -313,6 +321,10 @@ class AgenticAlertSummary:
     pass_power_k: bool
     best: AlertRunResult
 
+    @property
+    def run_passes(self) -> tuple[bool, ...]:
+        return tuple(r.eval.strict_pass for r in self.run_results)
+
 
 def _case_insensitive_get(d: dict, *keys: str) -> Any:
     """Look up a value by key, preferring an exact match then a case-insensitive one."""
@@ -548,6 +560,7 @@ def evaluate_agentic_alert_skill(
     question: str,
     expected_output: dict,
     k: int = _DEFAULT_K,
+    gate: EvalGate = DEFAULT_GATE,
     max_iterations: int = _DEFAULT_MAX_ITERATIONS,
     initial_conversation_id: str | None = None,
     langfuse: object | None = None,
@@ -598,6 +611,7 @@ def evaluate_agentic_alert_skill(
             run_metadata_extra,
             reasoning_effort,
         )
+        stamp_gate_metadata(run_metadata, k=len(summary.run_results), gate=gate)
         traces_by_conv = find_traces_per_conversation(
             langfuse,
             [r.conversation_id for r in summary.run_results],
@@ -620,6 +634,13 @@ def evaluate_agentic_alert_skill(
             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")
+                log_gate_scores(
+                    langfuse,
+                    tid,
+                    gate=gate,
+                    pass_at_k=summary.pass_at_k,
+                    pass_power_k=summary.pass_power_k,
+                )
                 log_quality_and_value_scores(
                     langfuse,
                     tid,
@@ -628,11 +649,12 @@ def evaluate_agentic_alert_skill(
                     cost_usd=pt.total_cost if pt else None,
                 )
 
-    if not summary.pass_at_k:
+    if not gate_passed(gate, pass_at_k=summary.pass_at_k, pass_power_k=summary.pass_power_k):
+        gate_note = gate_failure_note(gate, summary.run_passes)
         best = summary.best
         ev = best.eval
         raise AlertSkillAssertionError(
-            f"Alert skill assertion failed. strict_pass={ev.strict_pass}. "
+            f"Alert skill assertion failed. {gate_note} strict_pass={ev.strict_pass}. "
             f"alert_created={ev.alert_created}, operator_correct={ev.operator_correct}, "
             f"threshold_correct={ev.threshold_correct}, trigger_correct={ev.trigger_correct}, "
             f"filters_correct={ev.filters_correct}, metric_correct={ev.metric_correct}, "

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

@@ -5,6 +5,14 @@
 
 from dataclasses import dataclass
 
+from gooddata_eval.core.agentic._gate import (
+    DEFAULT_GATE,
+    EvalGate,
+    gate_failure_note,
+    gate_passed,
+    log_gate_scores,
+    stamp_gate_metadata,
+)
 from gooddata_eval.core.chat.sse_client import ChatClient
 from gooddata_eval.core.config import ReasoningEffort
 from gooddata_eval.core.evaluators._llm_judge import LLMJudge
@@ -63,6 +71,10 @@ class AgenticGeneralQuestionSummary:
     pass_power_k: bool
     best: GeneralQuestionResult
 
+    @property
+    def run_passes(self) -> tuple[bool, ...]:
+        return tuple(r.passed for r in self.run_results)
+
 
 def run_agentic_general_question(
     host: str,
@@ -148,6 +160,7 @@ def evaluate_agentic_general_question(
     question: str,
     expected_output: str,
     k: int = _DEFAULT_K,
+    gate: EvalGate = DEFAULT_GATE,
     initial_conversation_id: str | None = None,
     langfuse: object | None = None,
     dataset_item_id: str = "",
@@ -196,6 +209,7 @@ def evaluate_agentic_general_question(
             run_metadata_extra,
             reasoning_effort,
         )
+        stamp_gate_metadata(run_metadata, k=len(summary.run_results), gate=gate)
         traces_by_conv = find_traces_per_conversation(
             langfuse,
             [r.conversation_id for r in summary.run_results],
@@ -208,6 +222,13 @@ def evaluate_agentic_general_question(
             with observe(langfuse, pt.id if pt else None, dataset_item_id, run_name, run_metadata) as tid:
                 score_safe(langfuse, tid, name="general_question_pass", value=float(run.passed), data_type="BOOLEAN")
                 score_safe(langfuse, tid, name="llm_judge_score", value=run.llm_judge_score, data_type="NUMERIC")
+                log_gate_scores(
+                    langfuse,
+                    tid,
+                    gate=gate,
+                    pass_at_k=summary.pass_at_k,
+                    pass_power_k=summary.pass_power_k,
+                )
                 log_quality_and_value_scores(
                     langfuse,
                     tid,
@@ -216,8 +237,9 @@ def evaluate_agentic_general_question(
                     cost_usd=pt.total_cost if pt else None,
                 )
 
-    if not summary.pass_at_k:
+    if not gate_passed(gate, pass_at_k=summary.pass_at_k, pass_power_k=summary.pass_power_k):
+        gate_note = gate_failure_note(gate, summary.run_passes)
         best = summary.best
         raise GeneralQuestionAssertionError(
-            f"General question assertion failed. passed={best.passed}. Reasoning: {best.reasoning}"
+            f"General question assertion failed. {gate_note} passed={best.passed}. Reasoning: {best.reasoning}"
         )