diff --git a/packages/client/README.md b/packages/client/README.md index b2607d4e..613ee816 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -82,6 +82,25 @@ sys.exit(asyncio.run(main())) Generation and criterion 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. +### Supply the dataset inline + +Pass a list of rows as `dataset`, instead of a dataset key, for datasets that live in code or are built at run time rather than stored in LaunchDarkly. Each row is a `DatasetRow` or a mapping in the upload wire shape — any of `input`, `expectedOutput`, `variables` and `metadata`, plus an optional `rowIdx` — and its index is its position in the list. + +```python +result = await evals.run( + project_key="my-project", + key="support-qa-2026-08-20", + dataset=[ + {"input": "How do I reset my password?", "expectedOutput": "Use the reset link."}, + {"input": "Where is order {{order_id}}?", "variables": {"order_id": "A-17"}}, + ], + handler=create_openai_messages_handler(), + generation={"provider": "OpenAI", "model": "gpt-4o"}, +) +``` + +The rows are uploaded to the run, in batches of up to 500, before any generation starts. They are uploaded unrendered, and `{{...}}` placeholders render exactly as they do for a stored dataset. Every row needs a non-empty `input`, and `variables` and `metadata` must be JSON-encodable with no NaN or Infinity; a malformed row fails the run before any records are created. If an upload batch is rejected, the harness cancels the run — the API offers no way to mark it failed — and raises the upload error. Events from an inline run carry no dataset id. + ### Score rows with judges and scorers Pass `criteria` to `run()` to score every generated row. A `Judge` references an AI Judge config that already exists in LaunchDarkly — the SDK creates no judges and ships none of its own — and a `Scorer` wraps a local function, so a run can mix model-graded and deterministic checks. Each criterion runs once per generated row, bounded by the same `concurrency` as generation, and emits one `$ld:ai:offline-evals:criterion` event per `(row, criterion)` carrying the criterion identity, the judge's variation key and version, the validated score, its reason, usage, and timings. diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/api.py b/packages/client/src/launchdarkly_ai_server/evaluations/api.py index 3a46d23d..52d5f6c0 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/api.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/api.py @@ -133,7 +133,15 @@ def request( path: str, body: Any = None, params: dict[str, Any] | None = None, + *, + idempotent: bool = False, ) -> Any: + """Send one request, retrying where a replay cannot duplicate work. + + ``idempotent`` marks a non-GET request the server applies at most once + per payload, so it may be replayed after a 5xx or transport failure. + """ + retry_safe = idempotent or method.upper() in RETRY_SAFE_METHODS headers = { "Authorization": self.api_token, "Accept": "application/json", @@ -152,10 +160,7 @@ def request( 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 - ): + if not retry_safe or attempt >= self._max_retries: raise EvaluationsError( f"LaunchDarkly API {method} {path} failed after retries: {error}" ) from error @@ -165,7 +170,7 @@ def request( # 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 + response.status >= 500 and retry_safe ) if retryable and attempt < self._max_retries: self._sleep(self._retry_delay(attempt, response)) @@ -190,5 +195,5 @@ def request( 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) + def post(self, path: str, body: Any = None, *, idempotent: bool = False) -> Any: + return self.request("POST", path, body=body, idempotent=idempotent) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/events.py b/packages/client/src/launchdarkly_ai_server/evaluations/events.py index 4098e61b..9418ff45 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/events.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/events.py @@ -37,14 +37,14 @@ class CriterionEventPayload: evaluation_id: str evaluation_run_id: str run_id: str - dataset_id: str + dataset_id: str | None row_index: int criterion_type: str kind: CriterionEventKind event_id: str emitted_at: str evaluation_key: str - dataset_key: str + dataset_key: str | None status: CriterionStatus started_at: str evaluated_at: str diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 4cc684aa..3f64e852 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -2,11 +2,12 @@ import asyncio import inspect +import json import logging import math import os import time -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from typing import Any, cast from ..lifecycle import get_client, init_client @@ -24,14 +25,122 @@ ToolImplementation, _provides_for, _segment, + render_row, +) +from .types import ( + AIConfig, + DatasetRef, + DatasetRow, + EvalRunResult, + GenerationConfig, + InlineDatasetRow, + RunSummary, ) -from .types import AIConfig, 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 +INLINE_ROW_FIELDS = ("rowIdx", "input", "expectedOutput", "variables", "metadata") + + +def _render_inline_row(row: DatasetRow) -> DatasetRow: + return render_row( + row.row_index, + input_value=row.input, + expected_value=row.expected_output, + variables_value=row.variables, + metadata_value=row.metadata, + ) + + +def _normalize_inline_rows(rows: Sequence[InlineDatasetRow]) -> list[DatasetRow]: + """Validate caller-supplied rows, returning them raw and indexed by position. + + Pure, so a malformed row fails before any records are created. Each row + is held to the upload schema, so the server cannot reject a batch the + harness has already accepted. The values stay unrendered: they are + uploaded as stored rows, which the server renders the same way it renders + a hosted dataset's. + """ + if not rows: + raise EvaluationsError("Inline dataset is empty") + normalized: list[DatasetRow] = [] + for position, row in enumerate(rows): + if isinstance(row, DatasetRow): + if row.row_index != position: + raise EvaluationsError( + f"Inline dataset row {position} has row_index {row.row_index}; " + "an inline row's index is its position in the list" + ) + values: Mapping[str, Any] = { + "input": row.input, + "expectedOutput": row.expected_output, + "variables": row.variables, + "metadata": row.metadata, + } + elif isinstance(row, Mapping): + unknown = sorted(str(key) for key in row if key not in INLINE_ROW_FIELDS) + if unknown: + raise EvaluationsError( + f"Inline dataset row {position} has unknown fields: " + + ", ".join(repr(key) for key in unknown) + + ". Expected any of: " + + ", ".join(repr(key) for key in INLINE_ROW_FIELDS) + ) + row_idx = row.get("rowIdx") + if row_idx is not None and ( + isinstance(row_idx, bool) or row_idx != position + ): + raise EvaluationsError( + f"Inline dataset row {position} has rowIdx {row_idx!r}; " + "an inline row's index is its position in the list" + ) + values = row + else: + raise EvaluationsError( + f"Inline dataset row {position} must be a DatasetRow or a mapping" + ) + row_input = values.get("input") + if not isinstance(row_input, str) or not row_input: + raise EvaluationsError( + f"Inline dataset row {position} input must be a non-empty string" + ) + expected_output = values.get("expectedOutput") + if expected_output is not None and not isinstance(expected_output, str): + raise EvaluationsError( + f"Inline dataset row {position} expectedOutput must be a string" + ) + for field_name in ("variables", "metadata"): + value = values.get(field_name) + if value is None: + continue + if not isinstance(value, Mapping): + raise EvaluationsError( + f"Inline dataset row {position} {field_name} must be a mapping" + ) + # NaN and Infinity are rejected: json.dumps would otherwise emit + # them as bare tokens, which are not valid JSON. + try: + json.dumps(value, allow_nan=False) + except (TypeError, ValueError) as error: + raise EvaluationsError( + f"Inline dataset row {position} {field_name} must be " + f"JSON-encodable without NaN or Infinity: {error}" + ) from error + variables = values.get("variables") + metadata = values.get("metadata") + normalized.append( + DatasetRow( + row_index=position, + input=row_input, + expected_output=expected_output, + variables=dict(variables) if variables else {}, + metadata=dict(metadata) if metadata is not None else None, + ) + ) + return normalized def _env(name: str) -> str | None: @@ -125,7 +234,7 @@ async def run( *, project_key: str, key: str, - dataset: str, + dataset: str | Sequence[InlineDatasetRow], handler: EvalHandler, generation: GenerationConfig | None = None, ai_config: AIConfig | None = None, @@ -145,6 +254,14 @@ async def run( generated row, and one evaluation event is emitted per ``(row, criterion)`` result. + ``dataset`` is either the key of a dataset stored in LaunchDarkly or + a sequence of inline rows. Inline rows are + :class:`DatasetRow` values or mappings in the dataset-rows wire shape + (``input``, ``expectedOutput``, ``variables``, ``metadata``, optional + ``rowIdx``); each row's index is its position in the list. They are + uploaded to the run before any generation starts, and templates in + them render exactly as a stored dataset's do. + A :class:`Judge` is an independent AI Config and may be served by a different provider or mode than ``generation``. ``handler`` runs a judge only when it provides for that judge's provider; pass handlers for any @@ -173,12 +290,12 @@ async def run( self._validate_run_args( project_key=project_key, key=key, - dataset=dataset, handler=handler, concurrency=concurrency, poll_interval_seconds=poll_interval_seconds, poll_timeout_seconds=poll_timeout_seconds, ) + inline_rows = self._validate_dataset_source(dataset) self._validate_config_source(generation=generation, ai_config=ai_config) pinned_tool_versions: dict[str, int] = {} config_label = "" @@ -238,12 +355,16 @@ async def run( resolved_judges = await self._runner._resolve_judges( project_key, ld_judges, handler, run_judge_handlers ) - 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 - ) + if isinstance(dataset, str): + dataset_ref = await asyncio.to_thread( + self._runner._fetch_dataset, project_key, dataset + ) + dataset_rows = await asyncio.to_thread( + self._runner._get_dataset_rows, project_key, dataset + ) + else: + dataset_ref = DatasetRef(id=None, key=None) + dataset_rows = [_render_inline_row(row) for row in inline_rows] evaluation = await asyncio.to_thread( self._runner._create_evaluation, project_key, @@ -258,9 +379,29 @@ async def run( evaluation.id, dataset_ref.id, ) + if not isinstance(dataset, str): + # Must finish before any event is tracked: the run starts with a + # placeholder row count of 1, so a result counted before the rows + # land would mark the run complete. + try: + await asyncio.to_thread( + self._runner._upload_dataset_rows, + project_key, + evaluation.id, + evaluation_run.id, + inline_rows, + ) + except Exception: + # The API cannot mark a run failed; cancelling is the only + # terminal state a client can set, and it keeps the run from + # sitting PENDING with a partial dataset. + await self._cancel_run_after_failed_upload( + project_key, evaluation.id, evaluation_run.id + ) + raise config = self._runner._build_handler_config(generation, resolved_tools) results = await self._runner._run_rows( - rows, + dataset_rows, handler, config, run_tools, @@ -431,7 +572,6 @@ def _validate_run_args( *, project_key: str, key: str, - dataset: str, handler: EvalHandler, concurrency: int, poll_interval_seconds: float, @@ -440,7 +580,6 @@ def _validate_run_args( for name, value in ( ("project_key", project_key), ("key", key), - ("dataset", dataset), ): if not value.strip(): raise EvaluationsError(f"{name} must not be blank") @@ -458,6 +597,45 @@ def _validate_run_args( if seconds < 0: raise EvaluationsError(f"{name} must not be negative") + async def _cancel_run_after_failed_upload( + self, project_key: str, evaluation_id: str, run_id: str + ) -> None: + try: + await asyncio.to_thread( + self._runner._cancel_evaluation_run, + project_key, + evaluation_id, + run_id, + ) + except Exception: + logger.warning( + "Failed to cancel evaluation run %s after its inline dataset " + "upload failed", + run_id, + exc_info=True, + ) + + @staticmethod + def _validate_dataset_source( + dataset: str | Sequence[InlineDatasetRow], + ) -> list[DatasetRow]: + """Validate the dataset source, returning any inline rows raw. + + A ``str`` is itself a ``Sequence``, so it is always read as a dataset + key, never as one row per character. The list is + empty for a hosted dataset; an inline one is never empty. + """ + if not isinstance(dataset, Sequence): + raise EvaluationsError( + "dataset must be a LaunchDarkly dataset key or a sequence of " + "inline rows" + ) + if isinstance(dataset, str): + if not dataset.strip(): + raise EvaluationsError("dataset must not be blank") + return [] + return _normalize_inline_rows(dataset) + @staticmethod def _validate_config_source( *, diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index 6240e980..a31e53c8 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -56,6 +56,7 @@ logger = logging.getLogger(__name__) DATASET_PAGE_SIZE = 200 +DATASET_UPLOAD_BATCH_SIZE = 500 GENERATION_EVENT_NAME = "$ld:ai:offline-evals:generation" CRITERION_EVENT_NAME = "$ld:ai:offline-evals:criterion" @@ -185,6 +186,39 @@ def _required_string(data: Mapping[str, Any], key: str, description: str) -> str return value +def render_row( + row_index: int, + *, + input_value: Any, + expected_value: Any, + variables_value: Any, + metadata_value: Any, +) -> DatasetRow: + """Render one stored dataset row for handler invocation. + + Shared by hosted and inline datasets so a row renders the same whichever + path supplied it. + """ + variables = dict(variables_value) if isinstance(variables_value, Mapping) else {} + 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 + return 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, + ) + + class ConcurrencyController: """Owns row-worker permits.""" @@ -464,38 +498,13 @@ def _get_dataset_rows(self, project_key: str, dataset_key: str) -> list[DatasetR 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 - ), + render_row( + row_index, + input_value=item.get("input"), + expected_value=item.get("expectedOutput"), + variables_value=item.get("variables"), + metadata_value=item.get("metadata"), ) ) offset += len(items) @@ -554,22 +563,75 @@ def _create_evaluation_run( self, project_key: str, evaluation_id: str, - dataset_id: str, + dataset_id: str | None, ) -> EvaluationRunRef: path = ( f"projects/{_segment(project_key)}/evaluations/" f"{_segment(evaluation_id)}/runs" ) - body: dict[str, Any] = { - "source": "api", - "datasetId": dataset_id, - } + body: dict[str, Any] = {"source": "api"} + if dataset_id is not None: + body["datasetId"] = dataset_id raw = _mapping( self._api.post(path, body=body), description="evaluation run", ) return self._run_ref(raw) + def _upload_dataset_rows( + self, + project_key: str, + evaluation_id: str, + run_id: str, + rows: list[DatasetRow], + ) -> None: + """Upload an inline dataset's raw rows to its run, in bounded batches. + + Uploads are idempotent per ``rowIdx``, so each batch is retried like a + GET. Batches go one at a time so a failure names exactly which rows + did not land. + """ + path = ( + f"projects/{_segment(project_key)}/evaluations/" + f"{_segment(evaluation_id)}/runs/{_segment(run_id)}/dataset-rows" + ) + for start in range(0, len(rows), DATASET_UPLOAD_BATCH_SIZE): + batch = rows[start : start + DATASET_UPLOAD_BATCH_SIZE] + body = { + "rows": [ + { + "rowIdx": row.row_index, + "input": row.input, + "expectedOutput": row.expected_output, + "variables": row.variables, + "metadata": row.metadata, + } + for row in batch + ] + } + try: + self._api.post(path, body=body, idempotent=True) + except EvaluationsError as error: + raise EvaluationsError( + f"Failed to upload inline dataset rows {start}-" + f"{start + len(batch) - 1} of {len(rows)} to evaluation run " + f"{run_id}: {error}" + ) from error + + def _cancel_evaluation_run( + self, + project_key: str, + evaluation_id: str, + run_id: str, + ) -> None: + """Cancel a run, retried like a GET: replaying a cancel cannot change + its outcome.""" + path = ( + f"projects/{_segment(project_key)}/evaluations/" + f"{_segment(evaluation_id)}/runs/{_segment(run_id)}/cancel" + ) + self._api.post(path, idempotent=True) + def _run_ref(self, raw: Mapping[str, Any]) -> EvaluationRunRef: return EvaluationRunRef( id=_required_string(raw, "id", "evaluation run"), @@ -711,6 +773,8 @@ def _emit_generation_events( "datasetId": dataset.id, "rowIndex": result["row_index"], } + if dataset.id is None: + del identity["datasetId"] event_id = hashlib.sha256( json.dumps(identity, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() @@ -748,12 +812,13 @@ def _emit_generation_events( "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 dataset.key is not None: + payload["datasetKey"] = dataset.key if generated["output"] is not None: payload["output"] = generated["output"] if generated["error"] is not None: @@ -1059,6 +1124,8 @@ def _emit_evaluation_events( "rowIndex": result["row_index"], "criterionType": result["criterion_type"], } + if dataset.id is None: + del identity["datasetId"] event_id = hashlib.sha256( json.dumps(identity, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/types.py b/packages/client/src/launchdarkly_ai_server/evaluations/types.py index 75af1aaa..f59beeb2 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/types.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/types.py @@ -40,10 +40,15 @@ class GenerationConfig(TypedDict, total=False): @dataclass class DatasetRef: - """Identifiers returned when resolving a dataset by key.""" + """Identifiers returned when resolving a dataset by key. - id: str - key: str + Both are ``None`` for an inline dataset, whose rows are uploaded to the run + rather than stored as a dataset. Its events must carry no dataset id: the + server drops any that name one as a mismatch. + """ + + id: str | None + key: str | None @dataclass @@ -57,6 +62,16 @@ class DatasetRow: metadata: dict[str, Any] | None = None +InlineDatasetRow = DatasetRow | Mapping[str, Any] +"""One caller-supplied row of an inline dataset. + +A mapping uses the dataset-rows wire shape: ``input``, ``expectedOutput``, +``variables``, ``metadata`` and an optional ``rowIdx``. A row's index is its +position in the list; a ``rowIdx`` or ``DatasetRow.row_index`` that disagrees +with that position is rejected. +""" + + @dataclass class ResolvedTool: """The schema and pinned version returned by the LaunchDarkly tool API.""" diff --git a/packages/client/tests/test_evaluations.py b/packages/client/tests/test_evaluations.py index e4c27226..f3c61a3f 100644 --- a/packages/client/tests/test_evaluations.py +++ b/packages/client/tests/test_evaluations.py @@ -330,6 +330,57 @@ def test_rate_limited_post_is_retried() -> None: assert len(transport.requests) == 2 +def test_idempotent_post_is_retried_after_a_server_error() -> None: + transport = RecordingTransport( + [ + HttpResponse(status=503, body='{"message": "unavailable"}'), + HttpResponse(status=200, body='{"ok": true}'), + ] + ) + client = LDApiClient( + api_token="api-token", + transport=transport, + max_retries=2, + sleep=lambda _: None, + random_value=lambda: 0.0, + ) + + assert client.post( + "runs/run-id/dataset-rows", body={"rows": []}, idempotent=True + ) == {"ok": True} + assert len(transport.requests) == 2 + + +def test_idempotent_post_is_replayed_after_a_transport_failure() -> None: + attempts: list[str] = [] + + def flaky_transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: + attempts.append(method) + if len(attempts) == 1: + raise TimeoutError("timed out") + return HttpResponse(status=200, body="") + + client = LDApiClient( + api_token="api-token", + transport=flaky_transport, + max_retries=2, + sleep=lambda _: None, + random_value=lambda: 0.0, + ) + + assert ( + client.post("runs/run-id/dataset-rows", body={"rows": []}, idempotent=True) + is None + ) + assert attempts == ["POST", "POST"] + + def test_forbidden_response_is_not_retried() -> None: transport = RecordingTransport( [HttpResponse(status=403, body='{"message": "forbidden"}')] diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index 4a4b8571..a4f282b1 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib import json from collections.abc import Callable from datetime import datetime @@ -2965,3 +2966,457 @@ def test_ai_config_variation_from_api_layers_the_model_config() -> None: unlinked = AIConfigVariation.from_api(latest) assert "provider" not in unlinked.generation assert unlinked.generation["parameters"] == {"temperature": 0.7} + + +INLINE_EVALUATION = response( + 201, {"id": "evaluation-id", "name": "inline-eval", "version": 1} +) +INLINE_RUN = response( + 201, + { + "id": "run-id", + "evaluationId": "evaluation-id", + "source": "api", + "state": "PENDING", + }, +) + + +def inline_summary(total: int) -> HttpResponse: + return response( + 200, + { + "statusCounts": { + "total": total, + "passed": total, + "failed": 0, + "error": 0, + "pending": 0, + } + }, + ) + + +async def echo_handler( + config: dict[str, Any], + user_input: str | None, + tool_handlers: dict[str, Callable[..., Any]], + variables: dict[str, Any], +) -> dict[str, Any]: + return {"output": f"generated: {user_input}"} + + +INLINE_GENERATION: Any = {"provider": "OpenAI", "model": "gpt-4o"} + + +def is_upload(request: dict[str, Any]) -> bool: + return bool( + request["method"] == "POST" and request["url"].endswith("/dataset-rows") + ) + + +@pytest.mark.asyncio +async def test_inline_dataset_uploads_rows_before_any_event_and_omits_dataset_id( + stub_sdk_client: MagicMock, +) -> None: + calls: list[str] = [] + transport = SequencedTransport( + [INLINE_EVALUATION, INLINE_RUN, response(200, {}), inline_summary(2)] + ) + + def logging_transport( + method: str, + url: str, + headers: dict[str, str], + body: bytes | None, + timeout: float, + ) -> HttpResponse: + calls.append(f"{method} {url.rsplit('/', 1)[-1]}") + return transport(method, url, headers, body, timeout) + + stub_sdk_client.track.side_effect = lambda *args: calls.append("track") + received_inputs: 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]: + received_inputs.append(user_input) + return {"output": f"generated: {user_input}"} + + result = await init_evaluations(api_token="token", transport=logging_transport).run( + project_key="proj", + key="inline-eval", + dataset=[ + { + "input": "How do I reset my password?", + "expectedOutput": "Use the link.", + }, + DatasetRow( + row_index=1, + input="Where is order {{order_id}}?", + variables={"order_id": "A-17"}, + metadata={"suite": "orders"}, + ), + ], + handler=handler, + generation=INLINE_GENERATION, + ) + + assert result.passed is True + assert not any("/datasets/" in request["url"] for request in transport.requests) + assert transport.requests[1]["url"].endswith("/evaluations/evaluation-id/runs") + assert transport.requests[1]["body"] == {"source": "api"} + upload = transport.requests[2] + assert upload["method"] == "POST" + assert upload["url"].endswith( + "/projects/proj/evaluations/evaluation-id/runs/run-id/dataset-rows" + ) + # Rows are stored raw; the server renders them as it renders a hosted dataset. + assert upload["body"] == { + "rows": [ + { + "rowIdx": 0, + "input": "How do I reset my password?", + "expectedOutput": "Use the link.", + "variables": {}, + "metadata": None, + }, + { + "rowIdx": 1, + "input": "Where is order {{order_id}}?", + "expectedOutput": None, + "variables": {"order_id": "A-17"}, + "metadata": {"suite": "orders"}, + }, + ] + } + assert sorted(received_inputs, key=str) == [ + "How do I reset my password?", + "Where is order A-17?", + ] + # Rows land before anything is counted, or the placeholder row count of 1 + # would let the first event mark the run complete. + assert calls.index("track") > calls.index("POST dataset-rows") + events = [call.args[2] for call in stub_sdk_client.track.call_args_list] + assert sorted(event["rowIndex"] for event in events) == [0, 1] + for event in events: + assert "datasetId" not in event + assert "datasetKey" not in event + + +@pytest.mark.asyncio +async def test_inline_dataset_uploads_in_batches_of_500() -> None: + transport = SequencedTransport( + [ + INLINE_EVALUATION, + INLINE_RUN, + response(200, {}), + response(200, {}), + response(200, {}), + inline_summary(1001), + ] + ) + + await init_evaluations(api_token="token", transport=transport).run( + project_key="proj", + key="inline-eval", + dataset=[{"rowIdx": index, "input": f"row {index}"} for index in range(1001)], + handler=echo_handler, + generation=INLINE_GENERATION, + ) + + uploads = [ + request["body"]["rows"] for request in transport.requests if is_upload(request) + ] + assert [len(batch) for batch in uploads] == [500, 500, 1] + assert [row["rowIdx"] for batch in uploads for row in batch] == list(range(1001)) + + +@pytest.mark.asyncio +async def test_inline_dataset_criterion_events_omit_dataset_id( + stub_sdk_client: MagicMock, +) -> None: + transport = SequencedTransport( + [INLINE_EVALUATION, INLINE_RUN, response(200, {}), inline_summary(1)] + ) + + await init_evaluations(api_token="token", transport=transport).run( + project_key="proj", + key="inline-eval", + dataset=[{"input": "hello"}], + handler=echo_handler, + generation=INLINE_GENERATION, + criteria=[Scorer(name="non-empty", fn=lambda row, output: bool(output))], + ) + + criterion_events = [ + call.args[2] + for call in stub_sdk_client.track.call_args_list + if call.args[0] == "$ld:ai:offline-evals:criterion" + ] + assert len(criterion_events) == 1 + assert criterion_events[0]["rowIndex"] == 0 + assert criterion_events[0]["criterionType"] == "non-empty" + assert "datasetId" not in criterion_events[0] + assert "datasetKey" not in criterion_events[0] + + +@pytest.mark.parametrize( + ("dataset", "message"), + [ + ([], "Inline dataset is empty"), + ([DatasetRow(row_index=3, input="hi")], "has row_index 3"), + ( + [{"input": "hi", "expected_output": "x"}], + "unknown fields: 'expected_output'", + ), + ([{"rowIdx": 1, "input": "hi"}], "has rowIdx 1"), + ([{"rowIdx": True, "input": "hi"}], "has rowIdx True"), + (["just a string"], "must be a DatasetRow or a mapping"), + ([{"input": 7}], "input must be a non-empty string"), + ([{}], "row 0 input must be a non-empty string"), + ([{"input": ""}], "input must be a non-empty string"), + ([DatasetRow(row_index=0)], "input must be a non-empty string"), + ([{"input": "hi", "expectedOutput": 7}], "expectedOutput must be a string"), + ([{"input": "hi", "variables": ["a"]}], "variables must be a mapping"), + ( + [{"input": "hi"}, {"input": "hi", "variables": {"score": float("nan")}}], + "row 1 variables must be JSON-encodable without NaN", + ), + ( + [{"input": "hi", "metadata": {"limit": float("inf")}}], + "metadata must be JSON-encodable without NaN", + ), + ( + [{"input": "hi", "variables": {"nested": [{"x": float("-inf")}]}}], + "variables must be JSON-encodable", + ), + ( + [{"input": "hi", "metadata": {"tags": {"a", "b"}}}], + "metadata must be JSON-encodable", + ), + ( + [DatasetRow(row_index=0, input="hi", variables={"when": datetime.now()})], + "variables must be JSON-encodable", + ), + ( + [{"input": "hi", "variables": {("a", "b"): 1}}], + "variables must be JSON-encodable", + ), + ], +) +@pytest.mark.asyncio +async def test_malformed_inline_rows_fail_before_any_request( + dataset: list[Any], message: str +) -> None: + evals = init_evaluations(api_token="token", transport=failing_transport) + + with pytest.raises(EvaluationsError, match=message): + await evals.run( + project_key="proj", + key="inline-eval", + dataset=dataset, + handler=echo_handler, + generation=INLINE_GENERATION, + ) + + +@pytest.mark.asyncio +async def test_inline_upload_is_retried_after_a_server_error() -> None: + transport = SequencedTransport( + [ + INLINE_EVALUATION, + INLINE_RUN, + response(503, {"message": "unavailable"}), + response(200, {}), + inline_summary(1), + ] + ) + evals = init_evaluations(api_token="token", transport=transport) + evals.api._sleep = lambda _: None + + result = await evals.run( + project_key="proj", + key="inline-eval", + dataset=[{"input": "hello"}], + handler=echo_handler, + generation=INLINE_GENERATION, + ) + + assert result.passed is True + assert sum(is_upload(request) for request in transport.requests) == 2 + + +@pytest.mark.asyncio +async def test_failed_inline_upload_stops_the_run_before_generation( + stub_sdk_client: MagicMock, +) -> None: + transport = SequencedTransport( + [ + INLINE_EVALUATION, + INLINE_RUN, + response(400, {"message": "bad rows"}), + response(204), + ] + ) + handler = AsyncMock() + + with pytest.raises( + EvaluationsError, match=r"rows 0-0 of 1 to evaluation run run-id" + ): + await init_evaluations(api_token="token", transport=transport).run( + project_key="proj", + key="inline-eval", + dataset=[{"input": "hello"}], + handler=handler, + generation=INLINE_GENERATION, + ) + + handler.assert_not_awaited() + stub_sdk_client.track.assert_not_called() + cancel = transport.requests[-1] + assert cancel["method"] == "POST" + assert cancel["url"].endswith( + "/projects/proj/evaluations/evaluation-id/runs/run-id/cancel" + ) + assert cancel["body"] is None + + +@pytest.mark.asyncio +async def test_run_is_cancelled_when_a_later_upload_batch_fails() -> None: + transport = SequencedTransport( + [ + INLINE_EVALUATION, + INLINE_RUN, + response(200, {}), + response(400, {"message": "bad rows"}), + response(204), + ] + ) + handler = AsyncMock() + + with pytest.raises( + EvaluationsError, match=r"rows 500-500 of 501 to evaluation run run-id" + ): + await init_evaluations(api_token="token", transport=transport).run( + project_key="proj", + key="inline-eval", + dataset=[{"input": f"row {index}"} for index in range(501)], + handler=handler, + generation=INLINE_GENERATION, + ) + + handler.assert_not_awaited() + assert sum(is_upload(request) for request in transport.requests) == 2 + assert transport.requests[-1]["url"].endswith("/runs/run-id/cancel") + + +@pytest.mark.asyncio +async def test_failed_cancel_does_not_mask_the_upload_error( + caplog: pytest.LogCaptureFixture, +) -> None: + transport = SequencedTransport( + [ + INLINE_EVALUATION, + INLINE_RUN, + response(400, {"message": "bad rows"}), + response(409, {"message": "already terminal"}), + ] + ) + + with pytest.raises(EvaluationsError, match="Failed to upload inline dataset"): + await init_evaluations(api_token="token", transport=transport).run( + project_key="proj", + key="inline-eval", + dataset=[{"input": "hello"}], + handler=echo_handler, + generation=INLINE_GENERATION, + ) + + assert transport.requests[-1]["url"].endswith("/runs/run-id/cancel") + assert "Failed to cancel evaluation run run-id" in caplog.text + + +@pytest.mark.asyncio +async def test_hosted_dataset_event_identity_is_unchanged( + stub_sdk_client: MagicMock, +) -> None: + transport = SequencedTransport( + [ + response(200, {"id": "dataset-id", "key": "golden"}), + response(200, dataset_page([{"rowIndex": 2, "input": "hello"}], total=1)), + INLINE_EVALUATION, + INLINE_RUN, + inline_summary(1), + ] + ) + + await init_evaluations(api_token="token", transport=transport).run( + project_key="proj", + key="inline-eval", + dataset="golden", + handler=echo_handler, + generation=INLINE_GENERATION, + ) + + assert transport.requests[3]["body"] == {"source": "api", "datasetId": "dataset-id"} + assert not any(is_upload(request) for request in transport.requests) + event = stub_sdk_client.track.call_args.args[2] + identity = { + "projectKey": "proj", + "evaluationId": "evaluation-id", + "evaluationRunId": "run-id", + "runId": "run-id", + "datasetId": "dataset-id", + "rowIndex": 2, + } + assert ( + event["eventId"] + == hashlib.sha256( + json.dumps(identity, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + ) + assert event["datasetId"] == "dataset-id" + assert event["datasetKey"] == "golden" + + +@pytest.mark.parametrize( + ("dataset", "message"), + [ + ("", "dataset must not be blank"), + (" ", "dataset must not be blank"), + (None, "dataset must be a LaunchDarkly dataset key or a sequence"), + ({"input": "hi"}, "dataset must be a LaunchDarkly dataset key or a sequence"), + (7, "dataset must be a LaunchDarkly dataset key or a sequence"), + ], +) +@pytest.mark.asyncio +async def test_invalid_dataset_source_fails_before_any_request( + dataset: Any, message: str +) -> None: + evals = init_evaluations(api_token="token", transport=failing_transport) + + with pytest.raises(EvaluationsError, match=message): + await evals.run( + project_key="proj", + key="inline-eval", + dataset=dataset, + handler=echo_handler, + generation=INLINE_GENERATION, + ) + + +@pytest.mark.asyncio +async def test_dataset_is_required() -> None: + evals = init_evaluations(api_token="token", transport=failing_transport) + + with pytest.raises(TypeError, match="dataset"): + await evals.run( # type: ignore[call-arg] + project_key="proj", + key="inline-eval", + handler=echo_handler, + generation=INLINE_GENERATION, + )