feat: record judge scores as gen_ai.evaluation.result

TELEMETRY-CONTRACT.md

@@ -95,6 +95,8 @@ able to tell from the trace which path ran.
 | `launchdarkly.run.id` | `TrackData.runId` | `set_ld_span_attributes` |
 | `launchdarkly.graph.key` | `TrackData.graphKey`, only when present | `set_ld_span_attributes` |
 | `launchdarkly.stream.abandoned` | `True`, only when abandoned | `end_span_once` |
+| `gen_ai.evaluation.name` | judge config key, judge roots only | `with_judge_evaluation`, see section 4a |
+| `gen_ai.evaluation.score.value` | numeric score, judge roots only | `with_judge_evaluation`, see section 4a |
 
 The root also carries one span event, `feature_flag`, with these event attributes:
 
@@ -212,6 +214,29 @@ when `conversation_id(...)` is bound.
 
 ---
 
+## 4a. Judge evaluation events
+
+A judge run is itself a tracked AI call (`invoke_agent` + `chat`). After the score is parsed, the
+SDK writes a `gen_ai.evaluation.result` span event on that `invoke_agent` span:
+
+| Event attribute | Value |
+|---|---|
+| `gen_ai.evaluation.name` | judge config key |
+| `gen_ai.evaluation.score.value` | numeric score, only when the judge returned a finite number |
+
+The same keys are mirrored as span attributes, so section 2 lists them too.
+`gen_ai.evaluation.score.label` is not invented.
+The existing `track(evaluationMetricKey)` call is unchanged and still feeds AI Config Monitoring —
+a judge that returns a non-numeric score emits no evaluation event but still tracks the metric.
+
+`gen_ai.evaluation.explanation` is deliberately **not** emitted. The judge's reasoning is
+model-generated prose about the user's conversation — content, under section 7 — and content
+attributes require `captureContent` / `capture_content`, a handler-factory option this layer does
+not receive. The reasoning is still returned to the caller in `judgeResults` / `judge_results`;
+only the telemetry copy is withheld. Exporting it needs its own opt-in.
+
+---
+
 ## 5. Finish reasons
 
 One vocabulary across all six handlers: `stop`, `length`, `content_filter`, `tool_calls`, `error`.

examples/conversation.py

@@ -0,0 +1,87 @@
+"""
+Example: a multi-turn conversation grouped under one ``gen_ai.conversation.id``, with inline
+judge evaluation on every turn.
+
+This is the end-to-end check for O11Y-1888. Run it, then open the printed conversation id in
+LaunchDarkly's Conversations view and confirm:
+
+  1. One conversation, three turns — not three conversations. Every span of every turn carries
+     the same id: root, ``chat``, ``execute_tool``, and the judge's own ``invoke_agent``.
+  2. Each turn shows a score badge, sourced from the ``gen_ai.evaluation.result`` span event on
+     the judge span.
+  3. No judge reasoning anywhere in the telemetry. The score and the judge's config key are
+     exported; the explanation is not, because it is model prose about the user's conversation
+     and content attributes require ``capture_content``. The reasoning IS printed below, straight
+     from ``judge_results`` — that is the caller's copy, and it is unaffected.
+
+The flag key must point at an AI Config with a ``judge_configuration``, otherwise there are no
+judge turns to look at.
+
+Usage (via main.py):
+    python main.py conversation <flag-key> "<opening message>"
+"""
+
+from __future__ import annotations
+
+import sys
+from typing import Any
+
+import examples.register  # noqa: F401 – side-effect: populate global_registry
+from examples.utils import new_context, new_conversation_id
+from launchdarkly_ai_server import config, conversation_id, global_registry
+
+FOLLOW_UPS = [
+    "Can you give me a concrete example of that?",
+    "What is the most common mistake teams make with it?",
+]
+
+
+async def run(key: str, user_input: str) -> None:
+    conversation = new_conversation_id("conversation-example")
+    ctx = new_context()
+    history: list[dict[str, Any]] = []
+
+    print(f"[conversation] {conversation}", file=sys.stderr)
+
+    turns = [user_input or "What is a feature flag?", *FOLLOW_UPS]
+
+    for index, prompt in enumerate(turns, start=1):
+        # One binding per turn, same id every time — that is what makes them one conversation
+        # rather than three. Re-binding per turn is the realistic shape: each turn is usually a
+        # separate inbound request that looks the id up from its own thread/session.
+        with conversation_id(conversation):
+            response = await config(
+                key=key,
+                registry=global_registry,
+            ).invoke(prompt, ctx, None, history)
+
+        text = (
+            response.response
+            if isinstance(response.response, str)
+            else str(response.response)
+        )
+        print(f"\n─── turn {index} ───\n> {prompt}\n{text}")
+
+        judge_results = response.judge_results or {}
+        for judge_key, result in judge_results.items():
+            # `response` here is the judge's reasoning. It reaches the caller and is deliberately
+            # absent from the span — see the module docstring.
+            score = getattr(result, "score", None)
+            reasoning = getattr(result, "response", None)
+            print(
+                f"[judge] {judge_key} score={score} reasoning={reasoning}",
+                file=sys.stderr,
+            )
+        if not judge_results:
+            print(
+                "[judge] no judges ran — does this AI Config have a judge_configuration?",
+                file=sys.stderr,
+            )
+
+        history.append({"role": "user", "content": prompt})
+        history.append({"role": "assistant", "content": text})
+
+    print(
+        f"\n[conversation] done — open {conversation} in the Conversations view",
+        file=sys.stderr,
+    )

