diff --git a/backend/apps/agent_evaluation_app.py b/backend/apps/agent_evaluation_app.py index 3f11b0131..b13a6318b 100644 --- a/backend/apps/agent_evaluation_app.py +++ b/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 " diff --git a/backend/apps/agent_evaluation_runtime_app.py b/backend/apps/agent_evaluation_runtime_app.py index 3c098a92f..77aada355 100644 --- a/backend/apps/agent_evaluation_runtime_app.py +++ b/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,6 +28,17 @@ 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 @@ -35,6 +46,41 @@ def _load_evaluation_executor(): 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, diff --git a/backend/services/agent_evaluation_service.py b/backend/services/agent_evaluation_service.py index 6bd28ff33..8ceaad8bf 100644 --- a/backend/services/agent_evaluation_service.py +++ b/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, diff --git a/backend/services/runtime_proxy_service.py b/backend/services/runtime_proxy_service.py index 4a5b1d90d..cde7e8dae 100644 --- a/backend/services/runtime_proxy_service.py +++ b/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, diff --git a/test/backend/app/test_agent_evaluation_app.py b/test/backend/app/test_agent_evaluation_app.py index cac7c5c55..fa5c27f22 100644 --- a/test/backend/app/test_agent_evaluation_app.py +++ b/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") ) ) diff --git a/test/backend/app/test_agent_evaluation_runtime_app.py b/test/backend/app/test_agent_evaluation_runtime_app.py index e13c7f94b..3842438a5 100644 --- a/test/backend/app/test_agent_evaluation_runtime_app.py +++ b/test/backend/app/test_agent_evaluation_runtime_app.py @@ -1,6 +1,6 @@ """Tests for runtime-owned evaluation dispatch.""" -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import pytest from fastapi import HTTPException @@ -8,12 +8,64 @@ 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")) diff --git a/test/backend/app/test_evaluation_delete_app.py b/test/backend/app/test_evaluation_delete_app.py index 57eb9866b..77c7ccc43 100644 --- a/test/backend/app/test_evaluation_delete_app.py +++ b/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", ), diff --git a/test/backend/services/test_agent_evaluation_service.py b/test/backend/services/test_agent_evaluation_service.py index 233fe2141..44e830e9b 100644 --- a/test/backend/services/test_agent_evaluation_service.py +++ b/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() ) diff --git a/test/backend/services/test_evaluation_pure_logic.py b/test/backend/services/test_evaluation_pure_logic.py index 58a3d12f4..a75f7d2e6 100644 --- a/test/backend/services/test_evaluation_pure_logic.py +++ b/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 diff --git a/test/backend/services/test_runtime_proxy_service.py b/test/backend/services/test_runtime_proxy_service.py index 0c8b20f5b..ce43de09e 100644 --- a/test/backend/services/test_runtime_proxy_service.py +++ b/test/backend/services/test_runtime_proxy_service.py @@ -142,6 +142,134 @@ def test_authorization_headers_maps_missing_jwt_configuration(monkeypatch): proxy._authorization_headers("user-a", "tenant-a") +@pytest.mark.asyncio +async def test_forward_agent_evaluation_trial_run_posts_runtime_request(monkeypatch): + captured = {} + + async def handler(request: httpx.Request): + captured["request"] = request + return httpx.Response(200, json={"answer": "ok", "scores": {"judge": 1.0}}) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + monkeypatch.setattr(proxy, "RUNTIME_SERVICE_URL", "http://runtime:5014") + monkeypatch.setattr(proxy, "generate_internal_runtime_jwt", lambda *_: "jwt") + + def create_client(**kwargs): + client.headers.update(kwargs["headers"]) + return client + + monkeypatch.setattr(proxy, "create_httpx_client", create_client) + + result = await proxy.forward_agent_evaluation_trial_run( + agent_id=7, + agent_version_no=3, + query="hello", + judge_model_id=99, + evaluator_ids=[5], + language="zh", + user_id="user-a", + tenant_id="tenant-a", + ) + + assert result == {"answer": "ok", "scores": {"judge": 1.0}} + request = captured["request"] + assert str(request.url) == "http://runtime:5014/api/agent-evaluations/internal/trial-run" + assert request.headers["authorization"] == "Bearer jwt" + assert json.loads(request.content) == { + "agent_id": 7, + "agent_version_no": 3, + "query": "hello", + "judge_model_id": 99, + "evaluator_ids": [5], + "language": "zh", + } + assert client.is_closed is True + + +@pytest.mark.asyncio +async def test_forward_agent_evaluation_trial_run_maps_upstream_error(monkeypatch): + client = httpx.AsyncClient( + transport=httpx.MockTransport( + lambda _: httpx.Response(500, content=b'{"detail":"failed"}') + ) + ) + monkeypatch.setattr(proxy, "generate_internal_runtime_jwt", lambda *_: "jwt") + monkeypatch.setattr(proxy, "create_httpx_client", lambda **_: client) + + with pytest.raises(RuntimeUpstreamError) as exc_info: + await proxy.forward_agent_evaluation_trial_run( + agent_id=7, + agent_version_no=3, + query="hello", + judge_model_id=99, + evaluator_ids=None, + language="zh", + user_id="user-a", + tenant_id="tenant-a", + ) + + assert exc_info.value.status_code == 500 + assert client.is_closed is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("transport_error", "expected_error"), + [ + (httpx.ReadTimeout("timed out"), RuntimeServiceTimeoutError), + (httpx.ConnectError("connection failed"), RuntimeServiceUnavailableError), + ], +) +async def test_forward_agent_evaluation_trial_run_maps_transport_errors( + monkeypatch, transport_error, expected_error, +): + async def handler(request: httpx.Request): + transport_error.request = request + raise transport_error + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + monkeypatch.setattr(proxy, "generate_internal_runtime_jwt", lambda *_: "jwt") + monkeypatch.setattr(proxy, "create_httpx_client", lambda **_: client) + + with pytest.raises(expected_error): + await proxy.forward_agent_evaluation_trial_run( + agent_id=7, + agent_version_no=3, + query="hello", + judge_model_id=99, + evaluator_ids=None, + language="zh", + user_id="user-a", + tenant_id="tenant-a", + ) + + assert client.is_closed is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize("content", [b"not-json", b"[]"]) +async def test_forward_agent_evaluation_trial_run_rejects_invalid_success_payload(monkeypatch, content): + client = httpx.AsyncClient( + transport=httpx.MockTransport(lambda _: httpx.Response(200, content=content)) + ) + monkeypatch.setattr(proxy, "generate_internal_runtime_jwt", lambda *_: "jwt") + monkeypatch.setattr(proxy, "create_httpx_client", lambda **_: client) + + with pytest.raises(RuntimeServiceUnavailableError): + await proxy.forward_agent_evaluation_trial_run( + agent_id=7, + agent_version_no=3, + query="hello", + judge_model_id=99, + evaluator_ids=None, + language="zh", + user_id="user-a", + tenant_id="tenant-a", + ) + + assert client.is_closed is True + + @pytest.mark.asyncio async def test_forward_agent_run_streams_body_and_closes_resources(monkeypatch): stream = TrackingStream([b"data: one\n\n", b"data: two\n\n"])