From 0ab27841ff3277201e71d6d00c8dd9566772c92e Mon Sep 17 00:00:00 2001 From: Namrata Ghadi Date: Mon, 19 Oct 2026 14:03:58 -0700 Subject: [PATCH 1/7] add jev evaluator --- evaluators/contrib/galileo/pyproject.toml | 1 + .../jev/__init__.py | 16 + .../jev/client.py | 182 +++++ .../jev/config.py | 39 ++ .../jev/evaluator.py | 222 +++++++ .../galileo/tests/test_jev_evaluator.py | 620 ++++++++++++++++++ 6 files changed, 1080 insertions(+) create mode 100644 evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/__init__.py create mode 100644 evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/client.py create mode 100644 evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/config.py create mode 100644 evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/evaluator.py create mode 100644 evaluators/contrib/galileo/tests/test_jev_evaluator.py diff --git a/evaluators/contrib/galileo/pyproject.toml b/evaluators/contrib/galileo/pyproject.toml index 90325c21..059498eb 100644 --- a/evaluators/contrib/galileo/pyproject.toml +++ b/evaluators/contrib/galileo/pyproject.toml @@ -28,6 +28,7 @@ dev = [ [project.entry-points."agent_control.evaluators"] "galileo.luna" = "agent_control_evaluator_galileo.luna:LunaEvaluator" +"galileo.jev" = "agent_control_evaluator_galileo.jev:JevEvaluator" [build-system] requires = ["hatchling"] diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/__init__.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/__init__.py new file mode 100644 index 00000000..9891fc6b --- /dev/null +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/__init__.py @@ -0,0 +1,16 @@ +"""Galileo Jev decision scorer evaluator.""" + +from .client import GalileoJevClient, JevInvokeInputs, JevInvokeRequest, JevInvokeResponse +from .config import JevEvaluatorConfig, JevOperator, JevPayloadField +from .evaluator import JevEvaluator + +__all__ = [ + "GalileoJevClient", + "JevEvaluator", + "JevEvaluatorConfig", + "JevInvokeInputs", + "JevInvokeRequest", + "JevInvokeResponse", + "JevOperator", + "JevPayloadField", +] diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/client.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/client.py new file mode 100644 index 00000000..85df2adb --- /dev/null +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/client.py @@ -0,0 +1,182 @@ +"""HTTP client for Galileo Jev scorer invocation.""" + +from __future__ import annotations + +import logging +import os +from base64 import urlsafe_b64encode +from hashlib import sha256 +from hmac import new as hmac_new +from json import dumps +from time import time +from typing import Any +from urllib.parse import urlsplit + +import httpx +from agent_control_models import JSONObject, JSONValue, Step +from pydantic import BaseModel, Field + +from ..luna.client import GalileoExecutionContext +from ..luna.config import scorer_invoke_runtime_enabled +from ..records import record_from_step +from .config import JevPayloadField + +logger = logging.getLogger(__name__) + +DEFAULT_TIMEOUT_SECS = 10.0 +DEFAULT_INTERNAL_TOKEN_TTL_SECS = 3600 +DEFAULT_INVOKE_PATH = "/api/v1/scorers/invoke" +LUNA_INVOKE_URL_ENV = "GALILEO_LUNA_INVOKE_URL" +AUTH_UPSTREAM_CA_FILE_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_CA_FILE" + + +def _b64url(data: bytes) -> str: + return urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _jev_internal_auth_token( + api_secret: str, + ttl_seconds: int = DEFAULT_INTERNAL_TOKEN_TTL_SECS, +) -> str: + """Create the internal JWT expected by Runners scorer-invoke routes.""" + now = int(time()) + header = {"alg": "HS256", "typ": "JWT"} + payload = { + "internal": True, + "scope": "scorers.invoke", + "iat": now, + "exp": now + ttl_seconds, + } + signing_input = ".".join( + [ + _b64url(dumps(header, separators=(",", ":")).encode("utf-8")), + _b64url(dumps(payload, separators=(",", ":")).encode("utf-8")), + ] + ) + signature = hmac_new(api_secret.encode("utf-8"), signing_input.encode("ascii"), sha256).digest() + return f"{signing_input}.{_b64url(signature)}" + + +def _normalize_invoke_url(raw_url: str) -> str: + """Append the Runners scorer-invoke path to a service-root URL.""" + url = raw_url.strip().rstrip("/") + parsed = urlsplit(url) + if parsed.path not in ("", "/") or parsed.query or parsed.fragment: + return url + return f"{url}{DEFAULT_INVOKE_PATH}" + + +class JevInvokeInputs(BaseModel): + """Legacy inputs accepted alongside the structured Galileo record.""" + + query: JSONValue = "" + response: JSONValue = "" + ground_truth: JSONValue = None + tools: list[JSONObject] | None = None + + +class JevInvokeRequest(BaseModel): + """Structured Runners scorer-invoke request.""" + + scorer_id: str = Field(min_length=1) + scorer_version_id: str = Field(min_length=1) + scorer_label: str | None = None + execution_context: GalileoExecutionContext + record: JSONObject + inputs: JevInvokeInputs + config: JSONObject = Field(default_factory=dict) + + def to_dict(self) -> JSONObject: + """Serialize the complete request while preserving nested record data.""" + request = self.model_dump(mode="json", exclude_none=True) + request["execution_context"] = self.execution_context.model_dump(mode="json") + return request + + +class JevInvokeResponse(BaseModel): + """Response returned by the Jev scorer-invoke runtime.""" + + score: JSONValue | None = None + status: str = "unknown" + scorer_label: str | None = None + execution_time: float | None = None + error_message: str | None = None + + +class GalileoJevClient: + """Small HTTP client for the Runners scorer-invoke endpoint.""" + + def __init__( + self, + api_secret: str | None = None, + invoke_url: str | None = None, + *, + timeout: float = DEFAULT_TIMEOUT_SECS, + ) -> None: + resolved_secret = ( + api_secret or os.getenv("GALILEO_API_SECRET_KEY") or os.getenv("GALILEO_API_SECRET") + ) + if not resolved_secret: + raise ValueError( + "GALILEO_API_SECRET_KEY or GALILEO_API_SECRET is required for galileo.jev" + ) + resolved_url = invoke_url or os.getenv(LUNA_INVOKE_URL_ENV) + if not resolved_url or not resolved_url.strip(): + raise ValueError(f"{LUNA_INVOKE_URL_ENV} is required for galileo.jev") + self.api_secret = resolved_secret + self.invoke_url = _normalize_invoke_url(resolved_url) + self.runtime_enabled = scorer_invoke_runtime_enabled() + self.timeout = timeout + self._client = httpx.AsyncClient( + headers={"Content-Type": "application/json"}, + verify=os.getenv(AUTH_UPSTREAM_CA_FILE_ENV) or True, + ) + + async def invoke( + self, + *, + scorer_id: str, + scorer_version_id: str, + scorer_label: str | None, + step: Step, + execution_context: GalileoExecutionContext, + inputs: JevInvokeInputs, + timeout: float, + selected_data: Any = None, + payload_field: JevPayloadField = "input", + ) -> JevInvokeResponse: + """Invoke the pinned scorer with the complete Galileo Core record.""" + if not self.runtime_enabled: + raise RuntimeError( + "Galileo scorer-invoke runtime is disabled; galileo.jev requires " + "GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME=enabled." + ) + record = record_from_step( + step, + selected_data=selected_data, + payload_field=payload_field, + ).model_dump(mode="json", exclude_none=True) + request = JevInvokeRequest( + scorer_id=scorer_id, + scorer_version_id=scorer_version_id, + scorer_label=scorer_label, + execution_context=execution_context, + record=record, + inputs=inputs, + config={"request_timeout_seconds": timeout * 0.8}, + ) + response = await self._client.post( + self.invoke_url, + json=request.to_dict(), + headers={"Authorization": f"Bearer {_jev_internal_auth_token(self.api_secret)}"}, + timeout=timeout, + ) + response.raise_for_status() + body: Any = response.json() + if not isinstance(body, dict): + raise RuntimeError("Invalid Jev scorer response: expected a JSON object") + return JevInvokeResponse.model_validate(body) + + async def close(self) -> None: + """Close the underlying HTTP client.""" + await self._client.aclose() diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/config.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/config.py new file mode 100644 index 00000000..c8f6c993 --- /dev/null +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/config.py @@ -0,0 +1,39 @@ +"""Configuration for Galileo Jev decision scorer evaluation.""" + +from __future__ import annotations + +from typing import Literal + +from agent_control_evaluators import EvaluatorConfig +from agent_control_models import JSONValue +from pydantic import Field, model_validator + +JevOperator = Literal["gt", "gte", "lt", "lte", "eq", "ne", "contains", "any"] +JevPayloadField = Literal["input", "output"] +_NUMERIC_OPERATORS = frozenset({"gt", "gte", "lt", "lte"}) + + +class JevEvaluatorConfig(EvaluatorConfig): + """Pinned Jev scorer and local result comparison settings.""" + + scorer_id: str = Field(min_length=1) + scorer_version_id: str = Field(min_length=1) + scorer_label: str | None = Field(default=None, min_length=1) + threshold: JSONValue = 0.5 + operator: JevOperator = "gte" + payload_field: JevPayloadField = Field( + default="input", + description="Record side to use when the selector returns a scalar value.", + ) + timeout_ms: int = Field(default=10000, ge=1, le=60000) + + @model_validator(mode="after") + def validate_threshold(self) -> JevEvaluatorConfig: + """Ensure ordered comparisons use a numeric threshold.""" + if self.operator in _NUMERIC_OPERATORS and ( + isinstance(self.threshold, bool) or not isinstance(self.threshold, (int, float)) + ): + raise ValueError(f"operator '{self.operator}' requires a numeric threshold") + if self.operator != "any" and self.threshold is None: + raise ValueError("threshold is required unless operator is 'any'") + return self diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/evaluator.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/evaluator.py new file mode 100644 index 00000000..4ada5469 --- /dev/null +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/evaluator.py @@ -0,0 +1,222 @@ +"""Galileo Jev decision-model evaluator.""" + +from __future__ import annotations + +import logging +from importlib.metadata import PackageNotFoundError, version +from typing import Any + +import httpx +from agent_control_evaluators import Evaluator, EvaluatorMetadata, register_evaluator +from agent_control_models import EvaluatorResult, JSONObject, JSONValue, Step + +from ..luna.client import GalileoExecutionContext +from ..luna.config import scorer_invoke_runtime_enabled +from .client import GalileoJevClient, JevInvokeInputs, JevInvokeResponse +from .config import JevEvaluatorConfig + +logger = logging.getLogger(__name__) + +try: + _PACKAGE_VERSION = version("agent-control-evaluator-galileo") +except PackageNotFoundError: + _PACKAGE_VERSION = "0.0.0.dev" + + +def _contains(value: JSONValue, threshold: JSONValue) -> bool: + if isinstance(value, str): + return str(threshold) in value + if isinstance(value, list): + return threshold in value + if isinstance(value, dict): + return threshold in value.values() + return False + + +@register_evaluator +class JevEvaluator(Evaluator[JevEvaluatorConfig]): + """Evaluate data using a pinned Galileo Jev decision scorer.""" + + metadata = EvaluatorMetadata( + name="galileo.jev", + version=_PACKAGE_VERSION, + description="Galileo Jev decision scorer evaluation", + requires_api_key=True, + timeout_ms=10000, + ) + config_model = JevEvaluatorConfig + + def __init__(self, config: JevEvaluatorConfig) -> None: + super().__init__(config) + self._client = GalileoJevClient() + + async def evaluate(self, data: Any) -> EvaluatorResult: + """Return an actionable error because Jev requires runtime context.""" + if not scorer_invoke_runtime_enabled() or not self._client.runtime_enabled: + return self._error( + RuntimeError( + "Galileo scorer-invoke runtime is disabled; galileo.jev requires " + "GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME=enabled." + ) + ) + return self._error(RuntimeError("galileo.jev requires Step context and execution_context")) + + async def evaluate_with_context(self, data: Any, step: Step) -> EvaluatorResult: + """Evaluate a step when an authenticated execution context is supplied by extensions.""" + return await self._evaluate(data, step, None) + + async def evaluate_with_extensions( + self, data: Any, step: Step, extensions: JSONObject | None + ) -> EvaluatorResult: + """Evaluate with execution identity forwarded from trusted runtime extensions.""" + try: + context = self._execution_context(extensions) + if context is None: + raise ValueError("galileo.jev requires execution_context") + return await self._evaluate(data, step, context) + except Exception as exc: + return self._error(exc) + + async def _evaluate( + self, data: Any, step: Step, execution_context: GalileoExecutionContext | None + ) -> EvaluatorResult: + try: + if not scorer_invoke_runtime_enabled() or not self._client.runtime_enabled: + raise RuntimeError( + "Galileo scorer-invoke runtime is disabled; galileo.jev requires " + "GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME=enabled." + ) + if execution_context is None: + raise ValueError("galileo.jev requires execution_context") + inputs = self._inputs(data, step) + response = await self._client.invoke( + scorer_id=self.config.scorer_id, + scorer_version_id=self.config.scorer_version_id, + scorer_label=self.config.scorer_label, + step=step, + execution_context=execution_context, + inputs=inputs, + timeout=self.get_timeout_seconds(), + selected_data=data, + payload_field=self.config.payload_field, + ) + if response.status.lower() != "success": + raise RuntimeError( + response.error_message or f"Jev scorer status: {response.status}" + ) + if response.score is None: + raise RuntimeError("Jev scorer response did not include a score") + matched = self._matches(response.score) + metadata = self._metadata(response) + confidence = ( + response.score + if isinstance(response.score, float) and 0 <= response.score <= 1 + else 1.0 + ) + return EvaluatorResult( + matched=matched, + confidence=confidence, + message=( + f"Jev result {response.score!r} {self.config.operator} " + f"threshold {self.config.threshold!r}: control " + f"{'triggered' if matched else 'not triggered'}." + ), + metadata=metadata, + ) + except Exception as exc: + logger.error("Jev evaluation error: %s", exc, exc_info=True) + return self._error(exc) + + @staticmethod + def _execution_context(extensions: JSONObject | None) -> GalileoExecutionContext | None: + if extensions is None: + return None + metadata = extensions.get("metadata") + metadata_obj = metadata if isinstance(metadata, dict) else {} + organization_id = extensions.get("namespace_key") + user_id = extensions.get("caller_id") + target_type = extensions.get("target_type") + target_id = extensions.get("target_id") + run_id = target_id if target_type in ("log_stream", "agent_stream") else None + project_id = metadata_obj.get("project_id") + if not isinstance(organization_id, str) or not organization_id: + raise ValueError("Authenticated execution metadata is missing organization_id") + return GalileoExecutionContext( + organization_id=organization_id, + user_id=user_id if isinstance(user_id, str) and user_id else None, + project_id=project_id if isinstance(project_id, str) else None, + run_id=run_id if isinstance(run_id, str) and run_id else None, + ) + + @staticmethod + def _inputs(data: Any, step: Step) -> JevInvokeInputs: + """Build the legacy inputs without rejecting valid child-only records.""" + if isinstance(data, dict) and ("input" in data or "output" in data): + query = data.get("input", "") + response = data.get("output", "") + elif data is not None and data != "": + query, response = data, "" + else: + query, response = step.input, step.output if step.output is not None else "" + return JevInvokeInputs( + query=query, + response=response, + ground_truth=step.ground_truth, + tools=step.tools, + ) + + def _matches(self, value: JSONValue) -> bool: + operator = self.config.operator + threshold = self.config.threshold + if operator == "any": + return bool(value) + if operator == "eq": + return value == threshold + if operator == "ne": + return value != threshold + if operator == "contains": + return _contains(value, threshold) + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"Jev result {value!r} is not numeric for operator '{operator}'") + if isinstance(threshold, bool) or not isinstance(threshold, (int, float)): + raise ValueError(f"Jev threshold {threshold!r} is not numeric") + return { + "gt": value > threshold, + "gte": value >= threshold, + "lt": value < threshold, + "lte": value <= threshold, + }[operator] + + def _metadata(self, response: JevInvokeResponse) -> dict[str, Any]: + return { + "scorer_id": self.config.scorer_id, + "scorer_version_id": self.config.scorer_version_id, + "scorer_label": response.scorer_label or self.config.scorer_label, + "score": response.score, + "threshold": self.config.threshold, + "operator": self.config.operator, + "status": response.status, + "execution_time_seconds": response.execution_time, + "error_message": response.error_message, + } + + def _error(self, error: Exception) -> EvaluatorResult: + detail = str(error) + metadata: dict[str, Any] = { + "scorer_id": self.config.scorer_id, + "scorer_version_id": self.config.scorer_version_id, + "error_type": type(error).__name__, + } + if isinstance(error, httpx.HTTPStatusError): + metadata["http_status_code"] = error.response.status_code + return EvaluatorResult( + matched=False, + confidence=0.0, + message=f"Jev evaluation error: {detail}", + metadata=metadata, + error=detail, + ) + + async def aclose(self) -> None: + """Close the Jev HTTP client.""" + await self._client.close() diff --git a/evaluators/contrib/galileo/tests/test_jev_evaluator.py b/evaluators/contrib/galileo/tests/test_jev_evaluator.py new file mode 100644 index 00000000..953530c4 --- /dev/null +++ b/evaluators/contrib/galileo/tests/test_jev_evaluator.py @@ -0,0 +1,620 @@ +"""Tests for the Galileo Jev evaluator and Runners client.""" + +from __future__ import annotations + +import json +from base64 import urlsafe_b64decode, urlsafe_b64encode +from hashlib import sha256 +from hmac import compare_digest +from hmac import new as hmac_new +from typing import Any +from unittest.mock import AsyncMock + +import httpx +import pytest +from agent_control_evaluator_galileo.jev import ( + GalileoJevClient, + JevEvaluator, + JevEvaluatorConfig, + JevInvokeInputs, + JevInvokeRequest, + JevInvokeResponse, +) +from agent_control_evaluator_galileo.luna.client import ( + GalileoExecutionContext, + ScorerInvokeInputs, + ScorerInvokeRequest, +) +from agent_control_models import Step +from pydantic import ValidationError + +EXTENSIONS = { + "namespace_key": "org-1", + "caller_id": "user-7", + "target_type": "log_stream", + "target_id": "run-9", + "metadata": {"project_id": "project-3"}, +} + + +def _decode_jwt_segment(segment: str) -> dict[str, Any]: + padded = segment + ("=" * (-len(segment) % 4)) + return json.loads(urlsafe_b64decode(padded.encode()).decode()) + + +def _evaluator(monkeypatch: pytest.MonkeyPatch, **config: Any) -> JevEvaluator: + monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") + monkeypatch.setenv("GALILEO_LUNA_INVOKE_URL", "http://runners:8090") + monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") + evaluator = JevEvaluator( + JevEvaluatorConfig(scorer_id="scorer-a", scorer_version_id="version-a", **config) + ) + return evaluator + + +def test_jev_registration_and_required_configuration() -> None: + # Given: the separate Jev entry point and required pinned scorer config + # Then: Jev is registered independently and both scorer identifiers are required + from agent_control_evaluators import get_evaluator + + assert get_evaluator("galileo.jev") is JevEvaluator + with pytest.raises(ValidationError, match="scorer_id"): + JevEvaluatorConfig(scorer_version_id="version-a") # type: ignore[call-arg] + with pytest.raises(ValidationError, match="scorer_version_id"): + JevEvaluatorConfig(scorer_id="scorer-a") # type: ignore[call-arg] + + +@pytest.mark.asyncio +async def test_client_url_resolution_prefers_explicit_url_then_luna_environment( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") + monkeypatch.setenv("GALILEO_LUNA_INVOKE_URL", "http://luna:8090") + + explicit_client = GalileoJevClient(invoke_url="http://explicit:8090") + environment_client = GalileoJevClient() + + expected_path = "/api/v1/scorers/invoke" + assert explicit_client.invoke_url == f"http://explicit:8090{expected_path}" + assert environment_client.invoke_url == f"http://luna:8090{expected_path}" + await explicit_client.close() + await environment_client.close() + + +def test_client_requires_luna_invoke_url_when_no_explicit_url_is_given( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") + monkeypatch.delenv("GALILEO_LUNA_INVOKE_URL", raising=False) + + with pytest.raises(ValueError, match="GALILEO_LUNA_INVOKE_URL"): + GalileoJevClient() + + +def test_invoke_request_requires_execution_context_and_serializes_nested_record() -> None: + context = GalileoExecutionContext( + organization_id="org-1", user_id="user-7", project_id="project-3", run_id="run-9" + ) + record = { + "type": "trace", + "input": "", + "output": "", + "spans": [{"type": "llm", "input": [{"role": "user", "content": "q"}]}], + } + request = JevInvokeRequest( + scorer_id="scorer-a", + scorer_version_id="version-a", + execution_context=context, + record=record, + inputs=JevInvokeInputs(), + config={"request_timeout_seconds": 8.0}, + ) + + payload = request.to_dict() + assert payload["execution_context"] == { + "organization_id": "org-1", + "user_id": "user-7", + "project_id": "project-3", + "run_id": "run-9", + } + assert payload["record"] == record + assert payload["scorer_version_id"] == "version-a" + assert payload["inputs"] == {"query": "", "response": ""} + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("score", "threshold", "operator", "expected"), + [ + (0.82, 0.5, "gte", True), + ("unsafe", "unsafe", "eq", True), + (3, 2, "gte", True), + ("safe", "safe", "ne", False), + (["unsafe"], "unsafe", "contains", True), + ], +) +async def test_jev_applies_local_comparison( + monkeypatch: pytest.MonkeyPatch, + score: Any, + threshold: Any, + operator: str, + expected: bool, +) -> None: + evaluator = _evaluator(monkeypatch, threshold=threshold, operator=operator) + evaluator._client.invoke = AsyncMock( + return_value=JevInvokeResponse(score=score, status="success") + ) + result = await evaluator.evaluate_with_extensions( + "", + Step( + type="trace", + name="root", + input="", + children=[Step(type="llm", name="answer", input="")], + ), + EXTENSIONS, + ) + assert result.matched is expected + await evaluator.aclose() + + +@pytest.mark.asyncio +async def test_child_only_trace_keeps_nested_span_and_execution_identity( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evaluator = _evaluator(monkeypatch) + invoke = AsyncMock(return_value=JevInvokeResponse(score=0.82, status="success")) + evaluator._client.invoke = invoke + step = Step( + type="trace", + name="root", + input="", + output="", + children=[ + Step( + type="tool", + name="search", + input={"query": "question"}, + children=[Step(type="retriever", name="docs", input="question", output=["doc"])], + ), + Step(type="llm", name="answer", input="question", output="answer"), + ], + ) + + result = await evaluator.evaluate_with_extensions("", step, EXTENSIONS) + + assert result.matched is True + call = invoke.await_args.kwargs + assert call["scorer_version_id"] == "version-a" + assert call["execution_context"].user_id == "user-7" + record = call["step"] + assert record.children[0].children[0].type == "retriever" + assert call["inputs"].query == "" + await evaluator.aclose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("record_type", ["llm", "tool", "retriever", "trace", "session"]) +async def test_supported_record_types_are_built_from_step_children( + monkeypatch: pytest.MonkeyPatch, record_type: str +) -> None: + evaluator = _evaluator(monkeypatch) + invoke = AsyncMock(return_value=JevInvokeResponse(score="yes", status="success")) + evaluator._client.invoke = invoke + if record_type == "session": + step = Step( + type="session", + name="session", + input="", + children=[ + Step( + type="trace", + name="trace", + input="", + children=[Step(type="llm", name="llm", input="")], + ) + ], + ) + elif record_type == "trace": + step = Step( + type="trace", name="trace", input="", children=[Step(type="llm", name="llm", input="")] + ) + elif record_type in {"tool", "retriever"}: + step = Step( + type=record_type, + name=record_type, + input={"query": "question"} if record_type == "tool" else "question", + children=[Step(type="llm", name="llm", input="")], + ) + else: + step = Step(type="llm", name="llm", input="") + await evaluator.evaluate_with_extensions("input", step, EXTENSIONS) + assert invoke.await_args.kwargs["step"].type == record_type + await evaluator.aclose() + + +@pytest.mark.asyncio +async def test_runtime_disabled_and_missing_context_return_errors_without_fallback( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evaluator = _evaluator(monkeypatch) + monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "disabled") + invoke = AsyncMock() + evaluator._client.invoke = invoke + + result = await evaluator.evaluate_with_extensions( + "question", Step(type="llm", name="answer", input="question"), EXTENSIONS + ) + + assert result.matched is False + assert "runtime is disabled" in (result.error or "") + assert invoke.await_count == 0 + direct_result = await evaluator.evaluate("question") + assert "runtime is disabled" in (direct_result.error or "") + missing_context = await evaluator.evaluate_with_extensions( + "question", Step(type="llm", name="answer", input=""), None + ) + assert missing_context.matched is False + assert "execution_context" in (missing_context.error or "") + await evaluator.aclose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("error", [TimeoutError("deadline"), RuntimeError("provider failed")]) +async def test_timeout_and_provider_errors_are_non_matching( + monkeypatch: pytest.MonkeyPatch, error: Exception +) -> None: + evaluator = _evaluator(monkeypatch) + evaluator._client.invoke = AsyncMock(side_effect=error) + result = await evaluator.evaluate_with_extensions( + "input", Step(type="llm", name="answer", input="input"), EXTENSIONS + ) + assert result.matched is False + assert result.error == str(error) + assert result.metadata["error_type"] == type(error).__name__ + await evaluator.aclose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("record_type", ["trace", "session", "llm", "tool", "retriever"]) +async def test_client_serializes_full_record_context_and_uses_timeout( + monkeypatch: pytest.MonkeyPatch, record_type: str +) -> None: + monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") + monkeypatch.setenv("GALILEO_LUNA_INVOKE_URL", "http://runners:8090") + monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") + client = GalileoJevClient() + seen: dict[str, Any] = {} + + async def handler(request: httpx.Request) -> httpx.Response: + seen["url"] = str(request.url) + seen["payload"] = request.read().decode() + seen["timeout"] = request.extensions.get("timeout") + seen["authorization"] = request.headers["Authorization"] + return httpx.Response(200, json={"score": 0.9, "status": "success"}) + + await client._client.aclose() + client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + llm = Step( + type="llm", + name="answer", + input=[{"role": "user", "content": "question"}], + output={"role": "assistant", "content": "answer"}, + tools=[{"name": "lookup", "description": "search"}], + ) + if record_type == "trace": + step = Step( + type="trace", + name="trace", + input="", + children=[ + llm, + Step( + type="tool", + name="lookup", + input={"query": "question"}, + output={"found": True}, + children=[ + Step( + type="retriever", + name="docs", + input="question", + output=[{"text": "document", "metadata": {"source": "kb"}}], + ) + ], + ), + ], + ) + elif record_type == "session": + step = Step( + type="session", + name="session", + input="", + children=[Step(type="trace", name="trace", input="question", children=[llm])], + ) + elif record_type == "tool": + step = Step( + type="tool", + name="lookup", + input={"query": "question"}, + output={"found": True}, + ) + elif record_type == "retriever": + step = Step( + type="retriever", + name="docs", + input="question", + output=[{"text": "document", "metadata": {"source": "kb"}}], + ) + else: + step = llm + response = await client.invoke( + scorer_id="scorer-a", + scorer_version_id="version-a", + scorer_label=None, + step=step, + execution_context=GalileoExecutionContext(organization_id="org-1", user_id="user-7"), + inputs=JevInvokeInputs(), + timeout=5.0, + ) + assert response.score == 0.9 + assert seen["url"] == "http://runners:8090/api/v1/scorers/invoke" + payload = json.loads(seen["payload"]) + assert payload["scorer_version_id"] == "version-a" + assert payload["execution_context"]["user_id"] == "user-7" + assert payload["record"]["type"] == record_type + record = payload["record"] + assert record["name"] == step.name + assert record["user_metadata"] == {} + assert "created_at" in record + if record_type == "llm": + assert record["input"] == [{"content": "question", "role": "user"}] + assert record["output"] == {"content": "answer", "role": "assistant"} + assert record["tools"] == [{"name": "lookup", "description": "search"}] + elif record_type == "tool": + assert record["input"] == '{"query": "question"}' + assert record["output"] == '{"found": true}' + elif record_type == "retriever": + assert record["input"] == "question" + assert record["output"] == [ + { + "content": '{"text": "document", "metadata": {"source": "kb"}}', + "metadata": {}, + } + ] + if record_type == "session": + trace = record["traces"][0] + assert trace["type"] == "trace" + assert trace["name"] == "trace" + assert trace["input"] == "question" + assert trace["user_metadata"] == {} + assert trace["spans"][0]["input"] == [{"content": "question", "role": "user"}] + assert trace["spans"][0]["output"] == {"content": "answer", "role": "assistant"} + assert trace["spans"][0]["tools"] == [{"name": "lookup", "description": "search"}] + elif record_type == "trace": + llm_span, tool_span = record["spans"] + assert llm_span["type"] == "llm" + assert llm_span["name"] == "answer" + assert llm_span["input"] == [{"content": "question", "role": "user"}] + assert llm_span["output"] == {"content": "answer", "role": "assistant"} + assert llm_span["tools"] == [{"name": "lookup", "description": "search"}] + assert tool_span["type"] == "tool" + assert tool_span["name"] == "lookup" + assert tool_span["input"] == '{"query": "question"}' + assert tool_span["output"] == '{"found": true}' + retriever_span = tool_span["spans"][0] + assert retriever_span["type"] == "retriever" + assert retriever_span["name"] == "docs" + assert retriever_span["input"] == "question" + assert retriever_span["output"] == [ + { + "content": '{"text": "document", "metadata": {"source": "kb"}}', + "metadata": {}, + } + ] + token_parts = seen["authorization"].removeprefix("Bearer ").split(".") + assert len(token_parts) == 3 + header = _decode_jwt_segment(token_parts[0]) + claims = _decode_jwt_segment(token_parts[1]) + assert header == {"alg": "HS256", "typ": "JWT"} + assert claims["internal"] is True + assert claims["scope"] == "scorers.invoke" + assert claims["exp"] - claims["iat"] == 3600 + signing_input = ".".join(token_parts[:2]).encode("ascii") + expected_signature = ( + urlsafe_b64encode(hmac_new(b"test-secret", signing_input, sha256).digest()) + .rstrip(b"=") + .decode("ascii") + ) + assert compare_digest(token_parts[2], expected_signature) + assert payload["config"]["request_timeout_seconds"] == 4.0 + await client.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("selector_case", "step_type", "data", "payload_field"), + [ + ( + "input selector", + "llm", + [{"role": "user", "content": "selected input"}], + "input", + ), + ("output selector", "llm", "selected output", "output"), + ("input.query selector", "tool", "selected query", "input"), + ( + "wildcard selector", + "trace", + { + "type": "trace", + "name": "trace", + "input": "selected trace input", + "output": "selected trace output", + "children": [{"type": "llm", "name": "child answer"}], + }, + "input", + ), + ( + "wildcard session selector", + "session", + { + "type": "session", + "name": "session", + "input": "selected session input", + "output": "selected session output", + "children": [{"type": "trace", "name": "trace"}], + }, + "input", + ), + ], +) +async def test_selector_values_update_structured_record_and_preserve_children( + monkeypatch: pytest.MonkeyPatch, + selector_case: str, + step_type: str, + data: Any, + payload_field: str, +) -> None: + evaluator = _evaluator(monkeypatch, payload_field=payload_field) + seen: dict[str, Any] = {} + + async def handler(request: httpx.Request) -> httpx.Response: + seen["payload"] = json.loads(request.read()) + return httpx.Response(200, json={"score": 0.9, "status": "success"}) + + await evaluator._client._client.aclose() + evaluator._client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + llm_child = Step(type="llm", name="child answer", input="child question", output="child") + if step_type == "tool": + step = Step( + type="tool", + name="lookup", + input={"query": "original query"}, + output="tool result", + children=[llm_child], + ) + elif step_type == "trace": + step = Step( + type="trace", + name="trace", + input="original trace input", + output="original trace output", + children=[llm_child], + ) + elif step_type == "session": + step = Step( + type="session", + name="session", + input="original session input", + output="original session output", + children=[Step(type="trace", name="trace", input="child trace", children=[llm_child])], + ) + else: + step = Step( + type="llm", + name="root answer", + input=[{"role": "user", "content": "original input"}], + output="original output", + ) + + result = await evaluator.evaluate_with_extensions(data, step, EXTENSIONS) + + assert result.matched is True, selector_case + record = seen["payload"]["record"] + if selector_case == "input selector": + assert record["input"] == [{"role": "user", "content": "selected input"}] + assert record["output"]["content"] == "original output" + elif selector_case == "output selector": + assert record["input"] == [{"role": "user", "content": "original input"}] + assert record["output"]["content"] == "selected output" + elif selector_case == "input.query selector": + assert record["input"] == "selected query" + assert record["output"] == "tool result" + assert record["spans"][0]["name"] == "child answer" + elif selector_case == "wildcard selector": + assert record["input"] == "selected trace input" + assert record["output"] == "selected trace output" + assert record["spans"][0]["name"] == "child answer" + else: + assert record["input"] == "selected session input" + assert record["output"] == "selected session output" + assert record["traces"][0]["spans"][0]["name"] == "child answer" + await evaluator.aclose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("score", "threshold", "expected"), + [(0, 0, True), (False, False, True)], +) +async def test_successful_zero_and_false_scores_are_valid( + monkeypatch: pytest.MonkeyPatch, + score: Any, + threshold: Any, + expected: bool, +) -> None: + evaluator = _evaluator(monkeypatch, threshold=threshold, operator="eq") + evaluator._client.invoke = AsyncMock( + return_value=JevInvokeResponse(score=score, status="success") + ) + result = await evaluator.evaluate_with_extensions( + "input", Step(type="llm", name="answer", input="input"), EXTENSIONS + ) + assert result.matched is expected + await evaluator.aclose() + + +@pytest.mark.asyncio +async def test_successful_response_without_score_is_an_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evaluator = _evaluator(monkeypatch) + evaluator._client.invoke = AsyncMock( + return_value=JevInvokeResponse(status="success", score=None) + ) + result = await evaluator.evaluate_with_extensions( + "input", Step(type="llm", name="answer", input="input"), EXTENSIONS + ) + assert result.matched is False + assert result.error == "Jev scorer response did not include a score" + await evaluator.aclose() + + +@pytest.mark.asyncio +async def test_non_success_response_without_score_uses_provider_status_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evaluator = _evaluator(monkeypatch) + evaluator._client.invoke = AsyncMock( + return_value=JevInvokeResponse( + status="error", score=None, error_message="provider rejected request" + ) + ) + result = await evaluator.evaluate_with_extensions( + "input", Step(type="llm", name="answer", input="input"), EXTENSIONS + ) + assert result.matched is False + assert result.error == "provider rejected request" + await evaluator.aclose() + + +def test_luna_and_jev_request_shape_parity() -> None: + context = GalileoExecutionContext(organization_id="org-1", user_id="user-7") + luna = ScorerInvokeRequest( + scorer_id="scorer-a", + scorer_version_id="version-a", + inputs=ScorerInvokeInputs(query="q", response="a"), + record={"type": "llm", "input": "q", "output": "a"}, + execution_context=context, + ).to_dict() + jev = JevInvokeRequest( + scorer_id="scorer-a", + scorer_version_id="version-a", + inputs=JevInvokeInputs(query="q", response="a"), + record={"type": "llm", "input": "q", "output": "a"}, + execution_context=context, + ).to_dict() + assert set(luna) == set(jev) + assert set(luna["inputs"]) == set(jev["inputs"]) + assert luna["execution_context"] == jev["execution_context"] From 21ac4d67df8cd2e80416445462b1fec3db68582d Mon Sep 17 00:00:00 2001 From: Namrata Ghadi Date: Mon, 19 Oct 2026 15:56:35 -0700 Subject: [PATCH 2/7] test: cover Galileo Jev edge cases --- .../galileo/tests/test_jev_evaluator.py | 221 ++++++++++++++++++ 1 file changed, 221 insertions(+) diff --git a/evaluators/contrib/galileo/tests/test_jev_evaluator.py b/evaluators/contrib/galileo/tests/test_jev_evaluator.py index 953530c4..ae346223 100644 --- a/evaluators/contrib/galileo/tests/test_jev_evaluator.py +++ b/evaluators/contrib/galileo/tests/test_jev_evaluator.py @@ -7,6 +7,8 @@ from hashlib import sha256 from hmac import compare_digest from hmac import new as hmac_new +from importlib import reload +from importlib.metadata import PackageNotFoundError from typing import Any from unittest.mock import AsyncMock @@ -20,6 +22,9 @@ JevInvokeRequest, JevInvokeResponse, ) +from agent_control_evaluator_galileo.jev import evaluator as jev_evaluator_module +from agent_control_evaluator_galileo.jev.client import _normalize_invoke_url +from agent_control_evaluator_galileo.jev.evaluator import _contains from agent_control_evaluator_galileo.luna.client import ( GalileoExecutionContext, ScorerInvokeInputs, @@ -64,6 +69,69 @@ def test_jev_registration_and_required_configuration() -> None: JevEvaluatorConfig(scorer_id="scorer-a") # type: ignore[call-arg] +def test_missing_distribution_metadata_uses_development_version( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def missing_distribution(_: str) -> str: + raise PackageNotFoundError + + from agent_control_evaluators import clear_evaluators, get_all_evaluators, register_evaluator + + registered_evaluators = get_all_evaluators() + monkeypatch.setattr(jev_evaluator_module, "version", missing_distribution) + try: + clear_evaluators() + reloaded_module = reload(jev_evaluator_module) + assert reloaded_module.JevEvaluator.metadata.version == "0.0.0.dev" + finally: + clear_evaluators() + for evaluator_class in registered_evaluators.values(): + register_evaluator(evaluator_class) + + +@pytest.mark.parametrize( + ("value", "threshold", "expected"), + [ + ("unsafe answer", "unsafe", True), + ({"label": "unsafe"}, "unsafe", True), + ({"label": "safe"}, "unsafe", False), + (42, "unsafe", False), + ], +) +def test_contains_comparison_handles_supported_and_unsupported_values( + value: Any, threshold: Any, expected: bool +) -> None: + assert _contains(value, threshold) is expected + + +def test_config_rejects_non_numeric_or_missing_thresholds() -> None: + with pytest.raises(ValidationError, match="requires a numeric threshold"): + JevEvaluatorConfig( + scorer_id="scorer-a", + scorer_version_id="version-a", + operator="gt", + threshold=True, + ) + with pytest.raises(ValidationError, match="threshold is required"): + JevEvaluatorConfig( + scorer_id="scorer-a", + scorer_version_id="version-a", + operator="eq", + threshold=None, + ) + + +def test_config_allows_missing_threshold_for_any_operator() -> None: + config = JevEvaluatorConfig( + scorer_id="scorer-a", + scorer_version_id="version-a", + operator="any", + threshold=None, + ) + + assert config.threshold is None + + @pytest.mark.asyncio async def test_client_url_resolution_prefers_explicit_url_then_luna_environment( monkeypatch: pytest.MonkeyPatch, @@ -599,6 +667,159 @@ async def test_non_success_response_without_score_uses_provider_status_error( await evaluator.aclose() +@pytest.mark.asyncio +async def test_direct_evaluation_and_context_method_report_missing_context( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evaluator = _evaluator(monkeypatch) + step = Step(type="llm", name="answer", input="question") + + direct_result = await evaluator.evaluate("question") + context_result = await evaluator.evaluate_with_context("question", step) + + assert "requires Step context" in (direct_result.error or "") + assert "requires execution_context" in (context_result.error or "") + await evaluator.aclose() + + +@pytest.mark.asyncio +async def test_extensions_without_organization_id_return_validation_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evaluator = _evaluator(monkeypatch) + extensions = {"caller_id": "user-7", "target_type": "log_stream"} + + result = await evaluator.evaluate_with_extensions( + "question", Step(type="llm", name="answer", input="question"), extensions + ) + + assert result.matched is False + assert "missing organization_id" in (result.error or "") + await evaluator.aclose() + + +@pytest.mark.parametrize( + ("score", "threshold", "operator", "expected"), + [("yes", None, "any", True), ("", None, "any", False)], +) +def test_any_operator_uses_score_truthiness( + monkeypatch: pytest.MonkeyPatch, + score: Any, + threshold: Any, + operator: str, + expected: bool, +) -> None: + evaluator = _evaluator(monkeypatch, threshold=threshold, operator=operator) + + assert evaluator._matches(score) is expected + + +def test_numeric_comparison_rejects_non_numeric_threshold_from_unvalidated_config( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") + monkeypatch.setenv("GALILEO_LUNA_INVOKE_URL", "http://runners:8090") + config = JevEvaluatorConfig.model_construct( + scorer_id="scorer-a", + scorer_version_id="version-a", + threshold="not numeric", + operator="gte", + ) + evaluator = JevEvaluator(config) + + with pytest.raises(ValueError, match="threshold .* is not numeric"): + evaluator._matches(0.8) + + +def test_http_status_errors_include_status_code_in_metadata( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evaluator = _evaluator(monkeypatch) + response = httpx.Response(401, request=httpx.Request("POST", "http://runners")) + error = httpx.HTTPStatusError("unauthorized", request=response.request, response=response) + + result = evaluator._error(error) + + assert result.metadata["http_status_code"] == 401 + + +def test_invoke_url_with_existing_path_is_preserved() -> None: + url = "https://runners.example/custom?tenant=one#invoke" + + assert _normalize_invoke_url(url) == url + + +def test_client_falls_back_to_legacy_api_secret_environment_variable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("GALILEO_API_SECRET_KEY", raising=False) + monkeypatch.setenv("GALILEO_API_SECRET", "legacy-secret") + + client = GalileoJevClient(invoke_url="http://runners:8090") + + assert client.api_secret == "legacy-secret" + + +def test_client_requires_api_secret_when_environment_variables_are_missing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("GALILEO_API_SECRET_KEY", raising=False) + monkeypatch.delenv("GALILEO_API_SECRET", raising=False) + + with pytest.raises(ValueError, match="GALILEO_API_SECRET_KEY or GALILEO_API_SECRET"): + GalileoJevClient(invoke_url="http://runners:8090") + + +@pytest.mark.asyncio +async def test_client_rejects_invoke_when_runtime_is_disabled( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") + monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "disabled") + client = GalileoJevClient(invoke_url="http://runners:8090") + + with pytest.raises(RuntimeError, match="runtime is disabled"): + await client.invoke( + scorer_id="scorer-a", + scorer_version_id="version-a", + scorer_label=None, + step=Step(type="llm", name="answer", input="question"), + execution_context=GalileoExecutionContext(organization_id="org-1"), + inputs=JevInvokeInputs(), + timeout=5.0, + ) + + await client.close() + + +@pytest.mark.asyncio +async def test_client_rejects_non_object_invoke_response( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") + monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") + client = GalileoJevClient(invoke_url="http://runners:8090") + + async def handler(_: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=[{"score": 0.9}]) + + await client._client.aclose() + client._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + with pytest.raises(RuntimeError, match="expected a JSON object"): + await client.invoke( + scorer_id="scorer-a", + scorer_version_id="version-a", + scorer_label=None, + step=Step(type="llm", name="answer", input="question"), + execution_context=GalileoExecutionContext(organization_id="org-1"), + inputs=JevInvokeInputs(), + timeout=5.0, + ) + + await client.close() + + def test_luna_and_jev_request_shape_parity() -> None: context = GalileoExecutionContext(organization_id="org-1", user_id="user-7") luna = ScorerInvokeRequest( From 821e0fa289d51c570f3ad9e80082b199db7f2b71 Mon Sep 17 00:00:00 2001 From: Namrata Ghadi Date: Mon, 19 Oct 2026 16:28:36 -0700 Subject: [PATCH 3/7] test: fix Jev metadata fallback coverage --- evaluators/contrib/galileo/tests/test_jev_evaluator.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/evaluators/contrib/galileo/tests/test_jev_evaluator.py b/evaluators/contrib/galileo/tests/test_jev_evaluator.py index ae346223..b50e6f09 100644 --- a/evaluators/contrib/galileo/tests/test_jev_evaluator.py +++ b/evaluators/contrib/galileo/tests/test_jev_evaluator.py @@ -7,6 +7,7 @@ from hashlib import sha256 from hmac import compare_digest from hmac import new as hmac_new +from importlib import metadata as importlib_metadata from importlib import reload from importlib.metadata import PackageNotFoundError from typing import Any @@ -78,7 +79,7 @@ def missing_distribution(_: str) -> str: from agent_control_evaluators import clear_evaluators, get_all_evaluators, register_evaluator registered_evaluators = get_all_evaluators() - monkeypatch.setattr(jev_evaluator_module, "version", missing_distribution) + monkeypatch.setattr(importlib_metadata, "version", missing_distribution) try: clear_evaluators() reloaded_module = reload(jev_evaluator_module) From dc6bba73b1009f2d560586e391d5693ab57ac3cb Mon Sep 17 00:00:00 2001 From: Namrata Ghadi Date: Tue, 20 Oct 2026 16:46:11 -0700 Subject: [PATCH 4/7] refactor: rename Galileo Jev evaluator to decision models --- evaluators/contrib/README.md | 2 +- evaluators/contrib/galileo/pyproject.toml | 4 +- .../decision_models/__init__.py | 25 ++++ .../{jev => decision_models}/client.py | 41 ++++--- .../{jev => decision_models}/config.py | 16 +-- .../{jev => decision_models}/evaluator.py | 62 +++++----- .../jev/__init__.py | 16 --- ...r.py => test_decision_models_evaluator.py} | 110 +++++++++--------- 8 files changed, 149 insertions(+), 127 deletions(-) create mode 100644 evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/__init__.py rename evaluators/contrib/galileo/src/agent_control_evaluator_galileo/{jev => decision_models}/client.py (83%) rename evaluators/contrib/galileo/src/agent_control_evaluator_galileo/{jev => decision_models}/config.py (69%) rename evaluators/contrib/galileo/src/agent_control_evaluator_galileo/{jev => decision_models}/evaluator.py (79%) delete mode 100644 evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/__init__.py rename evaluators/contrib/galileo/tests/{test_jev_evaluator.py => test_decision_models_evaluator.py} (88%) diff --git a/evaluators/contrib/README.md b/evaluators/contrib/README.md index 4b852070..5ae14c6a 100644 --- a/evaluators/contrib/README.md +++ b/evaluators/contrib/README.md @@ -2,7 +2,7 @@ Contributed evaluators and templates for extending Agent Control. -- `galileo/` — Galileo evaluator integrations (Luna, LLM) +- `galileo/` — Galileo evaluator integrations (Luna, Decision Models, LLM-as-judge) - `template/` — Starter template for adding new evaluators Full guide: https://docs.agentcontrol.dev/concepts/evaluators/custom-evaluators diff --git a/evaluators/contrib/galileo/pyproject.toml b/evaluators/contrib/galileo/pyproject.toml index 721b0cbc..69f4af58 100644 --- a/evaluators/contrib/galileo/pyproject.toml +++ b/evaluators/contrib/galileo/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "agent-control-evaluator-galileo" version = "8.14.0" -description = "Galileo evaluators (Luna, LLM-as-judge) for agent-control" +description = "Galileo evaluators (Luna, Decision Models, LLM-as-judge) for agent-control" readme = "README.md" requires-python = ">=3.12" license = { text = "Apache-2.0" } @@ -28,7 +28,7 @@ dev = [ [project.entry-points."agent_control.evaluators"] "galileo.luna" = "agent_control_evaluator_galileo.luna:LunaEvaluator" -"galileo.jev" = "agent_control_evaluator_galileo.jev:JevEvaluator" +"galileo.decision_models" = "agent_control_evaluator_galileo.decision_models:DecisionModelsEvaluator" "galileo.llm" = "agent_control_evaluator_galileo.llm:LlmEvaluator" [build-system] diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/__init__.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/__init__.py new file mode 100644 index 00000000..90c844e4 --- /dev/null +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/__init__.py @@ -0,0 +1,25 @@ +"""Galileo Decision Models evaluator.""" + +from .client import ( + DecisionModelsInvokeInputs, + DecisionModelsInvokeRequest, + DecisionModelsInvokeResponse, + GalileoDecisionModelsClient, +) +from .config import ( + DecisionModelsEvaluatorConfig, + DecisionModelsOperator, + DecisionModelsPayloadField, +) +from .evaluator import DecisionModelsEvaluator + +__all__ = [ + "GalileoDecisionModelsClient", + "DecisionModelsEvaluator", + "DecisionModelsEvaluatorConfig", + "DecisionModelsInvokeInputs", + "DecisionModelsInvokeRequest", + "DecisionModelsInvokeResponse", + "DecisionModelsOperator", + "DecisionModelsPayloadField", +] diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/client.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py similarity index 83% rename from evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/client.py rename to evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py index 85df2adb..a592d351 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/client.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py @@ -1,4 +1,4 @@ -"""HTTP client for Galileo Jev scorer invocation.""" +"""HTTP client for Galileo Decision Models scorer invocation.""" from __future__ import annotations @@ -19,7 +19,7 @@ from ..luna.client import GalileoExecutionContext from ..luna.config import scorer_invoke_runtime_enabled from ..records import record_from_step -from .config import JevPayloadField +from .config import DecisionModelsPayloadField logger = logging.getLogger(__name__) @@ -34,7 +34,7 @@ def _b64url(data: bytes) -> str: return urlsafe_b64encode(data).rstrip(b"=").decode("ascii") -def _jev_internal_auth_token( +def _decision_models_internal_auth_token( api_secret: str, ttl_seconds: int = DEFAULT_INTERNAL_TOKEN_TTL_SECS, ) -> str: @@ -66,7 +66,7 @@ def _normalize_invoke_url(raw_url: str) -> str: return f"{url}{DEFAULT_INVOKE_PATH}" -class JevInvokeInputs(BaseModel): +class DecisionModelsInvokeInputs(BaseModel): """Legacy inputs accepted alongside the structured Galileo record.""" query: JSONValue = "" @@ -75,7 +75,7 @@ class JevInvokeInputs(BaseModel): tools: list[JSONObject] | None = None -class JevInvokeRequest(BaseModel): +class DecisionModelsInvokeRequest(BaseModel): """Structured Runners scorer-invoke request.""" scorer_id: str = Field(min_length=1) @@ -83,7 +83,7 @@ class JevInvokeRequest(BaseModel): scorer_label: str | None = None execution_context: GalileoExecutionContext record: JSONObject - inputs: JevInvokeInputs + inputs: DecisionModelsInvokeInputs config: JSONObject = Field(default_factory=dict) def to_dict(self) -> JSONObject: @@ -93,8 +93,8 @@ def to_dict(self) -> JSONObject: return request -class JevInvokeResponse(BaseModel): - """Response returned by the Jev scorer-invoke runtime.""" +class DecisionModelsInvokeResponse(BaseModel): + """Response returned by the Decision Models scorer-invoke runtime.""" score: JSONValue | None = None status: str = "unknown" @@ -103,7 +103,7 @@ class JevInvokeResponse(BaseModel): error_message: str | None = None -class GalileoJevClient: +class GalileoDecisionModelsClient: """Small HTTP client for the Runners scorer-invoke endpoint.""" def __init__( @@ -118,11 +118,12 @@ def __init__( ) if not resolved_secret: raise ValueError( - "GALILEO_API_SECRET_KEY or GALILEO_API_SECRET is required for galileo.jev" + "GALILEO_API_SECRET_KEY or GALILEO_API_SECRET is required for " + "galileo.decision_models" ) resolved_url = invoke_url or os.getenv(LUNA_INVOKE_URL_ENV) if not resolved_url or not resolved_url.strip(): - raise ValueError(f"{LUNA_INVOKE_URL_ENV} is required for galileo.jev") + raise ValueError(f"{LUNA_INVOKE_URL_ENV} is required for galileo.decision_models") self.api_secret = resolved_secret self.invoke_url = _normalize_invoke_url(resolved_url) self.runtime_enabled = scorer_invoke_runtime_enabled() @@ -140,15 +141,15 @@ async def invoke( scorer_label: str | None, step: Step, execution_context: GalileoExecutionContext, - inputs: JevInvokeInputs, + inputs: DecisionModelsInvokeInputs, timeout: float, selected_data: Any = None, - payload_field: JevPayloadField = "input", - ) -> JevInvokeResponse: + payload_field: DecisionModelsPayloadField = "input", + ) -> DecisionModelsInvokeResponse: """Invoke the pinned scorer with the complete Galileo Core record.""" if not self.runtime_enabled: raise RuntimeError( - "Galileo scorer-invoke runtime is disabled; galileo.jev requires " + "Galileo scorer-invoke runtime is disabled; galileo.decision_models requires " "GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME=enabled." ) record = record_from_step( @@ -156,7 +157,7 @@ async def invoke( selected_data=selected_data, payload_field=payload_field, ).model_dump(mode="json", exclude_none=True) - request = JevInvokeRequest( + request = DecisionModelsInvokeRequest( scorer_id=scorer_id, scorer_version_id=scorer_version_id, scorer_label=scorer_label, @@ -168,14 +169,16 @@ async def invoke( response = await self._client.post( self.invoke_url, json=request.to_dict(), - headers={"Authorization": f"Bearer {_jev_internal_auth_token(self.api_secret)}"}, + headers={ + "Authorization": f"Bearer {_decision_models_internal_auth_token(self.api_secret)}" + }, timeout=timeout, ) response.raise_for_status() body: Any = response.json() if not isinstance(body, dict): - raise RuntimeError("Invalid Jev scorer response: expected a JSON object") - return JevInvokeResponse.model_validate(body) + raise RuntimeError("Invalid Decision Models scorer response: expected a JSON object") + return DecisionModelsInvokeResponse.model_validate(body) async def close(self) -> None: """Close the underlying HTTP client.""" diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/config.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py similarity index 69% rename from evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/config.py rename to evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py index c8f6c993..3daf8e80 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/config.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py @@ -1,4 +1,4 @@ -"""Configuration for Galileo Jev decision scorer evaluation.""" +"""Configuration for Galileo Decision Models scoring.""" from __future__ import annotations @@ -8,27 +8,27 @@ from agent_control_models import JSONValue from pydantic import Field, model_validator -JevOperator = Literal["gt", "gte", "lt", "lte", "eq", "ne", "contains", "any"] -JevPayloadField = Literal["input", "output"] +DecisionModelsOperator = Literal["gt", "gte", "lt", "lte", "eq", "ne", "contains", "any"] +DecisionModelsPayloadField = Literal["input", "output"] _NUMERIC_OPERATORS = frozenset({"gt", "gte", "lt", "lte"}) -class JevEvaluatorConfig(EvaluatorConfig): - """Pinned Jev scorer and local result comparison settings.""" +class DecisionModelsEvaluatorConfig(EvaluatorConfig): + """Pinned Decision Models scorer and local result comparison settings.""" scorer_id: str = Field(min_length=1) scorer_version_id: str = Field(min_length=1) scorer_label: str | None = Field(default=None, min_length=1) threshold: JSONValue = 0.5 - operator: JevOperator = "gte" - payload_field: JevPayloadField = Field( + operator: DecisionModelsOperator = "gte" + payload_field: DecisionModelsPayloadField = Field( default="input", description="Record side to use when the selector returns a scalar value.", ) timeout_ms: int = Field(default=10000, ge=1, le=60000) @model_validator(mode="after") - def validate_threshold(self) -> JevEvaluatorConfig: + def validate_threshold(self) -> DecisionModelsEvaluatorConfig: """Ensure ordered comparisons use a numeric threshold.""" if self.operator in _NUMERIC_OPERATORS and ( isinstance(self.threshold, bool) or not isinstance(self.threshold, (int, float)) diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/evaluator.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py similarity index 79% rename from evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/evaluator.py rename to evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py index 4ada5469..84ce2f2b 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/evaluator.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py @@ -1,4 +1,4 @@ -"""Galileo Jev decision-model evaluator.""" +"""Galileo Decision Models evaluator.""" from __future__ import annotations @@ -12,8 +12,12 @@ from ..luna.client import GalileoExecutionContext from ..luna.config import scorer_invoke_runtime_enabled -from .client import GalileoJevClient, JevInvokeInputs, JevInvokeResponse -from .config import JevEvaluatorConfig +from .client import ( + DecisionModelsInvokeInputs, + DecisionModelsInvokeResponse, + GalileoDecisionModelsClient, +) +from .config import DecisionModelsEvaluatorConfig logger = logging.getLogger(__name__) @@ -34,32 +38,34 @@ def _contains(value: JSONValue, threshold: JSONValue) -> bool: @register_evaluator -class JevEvaluator(Evaluator[JevEvaluatorConfig]): - """Evaluate data using a pinned Galileo Jev decision scorer.""" +class DecisionModelsEvaluator(Evaluator[DecisionModelsEvaluatorConfig]): + """Evaluate data using a pinned Galileo Decision Models scorer.""" metadata = EvaluatorMetadata( - name="galileo.jev", + name="galileo.decision_models", version=_PACKAGE_VERSION, - description="Galileo Jev decision scorer evaluation", + description="Galileo Decision Models scoring", requires_api_key=True, timeout_ms=10000, ) - config_model = JevEvaluatorConfig + config_model = DecisionModelsEvaluatorConfig - def __init__(self, config: JevEvaluatorConfig) -> None: + def __init__(self, config: DecisionModelsEvaluatorConfig) -> None: super().__init__(config) - self._client = GalileoJevClient() + self._client = GalileoDecisionModelsClient() async def evaluate(self, data: Any) -> EvaluatorResult: - """Return an actionable error because Jev requires runtime context.""" + """Return an actionable error because Decision Models requires runtime context.""" if not scorer_invoke_runtime_enabled() or not self._client.runtime_enabled: return self._error( RuntimeError( - "Galileo scorer-invoke runtime is disabled; galileo.jev requires " + "Galileo scorer-invoke runtime is disabled; galileo.decision_models requires " "GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME=enabled." ) ) - return self._error(RuntimeError("galileo.jev requires Step context and execution_context")) + return self._error( + RuntimeError("galileo.decision_models requires Step context and execution_context") + ) async def evaluate_with_context(self, data: Any, step: Step) -> EvaluatorResult: """Evaluate a step when an authenticated execution context is supplied by extensions.""" @@ -72,7 +78,7 @@ async def evaluate_with_extensions( try: context = self._execution_context(extensions) if context is None: - raise ValueError("galileo.jev requires execution_context") + raise ValueError("galileo.decision_models requires execution_context") return await self._evaluate(data, step, context) except Exception as exc: return self._error(exc) @@ -83,11 +89,11 @@ async def _evaluate( try: if not scorer_invoke_runtime_enabled() or not self._client.runtime_enabled: raise RuntimeError( - "Galileo scorer-invoke runtime is disabled; galileo.jev requires " + "Galileo scorer-invoke runtime is disabled; galileo.decision_models requires " "GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME=enabled." ) if execution_context is None: - raise ValueError("galileo.jev requires execution_context") + raise ValueError("galileo.decision_models requires execution_context") inputs = self._inputs(data, step) response = await self._client.invoke( scorer_id=self.config.scorer_id, @@ -102,10 +108,10 @@ async def _evaluate( ) if response.status.lower() != "success": raise RuntimeError( - response.error_message or f"Jev scorer status: {response.status}" + response.error_message or f"Decision Models scorer status: {response.status}" ) if response.score is None: - raise RuntimeError("Jev scorer response did not include a score") + raise RuntimeError("Decision Models scorer response did not include a score") matched = self._matches(response.score) metadata = self._metadata(response) confidence = ( @@ -117,14 +123,14 @@ async def _evaluate( matched=matched, confidence=confidence, message=( - f"Jev result {response.score!r} {self.config.operator} " + f"Decision Models result {response.score!r} {self.config.operator} " f"threshold {self.config.threshold!r}: control " f"{'triggered' if matched else 'not triggered'}." ), metadata=metadata, ) except Exception as exc: - logger.error("Jev evaluation error: %s", exc, exc_info=True) + logger.error("Decision Models evaluation error: %s", exc, exc_info=True) return self._error(exc) @staticmethod @@ -149,7 +155,7 @@ def _execution_context(extensions: JSONObject | None) -> GalileoExecutionContext ) @staticmethod - def _inputs(data: Any, step: Step) -> JevInvokeInputs: + def _inputs(data: Any, step: Step) -> DecisionModelsInvokeInputs: """Build the legacy inputs without rejecting valid child-only records.""" if isinstance(data, dict) and ("input" in data or "output" in data): query = data.get("input", "") @@ -158,7 +164,7 @@ def _inputs(data: Any, step: Step) -> JevInvokeInputs: query, response = data, "" else: query, response = step.input, step.output if step.output is not None else "" - return JevInvokeInputs( + return DecisionModelsInvokeInputs( query=query, response=response, ground_truth=step.ground_truth, @@ -177,9 +183,11 @@ def _matches(self, value: JSONValue) -> bool: if operator == "contains": return _contains(value, threshold) if isinstance(value, bool) or not isinstance(value, (int, float)): - raise ValueError(f"Jev result {value!r} is not numeric for operator '{operator}'") + raise ValueError( + f"Decision Models result {value!r} is not numeric for operator '{operator}'" + ) if isinstance(threshold, bool) or not isinstance(threshold, (int, float)): - raise ValueError(f"Jev threshold {threshold!r} is not numeric") + raise ValueError(f"Decision Models threshold {threshold!r} is not numeric") return { "gt": value > threshold, "gte": value >= threshold, @@ -187,7 +195,7 @@ def _matches(self, value: JSONValue) -> bool: "lte": value <= threshold, }[operator] - def _metadata(self, response: JevInvokeResponse) -> dict[str, Any]: + def _metadata(self, response: DecisionModelsInvokeResponse) -> dict[str, Any]: return { "scorer_id": self.config.scorer_id, "scorer_version_id": self.config.scorer_version_id, @@ -212,11 +220,11 @@ def _error(self, error: Exception) -> EvaluatorResult: return EvaluatorResult( matched=False, confidence=0.0, - message=f"Jev evaluation error: {detail}", + message=f"Decision Models evaluation error: {detail}", metadata=metadata, error=detail, ) async def aclose(self) -> None: - """Close the Jev HTTP client.""" + """Close the Decision Models HTTP client.""" await self._client.close() diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/__init__.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/__init__.py deleted file mode 100644 index 9891fc6b..00000000 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/jev/__init__.py +++ /dev/null @@ -1,16 +0,0 @@ -"""Galileo Jev decision scorer evaluator.""" - -from .client import GalileoJevClient, JevInvokeInputs, JevInvokeRequest, JevInvokeResponse -from .config import JevEvaluatorConfig, JevOperator, JevPayloadField -from .evaluator import JevEvaluator - -__all__ = [ - "GalileoJevClient", - "JevEvaluator", - "JevEvaluatorConfig", - "JevInvokeInputs", - "JevInvokeRequest", - "JevInvokeResponse", - "JevOperator", - "JevPayloadField", -] diff --git a/evaluators/contrib/galileo/tests/test_jev_evaluator.py b/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py similarity index 88% rename from evaluators/contrib/galileo/tests/test_jev_evaluator.py rename to evaluators/contrib/galileo/tests/test_decision_models_evaluator.py index b50e6f09..1dc8603d 100644 --- a/evaluators/contrib/galileo/tests/test_jev_evaluator.py +++ b/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py @@ -1,4 +1,4 @@ -"""Tests for the Galileo Jev evaluator and Runners client.""" +"""Tests for the Galileo Decision Models evaluator and Runners client.""" from __future__ import annotations @@ -15,17 +15,19 @@ import httpx import pytest -from agent_control_evaluator_galileo.jev import ( - GalileoJevClient, - JevEvaluator, - JevEvaluatorConfig, - JevInvokeInputs, - JevInvokeRequest, - JevInvokeResponse, +from agent_control_evaluator_galileo.decision_models import ( + DecisionModelsEvaluator, + DecisionModelsEvaluatorConfig, + DecisionModelsInvokeInputs, + DecisionModelsInvokeRequest, + DecisionModelsInvokeResponse, + GalileoDecisionModelsClient, ) -from agent_control_evaluator_galileo.jev import evaluator as jev_evaluator_module -from agent_control_evaluator_galileo.jev.client import _normalize_invoke_url -from agent_control_evaluator_galileo.jev.evaluator import _contains +from agent_control_evaluator_galileo.decision_models import ( + evaluator as decision_models_evaluator_module, +) +from agent_control_evaluator_galileo.decision_models.client import _normalize_invoke_url +from agent_control_evaluator_galileo.decision_models.evaluator import _contains from agent_control_evaluator_galileo.luna.client import ( GalileoExecutionContext, ScorerInvokeInputs, @@ -48,26 +50,26 @@ def _decode_jwt_segment(segment: str) -> dict[str, Any]: return json.loads(urlsafe_b64decode(padded.encode()).decode()) -def _evaluator(monkeypatch: pytest.MonkeyPatch, **config: Any) -> JevEvaluator: +def _evaluator(monkeypatch: pytest.MonkeyPatch, **config: Any) -> DecisionModelsEvaluator: monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") monkeypatch.setenv("GALILEO_LUNA_INVOKE_URL", "http://runners:8090") monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") - evaluator = JevEvaluator( - JevEvaluatorConfig(scorer_id="scorer-a", scorer_version_id="version-a", **config) + evaluator = DecisionModelsEvaluator( + DecisionModelsEvaluatorConfig(scorer_id="scorer-a", scorer_version_id="version-a", **config) ) return evaluator -def test_jev_registration_and_required_configuration() -> None: - # Given: the separate Jev entry point and required pinned scorer config - # Then: Jev is registered independently and both scorer identifiers are required +def test_decision_models_registration_and_required_configuration() -> None: + # Given: the separate Decision Models entry point and required pinned scorer config + # Then: Decision Models is registered independently and both scorer identifiers are required from agent_control_evaluators import get_evaluator - assert get_evaluator("galileo.jev") is JevEvaluator + assert get_evaluator("galileo.decision_models") is DecisionModelsEvaluator with pytest.raises(ValidationError, match="scorer_id"): - JevEvaluatorConfig(scorer_version_id="version-a") # type: ignore[call-arg] + DecisionModelsEvaluatorConfig(scorer_version_id="version-a") # type: ignore[call-arg] with pytest.raises(ValidationError, match="scorer_version_id"): - JevEvaluatorConfig(scorer_id="scorer-a") # type: ignore[call-arg] + DecisionModelsEvaluatorConfig(scorer_id="scorer-a") # type: ignore[call-arg] def test_missing_distribution_metadata_uses_development_version( @@ -82,8 +84,8 @@ def missing_distribution(_: str) -> str: monkeypatch.setattr(importlib_metadata, "version", missing_distribution) try: clear_evaluators() - reloaded_module = reload(jev_evaluator_module) - assert reloaded_module.JevEvaluator.metadata.version == "0.0.0.dev" + reloaded_module = reload(decision_models_evaluator_module) + assert reloaded_module.DecisionModelsEvaluator.metadata.version == "0.0.0.dev" finally: clear_evaluators() for evaluator_class in registered_evaluators.values(): @@ -107,14 +109,14 @@ def test_contains_comparison_handles_supported_and_unsupported_values( def test_config_rejects_non_numeric_or_missing_thresholds() -> None: with pytest.raises(ValidationError, match="requires a numeric threshold"): - JevEvaluatorConfig( + DecisionModelsEvaluatorConfig( scorer_id="scorer-a", scorer_version_id="version-a", operator="gt", threshold=True, ) with pytest.raises(ValidationError, match="threshold is required"): - JevEvaluatorConfig( + DecisionModelsEvaluatorConfig( scorer_id="scorer-a", scorer_version_id="version-a", operator="eq", @@ -123,7 +125,7 @@ def test_config_rejects_non_numeric_or_missing_thresholds() -> None: def test_config_allows_missing_threshold_for_any_operator() -> None: - config = JevEvaluatorConfig( + config = DecisionModelsEvaluatorConfig( scorer_id="scorer-a", scorer_version_id="version-a", operator="any", @@ -140,8 +142,8 @@ async def test_client_url_resolution_prefers_explicit_url_then_luna_environment( monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") monkeypatch.setenv("GALILEO_LUNA_INVOKE_URL", "http://luna:8090") - explicit_client = GalileoJevClient(invoke_url="http://explicit:8090") - environment_client = GalileoJevClient() + explicit_client = GalileoDecisionModelsClient(invoke_url="http://explicit:8090") + environment_client = GalileoDecisionModelsClient() expected_path = "/api/v1/scorers/invoke" assert explicit_client.invoke_url == f"http://explicit:8090{expected_path}" @@ -157,7 +159,7 @@ def test_client_requires_luna_invoke_url_when_no_explicit_url_is_given( monkeypatch.delenv("GALILEO_LUNA_INVOKE_URL", raising=False) with pytest.raises(ValueError, match="GALILEO_LUNA_INVOKE_URL"): - GalileoJevClient() + GalileoDecisionModelsClient() def test_invoke_request_requires_execution_context_and_serializes_nested_record() -> None: @@ -170,12 +172,12 @@ def test_invoke_request_requires_execution_context_and_serializes_nested_record( "output": "", "spans": [{"type": "llm", "input": [{"role": "user", "content": "q"}]}], } - request = JevInvokeRequest( + request = DecisionModelsInvokeRequest( scorer_id="scorer-a", scorer_version_id="version-a", execution_context=context, record=record, - inputs=JevInvokeInputs(), + inputs=DecisionModelsInvokeInputs(), config={"request_timeout_seconds": 8.0}, ) @@ -202,7 +204,7 @@ def test_invoke_request_requires_execution_context_and_serializes_nested_record( (["unsafe"], "unsafe", "contains", True), ], ) -async def test_jev_applies_local_comparison( +async def test_decision_models_applies_local_comparison( monkeypatch: pytest.MonkeyPatch, score: Any, threshold: Any, @@ -211,7 +213,7 @@ async def test_jev_applies_local_comparison( ) -> None: evaluator = _evaluator(monkeypatch, threshold=threshold, operator=operator) evaluator._client.invoke = AsyncMock( - return_value=JevInvokeResponse(score=score, status="success") + return_value=DecisionModelsInvokeResponse(score=score, status="success") ) result = await evaluator.evaluate_with_extensions( "", @@ -232,7 +234,7 @@ async def test_child_only_trace_keeps_nested_span_and_execution_identity( monkeypatch: pytest.MonkeyPatch, ) -> None: evaluator = _evaluator(monkeypatch) - invoke = AsyncMock(return_value=JevInvokeResponse(score=0.82, status="success")) + invoke = AsyncMock(return_value=DecisionModelsInvokeResponse(score=0.82, status="success")) evaluator._client.invoke = invoke step = Step( type="trace", @@ -268,7 +270,7 @@ async def test_supported_record_types_are_built_from_step_children( monkeypatch: pytest.MonkeyPatch, record_type: str ) -> None: evaluator = _evaluator(monkeypatch) - invoke = AsyncMock(return_value=JevInvokeResponse(score="yes", status="success")) + invoke = AsyncMock(return_value=DecisionModelsInvokeResponse(score="yes", status="success")) evaluator._client.invoke = invoke if record_type == "session": step = Step( @@ -352,7 +354,7 @@ async def test_client_serializes_full_record_context_and_uses_timeout( monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") monkeypatch.setenv("GALILEO_LUNA_INVOKE_URL", "http://runners:8090") monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") - client = GalileoJevClient() + client = GalileoDecisionModelsClient() seen: dict[str, Any] = {} async def handler(request: httpx.Request) -> httpx.Response: @@ -423,7 +425,7 @@ async def handler(request: httpx.Request) -> httpx.Response: scorer_label=None, step=step, execution_context=GalileoExecutionContext(organization_id="org-1", user_id="user-7"), - inputs=JevInvokeInputs(), + inputs=DecisionModelsInvokeInputs(), timeout=5.0, ) assert response.score == 0.9 @@ -625,7 +627,7 @@ async def test_successful_zero_and_false_scores_are_valid( ) -> None: evaluator = _evaluator(monkeypatch, threshold=threshold, operator="eq") evaluator._client.invoke = AsyncMock( - return_value=JevInvokeResponse(score=score, status="success") + return_value=DecisionModelsInvokeResponse(score=score, status="success") ) result = await evaluator.evaluate_with_extensions( "input", Step(type="llm", name="answer", input="input"), EXTENSIONS @@ -640,13 +642,13 @@ async def test_successful_response_without_score_is_an_error( ) -> None: evaluator = _evaluator(monkeypatch) evaluator._client.invoke = AsyncMock( - return_value=JevInvokeResponse(status="success", score=None) + return_value=DecisionModelsInvokeResponse(status="success", score=None) ) result = await evaluator.evaluate_with_extensions( "input", Step(type="llm", name="answer", input="input"), EXTENSIONS ) assert result.matched is False - assert result.error == "Jev scorer response did not include a score" + assert result.error == "Decision Models scorer response did not include a score" await evaluator.aclose() @@ -656,7 +658,7 @@ async def test_non_success_response_without_score_uses_provider_status_error( ) -> None: evaluator = _evaluator(monkeypatch) evaluator._client.invoke = AsyncMock( - return_value=JevInvokeResponse( + return_value=DecisionModelsInvokeResponse( status="error", score=None, error_message="provider rejected request" ) ) @@ -720,13 +722,13 @@ def test_numeric_comparison_rejects_non_numeric_threshold_from_unvalidated_confi ) -> None: monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") monkeypatch.setenv("GALILEO_LUNA_INVOKE_URL", "http://runners:8090") - config = JevEvaluatorConfig.model_construct( + config = DecisionModelsEvaluatorConfig.model_construct( scorer_id="scorer-a", scorer_version_id="version-a", threshold="not numeric", operator="gte", ) - evaluator = JevEvaluator(config) + evaluator = DecisionModelsEvaluator(config) with pytest.raises(ValueError, match="threshold .* is not numeric"): evaluator._matches(0.8) @@ -756,7 +758,7 @@ def test_client_falls_back_to_legacy_api_secret_environment_variable( monkeypatch.delenv("GALILEO_API_SECRET_KEY", raising=False) monkeypatch.setenv("GALILEO_API_SECRET", "legacy-secret") - client = GalileoJevClient(invoke_url="http://runners:8090") + client = GalileoDecisionModelsClient(invoke_url="http://runners:8090") assert client.api_secret == "legacy-secret" @@ -768,7 +770,7 @@ def test_client_requires_api_secret_when_environment_variables_are_missing( monkeypatch.delenv("GALILEO_API_SECRET", raising=False) with pytest.raises(ValueError, match="GALILEO_API_SECRET_KEY or GALILEO_API_SECRET"): - GalileoJevClient(invoke_url="http://runners:8090") + GalileoDecisionModelsClient(invoke_url="http://runners:8090") @pytest.mark.asyncio @@ -777,7 +779,7 @@ async def test_client_rejects_invoke_when_runtime_is_disabled( ) -> None: monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "disabled") - client = GalileoJevClient(invoke_url="http://runners:8090") + client = GalileoDecisionModelsClient(invoke_url="http://runners:8090") with pytest.raises(RuntimeError, match="runtime is disabled"): await client.invoke( @@ -786,7 +788,7 @@ async def test_client_rejects_invoke_when_runtime_is_disabled( scorer_label=None, step=Step(type="llm", name="answer", input="question"), execution_context=GalileoExecutionContext(organization_id="org-1"), - inputs=JevInvokeInputs(), + inputs=DecisionModelsInvokeInputs(), timeout=5.0, ) @@ -799,7 +801,7 @@ async def test_client_rejects_non_object_invoke_response( ) -> None: monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") - client = GalileoJevClient(invoke_url="http://runners:8090") + client = GalileoDecisionModelsClient(invoke_url="http://runners:8090") async def handler(_: httpx.Request) -> httpx.Response: return httpx.Response(200, json=[{"score": 0.9}]) @@ -814,14 +816,14 @@ async def handler(_: httpx.Request) -> httpx.Response: scorer_label=None, step=Step(type="llm", name="answer", input="question"), execution_context=GalileoExecutionContext(organization_id="org-1"), - inputs=JevInvokeInputs(), + inputs=DecisionModelsInvokeInputs(), timeout=5.0, ) await client.close() -def test_luna_and_jev_request_shape_parity() -> None: +def test_luna_and_decision_models_request_shape_parity() -> None: context = GalileoExecutionContext(organization_id="org-1", user_id="user-7") luna = ScorerInvokeRequest( scorer_id="scorer-a", @@ -830,13 +832,13 @@ def test_luna_and_jev_request_shape_parity() -> None: record={"type": "llm", "input": "q", "output": "a"}, execution_context=context, ).to_dict() - jev = JevInvokeRequest( + decision_models = DecisionModelsInvokeRequest( scorer_id="scorer-a", scorer_version_id="version-a", - inputs=JevInvokeInputs(query="q", response="a"), + inputs=DecisionModelsInvokeInputs(query="q", response="a"), record={"type": "llm", "input": "q", "output": "a"}, execution_context=context, ).to_dict() - assert set(luna) == set(jev) - assert set(luna["inputs"]) == set(jev["inputs"]) - assert luna["execution_context"] == jev["execution_context"] + assert set(luna) == set(decision_models) + assert set(luna["inputs"]) == set(decision_models["inputs"]) + assert luna["execution_context"] == decision_models["execution_context"] From 88f5bcfaf269105eead28f99ac56f642e40d4179 Mon Sep 17 00:00:00 2001 From: Namrata Ghadi Date: Wed, 21 Oct 2026 10:10:33 -0700 Subject: [PATCH 5/7] refactor: decouple Decision Models from Luna --- .../decision_models/__init__.py | 2 ++ .../decision_models/client.py | 17 +++++++++++----- .../decision_models/config.py | 7 +++++++ .../decision_models/evaluator.py | 11 +++++----- .../tests/test_decision_models_evaluator.py | 20 ++++++++++++------- 5 files changed, 39 insertions(+), 18 deletions(-) diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/__init__.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/__init__.py index 90c844e4..1b369eac 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/__init__.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/__init__.py @@ -1,6 +1,7 @@ """Galileo Decision Models evaluator.""" from .client import ( + DecisionModelsExecutionContext, DecisionModelsInvokeInputs, DecisionModelsInvokeRequest, DecisionModelsInvokeResponse, @@ -14,6 +15,7 @@ from .evaluator import DecisionModelsEvaluator __all__ = [ + "DecisionModelsExecutionContext", "GalileoDecisionModelsClient", "DecisionModelsEvaluator", "DecisionModelsEvaluatorConfig", diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py index a592d351..9efc3d4b 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py @@ -16,10 +16,8 @@ from agent_control_models import JSONObject, JSONValue, Step from pydantic import BaseModel, Field -from ..luna.client import GalileoExecutionContext -from ..luna.config import scorer_invoke_runtime_enabled from ..records import record_from_step -from .config import DecisionModelsPayloadField +from .config import DecisionModelsPayloadField, scorer_invoke_runtime_enabled logger = logging.getLogger(__name__) @@ -30,6 +28,15 @@ AUTH_UPSTREAM_CA_FILE_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_CA_FILE" +class DecisionModelsExecutionContext(BaseModel): + """Authenticated organization and optional identity fields for scorer invocation.""" + + organization_id: str = Field(min_length=1) + user_id: str | None = Field(default=None, min_length=1) + project_id: str | None = Field(default=None, min_length=1) + run_id: str | None = Field(default=None, min_length=1) + + def _b64url(data: bytes) -> str: return urlsafe_b64encode(data).rstrip(b"=").decode("ascii") @@ -81,7 +88,7 @@ class DecisionModelsInvokeRequest(BaseModel): scorer_id: str = Field(min_length=1) scorer_version_id: str = Field(min_length=1) scorer_label: str | None = None - execution_context: GalileoExecutionContext + execution_context: DecisionModelsExecutionContext record: JSONObject inputs: DecisionModelsInvokeInputs config: JSONObject = Field(default_factory=dict) @@ -140,7 +147,7 @@ async def invoke( scorer_version_id: str, scorer_label: str | None, step: Step, - execution_context: GalileoExecutionContext, + execution_context: DecisionModelsExecutionContext, inputs: DecisionModelsInvokeInputs, timeout: float, selected_data: Any = None, diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py index 3daf8e80..45f1e29a 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os from typing import Literal from agent_control_evaluators import EvaluatorConfig @@ -11,6 +12,12 @@ DecisionModelsOperator = Literal["gt", "gte", "lt", "lte", "eq", "ne", "contains", "any"] DecisionModelsPayloadField = Literal["input", "output"] _NUMERIC_OPERATORS = frozenset({"gt", "gte", "lt", "lte"}) +SCORER_INVOKE_RUNTIME_FLAG_ENV = "GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME" + + +def scorer_invoke_runtime_enabled() -> bool: + """Return whether structured runtime data is enabled for Decision Models requests.""" + return os.getenv(SCORER_INVOKE_RUNTIME_FLAG_ENV) == "enabled" class DecisionModelsEvaluatorConfig(EvaluatorConfig): diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py index 84ce2f2b..0545d1c5 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py @@ -10,14 +10,13 @@ from agent_control_evaluators import Evaluator, EvaluatorMetadata, register_evaluator from agent_control_models import EvaluatorResult, JSONObject, JSONValue, Step -from ..luna.client import GalileoExecutionContext -from ..luna.config import scorer_invoke_runtime_enabled from .client import ( + DecisionModelsExecutionContext, DecisionModelsInvokeInputs, DecisionModelsInvokeResponse, GalileoDecisionModelsClient, ) -from .config import DecisionModelsEvaluatorConfig +from .config import DecisionModelsEvaluatorConfig, scorer_invoke_runtime_enabled logger = logging.getLogger(__name__) @@ -84,7 +83,7 @@ async def evaluate_with_extensions( return self._error(exc) async def _evaluate( - self, data: Any, step: Step, execution_context: GalileoExecutionContext | None + self, data: Any, step: Step, execution_context: DecisionModelsExecutionContext | None ) -> EvaluatorResult: try: if not scorer_invoke_runtime_enabled() or not self._client.runtime_enabled: @@ -134,7 +133,7 @@ async def _evaluate( return self._error(exc) @staticmethod - def _execution_context(extensions: JSONObject | None) -> GalileoExecutionContext | None: + def _execution_context(extensions: JSONObject | None) -> DecisionModelsExecutionContext | None: if extensions is None: return None metadata = extensions.get("metadata") @@ -147,7 +146,7 @@ def _execution_context(extensions: JSONObject | None) -> GalileoExecutionContext project_id = metadata_obj.get("project_id") if not isinstance(organization_id, str) or not organization_id: raise ValueError("Authenticated execution metadata is missing organization_id") - return GalileoExecutionContext( + return DecisionModelsExecutionContext( organization_id=organization_id, user_id=user_id if isinstance(user_id, str) and user_id else None, project_id=project_id if isinstance(project_id, str) else None, diff --git a/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py b/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py index 1dc8603d..d48c4c7a 100644 --- a/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py +++ b/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py @@ -16,6 +16,7 @@ import httpx import pytest from agent_control_evaluator_galileo.decision_models import ( + DecisionModelsExecutionContext, DecisionModelsEvaluator, DecisionModelsEvaluatorConfig, DecisionModelsInvokeInputs, @@ -163,7 +164,7 @@ def test_client_requires_luna_invoke_url_when_no_explicit_url_is_given( def test_invoke_request_requires_execution_context_and_serializes_nested_record() -> None: - context = GalileoExecutionContext( + context = DecisionModelsExecutionContext( organization_id="org-1", user_id="user-7", project_id="project-3", run_id="run-9" ) record = { @@ -424,7 +425,9 @@ async def handler(request: httpx.Request) -> httpx.Response: scorer_version_id="version-a", scorer_label=None, step=step, - execution_context=GalileoExecutionContext(organization_id="org-1", user_id="user-7"), + execution_context=DecisionModelsExecutionContext( + organization_id="org-1", user_id="user-7" + ), inputs=DecisionModelsInvokeInputs(), timeout=5.0, ) @@ -787,7 +790,7 @@ async def test_client_rejects_invoke_when_runtime_is_disabled( scorer_version_id="version-a", scorer_label=None, step=Step(type="llm", name="answer", input="question"), - execution_context=GalileoExecutionContext(organization_id="org-1"), + execution_context=DecisionModelsExecutionContext(organization_id="org-1"), inputs=DecisionModelsInvokeInputs(), timeout=5.0, ) @@ -815,7 +818,7 @@ async def handler(_: httpx.Request) -> httpx.Response: scorer_version_id="version-a", scorer_label=None, step=Step(type="llm", name="answer", input="question"), - execution_context=GalileoExecutionContext(organization_id="org-1"), + execution_context=DecisionModelsExecutionContext(organization_id="org-1"), inputs=DecisionModelsInvokeInputs(), timeout=5.0, ) @@ -824,20 +827,23 @@ async def handler(_: httpx.Request) -> httpx.Response: def test_luna_and_decision_models_request_shape_parity() -> None: - context = GalileoExecutionContext(organization_id="org-1", user_id="user-7") + luna_context = GalileoExecutionContext(organization_id="org-1", user_id="user-7") + decision_models_context = DecisionModelsExecutionContext( + organization_id="org-1", user_id="user-7" + ) luna = ScorerInvokeRequest( scorer_id="scorer-a", scorer_version_id="version-a", inputs=ScorerInvokeInputs(query="q", response="a"), record={"type": "llm", "input": "q", "output": "a"}, - execution_context=context, + execution_context=luna_context, ).to_dict() decision_models = DecisionModelsInvokeRequest( scorer_id="scorer-a", scorer_version_id="version-a", inputs=DecisionModelsInvokeInputs(query="q", response="a"), record={"type": "llm", "input": "q", "output": "a"}, - execution_context=context, + execution_context=decision_models_context, ).to_dict() assert set(luna) == set(decision_models) assert set(luna["inputs"]) == set(decision_models["inputs"]) From 9aa041ad3b5f4ff24430bb3122979d2ed45f4a31 Mon Sep 17 00:00:00 2001 From: Namrata Ghadi Date: Thu, 22 Oct 2026 13:36:56 -0700 Subject: [PATCH 6/7] fix(galileo): address decision models review feedback --- .../decision_models/client.py | 116 +++++++++++++++++- .../decision_models/evaluator.py | 29 ++++- .../tests/test_decision_models_evaluator.py | 92 +++++++++++++- 3 files changed, 222 insertions(+), 15 deletions(-) diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py index 9efc3d4b..e5db9cc2 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py @@ -4,6 +4,7 @@ import logging import os +from asyncio import Lock from base64 import urlsafe_b64encode from hashlib import sha256 from hmac import new as hmac_new @@ -26,6 +27,62 @@ DEFAULT_INVOKE_PATH = "/api/v1/scorers/invoke" LUNA_INVOKE_URL_ENV = "GALILEO_LUNA_INVOKE_URL" AUTH_UPSTREAM_CA_FILE_ENV = "AGENT_CONTROL_AUTH_UPSTREAM_CA_FILE" +DEFAULT_KEEPALIVE_EXPIRY_SECS = 1.0 +DEFAULT_MAX_CONNECTIONS = 100 +DEFAULT_MAX_KEEPALIVE_CONNECTIONS = 20 +DEFAULT_CLIENT_POOL_SIZE = 1 +LUNA_KEEPALIVE_EXPIRY_ENV = "GALILEO_LUNA_KEEPALIVE_EXPIRY_SECONDS" +LUNA_MAX_CONNECTIONS_ENV = "GALILEO_LUNA_MAX_CONNECTIONS" +LUNA_MAX_KEEPALIVE_CONNECTIONS_ENV = "GALILEO_LUNA_MAX_KEEPALIVE_CONNECTIONS" +LUNA_CLIENT_POOL_SIZE_ENV = "GALILEO_LUNA_CLIENT_POOL_SIZE" + + +def _load_float_env(env_name: str, default: float) -> float: + raw = os.getenv(env_name) + if raw is None or raw.strip() == "": + return default + try: + return float(raw) + except ValueError as exc: + raise ValueError(f"{env_name}={raw!r} is not a number.") from exc + + +def _load_int_env(env_name: str, default: int) -> int: + raw = os.getenv(env_name) + if raw is None or raw.strip() == "": + return default + try: + return int(raw) + except ValueError as exc: + raise ValueError(f"{env_name}={raw!r} is not an integer.") from exc + + +def _validate_connection_config( + *, + keepalive_expiry_seconds: float, + max_connections: int, + max_keepalive_connections: int, + client_pool_size: int, +) -> None: + if keepalive_expiry_seconds < 0: + raise ValueError( + f"{LUNA_KEEPALIVE_EXPIRY_ENV}={keepalive_expiry_seconds} " + "must be greater than or equal to 0." + ) + if max_connections <= 0: + raise ValueError(f"{LUNA_MAX_CONNECTIONS_ENV}={max_connections} must be greater than 0.") + if max_keepalive_connections < 0: + raise ValueError( + f"{LUNA_MAX_KEEPALIVE_CONNECTIONS_ENV}={max_keepalive_connections} " + "must be greater than or equal to 0." + ) + if max_keepalive_connections > max_connections: + raise ValueError( + f"{LUNA_MAX_KEEPALIVE_CONNECTIONS_ENV}={max_keepalive_connections} " + f"must be less than or equal to {LUNA_MAX_CONNECTIONS_ENV}={max_connections}." + ) + if client_pool_size <= 0: + raise ValueError(f"{LUNA_CLIENT_POOL_SIZE_ENV}={client_pool_size} must be greater than 0.") class DecisionModelsExecutionContext(BaseModel): @@ -135,11 +192,58 @@ def __init__( self.invoke_url = _normalize_invoke_url(resolved_url) self.runtime_enabled = scorer_invoke_runtime_enabled() self.timeout = timeout - self._client = httpx.AsyncClient( + self.keepalive_expiry_seconds = _load_float_env( + LUNA_KEEPALIVE_EXPIRY_ENV, DEFAULT_KEEPALIVE_EXPIRY_SECS + ) + self.max_connections = _load_int_env(LUNA_MAX_CONNECTIONS_ENV, DEFAULT_MAX_CONNECTIONS) + self.max_keepalive_connections = _load_int_env( + LUNA_MAX_KEEPALIVE_CONNECTIONS_ENV, DEFAULT_MAX_KEEPALIVE_CONNECTIONS + ) + self.client_pool_size = _load_int_env(LUNA_CLIENT_POOL_SIZE_ENV, DEFAULT_CLIENT_POOL_SIZE) + _validate_connection_config( + keepalive_expiry_seconds=self.keepalive_expiry_seconds, + max_connections=self.max_connections, + max_keepalive_connections=self.max_keepalive_connections, + client_pool_size=self.client_pool_size, + ) + self._client = self._create_client() + self._clients = [self._client] + self._next_client_index = 0 + self._client_lock = Lock() + + def _create_client(self) -> httpx.AsyncClient: + """Create an HTTP client with configured connection limits.""" + return httpx.AsyncClient( headers={"Content-Type": "application/json"}, + timeout=httpx.Timeout(self.timeout), + limits=httpx.Limits( + max_connections=self.max_connections, + max_keepalive_connections=self.max_keepalive_connections, + keepalive_expiry=self.keepalive_expiry_seconds, + ), verify=os.getenv(AUTH_UPSTREAM_CA_FILE_ENV) or True, ) + def _select_pooled_client(self) -> httpx.AsyncClient: + client = self._clients[self._next_client_index % len(self._clients)] + self._next_client_index = (self._next_client_index + 1) % len(self._clients) + return client + + async def _get_client(self) -> httpx.AsyncClient: + """Get or create the next HTTP client in the configured pool.""" + async with self._client_lock: + self._clients = [client for client in self._clients if not client.is_closed] + if self.client_pool_size == 1: + if self._client.is_closed: + self._client = self._create_client() + self._clients = [self._client] + return self._client + + while len(self._clients) < self.client_pool_size: + self._clients.append(self._create_client()) + self._client = self._clients[0] + return self._select_pooled_client() + async def invoke( self, *, @@ -173,7 +277,8 @@ async def invoke( inputs=inputs, config={"request_timeout_seconds": timeout * 0.8}, ) - response = await self._client.post( + client = await self._get_client() + response = await client.post( self.invoke_url, json=request.to_dict(), headers={ @@ -188,5 +293,8 @@ async def invoke( return DecisionModelsInvokeResponse.model_validate(body) async def close(self) -> None: - """Close the underlying HTTP client.""" - await self._client.aclose() + """Close all HTTP clients in the connection pool.""" + for client in self._clients: + if not client.is_closed: + await client.aclose() + self._clients.clear() diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py index 0545d1c5..73acb315 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py @@ -49,6 +49,11 @@ class DecisionModelsEvaluator(Evaluator[DecisionModelsEvaluatorConfig]): ) config_model = DecisionModelsEvaluatorConfig + @classmethod + def is_available(cls) -> bool: + """Expose Decision Models only when the scorer-invoke runtime is enabled.""" + return scorer_invoke_runtime_enabled() + def __init__(self, config: DecisionModelsEvaluatorConfig) -> None: super().__init__(config) self._client = GalileoDecisionModelsClient() @@ -63,11 +68,19 @@ async def evaluate(self, data: Any) -> EvaluatorResult: ) ) return self._error( - RuntimeError("galileo.decision_models requires Step context and execution_context") + RuntimeError( + "galileo.decision_models requires Step context and authenticated execution " + "metadata from runtime extensions" + ) ) async def evaluate_with_context(self, data: Any, step: Step) -> EvaluatorResult: - """Evaluate a step when an authenticated execution context is supplied by extensions.""" + """Evaluate a step when only step context is available. + + Decision Models also needs authenticated execution metadata, which this method does not + receive. The explicit error helps callers distinguish missing runtime metadata from a + missing step. + """ return await self._evaluate(data, step, None) async def evaluate_with_extensions( @@ -77,7 +90,10 @@ async def evaluate_with_extensions( try: context = self._execution_context(extensions) if context is None: - raise ValueError("galileo.decision_models requires execution_context") + raise ValueError( + "galileo.decision_models requires authenticated execution metadata from " + "runtime extensions" + ) return await self._evaluate(data, step, context) except Exception as exc: return self._error(exc) @@ -92,7 +108,10 @@ async def _evaluate( "GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME=enabled." ) if execution_context is None: - raise ValueError("galileo.decision_models requires execution_context") + raise ValueError( + "galileo.decision_models requires authenticated execution metadata from " + "runtime extensions; Step context alone is not sufficient" + ) inputs = self._inputs(data, step) response = await self._client.invoke( scorer_id=self.config.scorer_id, @@ -149,7 +168,7 @@ def _execution_context(extensions: JSONObject | None) -> DecisionModelsExecution return DecisionModelsExecutionContext( organization_id=organization_id, user_id=user_id if isinstance(user_id, str) and user_id else None, - project_id=project_id if isinstance(project_id, str) else None, + project_id=project_id if isinstance(project_id, str) and project_id else None, run_id=run_id if isinstance(run_id, str) and run_id else None, ) diff --git a/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py b/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py index d48c4c7a..43b0a47c 100644 --- a/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py +++ b/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py @@ -61,11 +61,14 @@ def _evaluator(monkeypatch: pytest.MonkeyPatch, **config: Any) -> DecisionModels return evaluator -def test_decision_models_registration_and_required_configuration() -> None: - # Given: the separate Decision Models entry point and required pinned scorer config - # Then: Decision Models is registered independently and both scorer identifiers are required - from agent_control_evaluators import get_evaluator +def test_decision_models_registration_and_required_configuration( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Given: the scorer-invoke runtime is available and pinned scorer config is required + from agent_control_evaluators import get_evaluator, register_evaluator + monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") + register_evaluator(DecisionModelsEvaluator) assert get_evaluator("galileo.decision_models") is DecisionModelsEvaluator with pytest.raises(ValidationError, match="scorer_id"): DecisionModelsEvaluatorConfig(scorer_version_id="version-a") # type: ignore[call-arg] @@ -73,6 +76,16 @@ def test_decision_models_registration_and_required_configuration() -> None: DecisionModelsEvaluatorConfig(scorer_id="scorer-a") # type: ignore[call-arg] +def test_decision_models_availability_follows_runtime_feature_flag( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", raising=False) + assert DecisionModelsEvaluator.is_available() is False + + monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") + assert DecisionModelsEvaluator.is_available() is True + + def test_missing_distribution_metadata_uses_development_version( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -327,10 +340,77 @@ async def test_runtime_disabled_and_missing_context_return_errors_without_fallba "question", Step(type="llm", name="answer", input=""), None ) assert missing_context.matched is False - assert "execution_context" in (missing_context.error or "") + assert "authenticated execution metadata" in (missing_context.error or "") + monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") + step_only_context = await evaluator.evaluate_with_context( + "question", Step(type="llm", name="answer", input="question") + ) + assert "Step context alone is not sufficient" in (step_only_context.error or "") await evaluator.aclose() +@pytest.mark.asyncio +async def test_empty_project_id_is_omitted_from_execution_context( + monkeypatch: pytest.MonkeyPatch, +) -> None: + evaluator = _evaluator(monkeypatch) + invoke = AsyncMock(return_value=DecisionModelsInvokeResponse(score=0.9, status="success")) + evaluator._client.invoke = invoke + extensions = {**EXTENSIONS, "metadata": {"project_id": ""}} + + result = await evaluator.evaluate_with_extensions( + "question", Step(type="llm", name="answer", input="question"), extensions + ) + + assert result.error is None + assert invoke.await_args.kwargs["execution_context"].project_id is None + await evaluator.aclose() + + +def test_client_applies_connection_pool_environment_overrides( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") + monkeypatch.setenv("GALILEO_LUNA_KEEPALIVE_EXPIRY_SECONDS", "0.25") + monkeypatch.setenv("GALILEO_LUNA_MAX_CONNECTIONS", "17") + monkeypatch.setenv("GALILEO_LUNA_MAX_KEEPALIVE_CONNECTIONS", "4") + monkeypatch.setenv("GALILEO_LUNA_CLIENT_POOL_SIZE", "3") + + client = GalileoDecisionModelsClient(invoke_url="http://runners:8090") + + assert client.keepalive_expiry_seconds == 0.25 + assert client.max_connections == 17 + assert client.max_keepalive_connections == 4 + assert client.client_pool_size == 3 + + +@pytest.mark.asyncio +async def test_client_pool_rotates_across_configured_http_clients( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") + monkeypatch.setenv("GALILEO_LUNA_CLIENT_POOL_SIZE", "2") + client = GalileoDecisionModelsClient(invoke_url="http://runners:8090") + + first = await client._get_client() + second = await client._get_client() + third = await client._get_client() + + assert first is third + assert first is not second + await client.close() + + +def test_client_rejects_invalid_connection_pool_environment( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") + monkeypatch.setenv("GALILEO_LUNA_MAX_CONNECTIONS", "0") + + with pytest.raises(ValueError, match="GALILEO_LUNA_MAX_CONNECTIONS"): + GalileoDecisionModelsClient(invoke_url="http://runners:8090") + + @pytest.mark.asyncio @pytest.mark.parametrize("error", [TimeoutError("deadline"), RuntimeError("provider failed")]) async def test_timeout_and_provider_errors_are_non_matching( @@ -684,7 +764,7 @@ async def test_direct_evaluation_and_context_method_report_missing_context( context_result = await evaluator.evaluate_with_context("question", step) assert "requires Step context" in (direct_result.error or "") - assert "requires execution_context" in (context_result.error or "") + assert "Step context alone is not sufficient" in (context_result.error or "") await evaluator.aclose() From 3c665dd3f5c13c04a2429de2a1dfbbafcc345dfd Mon Sep 17 00:00:00 2001 From: Namrata Ghadi Date: Thu, 22 Oct 2026 15:54:15 -0700 Subject: [PATCH 7/7] feat(galileo): add decision models evaluator flag --- .../decision_models/config.py | 6 ++++++ .../decision_models/evaluator.py | 10 +++++++--- .../galileo/tests/test_decision_models_evaluator.py | 12 +++++++++--- 3 files changed, 22 insertions(+), 6 deletions(-) diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py index 45f1e29a..df50266d 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py @@ -13,6 +13,7 @@ DecisionModelsPayloadField = Literal["input", "output"] _NUMERIC_OPERATORS = frozenset({"gt", "gte", "lt", "lte"}) SCORER_INVOKE_RUNTIME_FLAG_ENV = "GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME" +DECISION_MODELS_EVALUATOR_FLAG_ENV = "GALILEO_FEATURE_FLAG_DECISION_MODELS_EVALUATOR" def scorer_invoke_runtime_enabled() -> bool: @@ -20,6 +21,11 @@ def scorer_invoke_runtime_enabled() -> bool: return os.getenv(SCORER_INVOKE_RUNTIME_FLAG_ENV) == "enabled" +def decision_models_evaluator_enabled() -> bool: + """Return whether the Decision Models evaluator is enabled for Agent Control.""" + return os.getenv(DECISION_MODELS_EVALUATOR_FLAG_ENV) == "enabled" + + class DecisionModelsEvaluatorConfig(EvaluatorConfig): """Pinned Decision Models scorer and local result comparison settings.""" diff --git a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py index 73acb315..e440c4a5 100644 --- a/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py @@ -16,7 +16,11 @@ DecisionModelsInvokeResponse, GalileoDecisionModelsClient, ) -from .config import DecisionModelsEvaluatorConfig, scorer_invoke_runtime_enabled +from .config import ( + DecisionModelsEvaluatorConfig, + decision_models_evaluator_enabled, + scorer_invoke_runtime_enabled, +) logger = logging.getLogger(__name__) @@ -51,8 +55,8 @@ class DecisionModelsEvaluator(Evaluator[DecisionModelsEvaluatorConfig]): @classmethod def is_available(cls) -> bool: - """Expose Decision Models only when the scorer-invoke runtime is enabled.""" - return scorer_invoke_runtime_enabled() + """Expose Decision Models only when its feature flag and scorer runtime are enabled.""" + return decision_models_evaluator_enabled() and scorer_invoke_runtime_enabled() def __init__(self, config: DecisionModelsEvaluatorConfig) -> None: super().__init__(config) diff --git a/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py b/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py index 43b0a47c..3b5fdcdf 100644 --- a/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py +++ b/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py @@ -55,6 +55,7 @@ def _evaluator(monkeypatch: pytest.MonkeyPatch, **config: Any) -> DecisionModels monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") monkeypatch.setenv("GALILEO_LUNA_INVOKE_URL", "http://runners:8090") monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") + monkeypatch.setenv("GALILEO_FEATURE_FLAG_DECISION_MODELS_EVALUATOR", "enabled") evaluator = DecisionModelsEvaluator( DecisionModelsEvaluatorConfig(scorer_id="scorer-a", scorer_version_id="version-a", **config) ) @@ -68,6 +69,7 @@ def test_decision_models_registration_and_required_configuration( from agent_control_evaluators import get_evaluator, register_evaluator monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") + monkeypatch.setenv("GALILEO_FEATURE_FLAG_DECISION_MODELS_EVALUATOR", "enabled") register_evaluator(DecisionModelsEvaluator) assert get_evaluator("galileo.decision_models") is DecisionModelsEvaluator with pytest.raises(ValidationError, match="scorer_id"): @@ -76,15 +78,19 @@ def test_decision_models_registration_and_required_configuration( DecisionModelsEvaluatorConfig(scorer_id="scorer-a") # type: ignore[call-arg] -def test_decision_models_availability_follows_runtime_feature_flag( +def test_decision_models_availability_requires_both_feature_flags( monkeypatch: pytest.MonkeyPatch, ) -> None: - monkeypatch.delenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", raising=False) + monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") + monkeypatch.delenv("GALILEO_FEATURE_FLAG_DECISION_MODELS_EVALUATOR", raising=False) assert DecisionModelsEvaluator.is_available() is False - monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "enabled") + monkeypatch.setenv("GALILEO_FEATURE_FLAG_DECISION_MODELS_EVALUATOR", "enabled") assert DecisionModelsEvaluator.is_available() is True + monkeypatch.setenv("GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME", "disabled") + assert DecisionModelsEvaluator.is_available() is False + def test_missing_distribution_metadata_uses_development_version( monkeypatch: pytest.MonkeyPatch,