🐛 Fix(evaluation): run trials in runtime service (#3954)

backend/apps/agent_evaluation_app.py

@@ -19,9 +19,9 @@
     get_evaluation_stats_impl,
     list_agent_evaluation_cases_impl,
     list_agent_evaluations_by_agent_impl,
-    trial_run_evaluator_impl,
 )
 from services.evaluation_report_service import generate_agent_evaluation_report_impl
+from services.runtime_proxy_service import forward_agent_evaluation_trial_run
 from utils.auth_utils import get_current_user_id, get_current_user_info
 
 
@@ -568,15 +568,15 @@ async def trial_run_api(
     """
     try:
         user_id, tenant_id = get_current_user_id(authorization)
-        result = await trial_run_evaluator_impl(
-            tenant_id=tenant_id,
-            user_id=user_id,
+        result = await forward_agent_evaluation_trial_run(
             agent_id=payload.agent_id,
             agent_version_no=payload.agent_version_no,
             query=payload.query,
             judge_model_id=payload.judge_model_id,
             evaluator_ids=payload.evaluator_ids,
             language=payload.language,
+            user_id=user_id,
+            tenant_id=tenant_id,
         )
         logger.info(
             "trial_run_api OK: tenant=%s user=%s agent_id=%s version=%s "

backend/apps/agent_evaluation_runtime_app.py

@@ -5,6 +5,7 @@
 from typing import Annotated
 
 from fastapi import APIRouter, Header, HTTPException
+from nexent.core.concurrency import ManagedTaskSpec
 from pydantic import BaseModel, Field
 
 from consts.evaluation_status import EvalRunStatus
@@ -13,7 +14,6 @@
     claim_agent_evaluation_run,
     get_agent_evaluation,
 )
-from nexent.core.concurrency import ManagedTaskSpec
 from services.thread_lifecycle_service import runtime_thread_manager
 from utils.auth_utils import verify_internal_runtime_jwt
 
@@ -28,13 +28,59 @@ class EvaluationRunRequest(BaseModel):
     agent_evaluation_id: int = Field(gt=0)
 
 
+class TrialRunRequest(BaseModel):
+    """Payload used by Config service for a non-persistent trial evaluation."""
+
+    agent_id: int
+    agent_version_no: int = 1
+    query: str
+    judge_model_id: int
+    evaluator_ids: list[int] | None = None
+    language: str = "zh"
+
+
 def _load_evaluation_executor():
     """Load the evaluation service only when a runtime run is dispatched."""
     from services.agent_evaluation_service import execute_agent_evaluation_run
 
     return execute_agent_evaluation_run
 
 
+def _load_trial_executor():
+    """Load the trial executor only when Runtime receives a trial request."""
+    from services.agent_evaluation_service import trial_run_evaluator_impl
+
+    return trial_run_evaluator_impl
+
+
+@router.post("/trial-run", include_in_schema=False)
+async def trial_run_evaluation_api(
+    payload: TrialRunRequest,
+    authorization: Annotated[str | None, Header()] = None,
+):
+    """Run one ad-hoc evaluation in the Runtime process."""
+    try:
+        user_id, tenant_id = verify_internal_runtime_jwt(authorization)
+    except Exception as exc:
+        logger.warning("Rejected unauthenticated trial evaluation: %s", exc)
+        raise HTTPException(
+            status_code=HTTPStatus.UNAUTHORIZED,
+            detail="Invalid internal runtime authorization",
+        ) from exc
+
+    trial_run_evaluator_impl = _load_trial_executor()
+    return await trial_run_evaluator_impl(
+        tenant_id=tenant_id,
+        user_id=user_id,
+        agent_id=payload.agent_id,
+        agent_version_no=payload.agent_version_no,
+        query=payload.query,
+        judge_model_id=payload.judge_model_id,
+        evaluator_ids=payload.evaluator_ids,
+        language=payload.language,
+    )
+
+
 @router.post("/run", include_in_schema=False, status_code=HTTPStatus.ACCEPTED)
 async def dispatch_evaluation_run_api(
     payload: EvaluationRunRequest,

backend/services/agent_evaluation_service.py

@@ -62,7 +62,10 @@
 from database.evaluator_db import get_evaluator
 from management.services.agent.service import prepare_agent_run
 from services.evaluation_set_service import resolve_latest_published_version_no
-from services.thread_lifecycle_service import runtime_thread_manager
+from services.thread_lifecycle_service import (
+    config_thread_manager,
+    runtime_thread_manager,
+)
 from utils.llm_utils import call_llm_for_system_prompt
 from utils.prompt_template_utils import get_prompt_template
 
@@ -1054,12 +1057,12 @@ def _check_run_limits(tenant_id: str) -> None:
 def _run_in_background(
     fn, *fn_args, tenant_id, user_id, agent_evaluation_id, language="zh"
 ):
-    """Submit fn to the Runtime evaluation lane and attach failure cleanup."""
-    execution = runtime_thread_manager.submit(
+    """Submit Config-owned preparation or dispatch work in the background."""
+    execution = config_thread_manager.submit(
         "evaluation",
         ManagedTaskSpec(
             task_name="agent-evaluation-run",
-            owner="runtime",
+            owner="config",
             run_id=str(agent_evaluation_id),
         ),
         fn,

backend/services/runtime_proxy_service.py

@@ -29,6 +29,7 @@
 _STREAM_TIMEOUT = httpx.Timeout(connect=10.0, read=None, write=30.0, pool=10.0)
 _REQUEST_TIMEOUT = httpx.Timeout(connect=10.0, read=30.0, write=30.0, pool=10.0)
 _EVALUATION_DISPATCH_TIMEOUT = httpx.Timeout(connect=10.0, read=30.0, write=30.0, pool=10.0)
+_EVALUATION_TRIAL_TIMEOUT = httpx.Timeout(connect=10.0, read=None, write=30.0, pool=10.0)
 _RUNTIME_SERVICE_UNAVAILABLE_MESSAGE = "Runtime service is unavailable"
 
 
@@ -101,6 +102,58 @@ def dispatch_agent_evaluation_run(
     return payload
 
 
+async def forward_agent_evaluation_trial_run(
+    *,
+    agent_id: int,
+    agent_version_no: int,
+    query: str,
+    judge_model_id: int,
+    evaluator_ids: list[int] | None,
+    language: str,
+    user_id: str,
+    tenant_id: str,
+) -> dict:
+    """Execute an ad-hoc evaluation in the Runtime process and return its result."""
+    try:
+        async with create_httpx_client(
+            headers=_authorization_headers(user_id, tenant_id),
+            timeout=_EVALUATION_TRIAL_TIMEOUT,
+        ) as client:
+            response = await client.post(
+                _runtime_url("/agent-evaluations/internal/trial-run"),
+                json={
+                    "agent_id": agent_id,
+                    "agent_version_no": agent_version_no,
+                    "query": query,
+                    "judge_model_id": judge_model_id,
+                    "evaluator_ids": evaluator_ids,
+                    "language": language,
+                },
+            )
+    except httpx.TimeoutException as exc:
+        raise RuntimeServiceTimeoutError("Runtime trial evaluation timed out") from exc
+    except httpx.RequestError as exc:
+        raise RuntimeServiceUnavailableError(_RUNTIME_SERVICE_UNAVAILABLE_MESSAGE) from exc
+
+    if response.status_code >= 400:
+        raise RuntimeUpstreamError(
+            status_code=response.status_code,
+            content=response.content,
+            headers=_forwarded_headers(response.headers),
+        )
+    try:
+        payload = response.json()
+    except ValueError as exc:
+        raise RuntimeServiceUnavailableError(
+            "Runtime trial evaluation response is not valid JSON"
+        ) from exc
+    if not isinstance(payload, dict):
+        raise RuntimeServiceUnavailableError(
+            "Runtime trial evaluation response is not a JSON object"
+        )
+    return payload
+
+
 async def forward_agent_run(
     agent_request: AgentRequest,
     user_id: str,

test/backend/app/test_agent_evaluation_app.py

@@ -92,6 +92,7 @@ def _register_package(name: str) -> types.ModuleType:
     "services",
     "services.agent_evaluation_service",
     "services.evaluation_report_service",
+    "services.runtime_proxy_service",
     "database",
     "database.agent_evaluation_db",
     "utils",
@@ -100,6 +101,7 @@ def _register_package(name: str) -> types.ModuleType:
     _register_package(_name)
 sys.modules["services.agent_evaluation_service"] = MagicMock(name="agent_eval_svc")
 sys.modules["services.evaluation_report_service"] = MagicMock(name="report_svc")
+sys.modules["services.runtime_proxy_service"] = MagicMock(name="runtime_proxy_svc")
 sys.modules["database.agent_evaluation_db"] = MagicMock(name="agent_eval_db")
 sys.modules["utils.auth_utils"] = MagicMock(name="auth_utils")
 
@@ -179,7 +181,7 @@ def _mock_impls(**overrides):
             return_value={"items": [], "total": 0}
         ),
         "list_agent_evaluations_by_agent_impl": MagicMock(return_value=[{"id": 1}]),
-        "trial_run_evaluator_impl": AsyncMock(return_value={"result": "ok"}),
+        "forward_agent_evaluation_trial_run": AsyncMock(return_value={"result": "ok"}),
         "generate_agent_evaluation_report_impl": MagicMock(
             return_value=(b"%PDF-1.4 fake", 0)
         ),
@@ -617,7 +619,7 @@ def test_runs_trial(self, client):
         )
         assert response.status_code == 200
         assert response.json()["data"] == {"result": "ok"}
-        assert app.trial_run_evaluator_impl.call_args.kwargs["query"] == "hello"
+        assert app.forward_agent_evaluation_trial_run.call_args.kwargs["query"] == "hello"
 
     def test_401_on_unauthorized(self, client):
         from consts.exceptions import UnauthorizedError
@@ -631,7 +633,7 @@ def test_401_on_unauthorized(self, client):
 
     def test_500_on_exception(self, client):
         _mock_impls(
-            trial_run_evaluator_impl=AsyncMock(side_effect=RuntimeError("boom"))
+            forward_agent_evaluation_trial_run=AsyncMock(side_effect=RuntimeError("boom"))
         )
         response = client.post(
             "/agent-evaluations/trial-run",
@@ -641,7 +643,7 @@ def test_500_on_exception(self, client):
 
     def test_app_exception_propagates(self, client):
         _mock_impls(
-            trial_run_evaluator_impl=AsyncMock(
+            forward_agent_evaluation_trial_run=AsyncMock(
                 side_effect=_exc(_code("COMMON_RESOURCE_NOT_FOUND"), "missing")
             )
         )

test/backend/app/test_agent_evaluation_runtime_app.py

@@ -1,19 +1,71 @@
 """Tests for runtime-owned evaluation dispatch."""
 
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 from fastapi import HTTPException
 
 import apps.agent_evaluation_runtime_app as runtime_app
 from apps.agent_evaluation_runtime_app import (
     EvaluationRunRequest,
+    TrialRunRequest,
     dispatch_evaluation_run_api,
+    trial_run_evaluation_api,
 )
 from consts.evaluation_status import EvalRunStatus
 from consts.exceptions import AppException
 
 
+@pytest.mark.asyncio
+async def test_trial_run_uses_internal_identity_and_runtime_executor(monkeypatch):
+    monkeypatch.setattr(runtime_app, "verify_internal_runtime_jwt", lambda _: ("u1", "t1"))
+    executor = AsyncMock(return_value={"answer": "ok", "scores": {"judge": 1.0}})
+    monkeypatch.setattr(runtime_app, "_load_trial_executor", lambda: executor)
+
+    result = await trial_run_evaluation_api(
+        TrialRunRequest(
+            agent_id=7,
+            agent_version_no=3,
+            query="hello",
+            judge_model_id=99,
+            evaluator_ids=[5],
+        ),
+        "internal-token",
+    )
+
+    assert result == {"answer": "ok", "scores": {"judge": 1.0}}
+    executor.assert_awaited_once_with(
+        tenant_id="t1",
+        user_id="u1",
+        agent_id=7,
+        agent_version_no=3,
+        query="hello",
+        judge_model_id=99,
+        evaluator_ids=[5],
+        language="zh",
+    )
+
+
+@pytest.mark.asyncio
+async def test_trial_run_rejects_missing_internal_token_without_executing(monkeypatch):
+    monkeypatch.setattr(
+        runtime_app,
+        "verify_internal_runtime_jwt",
+        MagicMock(side_effect=ValueError("invalid token")),
+    )
+    load_executor = MagicMock()
+    monkeypatch.setattr(runtime_app, "_load_trial_executor", load_executor)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await trial_run_evaluation_api(
+            TrialRunRequest(agent_id=7, query="hello", judge_model_id=99),
+            None,
+        )
+
+    assert exc_info.value.status_code == 401
+    load_executor.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_dispatch_claims_pending_run_and_submits_runtime_worker(monkeypatch):
     monkeypatch.setattr(runtime_app, "verify_internal_runtime_jwt", lambda _: ("u1", "t1"))

test/backend/app/test_evaluation_delete_app.py

@@ -41,7 +41,7 @@
         "update_evaluation_set_case_impl",
     ),
     "database.agent_evaluation_db": ("update_annotation_schema_ids",),
-    "utils.auth_utils": ("get_current_user_id", "get_current_user_info"),
+    "utils.auth_utils": ("get_current_user_id", "get_current_user_info", "generate_internal_runtime_jwt"),
     "utils.evaluation_set_excel_utils": (
         "build_evaluation_set_excel_template_bytes", "parse_evaluation_cases_from_excel",
     ),

test/backend/services/test_agent_evaluation_service.py

@@ -448,6 +448,7 @@ def __init__(self, error_code=None, message=None, *args, **kwargs):
     "services.thread_lifecycle_service"
 )
 _thread_lifecycle_service_module.runtime_thread_manager = MagicMock()
+_thread_lifecycle_service_module.config_thread_manager = MagicMock()
 sys.modules["services.thread_lifecycle_service"] = _thread_lifecycle_service_module
 _services_pkg.thread_lifecycle_service = _thread_lifecycle_service_module
 
@@ -1150,7 +1151,7 @@ def test_create_agent_evaluation_run_happy_path(service_module):
     pool_mock = MagicMock()
     future = MagicMock()
     pool_mock.submit.return_value = types.SimpleNamespace(future=future)
-    service_module.runtime_thread_manager = pool_mock
+    service_module.config_thread_manager = pool_mock
 
     run = service_module.create_agent_evaluation_run_impl(
         tenant_id="t1",
@@ -1179,6 +1180,7 @@ def test_create_agent_evaluation_run_happy_path(service_module):
     assert len(kwargs["set_cases"]) == 3
 
     pool_mock.submit.assert_called_once()
+    service_module.runtime_thread_manager.submit.assert_not_called()
     future.add_done_callback.assert_called_once()
     # Done-callback signature should be a callable wrapping the run id + tenant.
     callback = future.add_done_callback.call_args.args[0]
@@ -1207,8 +1209,8 @@ def test_create_agent_evaluation_run_uses_resolved_version_no(service_module):
     """The published version number flows from ``resolve_latest_published_version_no``."""
     create_mock = _wire_full_db_module(service_module)
     service_module.resolve_latest_published_version_no.return_value = 13
-    service_module.runtime_thread_manager = MagicMock()
-    service_module.runtime_thread_manager.submit.return_value = types.SimpleNamespace(
+    service_module.config_thread_manager = MagicMock()
+    service_module.config_thread_manager.submit.return_value = types.SimpleNamespace(
         future=MagicMock()
     )
 

test/backend/services/test_evaluation_pure_logic.py

@@ -371,6 +371,7 @@ def __init__(self, code: Any, msg: str = ""):
     _services_pkg.evaluation_set_service = _ess_mod
     _tls_mod = _mk_mod(
         "services.thread_lifecycle_service",
+        config_thread_manager=MagicMock(),
         runtime_thread_manager=MagicMock(),
     )
     _services_pkg.thread_lifecycle_service = _tls_mod