feat(evaluations): export each offline row as one trace

packages/client/README.md

@@ -52,6 +52,8 @@ No code changes are required — `init_client()` detects the packages at runtime
 
 The evaluations harness reads an LD-hosted dataset, creates a new evaluation and API-source run, and invokes your handler once per row. Rows can then be scored by LaunchDarkly judges and local scorer functions; see [Score rows with judges and scorers](#score-rows-with-judges-and-scorers). Each success or error queues a `$ld:ai:offline-evals:generation` custom event containing the evaluation, run, dataset, and row identifiers plus output or error (`errorMessage` is included for `ERROR` rows), nested `usage.inputTokens`/`usage.outputTokens`, timing, and stable hashes. Dataset-owned input, expected output, metadata, and variables are not duplicated in the event. Each queued event is logged at `INFO` on the `launchdarkly_ai_server.evaluations.runner` logger with its RFC3339 UTC `emittedAt` timestamp and stable `eventId`, making it possible to compare SDK emission time with ClickHouse arrival time once that logger is enabled. The same `emittedAt` value is included in the event payload. Events are flushed before the summary is fetched and the call returns; handlers are never rerun to retry event delivery. Pass/fail is derived from LaunchDarkly's run summary.
 
+Each row is also exported as one trace on the tracer provider `init_client` registered: a `launchdarkly.dataset_row` root with the generation, its tool calls, and each criterion's `evaluate` span beneath it, all sharing the row's `gen_ai.conversation.id`. A provider handler traces its own call; for a plain function the runner writes the `chat` and `execute_tool` spans. Prompts, output, tool payloads, and judge explanations follow the same [`capture_content`](#process-wide-defaults) gate as every other span, so they are omitted unless capture is on.
+
 Management API requests go to the `api_base_uri` option, then `LD_API_BASE_URI`, then `https://app.launchdarkly.com`; `LD_BASE_URI` is never used, because it may point at a relay proxy. Result links use `ui_base_uri`, then `LD_UI_BASE_URI`, then `https://app.launchdarkly.com`; this is independent of `LD_API_BASE_URI`. At either level a blank value counts as unset, and a trailing `/` is stripped. After flushing generation events, the harness polls the run summary endpoint until passed + failed + error rows fully account for a nonzero total with no pending rows, polling every `poll_interval_seconds` (default 2s) up to `poll_timeout_seconds` (default 180s); pass either to `run()` to widen both for large datasets. The summary endpoint does not return run state, so `RunSummary` exposes row counts only. A run passes only when the completed summary has no failed, error, or pending rows. `failed_rows` counts rows whose criteria were scored and did not meet their threshold, so a gate that ignored it would exit 0 on a run where every row failed its judge. Evaluation keys must be unique because every call creates a new evaluation with `POST`.
 
 ```python

packages/client/src/launchdarkly_ai_server/conversation.py

@@ -6,7 +6,13 @@
 
 from __future__ import annotations
 
-from collections.abc import AsyncGenerator, AsyncIterator, Callable, Iterator
+from collections.abc import (
+    AsyncGenerator,
+    AsyncIterator,
+    Callable,
+    Iterator,
+    Mapping,
+)
 from contextlib import aclosing, asynccontextmanager, contextmanager
 from dataclasses import dataclass
 from time import time_ns
@@ -28,6 +34,8 @@
 
 _CONV_KEY = otel_context.create_key("launchdarkly.gen_ai.conversation.id")
 _EVAL_KEY = otel_context.create_key("launchdarkly.judge.evaluation")
+_STAMP_KEY = otel_context.create_key("launchdarkly.span.stamped_attributes")
+_COUNTER_KEY = otel_context.create_key("launchdarkly.span.counter")
 
 
 @dataclass
@@ -82,6 +90,53 @@ def _conversation_id_from(ctx: otel_context.Context | None) -> str | None:
     return value if isinstance(value, str) and value else None
 
 
+def _stamped_attributes_from(ctx: otel_context.Context | None) -> Mapping[str, Any]:
+    value = otel_context.get_value(_STAMP_KEY, ctx) if ctx is not None else None
+    if value is None:
+        value = otel_context.get_value(_STAMP_KEY)
+    return value if isinstance(value, Mapping) else {}
+
+
+@contextmanager
+def stamp_span_attributes(attributes: Mapping[str, str | int]) -> Iterator[None]:
+    """Write ``attributes`` onto every SDK span started in this block.
+
+    Each key is written only when the span does not already carry it, and only
+    onto spans from this SDK's tracers. Values must be strings or integers;
+    anything else is skipped. A nested block adds to the outer set, and its own
+    values win on a shared key. Relies on :class:`ConversationIdSpanProcessor`,
+    which ``init_client`` registers.
+    """
+    merged = {**_stamped_attributes_from(otel_context.get_current()), **attributes}
+    token = otel_context.attach(otel_context.set_value(_STAMP_KEY, merged))
+    try:
+        yield
+    finally:
+        otel_context.detach(token)
+
+
+@dataclass
+class SpanCounter:
+    """Number of SDK spans started while :func:`count_sdk_spans` is bound."""
+
+    count: int = 0
+
+
+@contextmanager
+def count_sdk_spans() -> Iterator[SpanCounter]:
+    """Count the spans this SDK's tracers start inside this block.
+
+    Only the innermost block counts a span. Relies on
+    :class:`ConversationIdSpanProcessor`, which ``init_client`` registers.
+    """
+    counter = SpanCounter()
+    token = otel_context.attach(otel_context.set_value(_COUNTER_KEY, counter))
+    try:
+        yield counter
+    finally:
+        otel_context.detach(token)
+
+
 def _record_evaluation(
     span: Any,
     name: str,
@@ -275,7 +330,7 @@ def record(
 
 
 class ConversationIdSpanProcessor(_SpanProcessorBase):
-    """Stamps ``gen_ai.conversation.id`` write-if-absent; delays judge ``invoke_agent`` end.
+    """Stamps the conversation id and bound attributes, counts SDK spans, delays judge ``invoke_agent`` end.
 
     Structurally a ``SpanProcessor``; the base is only real to a type checker so that an
     api-only install (no ``[otel]`` extra) still imports this module.
@@ -294,6 +349,20 @@ def on_start(
         if conv and _is_launchdarkly_span(span):
             set_conversation_id_if_absent(span, conv)
 
+        counter = otel_context.get_value(_COUNTER_KEY, ctx)
+        if isinstance(counter, SpanCounter) and _is_launchdarkly_span(span):
+            counter.count += 1
+
+        stamped = _stamped_attributes_from(ctx)
+        if stamped and _is_launchdarkly_span(span):
+            for key, value in stamped.items():
+                if not isinstance(key, str) or _read_attribute(span, key) is not None:
+                    continue
+                if isinstance(value, str):
+                    span.set_attribute(key, value)
+                elif isinstance(value, int) and not isinstance(value, bool):
+                    span.set_attribute(key, value)
+
         capture = otel_context.get_value(_EVAL_KEY, ctx)
         if capture is None:
             capture = otel_context.get_value(_EVAL_KEY)

packages/client/src/launchdarkly_ai_server/evaluations/module.py

@@ -10,7 +10,7 @@
 from collections.abc import Mapping, Sequence
 from typing import Any, Literal, cast, overload
 
-from ..lifecycle import get_client, init_client
+from ..lifecycle import _flush_tracer_provider, get_client, init_client
 from ..utils import normalize_mode
 from .api import (
     DEFAULT_BASE_URI,
@@ -482,14 +482,17 @@ async def run(
                 )
                 raise
         config = self._runner._build_handler_config(run_generation, run_tools)
-        results = await self._runner._run_rows(
-            dataset_rows,
-            generation_handler,
-            config,
-            run_tool_handlers,
-            concurrency,
-        )
+        row_spans: dict[int, Any] = {}
         try:
+            results = await self._runner._run_rows(
+                dataset_rows,
+                generation_handler,
+                config,
+                run_tool_handlers,
+                concurrency,
+                evaluation_run_id=evaluation_run.id,
+                row_spans=row_spans,
+            )
             self._runner._emit_generation_events(
                 client,
                 project_key=self._project_key,
@@ -504,6 +507,8 @@ async def run(
                     run_criteria,
                     resolved_judges,
                     concurrency,
+                    evaluation_run_id=evaluation_run.id,
+                    row_spans=row_spans,
                 )
                 self._runner._emit_evaluation_events(
                     client,
@@ -515,10 +520,19 @@ async def run(
                 )
         finally:
             # Generation results already queued on the SDK event buffer must
-            # reach LaunchDarkly even when the criteria phase fails.
-            flush_result = client.flush()
-            if inspect.isawaitable(flush_result):
-                await flush_result
+            # reach LaunchDarkly even when the criteria phase fails. Spans
+            # follow on the provider init_client already registered. A span
+            # that fails to end must not skip the event flush.
+            try:
+                for row_span in row_spans.values():
+                    row_span.end()
+            finally:
+                try:
+                    flush_result = client.flush()
+                    if inspect.isawaitable(flush_result):
+                        await flush_result
+                finally:
+                    _flush_tracer_provider()
         summary = await self._poll_summary_until_terminal(
             evaluation.id,
             evaluation_run.id,