diff --git a/AGENTS.md b/AGENTS.md index 6d239fc..e828b2c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -703,3 +703,10 @@ response = await graph( }, ).invoke(user_input, context) ``` + +## Maintaining this file + +Keep this file for knowledge useful to almost every future agent session in this project. +Do not repeat what the codebase already shows; point to the authoritative file or command instead. +Prefer rewriting or pruning existing entries over appending new ones. +When updating this file, preserve this bar for all agents and keep entries concise. diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..a9d4d26 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,2 @@ + +@AGENTS.md diff --git a/packages/ai/README.md b/packages/ai/README.md index e35135b..7bbf073 100644 --- a/packages/ai/README.md +++ b/packages/ai/README.md @@ -52,6 +52,25 @@ if result["enabled"]: Never raises. Returns `{"enabled": bool, "config": dict | None, "meta": dict | None}`. +## Evaluations from code + +`init_evaluations` and the evaluations result types are also re-exported: + +```python +from launchdarkly_ai_python import init_evaluations + +evals = init_evaluations() +result = await evals.run( + project_key="my-project", + key="unique-evaluation-key", + dataset="golden-dataset", + handler=my_handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, +) +``` + +`LD_API_TOKEN` is required. Configure `LD_SDK_KEY` — or initialize your own client with `init_client(client=...)` — to emit one `$ld:ai:offline-evals:generation` event per generated row through the standard SDK event transport. Use `LD_API_BASE_URI` for staging or local management API traffic; it is separate from the SDK delivery setting `LD_BASE_URI`. Evaluation-run links use the explicit `ui_base_uri` option or `LD_UI_BASE_URI` (for example, `https://ld-stg.launchdarkly.com` in staging), defaulting to `https://app.launchdarkly.com`. See the [core evaluations guide](../client/README.md#run-an-evaluation-from-code). + --- All exports, types, and behaviors are identical to `launchdarkly-ai-server`. See the [core client README](../client/README.md) for the full API reference. diff --git a/packages/client/README.md b/packages/client/README.md index 8691bab..852f5ef 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -39,6 +39,48 @@ No code changes are required — `init_client()` detects the packages at runtime | `LD_SERVICE_NAME` | No | OTel `service.name` resource attribute (default: `python-sdk`) | | `LD_ENVIRONMENT` | No | `deployment.environment` resource attribute attached to telemetry | | `OTEL_EXPORTER_OTLP_ENDPOINT` | No | OTLP endpoint override (default: LaunchDarkly Observability backend) | +| `LD_API_TOKEN` | For evaluations | API access token used by the evaluations management API | +| `LD_SDK_KEY` | For evaluations | SDK key whose event transport carries generation results to LaunchDarkly | +| `LD_API_BASE_URI` | No | Evaluations management API host override; intentionally separate from `LD_BASE_URI` | +| `LD_UI_BASE_URI` | No | LaunchDarkly application host for evaluation-run links (default: `https://app.launchdarkly.com`; staging: `https://ld-stg.launchdarkly.com`) | + +### Run an evaluation from code + +The generation-only evaluations harness reads an LD-hosted dataset, creates a new evaluation and API-source run, and invokes your handler once per row. Each success or error queues a `$ld:ai:offline-evals:generation` custom event containing the evaluation, run, dataset, and row identifiers plus output or error (`errorMessage` is included for `ERROR` rows), nested `usage.inputTokens`/`usage.outputTokens`, timing, and stable hashes. Dataset-owned input, expected output, metadata, and variables are not duplicated in the event. Each queued event prints a line to stdout with its RFC3339 UTC `emittedAt` timestamp and stable `eventId`, making it possible to compare SDK emission time with ClickHouse arrival time. The same `emittedAt` value is included in the event payload. Events are flushed before the summary is fetched and the call returns; handlers are never rerun to retry event delivery. Pass/fail is derived from LaunchDarkly's run summary. + +Result links use `ui_base_uri`, then `LD_UI_BASE_URI`, then `https://app.launchdarkly.com`; this is independent of `LD_API_BASE_URI`. After flushing generation events, the harness polls the run summary endpoint until passed + failed + error rows fully account for a nonzero total with no pending rows, polling every `poll_interval_seconds` (default 2s) up to `poll_timeout_seconds` (default 180s); pass either to `run()` to widen both for large datasets. The summary endpoint does not return run state, so `RunSummary` exposes row counts only. A generation result passes only when the completed summary has no error or pending rows. Evaluation keys must be unique because every call creates a new evaluation with `POST`. + +```python +import asyncio +import sys + +from launchdarkly_ai_openai_messages import create_openai_messages_handler +from launchdarkly_ai_server import init_evaluations + + +async def main() -> int: + evals = init_evaluations() # LD_API_TOKEN required; LD_SDK_KEY unless a client is already initialized + result = await evals.run( + project_key="my-project", + key="support-qa-2026-08-20", + dataset="support-golden", + handler=create_openai_messages_handler(), + generation={ + "provider": "OpenAI", + "model": "gpt-4o", + "instructions": "You are a support agent.", + }, + ) + print(result.url, result.summary) + return 0 if result.passed else 1 + + +sys.exit(asyncio.run(main())) +``` + +`project_key` is supplied per run rather than during initialization. `generation.instructions` is shorthand for one system message; use `generation.messages` instead for a full message list, but do not supply both. The harness never retries a handler invocation because doing so could repeat tool side effects. Its retries apply only to LaunchDarkly management API requests. + +Generation events are the only path by which row results reach LaunchDarkly, so `init_evaluations()` raises rather than creating a run that can never complete unless it can resolve an event transport: either an SDK key (`sdk_key` or `LD_SDK_KEY`) or a client already initialized through `init_client(client=...)`. Bringing your own client lets a process emit evaluation events without an SDK key in scope. Every generated row is emitted and flushed unconditionally; no feature flag gates event publishing. The harness then polls the summary endpoint until row accounting shows processing is complete. The client uses **lazy initialization**: importing the package does not connect to LaunchDarkly. The singleton is created automatically on the first API call that needs it (`config().invoke()`, `graph().invoke()`, `resolve_graph()`, etc.), as long as `LD_SDK_KEY` is set in the environment. diff --git a/packages/client/agents.md b/packages/client/agents.md index 9f143ac..3ef0c16 100644 --- a/packages/client/agents.md +++ b/packages/client/agents.md @@ -31,6 +31,7 @@ No other `launchdarkly-ai-*` package may define or duplicate these. They import | `src/launchdarkly_ai_server/utils.py` | `parse_template`, `parse_json_with_possible_fences`, `create_handler`, `parse_usage`, `make_track_data`, `to_ld_context` | | `src/launchdarkly_ai_server/registry.py` | `Registry`, `global_registry`, `compose`, `resolve_handlers`, `resolve_tools` | | `src/launchdarkly_ai_server/judges.py` | `run_judges`, `build_judge_tasks`, `run_judge` | +| `src/launchdarkly_ai_server/evaluations/` | `init_evaluations`, the private management API operations, and generation-only `EvaluationsModule.run()` orchestration | | `src/launchdarkly_ai_server/__init__.py` | Public barrel — the only surface handler packages import from | --- @@ -68,7 +69,7 @@ from launchdarkly_ai_server import Registry, global_registry, compose, resolve_h from launchdarkly_ai_server import execute_and_track, execute_and_stream, wrap_tool_handlers # Entry points -from launchdarkly_ai_server import config, graph, resolve_graph +from launchdarkly_ai_server import config, graph, resolve_graph, init_evaluations ``` When adding a new export, add it to `__init__.py`'s imports and `__all__`. Handler packages must never import from sub-paths (e.g. `launchdarkly_ai_server.client`). @@ -125,6 +126,14 @@ Handlers may return any of these — the client normalizes them before emitting --- +## SDK-run evaluations + +`init_evaluations()` creates an evaluations harness using `LD_API_TOKEN` and the management API host `LD_API_BASE_URI`. Do not reuse `LD_BASE_URI`: that variable configures SDK delivery and may point at a relay proxy. Evaluation-run links use the separate `ui_base_uri` option, then `LD_UI_BASE_URI`, then `https://app.launchdarkly.com`; do not derive their host from `LD_API_BASE_URI`. An event transport is resolved in `init_evaluations()`, which raises before any network I/O when it finds neither an SDK key (`sdk_key` or `LD_SDK_KEY`) nor an already-initialized event-capable client: generation events are the only ingest path for row results, so a run without a transport could never complete. The lifecycle module's bring-your-own-client path (`init_client(client=...)`) therefore satisfies the check on its own, and `run()` reuses that singleton through `_resolve_client`; `run()` raises if the client disappears before it emits. Both polling arguments reject NaN, which would otherwise never compare past a deadline and hang the run. The harness always queues one `$ld:ai:offline-evals:generation` custom event per row through the standard SDK event transport and flushes before returning. No feature flag gates event emission. The harness polls the run summary endpoint until a nonzero `total_rows` has `pending_rows == 0` and `passed + failed + error` rows accounting for the total, polling every `poll_interval_seconds` (default 2s) until `poll_timeout_seconds` (default 180s); both are `run()` arguments so large datasets can widen them. The summary endpoint does not return run state, so `RunSummary` exposes row counts only. + +`await EvaluationsModule.run(...)` takes `project_key` per call. Dataset lookup/row pagination, evaluation creation, and run creation are private helpers; only `run()` is public. Each call creates a new evaluation with `POST` and a run with `source="api"`, so its key must be unique. The harness directly invokes the supplied handler once per row and never retries it — event delivery is never a reason to rerun a handler because that would repeat tool side effects; retries apply only to management API requests. A 429 is replayed for any method, but 5xx responses and transport failures are replayed only for `GET`/`HEAD`, so an evaluation or run `POST` that the server may already have applied is never duplicated. Management API calls run in a worker thread (`asyncio.to_thread`) because the client is synchronous; the caller's event loop stays free. Generation events go through the already-initialized SDK client when the application has one — `init_client` is idempotent, so an existing singleton wins and the evaluations SDK key is ignored with a warning. Dataset-owned `input`, `expected_output`, `metadata`, and `variables` are deliberately excluded from the event payload. The harness flushes events, polls the run summary endpoint until row accounting is complete (`total_rows > 0`, `pending_rows == 0`, and `passed + failed + error == total_rows`), and raises a timeout once `poll_timeout_seconds` elapses if the backend never reaches one. `RunSummary` includes row counts only, and `EvalRunResult.passed` is true only when error and pending row counts are both zero. + +--- + ## Conversation grouping LaunchDarkly's conversation view groups spans on `gen_ai.conversation.id`. Bind a caller-supplied id around any `invoke()` / `stream()` / `graph().invoke()` call: diff --git a/packages/client/src/launchdarkly_ai_server/__init__.py b/packages/client/src/launchdarkly_ai_server/__init__.py index 80b959b..9a07b05 100644 --- a/packages/client/src/launchdarkly_ai_server/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/__init__.py @@ -21,6 +21,14 @@ conversation_id, set_conversation_id_if_absent, ) +from .evaluations import ( + EvalRunResult, + EvaluationsError, + EvaluationsModule, + GenerationConfig, + RunSummary, + init_evaluations, +) from .graph import GraphInstance, graph, resolve_graph from .judges import build_judge_tasks, run_judge, run_judges from .lifecycle import ( @@ -155,6 +163,13 @@ "text_message", "to_semconv_finish_reason", "VariationMeta", + # evaluations + "EvalRunResult", + "EvaluationsError", + "EvaluationsModule", + "GenerationConfig", + "RunSummary", + "init_evaluations", # utils "create_handler", "make_track_data", diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py new file mode 100644 index 0000000..6516f4a --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py @@ -0,0 +1,29 @@ +"""Run LaunchDarkly evaluations from your own environment.""" + +from .api import ( + DEFAULT_BASE_URI, + EvaluationsError, + HttpResponse, + LDApiClient, + LDApiError, + Transport, + urllib_transport, +) +from .module import EvaluationsModule, init_evaluations +from .types import EvalRunResult, GenerationConfig, RunSummary, Usage + +__all__ = [ + "DEFAULT_BASE_URI", + "EvalRunResult", + "EvaluationsError", + "EvaluationsModule", + "GenerationConfig", + "HttpResponse", + "LDApiClient", + "LDApiError", + "RunSummary", + "Transport", + "Usage", + "init_evaluations", + "urllib_transport", +] diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/api.py b/packages/client/src/launchdarkly_ai_server/evaluations/api.py new file mode 100644 index 0000000..0a6aeb4 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/api.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import json +import random +import time +import urllib.error +import urllib.parse +import urllib.request +from collections.abc import Callable +from dataclasses import dataclass, field +from datetime import UTC, datetime +from email.utils import parsedate_to_datetime +from typing import Any, Protocol + +DEFAULT_BASE_URI = "https://app.launchdarkly.com" + +# Only these methods are replayed after a 5xx or a transport failure: a POST that +# timed out may still have created a record server-side. +RETRY_SAFE_METHODS = frozenset({"GET", "HEAD"}) + + +class EvaluationsError(Exception): + """Base error for the evaluations harness.""" + + +class LDApiError(EvaluationsError): + """A non-2xx response from the LaunchDarkly API.""" + + def __init__(self, status: int, method: str, path: str, body: str) -> None: + super().__init__( + f"LaunchDarkly API {method} {path} failed with {status}: {body}" + ) + self.status = status + self.method = method + self.path = path + self.body = body + + +@dataclass +class HttpResponse: + status: int + body: str + headers: dict[str, str] = field(default_factory=dict) + + +class Transport(Protocol): + """Seam the API client sends requests through; replaced in tests.""" + + def __call__( + self, + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: ... + + +def urllib_transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, +) -> HttpResponse: + request = urllib.request.Request(url, data=body, headers=headers, method=method) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + return HttpResponse( + status=response.status, + body=response.read().decode("utf-8"), + headers={k.lower(): v for k, v in response.headers.items()}, + ) + except urllib.error.HTTPError as error: + return HttpResponse( + status=error.code, + body=error.read().decode("utf-8"), + headers={k.lower(): v for k, v in error.headers.items()}, + ) + + +class LDApiClient: + """Minimal retrying client for the LaunchDarkly public management API.""" + + def __init__( + self, + api_token: str, + base_uri: str = DEFAULT_BASE_URI, + transport: Transport = urllib_transport, + timeout: float = 30.0, + max_retries: int = 3, + sleep: Callable[[float], None] = time.sleep, + random_value: Callable[[], float] = random.random, + ) -> None: + self.api_token = api_token + self.base_uri = base_uri.rstrip("/") + self._transport = transport + self._timeout = timeout + self._max_retries = max(0, max_retries) + self._sleep = sleep + self._random_value = random_value + + def url_for(self, path: str, params: dict[str, Any] | None = None) -> str: + url = f"{self.base_uri}/api/v2/{path.lstrip('/')}" + if params: + query = {k: str(v) for k, v in params.items() if v is not None} + if query: + url = f"{url}?{urllib.parse.urlencode(query)}" + return url + + def _retry_delay(self, attempt: int, response: HttpResponse | None = None) -> float: + if response is not None: + retry_after = response.headers.get("retry-after") or response.headers.get( + "Retry-After" + ) + if retry_after: + try: + return max(0.0, float(retry_after)) + except ValueError: + try: + when: datetime = parsedate_to_datetime(retry_after) + now = datetime.now(UTC) + return max(0.0, (when - now).total_seconds()) + except (TypeError, ValueError, OverflowError): + pass + exponential = float(min(30.0, 0.5 * (2**attempt))) + jitter = float(self._random_value()) * min(1.0, exponential) + return exponential + jitter + + def request( + self, + method: str, + path: str, + body: Any = None, + params: dict[str, Any] | None = None, + ) -> Any: + headers = { + "Authorization": self.api_token, + "Accept": "application/json", + "User-Agent": "launchdarkly-ai-evaluations-python", + } + payload: bytes | None = None + if body is not None: + headers["Content-Type"] = "application/json" + payload = json.dumps(body).encode("utf-8") + + response: HttpResponse | None = None + for attempt in range(self._max_retries + 1): + try: + response = self._transport( + method, self.url_for(path, params), headers, payload, self._timeout + ) + except (TimeoutError, urllib.error.URLError) as error: + if ( + method.upper() not in RETRY_SAFE_METHODS + or attempt >= self._max_retries + ): + raise EvaluationsError( + f"LaunchDarkly API {method} {path} failed after retries: {error}" + ) from error + self._sleep(self._retry_delay(attempt)) + continue + + # A 429 is rejected before the server acts on it, so it is safe to + # replay for any method. + retryable = response.status == 429 or ( + response.status >= 500 and method.upper() in RETRY_SAFE_METHODS + ) + if retryable and attempt < self._max_retries: + self._sleep(self._retry_delay(attempt, response)) + continue + break + + if response is None: + raise EvaluationsError( + f"LaunchDarkly API {method} {path} returned no response" + ) + if response.status < 200 or response.status >= 300: + raise LDApiError(response.status, method, path, response.body) + if not response.body: + return None + try: + return json.loads(response.body) + except json.JSONDecodeError as error: + raise EvaluationsError( + f"LaunchDarkly API {method} {path} returned invalid JSON" + ) from error + + def get(self, path: str, params: dict[str, Any] | None = None) -> Any: + return self.request("GET", path, params=params) + + def post(self, path: str, body: Any = None) -> Any: + return self.request("POST", path, body=body) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py new file mode 100644 index 0000000..4a3ffad --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -0,0 +1,326 @@ +from __future__ import annotations + +import asyncio +import inspect +import logging +import math +import os +import time +from collections.abc import Mapping +from typing import Any + +from ..lifecycle import get_client, init_client +from .api import ( + DEFAULT_BASE_URI, + EvaluationsError, + LDApiClient, + Transport, + urllib_transport, +) +from .runner import EvalHandler, EvaluationsRunner, ToolImplementation, _segment +from .types import EvalRunResult, GenerationConfig, RunSummary + +logger = logging.getLogger(__name__) + +DEFAULT_UI_BASE_URI = "https://app.launchdarkly.com" +SUMMARY_POLL_INTERVAL_SECONDS = 2.0 +SUMMARY_POLL_TIMEOUT_SECONDS = 180.0 + + +def _env(name: str) -> str | None: + """Read an env var, treating blank/whitespace-only values as unset.""" + value = os.environ.get(name, "").strip() + return value if value else None + + +def _initialized_client() -> Any | None: + """Return the SDK singleton when one is initialized, otherwise ``None``.""" + try: + return get_client() + except RuntimeError: + return None + + +def _can_emit_events(client: Any) -> bool: + return callable(getattr(client, "track", None)) and callable( + getattr(client, "flush", None) + ) + + +def _is_terminal_summary(summary: RunSummary) -> bool: + accounted_rows = summary.passed_rows + summary.failed_rows + summary.error_rows + return ( + summary.total_rows > 0 + and summary.pending_rows == 0 + and accounted_rows == summary.total_rows + ) + + +class EvaluationsModule: + """Entry point for running LaunchDarkly evaluations from customer code.""" + + def __init__( + self, + api_client: LDApiClient, + sdk_key: str | None, + ui_base_uri: str = DEFAULT_UI_BASE_URI, + ) -> None: + self._api = api_client + self._sdk_key = sdk_key + self._ui_base_uri = ui_base_uri.rstrip("/") + self._runner = EvaluationsRunner(api_client) + + @property + def api(self) -> LDApiClient: + return self._api + + @property + def sdk_key(self) -> str | None: + """SDK key whose event transport carries generation results to LaunchDarkly.""" + return self._sdk_key + + @property + def ui_base_uri(self) -> str: + """LaunchDarkly application host used for evaluation-run links.""" + return self._ui_base_uri + + async def run( + self, + *, + project_key: str, + key: str, + dataset: str, + handler: EvalHandler, + generation: GenerationConfig, + tools: Mapping[str, ToolImplementation] | None = None, + concurrency: int = 10, + poll_interval_seconds: float | None = None, + poll_timeout_seconds: float | None = None, + ) -> EvalRunResult: + """ + Create and run a generation-only evaluation in the caller's process. + + The returned pass/fail result is derived from LaunchDarkly's run summary. + A CI script can exit with ``0 if result.passed else 1`` after awaiting + this method. Large datasets may need a longer ``poll_timeout_seconds`` + and a wider ``poll_interval_seconds``; both default to + ``SUMMARY_POLL_TIMEOUT_SECONDS`` / ``SUMMARY_POLL_INTERVAL_SECONDS``. + """ + if poll_interval_seconds is None: + poll_interval_seconds = SUMMARY_POLL_INTERVAL_SECONDS + if poll_timeout_seconds is None: + poll_timeout_seconds = SUMMARY_POLL_TIMEOUT_SECONDS + self._validate_run_args( + project_key=project_key, + key=key, + dataset=dataset, + handler=handler, + generation=generation, + concurrency=concurrency, + poll_interval_seconds=poll_interval_seconds, + poll_timeout_seconds=poll_timeout_seconds, + ) + run_tools = dict(tools or {}) + client = await self._resolve_client() + + # The management API client is synchronous; running it in a worker thread + # keeps the caller's event loop free. + # Tool verification is deliberately first: a typo must not create records. + resolved_tools = await asyncio.to_thread( + self._runner._resolve_tools, project_key, run_tools + ) + dataset_ref = await asyncio.to_thread( + self._runner._fetch_dataset, project_key, dataset + ) + rows = await asyncio.to_thread( + self._runner._get_dataset_rows, project_key, dataset + ) + evaluation = await asyncio.to_thread( + self._runner._create_evaluation, + project_key, + key, + generation, + resolved_tools, + ) + evaluation_run = await asyncio.to_thread( + self._runner._create_evaluation_run, + project_key, + evaluation.id, + len(rows), + dataset_ref.id, + ) + config = self._runner._build_handler_config(generation, resolved_tools) + results = await self._runner._run_rows( + rows, + handler, + config, + run_tools, + concurrency, + ) + self._runner._emit_generation_events( + client, + project_key=project_key, + evaluation=evaluation, + evaluation_run=evaluation_run, + dataset=dataset_ref, + results=results, + ) + flush_result = client.flush() + if inspect.isawaitable(flush_result): + await flush_result + summary = await self._poll_summary_until_terminal( + project_key, + evaluation.id, + evaluation_run.id, + poll_interval_seconds, + poll_timeout_seconds, + ) + url = ( + f"{self._ui_base_uri}/projects/{_segment(project_key)}/ai/evaluations/" + f"{_segment(evaluation.id)}/runs/{_segment(evaluation_run.id)}" + ) + return EvalRunResult( + passed=(summary.error_rows == 0 and summary.pending_rows == 0), + url=url, + run_id=evaluation_run.id, + summary=summary, + ) + + async def _poll_summary_until_terminal( + self, + project_key: str, + evaluation_id: str, + run_id: str, + poll_interval_seconds: float, + poll_timeout_seconds: float, + ) -> RunSummary: + deadline = time.monotonic() + poll_timeout_seconds + last_summary = None + while True: + last_summary = await asyncio.to_thread( + self._runner._get_summary, project_key, evaluation_id, run_id + ) + if _is_terminal_summary(last_summary): + return last_summary + remaining = deadline - time.monotonic() + if remaining <= 0: + accounted_rows = ( + last_summary.passed_rows + + last_summary.failed_rows + + last_summary.error_rows + ) + raise EvaluationsError( + "Timed out after " + f"{poll_timeout_seconds:g} seconds waiting for evaluation " + f"run {run_id} summary rows to be fully accounted " + f"(total_rows={last_summary.total_rows}, " + f"accounted_rows={accounted_rows}, " + f"pending_rows={last_summary.pending_rows})" + ) + await asyncio.sleep(min(poll_interval_seconds, remaining)) + + async def _resolve_client(self) -> Any: + """ + Return the SDK client used for generation events. + + ``init_client`` is idempotent, so an application that already holds a + client keeps it and the evaluations SDK key is not applied. + """ + existing = _initialized_client() + if existing is not None: + if self._sdk_key: + logger.warning( + "A LaunchDarkly client is already initialized; evaluation " + "events are sent with it and the evaluations SDK key is " + "ignored. Both must point at the project under evaluation." + ) + return existing + if not self._sdk_key: + raise EvaluationsError( + "No LaunchDarkly SDK key provided and no initialized " + "LaunchDarkly client is available to deliver generation events." + ) + return await init_client({"sdkKey": self._sdk_key}) + + @staticmethod + def _validate_run_args( + *, + project_key: str, + key: str, + dataset: str, + handler: EvalHandler, + generation: GenerationConfig, + concurrency: int, + poll_interval_seconds: float, + poll_timeout_seconds: float, + ) -> None: + for name, value in ( + ("project_key", project_key), + ("key", key), + ("dataset", dataset), + ): + if not value.strip(): + raise EvaluationsError(f"{name} must not be blank") + if not callable(handler): + raise EvaluationsError("handler must be callable") + provider = generation.get("provider") + model = generation.get("model") + if not isinstance(provider, str) or not provider.strip(): + raise EvaluationsError("generation.provider is required") + if not isinstance(model, str) or not model.strip(): + raise EvaluationsError("generation.model is required") + if "instructions" in generation and "messages" in generation: + raise EvaluationsError( + "generation.instructions and generation.messages are mutually exclusive" + ) + if concurrency < 1: + raise EvaluationsError("concurrency must be at least 1") + for name, seconds in ( + ("poll_interval_seconds", poll_interval_seconds), + ("poll_timeout_seconds", poll_timeout_seconds), + ): + # NaN comparisons are always false, so a NaN would poll forever. + if math.isnan(seconds): + raise EvaluationsError(f"{name} must be a number") + if seconds < 0: + raise EvaluationsError(f"{name} must not be negative") + + +def init_evaluations( + api_token: str | None = None, + sdk_key: str | None = None, + base_uri: str | None = None, + ui_base_uri: str | None = None, + transport: Transport = urllib_transport, +) -> EvaluationsModule: + """Resolve credentials and construct the evaluations module.""" + token = api_token or _env("LD_API_TOKEN") + if not token: + raise EvaluationsError( + "No LaunchDarkly API access token provided. Set the LD_API_TOKEN " + "environment variable or pass api_token to init_evaluations()." + ) + + resolved_sdk_key = sdk_key or _env("LD_SDK_KEY") + if not resolved_sdk_key: + byoc_client = _initialized_client() + if byoc_client is None or not _can_emit_events(byoc_client): + raise EvaluationsError( + "No LaunchDarkly SDK key provided and no initialized " + "LaunchDarkly client to emit events with. Generation results " + "reach LaunchDarkly through the SDK event transport, so a run " + "cannot complete without one: set the LD_SDK_KEY environment " + "variable, pass sdk_key to init_evaluations(), or initialize a " + "client first with init_client(client=...)." + ) + + api_client = LDApiClient( + api_token=token, + base_uri=base_uri or _env("LD_API_BASE_URI") or DEFAULT_BASE_URI, + transport=transport, + ) + return EvaluationsModule( + api_client=api_client, + sdk_key=resolved_sdk_key, + ui_base_uri=ui_base_uri or _env("LD_UI_BASE_URI") or DEFAULT_UI_BASE_URI, + ) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py new file mode 100644 index 0000000..951213b --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -0,0 +1,492 @@ +from __future__ import annotations + +import asyncio +import hashlib +import json +import time +import urllib.parse +from collections.abc import Awaitable, Callable, Mapping +from datetime import UTC, datetime +from typing import Any + +from ..types import NativeTool +from ..utils import parse_template, parse_usage, to_ld_context +from .api import EvaluationsError, LDApiClient, LDApiError +from .types import ( + DatasetRef, + DatasetRow, + EvaluationRef, + EvaluationRunRef, + GenerationConfig, + ResolvedTool, + RunSummary, +) + +DATASET_PAGE_SIZE = 200 +GENERATION_EVENT_NAME = "$ld:ai:offline-evals:generation" + +EvalHandler = Callable[..., Awaitable[dict[str, Any]]] +ToolImplementation = Callable[..., Any] | NativeTool + + +def _segment(value: str) -> str: + return urllib.parse.quote(value, safe="") + + +def _mapping(value: Any, *, description: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise EvaluationsError( + f"LaunchDarkly returned an invalid {description} response" + ) + return value + + +def _required_string(data: Mapping[str, Any], key: str, description: str) -> str: + value = data.get(key) + if not isinstance(value, str) or not value: + raise EvaluationsError( + f"LaunchDarkly {description} response is missing string field {key!r}" + ) + return value + + +class ConcurrencyController: + """Owns row-worker permits.""" + + def __init__(self, limit: int = 10) -> None: + if limit < 1: + raise EvaluationsError("concurrency must be at least 1") + self._semaphore = asyncio.Semaphore(limit) + + async def acquire(self, provider: str | None = None) -> None: + del provider + await self._semaphore.acquire() + + def release(self) -> None: + self._semaphore.release() + + def record_success( + self, + provider: str | None = None, + headers: Mapping[str, str] | None = None, + ) -> None: + del provider, headers + + def record_rate_limit( + self, + provider: str | None = None, + retry_after: float | None = None, + ) -> None: + del provider, retry_after + + +class EvaluationsRunner: + """Private API operations and orchestration used by EvaluationsModule.run().""" + + def __init__(self, api: LDApiClient) -> None: + self._api = api + + def _resolve_tools( + self, + project_key: str, + tools: Mapping[str, ToolImplementation], + ) -> dict[str, ResolvedTool]: + resolved: dict[str, ResolvedTool] = {} + for key, implementation in tools.items(): + if not callable(implementation) and not isinstance( + implementation, NativeTool + ): + raise EvaluationsError( + f"Tool {key!r} must be callable or a NativeTool instance" + ) + path = f"projects/{_segment(project_key)}/ai-tools/{_segment(key)}" + try: + raw = _mapping(self._api.get(path), description=f"tool {key!r}") + except LDApiError as error: + if error.status == 404: + raise EvaluationsError( + f"LaunchDarkly AI tool {key!r} was not found in project {project_key!r}" + ) from error + raise + version = raw.get("version") + if not isinstance(version, int): + raise EvaluationsError( + f"LaunchDarkly AI tool {key!r} has no integer version" + ) + schema = raw.get("schema") + if not isinstance(schema, Mapping): + schema = {} + resolved[key] = ResolvedTool( + key=key, + version=version, + description=str(raw.get("description") or ""), + schema=dict(schema), + ) + return resolved + + def _fetch_dataset(self, project_key: str, dataset_key: str) -> DatasetRef: + path = f"projects/{_segment(project_key)}/datasets/{_segment(dataset_key)}" + try: + raw = _mapping(self._api.get(path), description=f"dataset {dataset_key!r}") + except LDApiError as error: + if error.status == 404: + raise EvaluationsError( + f"LaunchDarkly dataset {dataset_key!r} was not found in project {project_key!r}" + ) from error + raise + dataset_id = _required_string(raw, "id", "dataset") + response_key = raw.get("key", raw.get("name", dataset_key)) + return DatasetRef(id=dataset_id, key=str(response_key)) + + def _fetch_dataset_rows_page( + self, + project_key: str, + dataset_key: str, + *, + offset: int, + ) -> Mapping[str, Any]: + path = f"projects/{_segment(project_key)}/datasets/{_segment(dataset_key)}/rows" + return _mapping( + self._api.get( + path, + params={ + "mode": "all", + "limit": DATASET_PAGE_SIZE, + "offset": offset, + }, + ), + description=f"rows for dataset {dataset_key!r}", + ) + + def _get_dataset_rows(self, project_key: str, dataset_key: str) -> list[DatasetRow]: + rows: list[DatasetRow] = [] + offset = 0 + total: int | None = None + while total is None or len(rows) < total: + page = self._fetch_dataset_rows_page( + project_key, dataset_key, offset=offset + ) + items = page.get("items") + page_total = page.get("totalCount") + if not isinstance(items, list) or not isinstance(page_total, int): + raise EvaluationsError( + f"LaunchDarkly returned invalid rows for dataset {dataset_key!r}" + ) + total = page_total + if not items: + break + for item_value in items: + item = _mapping(item_value, description="dataset row") + row_index = item.get("rowIndex") + if not isinstance(row_index, int): + raise EvaluationsError( + "A dataset row is missing its integer rowIndex" + ) + variables_value = item.get("variables") + variables = ( + dict(variables_value) + if isinstance(variables_value, Mapping) + else {} + ) + input_value = item.get("input") + expected_value = item.get("expectedOutput") + rendered_input = ( + parse_template(input_value, variables) + if isinstance(input_value, str) + else None + ) + rendered_expected = ( + parse_template(expected_value, variables) + if isinstance(expected_value, str) + else None + ) + variables["input"] = rendered_input + variables["expected_output"] = rendered_expected + metadata_value = item.get("metadata") + rows.append( + DatasetRow( + row_index=row_index, + input=rendered_input, + expected_output=rendered_expected, + variables=variables, + metadata=( + dict(metadata_value) + if isinstance(metadata_value, Mapping) + else None + ), + ) + ) + offset += len(items) + if not rows: + raise EvaluationsError(f"Dataset {dataset_key!r} is empty") + if total is not None and len(rows) != total: + raise EvaluationsError( + f"Dataset {dataset_key!r} returned {len(rows)} of {total} rows" + ) + return rows + + def _create_evaluation( + self, + project_key: str, + key: str, + generation: GenerationConfig, + tools: Mapping[str, ResolvedTool], + ) -> EvaluationRef: + body: dict[str, Any] = { + "name": key, + "generationProvider": generation["provider"], + "generationModel": generation["model"], + } + if "parameters" in generation: + body["parameters"] = generation["parameters"] + if "instructions" in generation: + body["messages"] = [ + {"role": "system", "content": generation["instructions"]} + ] + elif "messages" in generation: + body["messages"] = generation["messages"] + else: + body["messages"] = [] + if "prompt_snippets" in generation: + body["promptSnippets"] = generation["prompt_snippets"] + if tools: + body["tools"] = [ + {"key": tool.key, "version": tool.version} for tool in tools.values() + ] + + path = f"projects/{_segment(project_key)}/evaluations" + raw = _mapping(self._api.post(path, body=body), description="evaluation") + evaluation_id = _required_string(raw, "id", "evaluation") + response_key = raw.get("name", raw.get("label", key)) + version = raw.get("version") + return EvaluationRef( + id=evaluation_id, + key=str(response_key), + version=version if isinstance(version, int) else None, + ) + + def _create_evaluation_run( + self, + project_key: str, + evaluation_id: str, + row_count: int, + dataset_id: str, + ) -> EvaluationRunRef: + path = ( + f"projects/{_segment(project_key)}/evaluations/" + f"{_segment(evaluation_id)}/runs" + ) + raw = _mapping( + self._api.post( + path, + body={ + "source": "api", + "rowCount": row_count, + "datasetId": dataset_id, + }, + ), + description="evaluation run", + ) + return self._run_ref(raw) + + def _run_ref(self, raw: Mapping[str, Any]) -> EvaluationRunRef: + return EvaluationRunRef( + id=_required_string(raw, "id", "evaluation run"), + evaluation_id=_required_string(raw, "evaluationId", "evaluation run"), + state=_required_string(raw, "state", "evaluation run"), + status_reason=( + str(raw["statusReason"]) + if raw.get("statusReason") is not None + else None + ), + ) + + def _build_handler_config( + self, + generation: GenerationConfig, + tools: Mapping[str, ResolvedTool], + ) -> dict[str, Any]: + parameters = generation.get("parameters") + config: dict[str, Any] = { + "provider": {"name": generation["provider"]}, + "model": {"name": generation["model"], "parameters": parameters}, + "tools": { + key: { + "description": tool.description, + "parameters": tool.schema, + } + for key, tool in tools.items() + }, + } + snippet_variables = {"snippet": generation.get("prompt_snippets", {})} + if "instructions" in generation: + config["instructions"] = parse_template( + generation["instructions"], snippet_variables + ) + elif "messages" in generation: + config["messages"] = [ + { + **message, + "content": parse_template(message["content"], snippet_variables) + if isinstance(message.get("content"), str) + else message.get("content"), + } + for message in generation["messages"] + ] + if "output_format" in generation: + config["outputFormat"] = generation["output_format"] + return config + + async def _run_rows( + self, + rows: list[DatasetRow], + handler: EvalHandler, + config: dict[str, Any], + tool_handlers: dict[str, ToolImplementation], + concurrency: int, + ) -> list[dict[str, Any]]: + controller = ConcurrencyController(concurrency) + + async def invoke(row: DatasetRow) -> dict[str, Any]: + await controller.acquire(config["provider"]["name"]) + started = datetime.now(UTC) + started_clock = time.perf_counter() + try: + result = await handler( + config, row.input, tool_handlers, dict(row.variables) + ) + if not isinstance(result, Mapping): + raise TypeError("handler result must be a mapping") + completed = datetime.now(UTC) + payload: dict[str, Any] = { + "row_index": row.row_index, + "input": row.input, + "expected_output": row.expected_output, + "variables": row.variables, + "metadata": row.metadata, + "output": result.get("output"), + "started_at": started.isoformat().replace("+00:00", "Z"), + "generated_at": completed.isoformat().replace("+00:00", "Z"), + "latency_ms": round((time.perf_counter() - started_clock) * 1000), + "status": "COMPLETE", + } + usage = result.get("usage") + if isinstance(usage, Mapping): + payload["usage"] = dict(usage) + controller.record_success(config["provider"]["name"]) + return payload + except Exception as error: + completed = datetime.now(UTC) + return { + "row_index": row.row_index, + "input": row.input, + "expected_output": row.expected_output, + "variables": row.variables, + "metadata": row.metadata, + "started_at": started.isoformat().replace("+00:00", "Z"), + "generated_at": completed.isoformat().replace("+00:00", "Z"), + "latency_ms": round((time.perf_counter() - started_clock) * 1000), + "status": "ERROR", + "error": {"code": 5001, "message": f"handler raised: {error}"}, + } + finally: + controller.release() + + return list(await asyncio.gather(*(invoke(row) for row in rows))) + + def _emit_generation_events( + self, + client: Any, + *, + project_key: str, + evaluation: EvaluationRef, + evaluation_run: EvaluationRunRef, + dataset: DatasetRef, + results: list[dict[str, Any]], + ) -> None: + """Queue one LD custom event for each executed dataset row.""" + context = to_ld_context( + client, + { + "kind": "evaluation", + "key": evaluation_run.id, + "projectKey": project_key, + "evaluationId": evaluation.id, + }, + ) + for result in results: + identity = { + "projectKey": project_key, + "evaluationId": evaluation.id, + "evaluationRunId": evaluation_run.id, + "runId": evaluation_run.id, + "datasetId": dataset.id, + "rowIndex": result["row_index"], + } + event_id = hashlib.sha256( + json.dumps(identity, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + error = result.get("error") + generated = { + "status": result["status"], + "output": result.get("output"), + "error": error, + } + if result["status"] == "ERROR": + if isinstance(error, Mapping): + message = error.get("message") + generated["errorMessage"] = ( + str(message) if message else "Unknown error" + ) + else: + generated["errorMessage"] = str(error) if error else "Unknown error" + usage = result.get("usage") + if isinstance(usage, Mapping): + normalized_usage = parse_usage(dict(usage)) + generated["usage"] = { + "inputTokens": normalized_usage["input"], + "outputTokens": normalized_usage["output"], + } + content_hash = hashlib.sha256( + json.dumps( + generated, sort_keys=True, separators=(",", ":"), default=str + ).encode() + ).hexdigest() + emitted_at = datetime.now(UTC).isoformat().replace("+00:00", "Z") + payload: dict[str, Any] = { + **identity, + "eventId": event_id, + "contentHash": content_hash, + "emittedAt": emitted_at, + "evaluationKey": evaluation.key, + "evaluationVersion": evaluation.version, + "datasetKey": dataset.key, + "status": result["status"], + "startedAt": result["started_at"], + "generatedAt": result["generated_at"], + "latencyMs": result["latency_ms"], + } + if generated["output"] is not None: + payload["output"] = generated["output"] + if generated["error"] is not None: + payload["error"] = generated["error"] + if generated.get("errorMessage") is not None: + payload["errorMessage"] = generated["errorMessage"] + if "usage" in generated: + payload["usage"] = generated["usage"] + client.track(GENERATION_EVENT_NAME, context, payload, 1) + print( + f"{GENERATION_EVENT_NAME} emittedAt={emitted_at} eventId={event_id}", + flush=True, + ) + + def _get_summary( + self, project_key: str, evaluation_id: str, run_id: str + ) -> RunSummary: + path = ( + f"projects/{_segment(project_key)}/evaluations/{_segment(evaluation_id)}" + f"/runs/{_segment(run_id)}/summary" + ) + return RunSummary.from_wire( + _mapping(self._api.get(path), description="evaluation run summary") + ) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/types.py b/packages/client/src/launchdarkly_ai_server/evaluations/types.py new file mode 100644 index 0000000..5f4f4d5 --- /dev/null +++ b/packages/client/src/launchdarkly_ai_server/evaluations/types.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any, TypedDict + + +@dataclass +class Usage: + """Token counts for one generation, using the ingest wire field names.""" + + input_tokens: int + output_tokens: int + + def to_wire(self) -> dict[str, int]: + return { + "input_tokens": self.input_tokens, + "output_tokens": self.output_tokens, + } + + @classmethod + def from_wire(cls, data: Mapping[str, Any]) -> Usage: + return cls( + input_tokens=int(data.get("input_tokens") or 0), + output_tokens=int(data.get("output_tokens") or 0), + ) + + +class GenerationConfig(TypedDict, total=False): + """Generation settings stored on the evaluation and passed to its handler.""" + + provider: str + model: str + parameters: dict[str, Any] + instructions: str + messages: list[dict[str, Any]] + prompt_snippets: dict[str, str] + output_format: dict[str, Any] + + +@dataclass +class DatasetRef: + """Identifiers returned when resolving a dataset by key.""" + + id: str + key: str + + +@dataclass +class DatasetRow: + """A rendered dataset row ready for handler invocation and ingest.""" + + row_index: int + input: str | None = None + expected_output: str | None = None + variables: dict[str, Any] = field(default_factory=dict) + metadata: dict[str, Any] | None = None + + +@dataclass +class ResolvedTool: + """The schema and pinned version returned by the LaunchDarkly tool API.""" + + key: str + version: int + description: str = "" + schema: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class EvaluationRef: + """Identifiers returned after creating an evaluation.""" + + id: str + key: str + version: int | None = None + + +@dataclass +class EvaluationRunRef: + """Identifiers and state returned by the evaluation-run API.""" + + id: str + evaluation_id: str + state: str + status_reason: str | None = None + + +@dataclass +class RunSummary: + """Row counts for an evaluation run. + + The summary endpoint does not return run state, so terminal completion + is derived from row accounting instead. + """ + + total_rows: int = 0 + passed_rows: int = 0 + failed_rows: int = 0 + error_rows: int = 0 + pending_rows: int = 0 + + @classmethod + def from_wire(cls, data: Mapping[str, Any] | None) -> RunSummary: + data = data or {} + counts_value = data.get("statusCounts") + counts = counts_value if isinstance(counts_value, Mapping) else data + return cls( + total_rows=int(counts.get("total", counts.get("total_rows", 0)) or 0), + passed_rows=int(counts.get("passed", counts.get("passed_rows", 0)) or 0), + failed_rows=int(counts.get("failed", counts.get("failed_rows", 0)) or 0), + error_rows=int(counts.get("error", counts.get("error_rows", 0)) or 0), + pending_rows=int(counts.get("pending", counts.get("pending_rows", 0)) or 0), + ) + + +@dataclass +class EvalRunResult: + """The result of an evaluation run, derived from its row summary.""" + + passed: bool + url: str + run_id: str + summary: RunSummary diff --git a/packages/client/tests/test_evaluations.py b/packages/client/tests/test_evaluations.py new file mode 100644 index 0000000..57edf18 --- /dev/null +++ b/packages/client/tests/test_evaluations.py @@ -0,0 +1,395 @@ +from __future__ import annotations + +import json +from collections.abc import Iterator +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +import launchdarkly_ai_server.lifecycle as lifecycle_module +from launchdarkly_ai_server.evaluations import ( + DEFAULT_BASE_URI, + EvalRunResult, + EvaluationsError, + HttpResponse, + LDApiClient, + LDApiError, + RunSummary, + Usage, + init_evaluations, +) + + +@pytest.fixture(autouse=True) +def reset_sdk_singleton() -> Iterator[None]: + lifecycle_module._reset_for_testing() + yield + lifecycle_module._reset_for_testing() + + +class RecordingTransport: + """Mocked LD API — records requests and replays canned responses.""" + + def __init__(self, responses: list[HttpResponse] | None = None) -> None: + self.requests: list[dict[str, Any]] = [] + self.responses = responses or [HttpResponse(status=200, body="{}")] + + def __call__( + self, + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: + self.requests.append( + { + "method": method, + "url": url, + "headers": headers, + "body": json.loads(body) if body else None, + "timeout": timeout, + } + ) + index = min(len(self.requests) - 1, len(self.responses) - 1) + return self.responses[index] + + +def failing_transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, +) -> HttpResponse: + raise AssertionError("no network I/O expected") + + +def test_init_resolves_credentials_from_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token-from-env") + monkeypatch.setenv("LD_SDK_KEY", "sdk-key-from-env") + + evals = init_evaluations(transport=RecordingTransport()) + + assert evals.api.api_token == "api-token-from-env" + assert evals.sdk_key == "sdk-key-from-env" + assert evals.api.base_uri == DEFAULT_BASE_URI + assert evals.ui_base_uri == "https://app.launchdarkly.com" + + +def test_init_prefers_explicit_credentials(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token-from-env") + monkeypatch.setenv("LD_SDK_KEY", "sdk-key-from-env") + + evals = init_evaluations( + api_token="explicit-token", + sdk_key="explicit-sdk-key", + transport=RecordingTransport(), + ) + + assert evals.api.api_token == "explicit-token" + assert evals.sdk_key == "explicit-sdk-key" + + +def test_missing_api_token_raises_before_network_io( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("LD_API_TOKEN", raising=False) + monkeypatch.setenv("LD_SDK_KEY", "sdk-key") + + with pytest.raises(EvaluationsError, match="LD_API_TOKEN"): + init_evaluations(transport=failing_transport) + + +def test_blank_api_token_env_is_treated_as_unset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LD_API_TOKEN", " ") + + with pytest.raises(EvaluationsError): + init_evaluations(transport=failing_transport) + + +def test_missing_sdk_key_raises_before_network_io( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token") + monkeypatch.delenv("LD_SDK_KEY", raising=False) + + with pytest.raises(EvaluationsError, match="LD_SDK_KEY"): + init_evaluations(transport=failing_transport) + + +def test_blank_sdk_key_env_is_treated_as_unset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token") + monkeypatch.setenv("LD_SDK_KEY", " ") + + with pytest.raises(EvaluationsError, match="LD_SDK_KEY"): + init_evaluations(transport=failing_transport) + + +def test_missing_sdk_key_is_allowed_with_a_byoc_client( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token") + monkeypatch.delenv("LD_SDK_KEY", raising=False) + byoc_client = MagicMock() + byoc_client.track = MagicMock() + byoc_client.flush = AsyncMock() + lifecycle_module._set_client_for_testing(byoc_client) + + evals = init_evaluations(transport=failing_transport) + + assert evals.sdk_key is None + + +def test_missing_sdk_key_raises_when_the_byoc_client_cannot_emit_events( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token") + monkeypatch.delenv("LD_SDK_KEY", raising=False) + lifecycle_module._set_client_for_testing(object()) + + with pytest.raises(EvaluationsError, match="LD_SDK_KEY"): + init_evaluations(transport=failing_transport) + + +def test_base_uri_override_isolated_from_sdk_delivery_uri( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token") + monkeypatch.setenv("LD_SDK_KEY", "sdk-key") + monkeypatch.setenv("LD_API_BASE_URI", "https://api.staging.example.com/") + monkeypatch.setenv("LD_BASE_URI", "https://relay.example.com/") + + from_env = init_evaluations(transport=RecordingTransport()) + explicit = init_evaluations( + base_uri="https://other.example.com", transport=RecordingTransport() + ) + + assert from_env.api.base_uri == "https://api.staging.example.com" + assert explicit.api.base_uri == "https://other.example.com" + + +def test_ui_base_uri_precedence_and_api_base_isolation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LD_API_TOKEN", "api-token") + monkeypatch.setenv("LD_SDK_KEY", "sdk-key") + monkeypatch.setenv("LD_API_BASE_URI", "https://api.staging.example.com") + monkeypatch.setenv("LD_UI_BASE_URI", "https://ld-stg.launchdarkly.com/") + + from_env = init_evaluations(transport=RecordingTransport()) + explicit = init_evaluations( + ui_base_uri="https://ui.example.com/", transport=RecordingTransport() + ) + + assert from_env.api.base_uri == "https://api.staging.example.com" + assert from_env.ui_base_uri == "https://ld-stg.launchdarkly.com" + assert explicit.ui_base_uri == "https://ui.example.com" + + +def test_requests_carry_token_auth_and_json_body() -> None: + transport = RecordingTransport([HttpResponse(status=201, body='{"key": "run-1"}')]) + client = LDApiClient(api_token="api-token", transport=transport) + + result = client.post("projects/proj/evaluations", body={"key": "support-qa"}) + + assert result == {"key": "run-1"} + request = transport.requests[0] + assert request["method"] == "POST" + assert request["url"] == f"{DEFAULT_BASE_URI}/api/v2/projects/proj/evaluations" + assert request["headers"]["Authorization"] == "api-token" + assert request["headers"]["Content-Type"] == "application/json" + assert request["body"] == {"key": "support-qa"} + + +def test_get_encodes_query_params_and_omits_none() -> None: + transport = RecordingTransport([HttpResponse(status=200, body='{"items": []}')]) + client = LDApiClient( + api_token="api-token", base_uri="https://ld.example.com", transport=transport + ) + + client.get("projects/proj/datasets/golden", params={"limit": 50, "offset": None}) + + request = transport.requests[0] + assert ( + request["url"] + == "https://ld.example.com/api/v2/projects/proj/datasets/golden?limit=50" + ) + assert "Content-Type" not in request["headers"] + + +def test_rate_limit_retries_and_honors_retry_after() -> None: + transport = RecordingTransport( + [ + HttpResponse( + status=429, + body='{"message": "slow down"}', + headers={"retry-after": "2"}, + ), + HttpResponse(status=200, body='{"items": []}'), + ] + ) + sleeps: list[float] = [] + client = LDApiClient( + api_token="api-token", + transport=transport, + max_retries=1, + sleep=sleeps.append, + random_value=lambda: 0.0, + ) + + assert client.get("projects/proj/datasets") == {"items": []} + assert len(transport.requests) == 2 + assert sleeps == [2.0] + + +def test_server_error_retries_get_but_not_post() -> None: + server_error = HttpResponse(status=503, body='{"message": "unavailable"}') + get_transport = RecordingTransport( + [server_error, HttpResponse(200, '{"ok": true}')] + ) + client = LDApiClient( + api_token="api-token", + transport=get_transport, + max_retries=2, + sleep=lambda _: None, + random_value=lambda: 0.0, + ) + + assert client.get("projects/proj/datasets") == {"ok": True} + assert len(get_transport.requests) == 2 + + post_transport = RecordingTransport([server_error]) + client = LDApiClient( + api_token="api-token", + transport=post_transport, + max_retries=2, + sleep=lambda _: None, + random_value=lambda: 0.0, + ) + + with pytest.raises(LDApiError) as excinfo: + client.post("projects/proj/evaluations", body={"name": "eval"}) + + assert excinfo.value.status == 503 + assert len(post_transport.requests) == 1 + + +def test_transport_failure_is_not_replayed_for_post() -> None: + attempts: list[str] = [] + + def timing_out_transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: + attempts.append(method) + raise TimeoutError("timed out") + + client = LDApiClient( + api_token="api-token", + transport=timing_out_transport, + max_retries=2, + sleep=lambda _: None, + random_value=lambda: 0.0, + ) + + with pytest.raises(EvaluationsError): + client.post("projects/proj/evaluations", body={"name": "eval"}) + + assert attempts == ["POST"] + + +def test_rate_limited_post_is_retried() -> None: + transport = RecordingTransport( + [ + HttpResponse(status=429, body='{"message": "slow down"}'), + HttpResponse(status=201, body='{"id": "eval-id"}'), + ] + ) + client = LDApiClient( + api_token="api-token", + transport=transport, + max_retries=1, + sleep=lambda _: None, + random_value=lambda: 0.0, + ) + + assert client.post("projects/proj/evaluations", body={"name": "eval"}) == { + "id": "eval-id" + } + assert len(transport.requests) == 2 + + +def test_forbidden_response_is_not_retried() -> None: + transport = RecordingTransport( + [HttpResponse(status=403, body='{"message": "forbidden"}')] + ) + client = LDApiClient(api_token="api-token", transport=transport, max_retries=3) + + with pytest.raises(LDApiError) as excinfo: + client.get("projects/proj/evaluations") + + assert excinfo.value.status == 403 + assert len(transport.requests) == 1 + + +def test_error_response_raises_ld_api_error() -> None: + transport = RecordingTransport( + [HttpResponse(status=404, body='{"message": "nope"}')] + ) + client = LDApiClient(api_token="api-token", transport=transport) + + with pytest.raises(LDApiError) as excinfo: + client.get("projects/proj/ai-tools/missing") + + assert excinfo.value.status == 404 + assert excinfo.value.path == "projects/proj/ai-tools/missing" + + +def test_empty_response_body_is_none() -> None: + transport = RecordingTransport([HttpResponse(status=204, body="")]) + client = LDApiClient(api_token="api-token", transport=transport) + + assert client.post("projects/proj/evaluations/support-qa/runs") is None + + +def test_usage_matches_ingest_wire_shape() -> None: + usage = Usage(input_tokens=812, output_tokens=96) + + assert usage.to_wire() == {"input_tokens": 812, "output_tokens": 96} + assert Usage.from_wire({"input_tokens": 1, "output_tokens": 2}) == Usage(1, 2) + assert Usage.from_wire({}) == Usage(0, 0) + + +def test_run_summary_and_result() -> None: + summary = RunSummary.from_wire( + { + "total_rows": 500, + "passed_rows": 497, + "failed_rows": 1, + "error_rows": 1, + "pending_rows": 1, + } + ) + result = EvalRunResult( + passed=False, + url="https://app.launchdarkly.com/run", + run_id="run-1", + summary=summary, + ) + + assert summary.total_rows == 500 + assert summary.error_rows == 1 + assert summary.pending_rows == 1 + assert RunSummary.from_wire({"pending": 2}).pending_rows == 2 + assert RunSummary.from_wire(None) == RunSummary() + assert result.passed is False + assert result.run_id == "run-1" diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py new file mode 100644 index 0000000..5784559 --- /dev/null +++ b/packages/client/tests/test_evaluations_run.py @@ -0,0 +1,1035 @@ +from __future__ import annotations + +import json +from collections.abc import Callable +from datetime import datetime +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from launchdarkly_ai_server.evaluations import ( + EvaluationsError, + HttpResponse, + init_evaluations, +) + + +@pytest.fixture(autouse=True) +def stub_sdk_client(monkeypatch: pytest.MonkeyPatch) -> MagicMock: + """Give every run a resolvable SDK client, since one is now required.""" + monkeypatch.setenv("LD_SDK_KEY", "sdk-key") + client = MagicMock() + client.flush = AsyncMock() + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.get_client", + MagicMock(side_effect=RuntimeError("client not initialized")), + ) + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.init_client", + AsyncMock(return_value=client), + ) + return client + + +class SequencedTransport: + """Records requests and returns one response for each expected request.""" + + def __init__(self, responses: list[HttpResponse]) -> None: + self.responses = responses + self.requests: list[dict[str, Any]] = [] + + def __call__( + self, + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: + index = len(self.requests) + self.requests.append( + { + "method": method, + "url": url, + "headers": headers, + "body": json.loads(body) if body else None, + "timeout": timeout, + } + ) + if index >= len(self.responses): + raise AssertionError(f"unexpected request: {method} {url}") + return self.responses[index] + + +def response(status: int, body: dict[str, Any] | None = None) -> HttpResponse: + return HttpResponse( + status=status, body=json.dumps(body) if body is not None else "" + ) + + +def failing_transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, +) -> HttpResponse: + raise AssertionError("no network I/O expected") + + +def dataset_page( + items: list[dict[str, Any]], total: int, next_href: str | None = None +) -> dict[str, Any]: + links: dict[str, Any] = {"self": {"href": "https://api.test/current"}} + if next_href: + links["next"] = {"href": next_href} + return {"items": items, "totalCount": total, "_links": links} + + +async def successful_handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], +) -> dict[str, Any]: + assert config["provider"] == {"name": "OpenAI"} + assert config["model"] == { + "name": "gpt-4o", + "parameters": {"temperature": 0.2}, + } + assert config["tools"]["lookup_order"] == { + "description": "Look up an order", + "parameters": {"type": "object"}, + } + assert "lookup_order" in tool_handlers + assert variables["input"] == user_input + return { + "output": f"generated: {user_input}", + "usage": {"input_tokens": 10, "output_tokens": 4}, + } + + +def lookup_order(order_id: str) -> str: + return order_id + + +@pytest.mark.asyncio +async def test_complete_run_with_zero_failed_and_error_rows_passes( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + monkeypatch.delenv("LD_SDK_KEY", raising=False) + init_client = AsyncMock() + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.init_client", init_client + ) + transport = SequencedTransport( + [ + response( + 200, + { + "key": "lookup_order", + "version": 7, + "description": "Look up an order", + "schema": {"type": "object"}, + }, + ), + response( + 200, + { + "id": "33333333-3333-3333-3333-333333333333", + "name": "golden", + }, + ), + response( + 200, + dataset_page( + [ + { + "rowIndex": 4, + "input": "Order {{order_id}}", + "expectedOutput": "Found {{order_id}}", + "variables": {"order_id": "A19"}, + "metadata": {"suite": "orders"}, + } + ], + total=2, + next_href="https://api.test/api/v2/projects/proj/datasets/key/golden/preview?limit=1&offset=1", + ), + ), + response( + 200, + dataset_page( + [ + { + "rowIndex": 9, + "input": "Order {{order_id}}", + "expectedOutput": None, + "variables": {"order_id": "B20"}, + "metadata": None, + } + ], + total=2, + ), + ), + response( + 201, + { + "id": "11111111-1111-1111-1111-111111111111", + "name": "support-qa-unique", + "version": 1, + }, + ), + response( + 201, + { + "id": "22222222-2222-2222-2222-222222222222", + "evaluationId": "11111111-1111-1111-1111-111111111111", + "evaluationVersion": 1, + "source": "api", + "state": "PENDING", + "createdAt": 1, + }, + ), + response( + 200, + { + "evaluationId": "11111111-1111-1111-1111-111111111111", + "evaluationVersion": 1, + "evaluationRunId": "22222222-2222-2222-2222-222222222222", + "state": "COMPLETE", + "statusCounts": { + "total": 2, + "passed": 2, + "failed": 0, + "error": 0, + "pending": 0, + }, + "createdAt": 1, + }, + ), + ] + ) + client = MagicMock() + client.variation = AsyncMock(return_value=True) + client.flush = AsyncMock() + init_client.return_value = client + evals = init_evaluations( + api_token="token", + sdk_key="sdk-key", + base_uri="https://api.example.com", + ui_base_uri="https://ui.example.com/", + transport=transport, + ) + assert evals.sdk_key == "sdk-key" + + result = await evals.run( + project_key="proj", + key="support-qa-unique", + dataset="golden", + handler=successful_handler, + tools={"lookup_order": lookup_order}, + generation={ + "provider": "OpenAI", + "model": "gpt-4o", + "parameters": {"temperature": 0.2}, + "instructions": "Help the user.", + }, + concurrency=2, + ) + + assert result.passed is True + assert result.run_id == "22222222-2222-2222-2222-222222222222" + assert result.url == ( + "https://ui.example.com/projects/proj/ai/evaluations/" + "11111111-1111-1111-1111-111111111111/runs/" + "22222222-2222-2222-2222-222222222222" + ) + assert result.summary.total_rows == 2 + + assert [request["method"] for request in transport.requests] == [ + "GET", + "GET", + "GET", + "GET", + "POST", + "POST", + "GET", + ] + assert transport.requests[0]["url"].endswith( + "/api/v2/projects/proj/ai-tools/lookup_order" + ) + assert transport.requests[1]["url"].endswith( + "/api/v2/projects/proj/datasets/golden" + ) + assert "/projects/proj/datasets/golden/rows" in transport.requests[2]["url"] + assert "mode=all" in transport.requests[2]["url"] + assert transport.requests[4]["body"] == { + "name": "support-qa-unique", + "generationProvider": "OpenAI", + "generationModel": "gpt-4o", + "parameters": {"temperature": 0.2}, + "messages": [{"role": "system", "content": "Help the user."}], + "tools": [{"key": "lookup_order", "version": 7}], + } + assert transport.requests[5]["url"].endswith( + "/api/v2/projects/proj/evaluations/11111111-1111-1111-1111-111111111111/runs" + ) + assert transport.requests[5]["body"] == { + "source": "api", + "rowCount": 2, + "datasetId": "33333333-3333-3333-3333-333333333333", + } + + assert not any( + request["url"].endswith("/generation-results") for request in transport.requests + ) + assert client.track.call_count == 2 + event_name, context, event, metric_value = client.track.call_args_list[0].args + assert event_name == "$ld:ai:offline-evals:generation" + assert context["key"] == "22222222-2222-2222-2222-222222222222" + assert metric_value == 1 + assert event["projectKey"] == "proj" + assert event["evaluationId"] == "11111111-1111-1111-1111-111111111111" + assert event["evaluationRunId"] == "22222222-2222-2222-2222-222222222222" + assert event["runId"] == event["evaluationRunId"] + assert event["datasetId"] == "33333333-3333-3333-3333-333333333333" + assert event["rowIndex"] == 4 + assert event["status"] == "COMPLETE" + assert event["output"] == "generated: Order A19" + assert event["usage"] == {"inputTokens": 10, "outputTokens": 4} + assert "generationOutput" not in event + assert "inputTokens" not in event + assert "outputTokens" not in event + assert len(event["eventId"]) == len(event["contentHash"]) == 64 + assert event["emittedAt"].endswith("Z") + assert datetime.fromisoformat(event["emittedAt"]).tzinfo is not None + assert {"input", "expected_output", "metadata", "variables"}.isdisjoint(event) + output_lines = capsys.readouterr().out.splitlines() + assert len(output_lines) == 2 + assert output_lines[0] == ( + "$ld:ai:offline-evals:generation " + f"emittedAt={event['emittedAt']} eventId={event['eventId']}" + ) + client.flush.assert_awaited_once_with() + client.variation.assert_not_awaited() + init_client.assert_awaited_once_with({"sdkKey": "sdk-key"}) + + +@pytest.mark.asyncio +async def test_generation_events_always_emit_without_flag_or_run_status_poll( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.SUMMARY_POLL_INTERVAL_SECONDS", 0 + ) + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [{"rowIndex": 3, "input": "hello", "variables": {}}], + total=1, + ), + ), + response(201, {"id": "evaluation-id", "name": "eval-key"}), + response( + 201, + { + "id": "run-id", + "evaluationId": "evaluation-id", + "state": "PENDING", + }, + ), + response( + 200, + { + "total": 1, + "passed": 0, + "failed": 0, + "error": 0, + "pending": 1, + }, + ), + response( + 200, + { + "total": 1, + "passed": 0, + "failed": 0, + "error": 1, + "pending": 0, + }, + ), + ] + ) + client = MagicMock() + client.variation = AsyncMock(side_effect=AssertionError("flag must not be read")) + + def flush_before_summary() -> None: + assert len(transport.requests) == 4 + assert transport.requests[-1]["url"].endswith("/evaluations/evaluation-id/runs") + + client.flush.side_effect = flush_before_summary + + async def fake_init_client(options: dict[str, Any]) -> MagicMock: + assert options == {"sdkKey": "sdk-key"} + return client + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.init_client", fake_init_client + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + assert result.passed is False + assert result.summary.error_rows == 1 + assert result.summary.pending_rows == 0 + request_urls = [request["url"] for request in transport.requests] + assert not any(url.endswith("/generation-results") for url in request_urls) + client.variation.assert_not_awaited() + client.track.assert_called_once() + client.flush.assert_called_once_with() + status_url = "/evaluations/evaluation-id/runs/run-id" + assert not any(url.endswith(status_url) for url in request_urls) + assert request_urls[-1].endswith(f"{status_url}/summary") + + +@pytest.mark.asyncio +async def test_summary_is_polled_until_rows_are_accounted( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.SUMMARY_POLL_INTERVAL_SECONDS", 0 + ) + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [{"rowIndex": 0, "input": "hello", "variables": {}}], total=1 + ), + ), + response(201, {"id": "evaluation-id", "name": "eval-key"}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + { + "state": "PENDING", + "statusCounts": {"total": 1, "passed": 0, "error": 0, "pending": 1}, + }, + ), + response( + 200, + { + "state": "COMPLETE", + "statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}, + }, + ), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + summary_requests = [ + request for request in transport.requests if request["url"].endswith("/summary") + ] + assert len(summary_requests) == 2 + assert result.passed is True + + +@pytest.mark.asyncio +async def test_summary_polling_completes_for_real_backend_summary_without_state( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.SUMMARY_POLL_INTERVAL_SECONDS", 0 + ) + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [{"rowIndex": 0, "input": "hello", "variables": {}}], total=1 + ), + ), + response(201, {"id": "evaluation-id", "name": "eval-key"}), + response( + 201, + { + "id": "run-id", + "evaluationId": "evaluation-id", + "state": "COMPLETE", + "rowCount": 10, + "selectedRowCount": 10, + }, + ), + response( + 200, + { + "statusCounts": { + "total": 10, + "passed": 10, + "failed": 0, + "error": 0, + "pending": 0, + }, + "estimatedRemainingWindowMs": 0, + }, + ), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + summary_requests = [ + request for request in transport.requests if request["url"].endswith("/summary") + ] + assert len(summary_requests) == 1 + assert result.summary.total_rows == 10 + assert result.summary.pending_rows == 0 + assert result.summary.passed_rows == 10 + assert result.summary.failed_rows == 0 + assert result.summary.error_rows == 0 + + +@pytest.mark.asyncio +async def test_summary_polling_ignores_missing_state_even_when_pending_is_zero( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.SUMMARY_POLL_INTERVAL_SECONDS", 0 + ) + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [{"rowIndex": 0, "input": "hello", "variables": {}}], total=1 + ), + ), + response(201, {"id": "evaluation-id", "name": "eval-key"}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response(200, {}), + response( + 200, + {"statusCounts": {"total": 1, "passed": 0, "error": 0, "pending": 0}}, + ), + response( + 200, + { + "state": "COMPLETE", + "statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}, + }, + ), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + summary_requests = [ + request for request in transport.requests if request["url"].endswith("/summary") + ] + assert len(summary_requests) == 3 + assert result.passed is True + + +@pytest.mark.asyncio +async def test_summary_polling_times_out_waiting_for_rows_to_be_accounted( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.SUMMARY_POLL_TIMEOUT_SECONDS", 0 + ) + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [{"rowIndex": 0, "input": "hello", "variables": {}}], total=1 + ), + ), + response(201, {"id": "evaluation-id", "name": "eval-key"}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + { + "state": "PENDING", + "statusCounts": {"total": 1, "passed": 0, "error": 0, "pending": 1}, + }, + ), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + with pytest.raises( + EvaluationsError, + match=r"Timed out after 0 seconds.*rows to be fully accounted.*pending_rows=1", + ): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + summary_requests = [ + request for request in transport.requests if request["url"].endswith("/summary") + ] + assert len(summary_requests) == 1 + + +@pytest.mark.asyncio +async def test_poll_timeout_and_interval_are_configurable_per_run() -> None: + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [{"rowIndex": 0, "input": "hello", "variables": {}}], total=1 + ), + ), + response(201, {"id": "evaluation-id", "name": "eval-key"}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + {"statusCounts": {"total": 1, "passed": 0, "error": 0, "pending": 1}}, + ), + response( + 200, + {"statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}}, + ), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + poll_interval_seconds=0, + poll_timeout_seconds=600, + ) + + assert result.passed is True + summary_requests = [ + request for request in transport.requests if request["url"].endswith("/summary") + ] + assert len(summary_requests) == 2 + + with pytest.raises(EvaluationsError, match="poll_timeout_seconds"): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + poll_timeout_seconds=-1, + ) + + +@pytest.mark.parametrize( + ("poll_interval_seconds", "poll_timeout_seconds"), + [(float("nan"), 1.0), (1.0, float("nan"))], + ids=["interval", "timeout"], +) +@pytest.mark.asyncio +async def test_nan_poll_values_are_rejected( + poll_interval_seconds: float, poll_timeout_seconds: float +) -> None: + evals = init_evaluations(api_token="token", transport=failing_transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + with pytest.raises(EvaluationsError, match="must be a number"): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + poll_interval_seconds=poll_interval_seconds, + poll_timeout_seconds=poll_timeout_seconds, + ) + + +@pytest.mark.asyncio +async def test_run_uses_a_byoc_client_when_no_sdk_key_is_configured( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("LD_SDK_KEY", raising=False) + byoc_client = MagicMock() + byoc_client.flush = AsyncMock() + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.get_client", + MagicMock(return_value=byoc_client), + ) + init_client = AsyncMock(side_effect=AssertionError("must reuse the BYOC client")) + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.init_client", init_client + ) + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [{"rowIndex": 0, "input": "hello", "variables": {}}], total=1 + ), + ), + response(201, {"id": "evaluation-id", "name": "eval-key"}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + {"statusCounts": {"total": 1, "passed": 1, "error": 0, "pending": 0}}, + ), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + assert evals.sdk_key is None + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + assert result.passed is True + init_client.assert_not_awaited() + byoc_client.track.assert_called_once() + byoc_client.flush.assert_awaited_once_with() + + +@pytest.mark.asyncio +async def test_run_raises_when_no_sdk_key_and_no_initialized_client( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("LD_SDK_KEY", raising=False) + byoc_client = MagicMock() + byoc_client.flush = AsyncMock() + get_client = MagicMock(return_value=byoc_client) + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.get_client", get_client + ) + evals = init_evaluations(api_token="token", transport=failing_transport) + get_client.side_effect = RuntimeError("client not initialized") + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + with pytest.raises(EvaluationsError, match="no initialized LaunchDarkly client"): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + +@pytest.mark.asyncio +async def test_generation_failed_rows_do_not_fail_the_result() -> None: + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "name": "golden"}), + response( + 200, + dataset_page( + [{"rowIndex": 0, "input": "hello", "variables": {}}], total=1 + ), + ), + response(201, {"id": "evaluation-id", "name": "eval-key"}), + response( + 201, + {"id": "run-id", "evaluationId": "evaluation-id", "state": "PENDING"}, + ), + response( + 200, + { + "state": "COMPLETE", + "statusCounts": { + "total": 1, + "passed": 0, + "failed": 1, + "error": 0, + "pending": 0, + }, + }, + ), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + + async def handler(*args: object) -> dict[str, Any]: + return {"output": "generated"} + + result = await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + assert result.summary.failed_rows == 1 + assert result.passed is True + + +@pytest.mark.asyncio +async def test_run_rejects_instructions_and_messages_before_network_io() -> None: + transport = SequencedTransport([]) + evals = init_evaluations(api_token="token", transport=transport) + + with pytest.raises(EvaluationsError, match=r"instructions.*messages"): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=successful_handler, + generation={ + "provider": "OpenAI", + "model": "gpt-4o", + "instructions": "System prompt", + "messages": [{"role": "user", "content": "{{input}}"}], + }, + ) + + assert transport.requests == [] + + +@pytest.mark.asyncio +async def test_missing_tool_aborts_before_any_mutating_request() -> None: + transport = SequencedTransport( + [response(404, {"code": "not_found", "message": "not found"})] + ) + evals = init_evaluations(api_token="token", transport=transport) + + with pytest.raises(EvaluationsError, match="missing_tool"): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=successful_handler, + tools={"missing_tool": lookup_order}, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + assert [request["method"] for request in transport.requests] == ["GET"] + + +@pytest.mark.asyncio +async def test_empty_dataset_fails_before_evaluation_or_run_creation() -> None: + transport = SequencedTransport( + [ + response( + 200, + { + "id": "33333333-3333-3333-3333-333333333333", + "name": "golden", + }, + ), + response(200, dataset_page([], total=0)), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + + with pytest.raises(EvaluationsError, match="empty"): + await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=successful_handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + assert [request["method"] for request in transport.requests] == ["GET", "GET"] + + +@pytest.mark.asyncio +async def test_complete_run_with_error_rows_does_not_pass( + monkeypatch: pytest.MonkeyPatch, +) -> None: + error_rows = 1 + calls: list[str | None] = [] + + async def handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], + ) -> dict[str, Any]: + calls.append(user_input) + if user_input == "bad": + raise RuntimeError("provider failed") + return {"output": "ok"} + + transport = SequencedTransport( + [ + response( + 200, + { + "id": "33333333-3333-3333-3333-333333333333", + "name": "golden", + }, + ), + response( + 200, + dataset_page( + [ + {"rowIndex": 0, "input": "bad", "variables": {}}, + {"rowIndex": 1, "input": "good", "variables": {}}, + ], + total=2, + ), + ), + response( + 201, + { + "id": "11111111-1111-1111-1111-111111111111", + "name": "eval-key", + "version": 1, + }, + ), + response( + 201, + { + "id": "22222222-2222-2222-2222-222222222222", + "evaluationId": "11111111-1111-1111-1111-111111111111", + "evaluationVersion": 1, + "source": "api", + "state": "PENDING", + "createdAt": 1, + }, + ), + response( + 200, + { + "evaluationId": "11111111-1111-1111-1111-111111111111", + "evaluationVersion": 1, + "evaluationRunId": "22222222-2222-2222-2222-222222222222", + "state": "COMPLETE", + "statusCounts": { + "total": 2, + "passed": 2 - error_rows, + "error": error_rows, + "pending": 0, + }, + "createdAt": 1, + }, + ), + ] + ) + client = MagicMock() + client.variation = AsyncMock(return_value=True) + + async def fake_init_client(options: dict[str, Any]) -> MagicMock: + assert options == {"sdkKey": "sdk-key"} + return client + + monkeypatch.setattr( + "launchdarkly_ai_server.evaluations.module.init_client", fake_init_client + ) + evals = init_evaluations(api_token="token", sdk_key="sdk-key", transport=transport) + + result = await evals.run( + project_key="proj", + key="eval-key", + dataset="golden", + handler=handler, + generation={"provider": "OpenAI", "model": "gpt-4o"}, + ) + + assert set(calls) == {"bad", "good"} + assert len(calls) == 2 + assert result.passed is False + client.variation.assert_not_awaited() + assert client.track.call_count == 2 + events = [call.args[2] for call in client.track.call_args_list] + assert {event["status"] for event in events} == {"COMPLETE", "ERROR"} + error_event = next(event for event in events if event["status"] == "ERROR") + assert error_event["rowIndex"] == 0 + assert "provider failed" in error_event["error"]["message"] + assert "provider failed" in error_event["errorMessage"] + assert "generationOutput" not in error_event + assert "output" not in error_event + assert "usage" not in error_event + assert "inputTokens" not in error_event + assert "outputTokens" not in error_event + assert {"input", "expected_output", "metadata", "variables"}.isdisjoint(error_event)