fix(gooddata-eval): let --gate decide an anomaly item, not pass@K alone

`run_agentic_anomaly_detection` already computed `pass_power_k`; the evaluator
never read it, asking `if not summary.pass_at_k` for the verdict, so `--gate
power` on a flaky item reported a pass on the strength of one good run out of K.

Wired the same way as the other multi-run kinds: the evaluator takes `gate`, the
dispatch passes it, `gate_passed(...)` decides, the gate's note goes into the
assertion message -- which matters because the body describes the BEST run, and
under pass^K that can be one that passed -- and pass@K/pass^K/gate_passed reach
Langfuse alongside the run metadata stamp.

The default is unchanged: without --gate this is pass@K exactly as before, pinned
by a test beside the power case. The power test fails against the previous code.

Found by CodeRabbit on #1798. Forecasting is fixed there; what-if, which is
already on master, in #1831.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
(cherry picked from commit 5632da8)

Author: Tomkess

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

@@ -326,6 +326,7 @@ def _dispatch_agentic(
             question=item.question,
             expected_output=eo if isinstance(eo, dict) else {},
             k=k,
+            gate=gate,
             agent_id=agent_id,
             **lf_kw,
         )

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

@@ -32,6 +32,14 @@
 from typing import Any
 
 from gooddata_eval.core.agentic._failed_runs import build_failed_runs
+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.agentic._trace_linker import (
     RunIdentity,
     RunTraceContext,
@@ -503,6 +511,7 @@ def evaluate_agentic_anomaly_detection(
     k: int = _DEFAULT_K,
     max_iterations: int = _DEFAULT_MAX_ITERATIONS,
     initial_conversation_id: str | None = None,
+    gate: EvalGate = DEFAULT_GATE,
     agent_id: str | None = None,
     langfuse: object | None = None,
     dataset_item_id: str = "",
@@ -533,6 +542,8 @@ def evaluate_agentic_anomaly_detection(
         window_end = utc_now()
 
         def _write_scores(ctx: RunTraceContext) -> None:
+            stamp_gate_metadata(ctx.run_metadata, k=len(summary.run_results), gate=gate)
+
             for run_idx, run in enumerate(summary.run_results):
                 pt = ctx.trace(run.conversation_id)
                 ev = run.evaluation
@@ -558,6 +569,7 @@ def _write_scores(ctx: RunTraceContext) -> None:
                 with ctx.observe(pt, run_idx) as tid:
                     for score_name, value in strict_checks.items():
                         ctx.score(tid, name=score_name, value=float(value), data_type="BOOLEAN")
+                    log_gate_scores(ctx, tid, gate=gate, pass_at_k=summary.pass_at_k, pass_power_k=summary.pass_power_k)
                     ctx.quality(
                         tid,
                         strict_checks=strict_checks,
@@ -610,9 +622,10 @@ def _write_scores(ctx: RunTraceContext) -> None:
     )
     runs_passed = sum(1 for r in summary.run_results if r.evaluation.strict_pass)
 
-    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, runs_passed, len(summary.run_results))
         message = (
-            f"Anomaly detection assertion failed. strict_pass={ev.strict_pass} "
+            f"Anomaly detection assertion failed. {gate_note} strict_pass={ev.strict_pass} "
             f"(triggered={ev.triggered}, executed={ev.executed}, success={ev.success}, "
             f"turn_completed={ev.turn_completed}, metric_correct={ev.metric_correct}, "
             f"granularity_correct={ev.granularity_correct}). "

packages/gooddata-eval/tests/test_agentic_anomaly_detection.py

@@ -410,6 +410,8 @@ def __init__(self):
         self.scores: dict[str, float] = {}
         # Not named `quality`: the method below would overwrite itself on first call.
         self.quality_call: dict = {}
+        # stamp_gate_metadata writes the gate and K here, the way every other kind does.
+        self.run_metadata: dict = {}
 
     def trace(self, _conversation_id):
         return None
@@ -479,3 +481,50 @@ def test_cost_is_reported_even_when_the_tool_was_never_reached():
     ev.triggered hid that and understated what the item cost."""
     ctx = _scored(_EXPECTED, calls=[])
     assert "cost_usd" in ctx.quality_call
+
+
+# ── the gate decides the verdict, not pass@K alone ──────────────────────────
+#
+# run_agentic_anomaly_detection computed pass^K and the evaluator never read it, so
+# `--gate power` on a flaky item passed on the strength of one good run out of K -- the
+# precise case the gate exists to catch.
+
+
+def _two_runs(gate=None):
+    """One passing run then one failing one, which pass@K clears and pass^K must not."""
+    client = MagicMock()
+    client.create_conversation.side_effect = ["conv-1", "conv-2"]
+    client.send_message.side_effect = [_chat(_pair()), *[_chat([])] * 8]
+    kwargs = {"gate": gate} if gate is not None else {}
+    with (
+        patch(f"{_MODULE}.ChatClient", return_value=client),
+        patch(f"{_MODULE}.generate_simulated_anomaly_response", return_value="use spend, monthly"),
+    ):
+        return evaluate_agentic_anomaly_detection(
+            host="http://h",
+            token="tok",
+            workspace_id="ws1",
+            question="Any anomalies in monthly spend?",
+            expected_output=_EXPECTED,
+            k=2,
+            **kwargs,
+        )
+
+
+def test_power_gate_fails_an_item_where_only_some_runs_passed():
+    with pytest.raises(AnomalyDetectionAssertionError) as excinfo:
+        _two_runs(gate="power")
+    assert "pass^2" in str(excinfo.value)
+    assert excinfo.value.runs_passed == 1
+    assert excinfo.value.runs_effective == 2
+
+
+def test_any_gate_still_passes_the_same_item():
+    """The default is unchanged: one passing run out of K clears pass@K."""
+    outcome = _two_runs(gate="any")
+    assert outcome.runs_passed == 1
+    assert outcome.runs_effective == 2
+
+
+def test_the_default_gate_is_pass_at_k():
+    assert _two_runs().runs_passed == 1