From 196d0b1e7cca46c9043fc8da13376fabbf76025e Mon Sep 17 00:00:00 2001 From: Adriana Knight Date: Thu, 1 Oct 2026 15:54:26 -0400 Subject: [PATCH 1/7] Add support for evaluation data to be defined in code --- packages/client/README.md | 19 + .../src/launchdarkly_ai_server/__init__.py | 2 + .../evaluations/__init__.py | 2 + .../launchdarkly_ai_server/evaluations/api.py | 19 +- .../evaluations/events.py | 4 +- .../evaluations/module.py | 177 ++++++++- .../evaluations/runner.py | 127 +++++-- .../evaluations/types.py | 21 +- packages/client/tests/test_evaluations.py | 51 +++ packages/client/tests/test_evaluations_run.py | 349 ++++++++++++++++++ 10 files changed, 709 insertions(+), 62 deletions(-) diff --git a/packages/client/README.md b/packages/client/README.md index b2607d4e..bb959210 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 `rows` instead of `dataset` for datasets that live in code or are built at run time rather than stored in LaunchDarkly. The two are mutually exclusive, and `run()` raises unless exactly one is given. 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", + rows=[ + {"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. A malformed row fails the run before any records are created. 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/__init__.py b/packages/client/src/launchdarkly_ai_server/__init__.py index 4001239e..c7bfb262 100644 --- a/packages/client/src/launchdarkly_ai_server/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/__init__.py @@ -33,6 +33,7 @@ EvaluationsError, EvaluationsModule, GenerationConfig, + InlineDatasetRow, Judge, RunSummary, Scorer, @@ -194,6 +195,7 @@ "EvaluationsError", "EvaluationsModule", "GenerationConfig", + "InlineDatasetRow", "Judge", "RunSummary", "Scorer", diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py index 9cd942d9..d3b9502a 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py @@ -16,6 +16,7 @@ DatasetRow, EvalRunResult, GenerationConfig, + InlineDatasetRow, RunSummary, Usage, ) @@ -30,6 +31,7 @@ "EvaluationsModule", "GenerationConfig", "HttpResponse", + "InlineDatasetRow", "Judge", "LDApiClient", "LDApiError", 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..fe5bb2e7 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -6,7 +6,7 @@ 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 +24,105 @@ 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. 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" + ) + for field_name in ("input", "expectedOutput"): + value = values.get(field_name) + if value is not None and not isinstance(value, str): + raise EvaluationsError( + f"Inline dataset row {position} {field_name} must be a string" + ) + for field_name in ("variables", "metadata"): + value = values.get(field_name) + if value is not None and not isinstance(value, Mapping): + raise EvaluationsError( + f"Inline dataset row {position} {field_name} must be a mapping" + ) + variables = values.get("variables") + metadata = values.get("metadata") + normalized.append( + DatasetRow( + row_index=position, + input=values.get("input"), + expected_output=values.get("expectedOutput"), + 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 +216,8 @@ async def run( *, project_key: str, key: str, - dataset: str, + dataset: str | None = None, + rows: Sequence[InlineDatasetRow] | None = None, handler: EvalHandler, generation: GenerationConfig | None = None, ai_config: AIConfig | None = None, @@ -145,6 +237,14 @@ async def run( generated row, and one evaluation event is emitted per ``(row, criterion)`` result. + Pass exactly one of ``dataset``, the key of a dataset stored in + LaunchDarkly, or ``rows``, an inline dataset. 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 +273,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=dataset, rows=rows) self._validate_config_source(generation=generation, ai_config=ai_config) pinned_tool_versions: dict[str, int] = {} config_label = "" @@ -238,12 +338,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 dataset is not None: + 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 +362,20 @@ async def run( evaluation.id, dataset_ref.id, ) + if dataset is None: + # 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. + await asyncio.to_thread( + self._runner._upload_dataset_rows, + project_key, + evaluation.id, + evaluation_run.id, + inline_rows, + ) config = self._runner._build_handler_config(generation, resolved_tools) results = await self._runner._run_rows( - rows, + dataset_rows, handler, config, run_tools, @@ -431,7 +546,6 @@ def _validate_run_args( *, project_key: str, key: str, - dataset: str, handler: EvalHandler, concurrency: int, poll_interval_seconds: float, @@ -440,7 +554,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 +571,44 @@ def _validate_run_args( if seconds < 0: raise EvaluationsError(f"{name} must not be negative") + @staticmethod + def _validate_dataset_source( + *, + dataset: str | None, + rows: Sequence[InlineDatasetRow] | None, + ) -> list[DatasetRow]: + """Require exactly one dataset source, returning any inline rows raw. + + Types are checked at runtime as well: a ``str`` is itself a + ``Sequence``, so a key passed as ``rows`` would otherwise be read as + one row per character. The list is empty for a hosted dataset; an + inline one is never empty. + """ + if dataset is not None and rows is not None: + raise EvaluationsError( + "dataset and rows are mutually exclusive: pass dataset for a " + "LaunchDarkly dataset key, or rows for an inline dataset" + ) + if dataset is not None: + if not isinstance(dataset, str): + raise EvaluationsError( + "dataset must be a LaunchDarkly dataset key; pass inline " + "rows with rows=" + ) + if not dataset.strip(): + raise EvaluationsError("dataset must not be blank") + return [] + if rows is None: + raise EvaluationsError( + "Pass dataset, a LaunchDarkly dataset key, or rows, an inline dataset" + ) + if isinstance(rows, str) or not isinstance(rows, Sequence): + raise EvaluationsError( + "rows must be a sequence of rows; pass a LaunchDarkly dataset " + "key with dataset=" + ) + return _normalize_inline_rows(rows) + @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..a674e76b 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,61 @@ 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 _run_ref(self, raw: Mapping[str, Any]) -> EvaluationRunRef: return EvaluationRunRef( id=_required_string(raw, "id", "evaluation run"), @@ -711,6 +759,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 +798,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 +1110,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..11efa571 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,351 @@ 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", + rows=[ + { + "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", + rows=[{"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", + rows=[{"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 string"), + ([{"variables": ["a"]}], "variables must be a mapping"), + ], +) +@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", + rows=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", + rows=[{"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"})] + ) + 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", + rows=[{"input": "hello"}], + handler=handler, + generation=INLINE_GENERATION, + ) + + handler.assert_not_awaited() + stub_sdk_client.track.assert_not_called() + + +@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( + ("sources", "message"), + [ + ({}, "Pass dataset, a LaunchDarkly dataset key, or rows"), + ({"dataset": "golden", "rows": [{"input": "hi"}]}, "mutually exclusive"), + ({"dataset": "golden", "rows": []}, "mutually exclusive"), + ({"dataset": [{"input": "hi"}]}, "pass inline rows with rows="), + ({"rows": "golden"}, "pass a LaunchDarkly dataset key with dataset="), + ({"rows": {"input": "hi"}}, "rows must be a sequence of rows"), + ], +) +@pytest.mark.asyncio +async def test_exactly_one_dataset_source_is_required_before_any_request( + sources: dict[str, 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", + handler=echo_handler, + generation=INLINE_GENERATION, + **sources, + ) From ad24861cc3fa5c55e71fc426a6d62f12677da96c Mon Sep 17 00:00:00 2001 From: Adriana Knight Date: Tue, 6 Oct 2026 10:47:52 -0400 Subject: [PATCH 2/7] consolidate rows arg into dataset --- packages/client/README.md | 4 +- .../evaluations/module.py | 50 ++++++------------- packages/client/tests/test_evaluations_run.py | 44 ++++++++++------ 3 files changed, 46 insertions(+), 52 deletions(-) diff --git a/packages/client/README.md b/packages/client/README.md index bb959210..0547fe78 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -84,13 +84,13 @@ Generation and criterion events are the only path by which row results reach Lau ### Supply the dataset inline -Pass `rows` instead of `dataset` for datasets that live in code or are built at run time rather than stored in LaunchDarkly. The two are mutually exclusive, and `run()` raises unless exactly one is given. 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. +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", - rows=[ + dataset=[ {"input": "How do I reset my password?", "expectedOutput": "Use the reset link."}, {"input": "Where is order {{order_id}}?", "variables": {"order_id": "A-17"}}, ], diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index fe5bb2e7..9550ca47 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -216,8 +216,7 @@ async def run( *, project_key: str, key: str, - dataset: str | None = None, - rows: Sequence[InlineDatasetRow] | None = None, + dataset: str | Sequence[InlineDatasetRow], handler: EvalHandler, generation: GenerationConfig | None = None, ai_config: AIConfig | None = None, @@ -237,8 +236,8 @@ async def run( generated row, and one evaluation event is emitted per ``(row, criterion)`` result. - Pass exactly one of ``dataset``, the key of a dataset stored in - LaunchDarkly, or ``rows``, an inline dataset. Inline rows are + ``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 @@ -278,7 +277,7 @@ async def run( poll_interval_seconds=poll_interval_seconds, poll_timeout_seconds=poll_timeout_seconds, ) - inline_rows = self._validate_dataset_source(dataset=dataset, rows=rows) + 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 = "" @@ -338,7 +337,7 @@ async def run( resolved_judges = await self._runner._resolve_judges( project_key, ld_judges, handler, run_judge_handlers ) - if dataset is not None: + if isinstance(dataset, str): dataset_ref = await asyncio.to_thread( self._runner._fetch_dataset, project_key, dataset ) @@ -362,7 +361,7 @@ async def run( evaluation.id, dataset_ref.id, ) - if dataset is None: + 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. @@ -573,41 +572,24 @@ def _validate_run_args( @staticmethod def _validate_dataset_source( - *, - dataset: str | None, - rows: Sequence[InlineDatasetRow] | None, + dataset: str | Sequence[InlineDatasetRow], ) -> list[DatasetRow]: - """Require exactly one dataset source, returning any inline rows raw. + """Validate the dataset source, returning any inline rows raw. - Types are checked at runtime as well: a ``str`` is itself a - ``Sequence``, so a key passed as ``rows`` would otherwise be read as - one row per character. The list is empty for a hosted dataset; an - inline one is never empty. + 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 dataset is not None and rows is not None: + if not isinstance(dataset, Sequence): raise EvaluationsError( - "dataset and rows are mutually exclusive: pass dataset for a " - "LaunchDarkly dataset key, or rows for an inline dataset" + "dataset must be a LaunchDarkly dataset key or a sequence of " + "inline rows" ) - if dataset is not None: - if not isinstance(dataset, str): - raise EvaluationsError( - "dataset must be a LaunchDarkly dataset key; pass inline " - "rows with rows=" - ) + if isinstance(dataset, str): if not dataset.strip(): raise EvaluationsError("dataset must not be blank") return [] - if rows is None: - raise EvaluationsError( - "Pass dataset, a LaunchDarkly dataset key, or rows, an inline dataset" - ) - if isinstance(rows, str) or not isinstance(rows, Sequence): - raise EvaluationsError( - "rows must be a sequence of rows; pass a LaunchDarkly dataset " - "key with dataset=" - ) - return _normalize_inline_rows(rows) + return _normalize_inline_rows(dataset) @staticmethod def _validate_config_source( diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index 11efa571..34e9e74b 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -3049,7 +3049,7 @@ async def handler( result = await init_evaluations(api_token="token", transport=logging_transport).run( project_key="proj", key="inline-eval", - rows=[ + dataset=[ { "input": "How do I reset my password?", "expectedOutput": "Use the link.", @@ -3123,7 +3123,7 @@ async def test_inline_dataset_uploads_in_batches_of_500() -> None: await init_evaluations(api_token="token", transport=transport).run( project_key="proj", key="inline-eval", - rows=[{"rowIdx": index, "input": f"row {index}"} for index in range(1001)], + dataset=[{"rowIdx": index, "input": f"row {index}"} for index in range(1001)], handler=echo_handler, generation=INLINE_GENERATION, ) @@ -3146,7 +3146,7 @@ async def test_inline_dataset_criterion_events_omit_dataset_id( await init_evaluations(api_token="token", transport=transport).run( project_key="proj", key="inline-eval", - rows=[{"input": "hello"}], + dataset=[{"input": "hello"}], handler=echo_handler, generation=INLINE_GENERATION, criteria=[Scorer(name="non-empty", fn=lambda row, output: bool(output))], @@ -3190,7 +3190,7 @@ async def test_malformed_inline_rows_fail_before_any_request( await evals.run( project_key="proj", key="inline-eval", - rows=dataset, + dataset=dataset, handler=echo_handler, generation=INLINE_GENERATION, ) @@ -3213,7 +3213,7 @@ async def test_inline_upload_is_retried_after_a_server_error() -> None: result = await evals.run( project_key="proj", key="inline-eval", - rows=[{"input": "hello"}], + dataset=[{"input": "hello"}], handler=echo_handler, generation=INLINE_GENERATION, ) @@ -3237,7 +3237,7 @@ async def test_failed_inline_upload_stops_the_run_before_generation( await init_evaluations(api_token="token", transport=transport).run( project_key="proj", key="inline-eval", - rows=[{"input": "hello"}], + dataset=[{"input": "hello"}], handler=handler, generation=INLINE_GENERATION, ) @@ -3290,27 +3290,39 @@ async def test_hosted_dataset_event_identity_is_unchanged( @pytest.mark.parametrize( - ("sources", "message"), + ("dataset", "message"), [ - ({}, "Pass dataset, a LaunchDarkly dataset key, or rows"), - ({"dataset": "golden", "rows": [{"input": "hi"}]}, "mutually exclusive"), - ({"dataset": "golden", "rows": []}, "mutually exclusive"), - ({"dataset": [{"input": "hi"}]}, "pass inline rows with rows="), - ({"rows": "golden"}, "pass a LaunchDarkly dataset key with dataset="), - ({"rows": {"input": "hi"}}, "rows must be a sequence of rows"), + ("", "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_exactly_one_dataset_source_is_required_before_any_request( - sources: dict[str, Any], message: str +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, - **sources, ) From e7e35783e5a14a3f72f76dcc505d2bb8c47511c8 Mon Sep 17 00:00:00 2001 From: Adriana Knight Date: Tue, 6 Oct 2026 11:06:36 -0400 Subject: [PATCH 3/7] Add json validation and attempt to cancel failed runs --- packages/client/README.md | 2 +- .../evaluations/module.py | 83 +++++++++++---- .../evaluations/runner.py | 14 +++ packages/client/tests/test_evaluations_run.py | 100 +++++++++++++++++- 4 files changed, 176 insertions(+), 23 deletions(-) diff --git a/packages/client/README.md b/packages/client/README.md index 0547fe78..613ee816 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -99,7 +99,7 @@ result = await evals.run( ) ``` -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. A malformed row fails the run before any records are created. Events from an inline run carry no dataset id. +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 diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 9550ca47..3f64e852 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -2,6 +2,7 @@ import asyncio import inspect +import json import logging import math import os @@ -57,9 +58,11 @@ def _render_inline_row(row: DatasetRow) -> DatasetRow: 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. The values - stay unrendered: they are uploaded as stored rows, which the server renders - the same way it renders a hosted dataset's. + 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") @@ -99,25 +102,40 @@ def _normalize_inline_rows(rows: Sequence[InlineDatasetRow]) -> list[DatasetRow] raise EvaluationsError( f"Inline dataset row {position} must be a DatasetRow or a mapping" ) - for field_name in ("input", "expectedOutput"): - value = values.get(field_name) - if value is not None and not isinstance(value, str): - raise EvaluationsError( - f"Inline dataset row {position} {field_name} must be a string" - ) + 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 not None and not isinstance(value, Mapping): + 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=values.get("input"), - expected_output=values.get("expectedOutput"), + input=row_input, + expected_output=expected_output, variables=dict(variables) if variables else {}, metadata=dict(metadata) if metadata is not None else None, ) @@ -365,13 +383,22 @@ async def run( # 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. - await asyncio.to_thread( - self._runner._upload_dataset_rows, - project_key, - evaluation.id, - evaluation_run.id, - inline_rows, - ) + 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( dataset_rows, @@ -570,6 +597,24 @@ 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], diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py index a674e76b..a31e53c8 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/runner.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/runner.py @@ -618,6 +618,20 @@ def _upload_dataset_rows( 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"), diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index 34e9e74b..a4f282b1 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -3176,8 +3176,36 @@ async def test_inline_dataset_criterion_events_omit_dataset_id( ([{"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 string"), - ([{"variables": ["a"]}], "variables must be 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 @@ -3227,7 +3255,12 @@ 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"})] + [ + INLINE_EVALUATION, + INLINE_RUN, + response(400, {"message": "bad rows"}), + response(204), + ] ) handler = AsyncMock() @@ -3244,6 +3277,67 @@ async def test_failed_inline_upload_stops_the_run_before_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 From c1e6b20d54b07a7f846625b0135720a542d63874 Mon Sep 17 00:00:00 2001 From: Adriana Knight Date: Tue, 6 Oct 2026 11:09:57 -0400 Subject: [PATCH 4/7] don't surface InlineDatasetRow in public module --- packages/client/src/launchdarkly_ai_server/__init__.py | 2 -- .../client/src/launchdarkly_ai_server/evaluations/__init__.py | 2 -- 2 files changed, 4 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/__init__.py b/packages/client/src/launchdarkly_ai_server/__init__.py index c7bfb262..4001239e 100644 --- a/packages/client/src/launchdarkly_ai_server/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/__init__.py @@ -33,7 +33,6 @@ EvaluationsError, EvaluationsModule, GenerationConfig, - InlineDatasetRow, Judge, RunSummary, Scorer, @@ -195,7 +194,6 @@ "EvaluationsError", "EvaluationsModule", "GenerationConfig", - "InlineDatasetRow", "Judge", "RunSummary", "Scorer", diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py index d3b9502a..9cd942d9 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/__init__.py @@ -16,7 +16,6 @@ DatasetRow, EvalRunResult, GenerationConfig, - InlineDatasetRow, RunSummary, Usage, ) @@ -31,7 +30,6 @@ "EvaluationsModule", "GenerationConfig", "HttpResponse", - "InlineDatasetRow", "Judge", "LDApiClient", "LDApiError", From 56df9132f015cf2fb29c867dd68b7f53904b546f Mon Sep 17 00:00:00 2001 From: Adriana Knight Date: Tue, 6 Oct 2026 16:50:21 -0400 Subject: [PATCH 5/7] ensure no coercion --- .../launchdarkly_ai_server/evaluations/module.py | 16 +++++++++++++++- packages/client/tests/test_evaluations_run.py | 16 ++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 3f64e852..98109a6c 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -55,6 +55,17 @@ def _render_inline_row(row: DatasetRow) -> DatasetRow: ) +def _check_string_keys(value: Any) -> None: + if isinstance(value, Mapping): + for key, item in value.items(): + if not isinstance(key, str): + raise TypeError(f"keys must be str, not {type(key).__name__}") + _check_string_keys(item) + elif isinstance(value, (list, tuple)): + for item in value: + _check_string_keys(item) + + def _normalize_inline_rows(rows: Sequence[InlineDatasetRow]) -> list[DatasetRow]: """Validate caller-supplied rows, returning them raw and indexed by position. @@ -121,9 +132,12 @@ def _normalize_inline_rows(rows: Sequence[InlineDatasetRow]) -> list[DatasetRow] 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. + # them as bare tokens, which are not valid JSON. Non-str keys are + # rejected rather than coerced; json.dumps runs first so circular + # references fail there instead of recursing in the key check. try: json.dumps(value, allow_nan=False) + _check_string_keys(value) except (TypeError, ValueError) as error: raise EvaluationsError( f"Inline dataset row {position} {field_name} must be " diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index a4f282b1..6a4057ae 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -3206,6 +3206,22 @@ async def test_inline_dataset_criterion_events_omit_dataset_id( [{"input": "hi", "variables": {("a", "b"): 1}}], "variables must be JSON-encodable", ), + ( + [{"input": "hi", "variables": {1: "a"}}], + "variables must be JSON-encodable.*keys must be str, not int", + ), + ( + [{"input": "hi", "metadata": {"nested": [{True: "x"}]}}], + "metadata must be JSON-encodable.*keys must be str, not bool", + ), + ( + [{"input": "hi", "variables": {"a": {None: 1}}}], + "variables must be JSON-encodable.*keys must be str, not NoneType", + ), + ( + [{"input": "hi", "metadata": {1.5: "x"}}], + "metadata must be JSON-encodable.*keys must be str, not float", + ), ], ) @pytest.mark.asyncio From 5ec5b04b02be83ddc7e8984c5568b2d671fe54d6 Mon Sep 17 00:00:00 2001 From: Adriana Knight Date: Tue, 6 Oct 2026 17:03:40 -0400 Subject: [PATCH 6/7] add extra test case --- packages/client/tests/test_evaluations_run.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index 6a4057ae..df66943b 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -3407,6 +3407,10 @@ async def test_hosted_dataset_event_identity_is_unchanged( (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"), + ( + ({"input": text} for text in ("hi", "there")), + "dataset must be a LaunchDarkly dataset key or a sequence", + ), ], ) @pytest.mark.asyncio From d956bc230fea26a7f99138e79d1ea2b231845690 Mon Sep 17 00:00:00 2001 From: Adriana Knight Date: Tue, 6 Oct 2026 17:08:28 -0400 Subject: [PATCH 7/7] stop reusing DatasetRow --- packages/client/README.md | 2 +- .../evaluations/module.py | 60 +++++++------------ .../evaluations/types.py | 6 +- packages/client/tests/test_evaluations_run.py | 21 +++---- 4 files changed, 36 insertions(+), 53 deletions(-) diff --git a/packages/client/README.md b/packages/client/README.md index 613ee816..80f77553 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -84,7 +84,7 @@ Generation and criterion events are the only path by which row results reach Lau ### 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. +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 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( diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/module.py b/packages/client/src/launchdarkly_ai_server/evaluations/module.py index 98109a6c..4c6628c1 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/module.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/module.py @@ -79,52 +79,34 @@ def _normalize_inline_rows(rows: Sequence[InlineDatasetRow]) -> list[DatasetRow] 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: + if not isinstance(row, Mapping): + raise EvaluationsError(f"Inline dataset row {position} must be a 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} must be a DatasetRow or a mapping" + f"Inline dataset row {position} has rowIdx {row_idx!r}; " + "an inline row's index is its position in the list" ) - row_input = values.get("input") + row_input = row.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") + expected_output = row.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) + value = row.get(field_name) if value is None: continue if not isinstance(value, Mapping): @@ -143,8 +125,8 @@ def _normalize_inline_rows(rows: Sequence[InlineDatasetRow]) -> list[DatasetRow] 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") + variables = row.get("variables") + metadata = row.get("metadata") normalized.append( DatasetRow( row_index=position, @@ -269,8 +251,8 @@ async def run( ``(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 + a sequence of inline rows. Inline rows are 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 diff --git a/packages/client/src/launchdarkly_ai_server/evaluations/types.py b/packages/client/src/launchdarkly_ai_server/evaluations/types.py index f59beeb2..7c1392fd 100644 --- a/packages/client/src/launchdarkly_ai_server/evaluations/types.py +++ b/packages/client/src/launchdarkly_ai_server/evaluations/types.py @@ -62,13 +62,13 @@ class DatasetRow: metadata: dict[str, Any] | None = None -InlineDatasetRow = DatasetRow | Mapping[str, Any] +InlineDatasetRow = 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. +position in the list; a ``rowIdx`` that disagrees with that position is +rejected. """ diff --git a/packages/client/tests/test_evaluations_run.py b/packages/client/tests/test_evaluations_run.py index df66943b..b37215e2 100644 --- a/packages/client/tests/test_evaluations_run.py +++ b/packages/client/tests/test_evaluations_run.py @@ -3054,12 +3054,11 @@ async def handler( "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"}, - ), + { + "input": "Where is order {{order_id}}?", + "variables": {"order_id": "A-17"}, + "metadata": {"suite": "orders"}, + }, ], handler=handler, generation=INLINE_GENERATION, @@ -3168,18 +3167,20 @@ async def test_inline_dataset_criterion_events_omit_dataset_id( ("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"), + (["just a string"], "row 0 must be a mapping"), + ( + [DatasetRow(row_index=0, input="hi")], + "row 0 must be 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"), ( @@ -3199,7 +3200,7 @@ async def test_inline_dataset_criterion_events_omit_dataset_id( "metadata must be JSON-encodable", ), ( - [DatasetRow(row_index=0, input="hi", variables={"when": datetime.now()})], + [{"input": "hi", "variables": {"when": datetime.now()}}], "variables must be JSON-encodable", ), (