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 fc1739c7..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,6 +28,7 @@ dev = [ [project.entry-points."agent_control.evaluators"] "galileo.luna" = "agent_control_evaluator_galileo.luna:LunaEvaluator" +"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..1b369eac --- /dev/null +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/__init__.py @@ -0,0 +1,27 @@ +"""Galileo Decision Models evaluator.""" + +from .client import ( + DecisionModelsExecutionContext, + DecisionModelsInvokeInputs, + DecisionModelsInvokeRequest, + DecisionModelsInvokeResponse, + GalileoDecisionModelsClient, +) +from .config import ( + DecisionModelsEvaluatorConfig, + DecisionModelsOperator, + DecisionModelsPayloadField, +) +from .evaluator import DecisionModelsEvaluator + +__all__ = [ + "DecisionModelsExecutionContext", + "GalileoDecisionModelsClient", + "DecisionModelsEvaluator", + "DecisionModelsEvaluatorConfig", + "DecisionModelsInvokeInputs", + "DecisionModelsInvokeRequest", + "DecisionModelsInvokeResponse", + "DecisionModelsOperator", + "DecisionModelsPayloadField", +] 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 new file mode 100644 index 00000000..e5db9cc2 --- /dev/null +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/client.py @@ -0,0 +1,300 @@ +"""HTTP client for Galileo Decision Models scorer invocation.""" + +from __future__ import annotations + +import logging +import os +from asyncio import Lock +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 ..records import record_from_step +from .config import DecisionModelsPayloadField, scorer_invoke_runtime_enabled + +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" +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): + """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") + + +def _decision_models_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 DecisionModelsInvokeInputs(BaseModel): + """Legacy inputs accepted alongside the structured Galileo record.""" + + query: JSONValue = "" + response: JSONValue = "" + ground_truth: JSONValue = None + tools: list[JSONObject] | None = None + + +class DecisionModelsInvokeRequest(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: DecisionModelsExecutionContext + record: JSONObject + inputs: DecisionModelsInvokeInputs + 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 DecisionModelsInvokeResponse(BaseModel): + """Response returned by the Decision Models 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 GalileoDecisionModelsClient: + """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.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.decision_models") + self.api_secret = resolved_secret + self.invoke_url = _normalize_invoke_url(resolved_url) + self.runtime_enabled = scorer_invoke_runtime_enabled() + self.timeout = timeout + 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, + *, + scorer_id: str, + scorer_version_id: str, + scorer_label: str | None, + step: Step, + execution_context: DecisionModelsExecutionContext, + inputs: DecisionModelsInvokeInputs, + timeout: float, + selected_data: Any = None, + 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.decision_models 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 = DecisionModelsInvokeRequest( + 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}, + ) + client = await self._get_client() + response = await client.post( + self.invoke_url, + json=request.to_dict(), + 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 Decision Models scorer response: expected a JSON object") + return DecisionModelsInvokeResponse.model_validate(body) + + async def close(self) -> None: + """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/config.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py new file mode 100644 index 00000000..df50266d --- /dev/null +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/config.py @@ -0,0 +1,52 @@ +"""Configuration for Galileo Decision Models scoring.""" + +from __future__ import annotations + +import os +from typing import Literal + +from agent_control_evaluators import EvaluatorConfig +from agent_control_models import JSONValue +from pydantic import Field, model_validator + +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" +DECISION_MODELS_EVALUATOR_FLAG_ENV = "GALILEO_FEATURE_FLAG_DECISION_MODELS_EVALUATOR" + + +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" + + +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.""" + + 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: 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) -> 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)) + ): + 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/decision_models/evaluator.py b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py new file mode 100644 index 00000000..e440c4a5 --- /dev/null +++ b/evaluators/contrib/galileo/src/agent_control_evaluator_galileo/decision_models/evaluator.py @@ -0,0 +1,252 @@ +"""Galileo Decision Models 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 .client import ( + DecisionModelsExecutionContext, + DecisionModelsInvokeInputs, + DecisionModelsInvokeResponse, + GalileoDecisionModelsClient, +) +from .config import ( + DecisionModelsEvaluatorConfig, + decision_models_evaluator_enabled, + scorer_invoke_runtime_enabled, +) + +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 DecisionModelsEvaluator(Evaluator[DecisionModelsEvaluatorConfig]): + """Evaluate data using a pinned Galileo Decision Models scorer.""" + + metadata = EvaluatorMetadata( + name="galileo.decision_models", + version=_PACKAGE_VERSION, + description="Galileo Decision Models scoring", + requires_api_key=True, + timeout_ms=10000, + ) + config_model = DecisionModelsEvaluatorConfig + + @classmethod + def is_available(cls) -> bool: + """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) + self._client = GalileoDecisionModelsClient() + + async def evaluate(self, data: Any) -> EvaluatorResult: + """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.decision_models requires " + "GALILEO_FEATURE_FLAG_SCORER_INVOKE_RUNTIME=enabled." + ) + ) + return self._error( + 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 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( + 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.decision_models requires authenticated execution metadata from " + "runtime extensions" + ) + 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: DecisionModelsExecutionContext | None + ) -> EvaluatorResult: + try: + if not scorer_invoke_runtime_enabled() or not self._client.runtime_enabled: + raise RuntimeError( + "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.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, + 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"Decision Models scorer status: {response.status}" + ) + if response.score is None: + raise RuntimeError("Decision Models 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"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("Decision Models evaluation error: %s", exc, exc_info=True) + return self._error(exc) + + @staticmethod + def _execution_context(extensions: JSONObject | None) -> DecisionModelsExecutionContext | 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 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) and project_id else None, + run_id=run_id if isinstance(run_id, str) and run_id else None, + ) + + @staticmethod + 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", "") + 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 DecisionModelsInvokeInputs( + 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"Decision Models result {value!r} is not numeric for operator '{operator}'" + ) + if isinstance(threshold, bool) or not isinstance(threshold, (int, float)): + raise ValueError(f"Decision Models threshold {threshold!r} is not numeric") + return { + "gt": value > threshold, + "gte": value >= threshold, + "lt": value < threshold, + "lte": value <= threshold, + }[operator] + + def _metadata(self, response: DecisionModelsInvokeResponse) -> 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"Decision Models evaluation error: {detail}", + metadata=metadata, + error=detail, + ) + + async def aclose(self) -> None: + """Close the Decision Models HTTP client.""" + await self._client.close() diff --git a/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py b/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py new file mode 100644 index 00000000..3b5fdcdf --- /dev/null +++ b/evaluators/contrib/galileo/tests/test_decision_models_evaluator.py @@ -0,0 +1,936 @@ +"""Tests for the Galileo Decision Models 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 importlib import metadata as importlib_metadata +from importlib import reload +from importlib.metadata import PackageNotFoundError +from typing import Any +from unittest.mock import AsyncMock + +import httpx +import pytest +from agent_control_evaluator_galileo.decision_models import ( + DecisionModelsExecutionContext, + DecisionModelsEvaluator, + DecisionModelsEvaluatorConfig, + DecisionModelsInvokeInputs, + DecisionModelsInvokeRequest, + DecisionModelsInvokeResponse, + GalileoDecisionModelsClient, +) +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, + 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) -> 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") + monkeypatch.setenv("GALILEO_FEATURE_FLAG_DECISION_MODELS_EVALUATOR", "enabled") + evaluator = DecisionModelsEvaluator( + DecisionModelsEvaluatorConfig(scorer_id="scorer-a", scorer_version_id="version-a", **config) + ) + return 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") + 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"): + DecisionModelsEvaluatorConfig(scorer_version_id="version-a") # type: ignore[call-arg] + with pytest.raises(ValidationError, match="scorer_version_id"): + DecisionModelsEvaluatorConfig(scorer_id="scorer-a") # type: ignore[call-arg] + + +def test_decision_models_availability_requires_both_feature_flags( + monkeypatch: pytest.MonkeyPatch, +) -> None: + 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_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, +) -> 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(importlib_metadata, "version", missing_distribution) + try: + clear_evaluators() + 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(): + 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"): + DecisionModelsEvaluatorConfig( + scorer_id="scorer-a", + scorer_version_id="version-a", + operator="gt", + threshold=True, + ) + with pytest.raises(ValidationError, match="threshold is required"): + DecisionModelsEvaluatorConfig( + scorer_id="scorer-a", + scorer_version_id="version-a", + operator="eq", + threshold=None, + ) + + +def test_config_allows_missing_threshold_for_any_operator() -> None: + config = DecisionModelsEvaluatorConfig( + 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, +) -> None: + monkeypatch.setenv("GALILEO_API_SECRET_KEY", "test-secret") + monkeypatch.setenv("GALILEO_LUNA_INVOKE_URL", "http://luna:8090") + + 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}" + 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"): + GalileoDecisionModelsClient() + + +def test_invoke_request_requires_execution_context_and_serializes_nested_record() -> None: + context = DecisionModelsExecutionContext( + 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 = DecisionModelsInvokeRequest( + scorer_id="scorer-a", + scorer_version_id="version-a", + execution_context=context, + record=record, + inputs=DecisionModelsInvokeInputs(), + 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_decision_models_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=DecisionModelsInvokeResponse(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=DecisionModelsInvokeResponse(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=DecisionModelsInvokeResponse(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 "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( + 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 = GalileoDecisionModelsClient() + 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=DecisionModelsExecutionContext( + organization_id="org-1", user_id="user-7" + ), + inputs=DecisionModelsInvokeInputs(), + 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=DecisionModelsInvokeResponse(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=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 == "Decision Models 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=DecisionModelsInvokeResponse( + 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() + + +@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 "Step context alone is not sufficient" 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 = DecisionModelsEvaluatorConfig.model_construct( + scorer_id="scorer-a", + scorer_version_id="version-a", + threshold="not numeric", + operator="gte", + ) + evaluator = DecisionModelsEvaluator(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 = GalileoDecisionModelsClient(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"): + GalileoDecisionModelsClient(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 = GalileoDecisionModelsClient(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=DecisionModelsExecutionContext(organization_id="org-1"), + inputs=DecisionModelsInvokeInputs(), + 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 = GalileoDecisionModelsClient(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=DecisionModelsExecutionContext(organization_id="org-1"), + inputs=DecisionModelsInvokeInputs(), + timeout=5.0, + ) + + await client.close() + + +def test_luna_and_decision_models_request_shape_parity() -> None: + 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=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=decision_models_context, + ).to_dict() + assert set(luna) == set(decision_models) + assert set(luna["inputs"]) == set(decision_models["inputs"]) + assert luna["execution_context"] == decision_models["execution_context"]