From 1cb19811007b70b3667796bde25c5b7afac691be Mon Sep 17 00:00:00 2001 From: Alexis Georges Date: Thu, 29 Oct 2026 23:12:07 -0400 Subject: [PATCH 1/4] feat(client): point judge spans at the run they graded A judge runs after the graded handler has ended its span, so each judge starts a trace of its own with its own launchdarkly.run.id, and nothing on that trace points to the run it graded. Only the judge's metric event carried the graded run id. run_judges (inline) and run_judge (deferred) now wrap each judge run in stamp_span_attributes, the same write-if-absent mechanism offline evaluations use for launchdarkly.evaluation_run.id. Every SDK span started during the judge run carries: - launchdarkly.judge.target.run.id - launchdarkly.judge.target.config.key - launchdarkly.judge.target.variation.key Inline judges read base_track_data, deferred judges read the task's parent_track_data. Each key is written only when its value is a non-empty string. Matches the JS turnkey SDK. --- .../src/launchdarkly_ai_server/judges.py | 327 ++++++++++-------- 1 file changed, 181 insertions(+), 146 deletions(-) diff --git a/packages/client/src/launchdarkly_ai_server/judges.py b/packages/client/src/launchdarkly_ai_server/judges.py index cf9ec3b8..36bf39a9 100644 --- a/packages/client/src/launchdarkly_ai_server/judges.py +++ b/packages/client/src/launchdarkly_ai_server/judges.py @@ -2,10 +2,11 @@ import logging import random -from collections.abc import Callable, Mapping +from collections.abc import Callable, Iterator, Mapping +from contextlib import contextmanager from typing import Any -from .conversation import with_judge_evaluation +from .conversation import stamp_span_attributes, with_judge_evaluation from .judge_scoring import ( build_message_history, numeric_score, @@ -82,6 +83,35 @@ def _handler_for( logger = logging.getLogger(__name__) +#: The run id of the config a judge graded, on every span of that judge's run. +LD_JUDGE_TARGET_RUN_ID = "launchdarkly.judge.target.run.id" +#: The key of the config a judge graded. +LD_JUDGE_TARGET_CONFIG_KEY = "launchdarkly.judge.target.config.key" +#: The variation key of the config a judge graded. +LD_JUDGE_TARGET_VARIATION_KEY = "launchdarkly.judge.target.variation.key" + + +@contextmanager +def _judge_target(target: TrackData | None) -> Iterator[None]: + """Point a judge's spans at the run it graded. + + A judge runs after the graded handler has ended its span, so its spans start + a trace of their own. Without these, the only tie back to the graded run is + the judge's metric event. Each key is written only when its value is a + non-empty string. + """ + attributes: dict[str, str | int] = {} + for attribute, field in ( + (LD_JUDGE_TARGET_RUN_ID, "runId"), + (LD_JUDGE_TARGET_CONFIG_KEY, "configKey"), + (LD_JUDGE_TARGET_VARIATION_KEY, "variationKey"), + ): + value = (target or {}).get(field) + if isinstance(value, str) and value: + attributes[attribute] = value + with stamp_span_attributes(attributes): + yield + async def run_judges( *, @@ -187,78 +217,79 @@ async def run_judges( output=llm_response, ) - async with with_judge_evaluation(judge_key) as record_evaluation: - result = await execute_and_track( - config_key=judge_key, - config=effective_judge_config, - meta=judge_meta, - user_context=user_context, - handler=judge_handler, - user_input=llm_response, - tool_handlers=None, - graph_key=graph_key, - variables={ - "message_history": message_history, - "response_to_evaluate": llm_response, - **({"input": user_input} if user_input else {}), - }, - evaluated_value=variation.get("value"), - ) + with _judge_target(base_track_data): + async with with_judge_evaluation(judge_key) as record_evaluation: + result = await execute_and_track( + config_key=judge_key, + config=effective_judge_config, + meta=judge_meta, + user_context=user_context, + handler=judge_handler, + user_input=llm_response, + tool_handlers=None, + graph_key=graph_key, + variables={ + "message_history": message_history, + "response_to_evaluate": llm_response, + **({"input": user_input} if user_input else {}), + }, + evaluated_value=variation.get("value"), + ) - usage = to_usage_dict(result["usage"]) - entries = typesafe_judge_entries(result["response"]) - if entries is not None: - # One Jev call. Every label is returned with that call's - # full usage, and each label is tracked under its eventKey. - from .lifecycle import get_client - - client = get_client() - for entry in entries: - result_key = f"{judge_key}.{entry['key']}" - score = entry["score"] - judge_results[result_key] = JudgeResult( - usage=usage, - response=entry["reason"], - score=score, - event_key=entry["eventKey"], + usage = to_usage_dict(result["usage"]) + entries = typesafe_judge_entries(result["response"]) + if entries is not None: + # One Jev call. Every label is returned with that call's + # full usage, and each label is tracked under its eventKey. + from .lifecycle import get_client + + client = get_client() + for entry in entries: + result_key = f"{judge_key}.{entry['key']}" + score = entry["score"] + judge_results[result_key] = JudgeResult( + usage=usage, + response=entry["reason"], + score=score, + event_key=entry["eventKey"], + ) + record_evaluation(score, None, result_key) + client.track( + entry["eventKey"], + to_ld_context(client, user_context), + {**base_track_data, "judgeConfigKey": result_key}, + score, + ) + continue + + score, reasoning = parse_judge_response(result["response"]) + judge_results[judge_key] = JudgeResult( + usage=usage, + response=reasoning, + score=score, + ) + metric_score = numeric_score(score) + if metric_score is not None: + record_evaluation( + metric_score, + reasoning if judge_handler.capture_content else None, ) - record_evaluation(score, None, result_key) + + evaluation_metric_key = ( + judge_ai_config.get("evaluationMetricKey") + if isinstance(judge_ai_config, dict) + else None + ) + if evaluation_metric_key and score is not None: + from .lifecycle import get_client + + client = get_client() client.track( - entry["eventKey"], + evaluation_metric_key, to_ld_context(client, user_context), - {**base_track_data, "judgeConfigKey": result_key}, + {**base_track_data, "judgeConfigKey": judge_key}, score, ) - continue - - score, reasoning = parse_judge_response(result["response"]) - judge_results[judge_key] = JudgeResult( - usage=usage, - response=reasoning, - score=score, - ) - metric_score = numeric_score(score) - if metric_score is not None: - record_evaluation( - metric_score, - reasoning if judge_handler.capture_content else None, - ) - - evaluation_metric_key = ( - judge_ai_config.get("evaluationMetricKey") - if isinstance(judge_ai_config, dict) - else None - ) - if evaluation_metric_key and score is not None: - from .lifecycle import get_client - - client = get_client() - client.track( - evaluation_metric_key, - to_ld_context(client, user_context), - {**base_track_data, "judgeConfigKey": judge_key}, - score, - ) except Exception as exc: logger.error("Judge '%s' failed: %s", judge_key, exc) @@ -446,90 +477,94 @@ async def run_judge( output=task.actual_output, ) - async with with_judge_evaluation(task.config_key) as record_evaluation: - result = await execute_and_track( - config_key=task.config_key, - config=effective_config, - meta=task.judge_meta, - user_context=task.user_context, - handler=judge_handler, - user_input=task.actual_output, - tool_handlers=None, - variables={ - **(task.variables or {}), - "message_history": message_history, - "response_to_evaluate": task.actual_output, - **({"input": task.user_input} if task.user_input else {}), - }, - evaluated_value=task.judge_value, - ) - - usage = to_usage_dict(result["usage"]) - try: - entries = typesafe_judge_entries(result["response"]) - except ValueError: - return None - if entries is not None: - first = entries[0] - result_key = f"{task.config_key}.{first['key']}" - expanded = { - f"{task.config_key}.{entry['key']}": JudgeResult( + with _judge_target(task.parent_track_data): + async with with_judge_evaluation(task.config_key) as record_evaluation: + result = await execute_and_track( + config_key=task.config_key, + config=effective_config, + meta=task.judge_meta, + user_context=task.user_context, + handler=judge_handler, + user_input=task.actual_output, + tool_handlers=None, + variables={ + **(task.variables or {}), + "message_history": message_history, + "response_to_evaluate": task.actual_output, + **({"input": task.user_input} if task.user_input else {}), + }, + evaluated_value=task.judge_value, + ) + + usage = to_usage_dict(result["usage"]) + try: + entries = typesafe_judge_entries(result["response"]) + except ValueError: + return None + if entries is not None: + first = entries[0] + result_key = f"{task.config_key}.{first['key']}" + expanded = { + f"{task.config_key}.{entry['key']}": JudgeResult( + usage=usage, + response=entry["reason"], + score=entry["score"], + event_key=entry["eventKey"], + ) + for entry in entries + } + metrics = [ + { + "eventKey": entry["eventKey"], + "score": entry["score"], + "judgeConfigKey": f"{task.config_key}.{entry['key']}", + } + for entry in entries + ] + for entry in entries: + record_evaluation( + entry["score"], None, f"{task.config_key}.{entry['key']}" + ) + # The score belongs to the evaluated config. Each metric carries its own judgeConfigKey. + typesafe_track_data: TrackData = { + **task.parent_track_data, + "judgeConfigKey": result_key, + } + return JudgeRunResult( + score=first["score"], + response=first["reason"], usage=usage, - response=entry["reason"], - score=entry["score"], - event_key=entry["eventKey"], + track_data=typesafe_track_data, + results=expanded, + metrics=metrics, + event_key=first["eventKey"], ) - for entry in entries - } - metrics = [ - { - "eventKey": entry["eventKey"], - "score": entry["score"], - "judgeConfigKey": f"{task.config_key}.{entry['key']}", - } - for entry in entries - ] - for entry in entries: + + try: + score, reasoning = parse_judge_response(result["response"]) + except ValueError: + return None + + # The score is reported as the judge gave it. A missing or null score + # is not a zero: coercing it would record a gen_ai.evaluation of 0 -- + # indistinguishable from a judge that scored the output a hard fail -- + # where every other non-numeric judge output skips the metric instead. + metric_score = numeric_score(score) + if metric_score is not None: record_evaluation( - entry["score"], None, f"{task.config_key}.{entry['key']}" + metric_score, + reasoning if judge_handler.capture_content else None, ) - # The score belongs to the evaluated config. Each metric carries its own judgeConfigKey. - typesafe_track_data: TrackData = { + # The score belongs to the evaluated (parent) config, exactly as in + # the inline run_judges path. + merged_track_data: TrackData = { **task.parent_track_data, - "judgeConfigKey": result_key, + "judgeConfigKey": task.config_key, } + return JudgeRunResult( - score=first["score"], - response=first["reason"], + score=score, + response=reasoning, usage=usage, - track_data=typesafe_track_data, - results=expanded, - metrics=metrics, - event_key=first["eventKey"], - ) - - try: - score, reasoning = parse_judge_response(result["response"]) - except ValueError: - return None - - # The score is reported as the judge gave it. A missing or null score - # is not a zero: coercing it would record a gen_ai.evaluation of 0 -- - # indistinguishable from a judge that scored the output a hard fail -- - # where every other non-numeric judge output skips the metric instead. - metric_score = numeric_score(score) - if metric_score is not None: - record_evaluation( - metric_score, - reasoning if judge_handler.capture_content else None, + track_data=merged_track_data, ) - # The score belongs to the evaluated (parent) config, exactly as in - # the inline run_judges path. - merged_track_data: TrackData = { - **task.parent_track_data, - "judgeConfigKey": task.config_key, - } - - return JudgeRunResult( - score=score, response=reasoning, usage=usage, track_data=merged_track_data - ) From 7231c3dcfe96ff4af6a69b1f38bc15c99a5c4b9f Mon Sep 17 00:00:00 2001 From: Alexis Georges Date: Thu, 29 Oct 2026 23:12:07 -0400 Subject: [PATCH 2/4] test(client): check judge spans carry the graded run's identity Uses a real tracer provider with ConversationIdSpanProcessor and an in-memory exporter. A stand-in judge handler opens invoke_agent and a chat span under it. Covers the inline path, the deferred path reading the task, and a span outside any judge run, which stays unstamped. --- .../client/tests/test_judge_target_run.py | 155 ++++++++++++++++++ 1 file changed, 155 insertions(+) create mode 100644 packages/client/tests/test_judge_target_run.py diff --git a/packages/client/tests/test_judge_target_run.py b/packages/client/tests/test_judge_target_run.py new file mode 100644 index 00000000..5e6c806e --- /dev/null +++ b/packages/client/tests/test_judge_target_run.py @@ -0,0 +1,155 @@ +"""A judge's spans point at the run it graded. + +A judge runs after the graded handler has ended its span, so its spans start a +trace of their own. Every span of a judge's run carries the run id, config key +and variation key of the run it graded, so the two traces can be joined exactly +instead of by timestamp. +""" + +from __future__ import annotations + +import random +from collections.abc import Iterator +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +import launchdarkly_ai_server.lifecycle as lifecycle_module +from launchdarkly_ai_server import ProviderHandler, run_judges +from launchdarkly_ai_server.conversation import ConversationIdSpanProcessor +from launchdarkly_ai_server.judges import ( + LD_JUDGE_TARGET_CONFIG_KEY, + LD_JUDGE_TARGET_RUN_ID, + LD_JUDGE_TARGET_VARIATION_KEY, + run_judge, +) +from launchdarkly_ai_server.types import JudgeTask + +_exporter = InMemorySpanExporter() +_provider = TracerProvider() +_provider.add_span_processor(ConversationIdSpanProcessor()) +_provider.add_span_processor(SimpleSpanProcessor(_exporter)) +_tracer = _provider.get_tracer("@launchdarkly/ai-claude-messages") + +CONTEXT = {"kind": "user", "key": "alexis"} + +GRADED_RUN = { + "runId": "graded-run-id", + "configKey": "pr-reviewer", + "variationKey": "opus", + "version": 3, + "modelName": "claude-opus-5-5", + "providerName": "Anthropic", +} + +JUDGE_CONFIG = { + "model": {"name": "claude-sonnet-5"}, + "provider": {"name": "Anthropic"}, + "instructions": "You are a judge.", + "evaluationMetricKey": "$ld:ai:judge:no-nitpicks", +} +JUDGE_META = { + "enabled": True, + "variationKey": "default", + "version": 1, + "mode": "messages", +} + + +async def _judge_fn( + config: Any, + user_input: Any, + tool_handlers: Any, + variables: Any, + history: Any = None, +) -> dict[str, Any]: + # Stands in for a judge handler: it opens the judge's invoke_agent and a + # chat span under it, on an SDK tracer, after the graded run is over. + with _tracer.start_as_current_span("invoke_agent"): + _tracer.start_span("chat").end() + return { + "output": '{"score": 1, "reasoning": "ok"}', + "usage": {"input_tokens": 1, "output_tokens": 1}, + } + + +HANDLER = ProviderHandler(fn=_judge_fn, provides_for=("Anthropic", "messages")) # type: ignore[arg-type] + + +@pytest.fixture(autouse=True) +def _reset_exporter() -> Iterator[None]: + _exporter.clear() + yield + _exporter.clear() + + +@pytest.fixture +def mock_ld_client() -> Iterator[MagicMock]: + client = MagicMock() + client.track = MagicMock() + client.flush = AsyncMock() + client.close = AsyncMock() + client.variation = AsyncMock(return_value={**JUDGE_CONFIG, "_ldMeta": JUDGE_META}) + lifecycle_module._set_client_for_testing(client) + yield client + lifecycle_module._reset_for_testing() + + +def _expect_every_span_points_at_the_graded_run() -> None: + spans = _exporter.get_finished_spans() + assert sorted(span.name for span in spans) == ["chat", "invoke_agent"] + for span in spans: + attributes = span.attributes or {} + assert attributes[LD_JUDGE_TARGET_RUN_ID] == "graded-run-id" + assert attributes[LD_JUDGE_TARGET_CONFIG_KEY] == "pr-reviewer" + assert attributes[LD_JUDGE_TARGET_VARIATION_KEY] == "opus" + + +class TestJudgeSpansPointAtTheRunTheyGraded: + async def test_stamps_every_span_of_an_inline_judge_with_the_graded_run( + self, mock_ld_client: MagicMock + ) -> None: + with patch.object(random, "random", return_value=0.0): + await run_judges( + config={ + **JUDGE_CONFIG, + "judgeConfiguration": { + "judges": [{"key": "no-nitpicks", "samplingRate": 1.0}] + }, + }, + user_context=CONTEXT, + handler=HANDLER, + user_input=None, + llm_response="Verdict: approve", + base_track_data=dict(GRADED_RUN), + ) + _expect_every_span_points_at_the_graded_run() + + async def test_stamps_every_span_of_a_deferred_judge_with_the_graded_run_from_the_task( + self, mock_ld_client: MagicMock + ) -> None: + task = JudgeTask( + config_key="no-nitpicks", + judge_config=dict(JUDGE_CONFIG), + judge_meta=dict(JUDGE_META), # type: ignore[arg-type] + actual_output="Verdict: approve", + user_context=CONTEXT, + judge_provider="Anthropic", + judge_mode="messages", + collapse_messages=False, + parent_track_data=dict(GRADED_RUN), + ) + await run_judge(task, [HANDLER]) + _expect_every_span_points_at_the_graded_run() + + def test_leaves_spans_outside_a_judge_run_alone(self) -> None: + _tracer.start_span("invoke_agent").end() + [span] = _exporter.get_finished_spans() + attributes = span.attributes or {} + assert LD_JUDGE_TARGET_RUN_ID not in attributes + assert LD_JUDGE_TARGET_CONFIG_KEY not in attributes + assert LD_JUDGE_TARGET_VARIATION_KEY not in attributes From bb98172afb6fc8ec4e2cd9dd0a2c473645dfebd7 Mon Sep 17 00:00:00 2001 From: Alexis Georges Date: Thu, 29 Oct 2026 23:12:07 -0400 Subject: [PATCH 3/4] docs: document the judge target span attributes Adds the three keys to the judge section of the telemetry contract, where they come from on each path, and that capture settings do not gate them because they are identifiers, not content. --- TELEMETRY-CONTRACT.md | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/TELEMETRY-CONTRACT.md b/TELEMETRY-CONTRACT.md index 212610b8..0fba00de 100644 --- a/TELEMETRY-CONTRACT.md +++ b/TELEMETRY-CONTRACT.md @@ -242,6 +242,26 @@ model-generated prose about the user's conversation, so it follows the judge han handler or the process default withholds it. The reasoning is always returned to the caller in `judgeResults` / `judge_results`; only the telemetry copy is gated. +### Judge spans point at the run they graded + +A judge runs after the graded handler has ended its span, so its spans start a trace of their own, +with their own `launchdarkly.run.id`. To tie them back, `run_judges` (inline) and `run_judge` +(deferred) wrap each judge's run in `stamp_span_attributes`, so every SDK span of that run carries +the graded run's identity, write-if-absent: + +| Span attribute | Value | +|---|---| +| `launchdarkly.judge.target.run.id` | graded run's `TrackData.runId` | +| `launchdarkly.judge.target.config.key` | graded run's `TrackData.configKey` | +| `launchdarkly.judge.target.variation.key` | graded run's `TrackData.variationKey` | + +The constants are `LD_JUDGE_TARGET_*` in `launchdarkly_ai_server/judges.py`. Inline judges read +`base_track_data`; deferred judges read `JudgeTask.parent_track_data`. Each key is written only when +its value is a non-empty string. They are identifiers, not content, so capture settings do not gate +them. The judge's metric event already carries the graded `runId` in its `TrackData`; these +attributes put the same link on the trace. Offline evaluations stamp `launchdarkly.evaluation_run.id` +on every span of a row the same way. + --- ## 5. Finish reasons From 2f3fe842609a8481a910f79bab2eb74121d59ef5 Mon Sep 17 00:00:00 2001 From: Alexis Georges Date: Thu, 29 Oct 2026 23:16:21 -0400 Subject: [PATCH 4/4] test(client): cover the rest of the judge target rule Adds the cases the cross-SDK rule requires beyond the inline, deferred and outside-a-run checks: - a span from a non-SDK tracer inside the judge run stays unstamped - the attributes are written with the judge handler's capture off - an empty variationKey leaves that one key off and writes the other two - a span that already carries a target key keeps its value - in graph().invoke(), a per-node judge points at the node's run, and the graph-level judge points at the graph run, graph key and graph variation key, each equal to what that judge's metric event names --- .../client/tests/test_judge_target_run.py | 285 ++++++++++++++++-- 1 file changed, 256 insertions(+), 29 deletions(-) diff --git a/packages/client/tests/test_judge_target_run.py b/packages/client/tests/test_judge_target_run.py index 5e6c806e..06d73263 100644 --- a/packages/client/tests/test_judge_target_run.py +++ b/packages/client/tests/test_judge_target_run.py @@ -14,12 +14,13 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest -from opentelemetry.sdk.trace import TracerProvider +from opentelemetry import trace +from opentelemetry.sdk.trace import ReadableSpan, TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter import launchdarkly_ai_server.lifecycle as lifecycle_module -from launchdarkly_ai_server import ProviderHandler, run_judges +from launchdarkly_ai_server import ProviderHandler, graph, run_judges from launchdarkly_ai_server.conversation import ConversationIdSpanProcessor from launchdarkly_ai_server.judges import ( LD_JUDGE_TARGET_CONFIG_KEY, @@ -34,6 +35,13 @@ _provider.add_span_processor(ConversationIdSpanProcessor()) _provider.add_span_processor(SimpleSpanProcessor(_exporter)) _tracer = _provider.get_tracer("@launchdarkly/ai-claude-messages") +_foreign_tracer = _provider.get_tracer("opentelemetry.instrumentation.httpx") + +TARGET_KEYS = ( + LD_JUDGE_TARGET_RUN_ID, + LD_JUDGE_TARGET_CONFIG_KEY, + LD_JUDGE_TARGET_VARIATION_KEY, +) CONTEXT = {"kind": "user", "key": "alexis"} @@ -59,6 +67,24 @@ "mode": "messages", } +JUDGE_OUTPUT = { + "output": '{"score": 1, "reasoning": "ok"}', + "usage": {"input_tokens": 1, "output_tokens": 1}, +} + + +def _open_judge_spans(label: str = "judge") -> None: + """Stands in for a judge handler's spans. + + Opens the judge's ``invoke_agent`` and a ``chat`` span under it on an SDK + tracer, after the graded run is over. ``label`` lets a test tell apart the + spans of different judges in one trace. + """ + with _tracer.start_as_current_span( + "invoke_agent", attributes={"test.label": label} + ): + _tracer.start_span("chat", attributes={"test.label": label}).end() + async def _judge_fn( config: Any, @@ -67,19 +93,17 @@ async def _judge_fn( variables: Any, history: Any = None, ) -> dict[str, Any]: - # Stands in for a judge handler: it opens the judge's invoke_agent and a - # chat span under it, on an SDK tracer, after the graded run is over. - with _tracer.start_as_current_span("invoke_agent"): - _tracer.start_span("chat").end() - return { - "output": '{"score": 1, "reasoning": "ok"}', - "usage": {"input_tokens": 1, "output_tokens": 1}, - } + _open_judge_spans() + return JUDGE_OUTPUT HANDLER = ProviderHandler(fn=_judge_fn, provides_for=("Anthropic", "messages")) # type: ignore[arg-type] +def _finished() -> list[ReadableSpan]: + return list(_exporter.get_finished_spans()) + + @pytest.fixture(autouse=True) def _reset_exporter() -> Iterator[None]: _exporter.clear() @@ -99,8 +123,30 @@ def mock_ld_client() -> Iterator[MagicMock]: lifecycle_module._reset_for_testing() +async def _run_inline( + handler: ProviderHandler = HANDLER, + base_track_data: dict[str, Any] | None = None, +) -> None: + with patch.object(random, "random", return_value=0.0): + await run_judges( + config={ + **JUDGE_CONFIG, + "judgeConfiguration": { + "judges": [{"key": "no-nitpicks", "samplingRate": 1.0}] + }, + }, + user_context=CONTEXT, + handler=handler, + user_input=None, + llm_response="Verdict: approve", + base_track_data=dict( + GRADED_RUN if base_track_data is None else base_track_data + ), + ) + + def _expect_every_span_points_at_the_graded_run() -> None: - spans = _exporter.get_finished_spans() + spans = _finished() assert sorted(span.name for span in spans) == ["chat", "invoke_agent"] for span in spans: attributes = span.attributes or {} @@ -113,20 +159,7 @@ class TestJudgeSpansPointAtTheRunTheyGraded: async def test_stamps_every_span_of_an_inline_judge_with_the_graded_run( self, mock_ld_client: MagicMock ) -> None: - with patch.object(random, "random", return_value=0.0): - await run_judges( - config={ - **JUDGE_CONFIG, - "judgeConfiguration": { - "judges": [{"key": "no-nitpicks", "samplingRate": 1.0}] - }, - }, - user_context=CONTEXT, - handler=HANDLER, - user_input=None, - llm_response="Verdict: approve", - base_track_data=dict(GRADED_RUN), - ) + await _run_inline() _expect_every_span_points_at_the_graded_run() async def test_stamps_every_span_of_a_deferred_judge_with_the_graded_run_from_the_task( @@ -148,8 +181,202 @@ async def test_stamps_every_span_of_a_deferred_judge_with_the_graded_run_from_th def test_leaves_spans_outside_a_judge_run_alone(self) -> None: _tracer.start_span("invoke_agent").end() - [span] = _exporter.get_finished_spans() + [span] = _finished() attributes = span.attributes or {} - assert LD_JUDGE_TARGET_RUN_ID not in attributes - assert LD_JUDGE_TARGET_CONFIG_KEY not in attributes - assert LD_JUDGE_TARGET_VARIATION_KEY not in attributes + assert not [key for key in TARGET_KEYS if key in attributes] + + async def test_leaves_spans_from_other_tracers_alone( + self, mock_ld_client: MagicMock + ) -> None: + async def fn(*args: Any, **kwargs: Any) -> dict[str, Any]: + _open_judge_spans() + _foreign_tracer.start_span("POST").end() + return JUDGE_OUTPUT + + await _run_inline( + ProviderHandler(fn=fn, provides_for=("Anthropic", "messages")) + ) # type: ignore[arg-type] + [foreign] = [span for span in _finished() if span.name == "POST"] + attributes = foreign.attributes or {} + assert not [key for key in TARGET_KEYS if key in attributes] + + async def test_writes_the_attributes_with_content_capture_off( + self, mock_ld_client: MagicMock + ) -> None: + handler = ProviderHandler( + fn=_judge_fn, # type: ignore[arg-type] + provides_for=("Anthropic", "messages"), + capture_content=False, + ) + await _run_inline(handler) + _expect_every_span_points_at_the_graded_run() + + async def test_omits_an_empty_value_and_writes_the_others( + self, mock_ld_client: MagicMock + ) -> None: + await _run_inline(base_track_data={**GRADED_RUN, "variationKey": ""}) + spans = _finished() + assert sorted(span.name for span in spans) == ["chat", "invoke_agent"] + for span in spans: + attributes = span.attributes or {} + assert LD_JUDGE_TARGET_VARIATION_KEY not in attributes + assert attributes[LD_JUDGE_TARGET_RUN_ID] == "graded-run-id" + assert attributes[LD_JUDGE_TARGET_CONFIG_KEY] == "pr-reviewer" + + async def test_keeps_a_value_the_span_already_has( + self, mock_ld_client: MagicMock + ) -> None: + async def fn(*args: Any, **kwargs: Any) -> dict[str, Any]: + _open_judge_spans() + _tracer.start_span("own", attributes={LD_JUDGE_TARGET_RUN_ID: "own"}).end() + return JUDGE_OUTPUT + + await _run_inline( + ProviderHandler(fn=fn, provides_for=("Anthropic", "messages")) + ) # type: ignore[arg-type] + [own] = [span for span in _finished() if span.name == "own"] + attributes = own.attributes or {} + assert attributes[LD_JUDGE_TARGET_RUN_ID] == "own" + assert attributes[LD_JUDGE_TARGET_CONFIG_KEY] == "pr-reviewer" + + +# --------------------------------------------------------------------------- +# Graphs +# --------------------------------------------------------------------------- + +GRAPH_VARIATION = { + "root": "root-node", + "edges": {}, + "_ldMeta": {"enabled": True, "variationKey": "graph-v2", "version": 2}, +} +NODE_INSTRUCTIONS = "I am the root node." +NODE_JUDGE_INSTRUCTIONS = "You judge the node." +GRAPH_JUDGE_INSTRUCTIONS = "You judge the graph." + + +def _graph_variation_for(key: str) -> dict[str, Any]: + if key == "graph-key": + return GRAPH_VARIATION + judge_meta = { + "enabled": True, + "variationKey": "j1", + "version": 1, + "mode": "messages", + } + if key == "node-judge": + return { + **JUDGE_CONFIG, + "provider": {"name": "TestProvider"}, + "instructions": NODE_JUDGE_INSTRUCTIONS, + "evaluationMetricKey": "$ld:ai:judge:node", + "_ldMeta": judge_meta, + } + if key == "graph-judge": + return { + **JUDGE_CONFIG, + "provider": {"name": "TestProvider"}, + "instructions": GRAPH_JUDGE_INSTRUCTIONS, + "evaluationMetricKey": "$ld:ai:judge:graph", + "_ldMeta": judge_meta, + } + return { + "model": {"name": "gpt-4"}, + "provider": {"name": "TestProvider"}, + "instructions": NODE_INSTRUCTIONS, + "judgeConfiguration": {"judges": [{"key": "node-judge", "samplingRate": 1.0}]}, + "_ldMeta": { + "enabled": True, + "variationKey": "node-v1", + "version": 1, + "mode": "messages", + }, + } + + +async def _graph_handler_fn( + config: Any, + user_input: Any, + tool_handlers: Any, + variables: Any, + history: Any = None, +) -> dict[str, Any]: + instructions = config.get("instructions", "") + if instructions == NODE_JUDGE_INSTRUCTIONS: + _open_judge_spans("node-judge") + return JUDGE_OUTPUT + if instructions == GRAPH_JUDGE_INSTRUCTIONS: + _open_judge_spans("graph-judge") + return JUDGE_OUTPUT + _open_judge_spans("node") + return {"output": "final", "usage": {"input_tokens": 1, "output_tokens": 1}} + + +@pytest.fixture +def graph_ld_client() -> Iterator[MagicMock]: + client = MagicMock() + client.track = MagicMock() + client.flush = AsyncMock() + client.close = AsyncMock() + + async def variation(key: str, ctx: Any, default: Any) -> Any: + return _graph_variation_for(key) + + client.variation = AsyncMock(side_effect=variation) + lifecycle_module._set_client_for_testing(client) + # graph.py opens the graph span on trace.get_tracer, and the global provider + # is claimed process-wide, so hand it this file's tracer instead. + with patch.object(trace, "get_tracer", return_value=_tracer): + yield client + lifecycle_module._reset_for_testing() + + +def _metric_track_data(client: MagicMock, event_key: str) -> dict[str, Any]: + [call] = [c for c in client.track.call_args_list if c.args[0] == event_key] + return dict(call.args[2]) + + +def _labelled(label: str) -> list[ReadableSpan]: + return [s for s in _finished() if (s.attributes or {}).get("test.label") == label] + + +class TestGraphJudgeTargets: + async def test_node_and_graph_judges_point_at_the_runs_they_graded( + self, graph_ld_client: MagicMock + ) -> None: + handler = ProviderHandler( + fn=_graph_handler_fn, # type: ignore[arg-type] + provides_for=("TestProvider", "messages"), + ) + await graph("graph-key", handlers=[handler], graph_judge="graph-judge").invoke( + "hi", CONTEXT + ) + + # A per-node judge points at the node's own run: the run its metric event names. + node_metric = _metric_track_data(graph_ld_client, "$ld:ai:judge:node") + assert node_metric["configKey"] == "root-node" + node_judge_spans = _labelled("node-judge") + assert sorted(s.name for s in node_judge_spans) == ["chat", "invoke_agent"] + for span in node_judge_spans: + attributes = span.attributes or {} + assert attributes[LD_JUDGE_TARGET_RUN_ID] == node_metric["runId"] + assert attributes[LD_JUDGE_TARGET_CONFIG_KEY] == "root-node" + assert attributes[LD_JUDGE_TARGET_VARIATION_KEY] == "node-v1" + + # The graph-level judge points at the graph run. + [graph_span] = [s for s in _finished() if s.name == "launchdarkly.graph"] + graph_run_id = (graph_span.attributes or {})["launchdarkly.run.id"] + graph_metric = _metric_track_data(graph_ld_client, "$ld:ai:judge:graph") + assert graph_metric["runId"] == graph_run_id + assert graph_run_id != node_metric["runId"] + graph_judge_spans = _labelled("graph-judge") + assert sorted(s.name for s in graph_judge_spans) == ["chat", "invoke_agent"] + for span in graph_judge_spans: + attributes = span.attributes or {} + assert attributes[LD_JUDGE_TARGET_RUN_ID] == graph_run_id + assert attributes[LD_JUDGE_TARGET_CONFIG_KEY] == "graph-key" + assert attributes[LD_JUDGE_TARGET_VARIATION_KEY] == "graph-v2" + + # The graded node's own spans are not a judge run. + for span in _labelled("node"): + attributes = span.attributes or {} + assert not [key for key in TARGET_KEYS if key in attributes]