Fix LLMJudgeMetric.evaluate crash inside a running event loop

openagent_eval/metrics/generation/llm_judge.py

@@ -6,6 +6,8 @@
 
 from __future__ import annotations
 
+import asyncio
+import inspect
 import json
 import re
 from dataclasses import dataclass
@@ -25,6 +27,37 @@
 _BARE_DECIMAL_RE = re.compile(r"^\s*(\d+\.?\d*)\s*$")
 
 
+def _run_coroutine_blocking(coro: Any) -> Any:
+    """Run a coroutine to completion from a synchronous context.
+
+    Works whether or not an asyncio event loop is already running. This mirrors
+    the pattern in ``openagent_eval.cicd.plugin``: if there is no running loop
+    we drive the coroutine directly with ``asyncio.run``; if we are already
+    inside a running loop (e.g. the async evaluation pipeline), running the
+    coroutine on that same loop would raise ``RuntimeError`` (or deadlock via
+    ``run_coroutine_threadsafe``), so we execute it in a separate thread that
+    owns its own fresh event loop.
+
+    Args:
+        coro: The coroutine object to run.
+
+    Returns:
+        The value the coroutine resolves to.
+    """
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        # No loop running in this thread — safe to drive the coroutine here.
+        return asyncio.run(coro)
+
+    # A loop is already running in this thread. Hand the coroutine off to a
+    # worker thread with its own event loop so we do not touch the live one.
+    import concurrent.futures
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+        return pool.submit(asyncio.run, coro).result()
+
+
 @dataclass(frozen=True)
 class JudgeCriteria:
     """Criteria for LLM-as-Judge evaluation.
@@ -146,13 +179,12 @@ def evaluate(self, **kwargs: Any) -> MetricResult:
         )
 
         try:
-            import asyncio
-            import inspect
-
             response = self._provider.generate(prompt)
-            # Handle both sync and async providers
+            # Handle both sync and async providers. When ``generate`` is a
+            # coroutine we must drive it to completion even if we are already
+            # inside a running event loop (the async evaluation pipeline).
             if inspect.iscoroutine(response):
-                response = asyncio.get_event_loop().run_until_complete(response)
+                response = _run_coroutine_blocking(response)
             score = self._parse_score(response)
         except Exception as e:
             logger.error("LLM judge failed: {}", e)

tests/unit/test_metrics/test_llm_judge.py

@@ -0,0 +1,109 @@
+"""Unit tests for the LLM-as-Judge metric.
+
+Regression coverage for issue #71: ``LLMJudgeMetric.evaluate`` must resolve an
+async provider's coroutine whether it is called from a plain synchronous
+context OR from inside an already-running asyncio event loop. The previous
+implementation used ``asyncio.get_event_loop().run_until_complete(...)``, which
+raises ``RuntimeError`` when a loop is already running.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from openagent_eval.metrics.generation.llm_judge import (
+    RELEVANCY_CRITERIA,
+    LLMJudgeMetric,
+)
+
+
+class _AsyncProvider:
+    """Minimal async LLM provider whose ``generate`` returns a coroutine."""
+
+    name = "async_stub"
+    description = "async stub provider"
+
+    def __init__(self, reply: str = "0.9") -> None:
+        self._reply = reply
+        self.calls = 0
+
+    async def generate(self, prompt: str, **kwargs: object) -> str:
+        self.calls += 1
+        return self._reply
+
+    async def get_token_count(self, text: str) -> int:  # pragma: no cover
+        return len(text.split())
+
+
+class _SyncProvider:
+    """Provider whose ``generate`` returns a plain string (not a coroutine)."""
+
+    name = "sync_stub"
+    description = "sync stub provider"
+
+    def __init__(self, reply: str = "0.8") -> None:
+        self._reply = reply
+
+    def generate(self, prompt: str, **kwargs: object) -> str:
+        return self._reply
+
+    async def get_token_count(self, text: str) -> int:  # pragma: no cover
+        return len(text.split())
+
+
+def _judge(provider: object) -> LLMJudgeMetric:
+    return LLMJudgeMetric(provider=provider, criteria=RELEVANCY_CRITERIA)
+
+
+def test_evaluate_async_provider_from_sync_context() -> None:
+    """Async provider resolved from a plain sync call (no running loop)."""
+    provider = _AsyncProvider(reply="0.9")
+    result = _judge(provider).evaluate(
+        premise="What color is the sky?",
+        hypothesis="The sky is blue.",
+    )
+
+    assert provider.calls == 1
+    assert result.score == 0.9
+    assert "error" not in result.metadata
+    assert result.metadata["provider"] == "async_stub"
+
+
+def test_evaluate_async_provider_inside_running_loop() -> None:
+    """Regression for #71: evaluate() must work inside a running event loop.
+
+    Before the fix this raised RuntimeError ("... event loop is already
+    running"), which was swallowed by the generic except and returned 0.0.
+    """
+
+    async def driver() -> object:
+        # We are now inside asyncio.run(...), i.e. a running event loop —
+        # exactly the async-pipeline scenario described in issue #71.
+        return _judge(_AsyncProvider(reply="0.75")).evaluate(
+            premise="What color is the sky?",
+            hypothesis="The sky is blue.",
+        )
+
+    result = asyncio.run(driver())
+
+    # The coroutine must have been driven to completion, not error out.
+    assert result.score == 0.75, result.reason
+    assert result.reason.startswith("LLM judge (relevancy)")
+
+
+def test_evaluate_sync_provider_from_sync_context() -> None:
+    """A synchronous provider (non-coroutine generate) still works."""
+    result = _judge(_SyncProvider(reply="0.8")).evaluate(
+        premise="What color is the sky?",
+        hypothesis="The sky is blue.",
+    )
+
+    assert result.score == 0.8
+    assert result.reason.startswith("LLM judge (relevancy)")
+
+
+def test_evaluate_missing_inputs_returns_zero() -> None:
+    """Missing premise/hypothesis short-circuits to a 0.0 result."""
+    result = _judge(_AsyncProvider()).evaluate(premise="", hypothesis="")
+    assert result.score == 0.0
+    assert result.reason == "Missing premise or hypothesis"