main.py

@@ -43,6 +43,7 @@
     "agent": "examples.agent",
     "streaming": "examples.streaming",
     "graph": "examples.graph_example",
+    "conversation": "examples.conversation",
     "history": "examples.history",
     "judge": "examples.judge_example",
     "claude-agents": "examples.claude_agents_example",

packages/claude-agents/src/launchdarkly_ai_claude_agents/handler.py

@@ -617,7 +617,12 @@ def _stream_impl(
             capture_content=capture_content,
         )
 
-    return create_handler(("Anthropic", "agent"), _call_impl, _stream_impl)  # type: ignore[arg-type]
+    return create_handler(
+        ("Anthropic", "agent"),
+        _call_impl,  # type: ignore[arg-type]
+        _stream_impl,  # type: ignore[arg-type]
+        capture_content=capture_content,
+    )
 
 
 async def _stream_gen(

packages/claude-messages/src/launchdarkly_ai_claude_messages/handler.py

@@ -383,7 +383,12 @@ def _stream_impl(
             capture_content=capture_content,
         )
 
-    return create_handler(("Anthropic", "messages"), _call_impl, _stream_impl)  # type: ignore[arg-type]
+    return create_handler(
+        ("Anthropic", "messages"),
+        _call_impl,  # type: ignore[arg-type]
+        _stream_impl,  # type: ignore[arg-type]
+        capture_content=capture_content,
+    )
 
 
 async def _stream_gen(

packages/client/agents.md

@@ -120,7 +120,7 @@ Handlers may return any of these — the client normalizes them before emitting
       - Calls `handler(config, user_input, tool_handlers, variables)`
       - On success: emits `$ld:ai:generation:success` + token tracks
       - On error: emits `$ld:ai:generation:error` then re-raises
-3. If `judge_configuration.judges` is present, runs each judge handler (sampled by `sampling_rate`) against the primary response and tracks `evaluation_metric_key`.
+3. If `judge_configuration.judges` is present, runs each judge handler (sampled by `sampling_rate`) against the primary response, tracks `evaluation_metric_key`, and emits a `gen_ai.evaluation.result` span event on the judge's `invoke_agent` span (`gen_ai.evaluation.name` / `.score.value` / `.explanation`).
 4. Returns `ProviderResponse`: `{ response: str, usage: UsageDict, track_data: TrackData, judge_results?: dict[str, JudgeResult], judge_tasks?: list[JudgeTask] }`. `judge_results` is populated when `skip_judges=False` (default) and judges ran; `judge_tasks` is populated when `skip_judges=True`.
 
 ---

packages/client/src/launchdarkly_ai_server/conversation.py

@@ -1,13 +1,15 @@
-"""Caller-supplied ``gen_ai.conversation.id``.
+"""Caller-supplied ``gen_ai.conversation.id`` and judge evaluation span events.
 
 A dedicated OTel context key, not W3C baggage: the id must not leak onto outbound
 provider HTTP calls. A multi-tenant process binds a different id per request.
 """
 
 from __future__ import annotations
 
-from collections.abc import AsyncGenerator, Iterator
-from contextlib import aclosing, contextmanager
+from collections.abc import AsyncGenerator, AsyncIterator, Callable, Iterator
+from contextlib import aclosing, asynccontextmanager, contextmanager
+from dataclasses import dataclass
+from time import time_ns
 from typing import TYPE_CHECKING, Any
 
 from opentelemetry import context as otel_context
@@ -25,6 +27,16 @@
 GEN_AI_CONVERSATION_ID = "gen_ai.conversation.id"
 
 _CONV_KEY = otel_context.create_key("launchdarkly.gen_ai.conversation.id")
+_EVAL_KEY = otel_context.create_key("launchdarkly.judge.evaluation")
+
+
+@dataclass
+class _JudgeEvalCapture:
+    name: str
+    released: bool = False
+    span: Any = None
+    pending_end: Callable[[], None] | None = None
+
 
 # Every tracer this SDK creates is named "@launchdarkly/ai-<package>". The processor is registered
 # on the *global* provider, so without this gate it stamps a caller-supplied id onto every span in
@@ -67,6 +79,59 @@ def _conversation_id_from(ctx: otel_context.Context | None) -> str | None:
     return value if isinstance(value, str) and value else None
 
 
+def _record_evaluation(
+    span: Any, name: str, score: float, explanation: str | None = None
+) -> None:
+    """Write the judge score as a ``gen_ai.evaluation.result`` event plus mirrored attributes.
+
+    ``explanation`` is passed only when the judge's own handler captures content — it is model
+    prose about the user's conversation, so it follows the same gate as every other content
+    attribute. The reasoning always reaches the caller in ``judge_results``.
+    """
+    if span is None or not span.is_recording():
+        return
+    attrs: dict[str, Any] = {
+        "gen_ai.evaluation.name": name,
+        "gen_ai.evaluation.score.value": score,
+    }
+    if explanation:
+        attrs["gen_ai.evaluation.explanation"] = explanation
+    span.add_event("gen_ai.evaluation.result", attrs)
+    for key, value in attrs.items():
+        span.set_attribute(key, value)
+
+
+def _delay_invoke_agent_end(span: Any, capture: _JudgeEvalCapture) -> None:
+    original_end = span.end
+    ended = False
+
+    def wrapped_end(*args: Any, **kwargs: Any) -> None:
+        nonlocal ended
+        if ended:
+            return
+        if capture.released:
+            ended = True
+            original_end(*args, **kwargs)
+            return
+        capture.span = span
+        # Freeze the end time at the handler's call. Replaying a no-arg end() later would let the
+        # SDK stamp time_ns() at release, inflating the judge span by the tracking and parsing
+        # work that runs between the handler ending the span and the score being recorded.
+        if not args and "end_time" not in kwargs:
+            kwargs = {**kwargs, "end_time": time_ns()}
+
+        def pending() -> None:
+            nonlocal ended
+            if ended:
+                return
+            ended = True
+            original_end(*args, **kwargs)
+
+        capture.pending_end = pending
+
+    span.end = wrapped_end
+
+
 @contextmanager
 def conversation_id(conversation: str | None) -> Iterator[None]:
     """Bind a caller-supplied conversation id for the duration of the ``with`` block.
@@ -88,6 +153,9 @@ def conversation_id(conversation: str | None) -> Iterator[None]:
         otel_context.detach(token)
 
 
+RecordEvaluation = Callable[..., None]
+
+
 async def _stream_with_bound_id(
     generator: AsyncGenerator[Any, None], conversation: str
 ) -> AsyncGenerator[Any, None]:
@@ -130,8 +198,35 @@ def bind_conversation_id(
     return _stream_with_bound_id(generator, conversation)
 
 
+@asynccontextmanager
+async def with_judge_evaluation(name: str) -> AsyncIterator[RecordEvaluation]:
+    """Hold the judge ``invoke_agent`` span open until ``record`` runs.
+
+    ``execute_and_track`` returns after the handler has already called ``span.end()``,
+    so without this delay the evaluation event would be dropped.
+    """
+    capture = _JudgeEvalCapture(name=name)
+
+    def record(score: float, explanation: str | None = None) -> None:
+        if capture.span is not None:
+            _record_evaluation(capture.span, capture.name, score, explanation)
+
+    token = otel_context.attach(otel_context.set_value(_EVAL_KEY, capture))
+    try:
+        yield record
+    finally:
+        capture.released = True
+        try:
+            if capture.pending_end is not None:
+                capture.pending_end()
+        finally:
+            # Detach even if ending the span raises (a user span processor's on_end can throw);
+            # otherwise the judge capture stays bound in this task's context for good.
+            otel_context.detach(token)
+
+
 class ConversationIdSpanProcessor(_SpanProcessorBase):
-    """Stamps ``gen_ai.conversation.id`` write-if-absent on every span.
+    """Stamps ``gen_ai.conversation.id`` write-if-absent; 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.
@@ -150,6 +245,15 @@ def on_start(
         if conv and _is_launchdarkly_span(span):
             set_conversation_id_if_absent(span, conv)
 
+        capture = otel_context.get_value(_EVAL_KEY, ctx)
+        if capture is None:
+            capture = otel_context.get_value(_EVAL_KEY)
+        if (
+            isinstance(capture, _JudgeEvalCapture)
+            and getattr(span, "name", None) == "invoke_agent"
+        ):
+            _delay_invoke_agent_end(span, capture)
+
     def on_end(self, span: Any) -> None:
         return None