[AIC-3363] Add support for evaluation data to be defined in code

packages/client/README.md

@@ -81,6 +81,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 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.

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_key,
             "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,8 +195,8 @@ 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)
 
 
 def segment(value: str) -> str:

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

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
@@ -19,15 +20,123 @@
     urllib_transport,
 )
 from .criteria import Criterion, Judge
-from .runner import EvalHandler, EvaluationsRunner, _provides_for
+from .runner import (
+    EvalHandler,
+    EvaluationsRunner,
+    _provides_for,
+    render_row,
+)
 from .tools import EvalTool, ToolsClient, tool_handlers, validate_tools
-from .types import AIConfig, EvalRunResult, GenerationConfig, RunSummary
+from .types import (
+    AIConfig,
+    DatasetRef,
+    DatasetRow,
+    EvalRunResult,
+    GenerationConfig,
+    InlineDatasetRow,
+    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 _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.
+
+    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 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} has rowIdx {row_idx!r}; "
+                "an inline row's index is its position in the list"
+            )
+        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 = 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 = row.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. 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 "
+                    f"JSON-encodable without NaN or Infinity: {error}"
+                ) from error
+        variables = row.get("variables")
+        metadata = row.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:
@@ -133,7 +242,7 @@ async def run(
         self,
         *,
         key: str,
-        dataset: str,
+        dataset: str | Sequence[InlineDatasetRow],
         handler: EvalHandler,
         generation: GenerationConfig | None = None,
         ai_config: AIConfig | None = None,
@@ -160,6 +269,14 @@ async def run(
         no tool from the API, and it checks the list before any network I/O.
         Handlers receive a ``{key: executable}`` map either way.
 
+        ``dataset`` is either the key of a dataset stored in LaunchDarkly or
+        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
+        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
@@ -187,12 +304,12 @@ async def run(
             poll_timeout_seconds = SUMMARY_POLL_TIMEOUT_SECONDS
         self._validate_run_args(
             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)
         run_tools = list(tools or [])
         validate_tools(run_tools, self._project_key)
@@ -264,12 +381,16 @@ async def run(
         resolved_judges = await self._runner._resolve_judges(
             self._project_key, ld_judges, handler, run_judge_handlers
         )
-        dataset_ref = await asyncio.to_thread(
-            self._runner._fetch_dataset, self._project_key, dataset
-        )
-        rows = await asyncio.to_thread(
-            self._runner._get_dataset_rows, self._project_key, dataset
-        )
+        if isinstance(dataset, str):
+            dataset_ref = await asyncio.to_thread(
+                self._runner._fetch_dataset, self._project_key, dataset
+            )
+            dataset_rows = await asyncio.to_thread(
+                self._runner._get_dataset_rows, self._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,
             self._project_key,
@@ -284,9 +405,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,
+                    self._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(
+                    self._project_key, evaluation.id, evaluation_run.id
+                )
+                raise
         config = self._runner._build_handler_config(generation, run_tools)
         results = await self._runner._run_rows(
-            rows,
+            dataset_rows,
             handler,
             config,
             run_tool_handlers,
@@ -454,16 +595,12 @@ def _validate_judge_handlers(judge_handlers: list[EvalHandler]) -> None:
     def _validate_run_args(
         *,
         key: str,
-        dataset: str,
         handler: EvalHandler,
         concurrency: int,
         poll_interval_seconds: float,
         poll_timeout_seconds: float,
     ) -> None:
-        for name, value in (
-            ("key", key),
-            ("dataset", dataset),
-        ):
+        for name, value in (("key", key),):
             if not value.strip():
                 raise EvaluationsError(f"{name} must not be blank")
         if not callable(handler):
@@ -480,6 +617,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(
         *,