feat(client): run shadow requests alongside the served variation

packages/claude-agents/tests/test_skills.py

@@ -672,6 +672,7 @@ async def during(kwargs: dict[str, Any]) -> None:
 async def ld_client() -> Any:
     await shutdown()
     client = MagicMock()
+    del client.variation_detail
     client.track = MagicMock()
     client.flush = MagicMock()
     client.close = MagicMock()

packages/client/README.md

@@ -379,6 +379,29 @@ elif result and task.evaluation_metric_key:
     get_client().track(task.evaluation_metric_key, context, result.track_data, result.score)
 ```
 
+#### Shadow requests
+
+A shadow request runs another variation of an AI Config on the same input and reports it on the normal AI metrics with `isShadow: true`. You attach shadow requests to a targeting rule or the fallthrough in LaunchDarkly. The shadow's response is discarded, and your response, latency and events are unchanged. Shadow runs use your provider account, so they are billed like real calls.
+
+`config().invoke()` and `config().stream()` run them in the background, each with a 30 second timeout that you can change with `shadow_timeout_s`. A shadow makes no tool calls and runs no judges. Graphs and judges never run shadows.
+
+With `skip_shadows=True`, `invoke()` returns `shadow_tasks` instead of running them, and `stream()` puts them on the `done` event. A task can be pickled, so you can run it later on your own queue or worker with `run_shadow(task, handlers)`, the same way as `run_judge`:
+
+```python
+from launchdarkly_ai_server import config, run_shadow
+
+result = await config(key="my-ai-config", handler=handlers, skip_shadows=True).invoke(
+    user_input, context
+)
+for task in result.shadow_tasks or []:
+    queue.put_nowait(task)
+
+# in the worker
+outcome = await run_shadow(task, handlers)
+```
+
+`run_shadow` never raises. A failure or timeout is in `outcome.error`, and the shadow's events are already tracked.
+
 ### `graph(key, **options)`
 
 Runs a multi-agent workflow defined in a LaunchDarkly agent graph flag. The SDK uses a **model-driven router**: it starts at the root node, presents outgoing edges as handoff choices to the model, and follows whichever edge the model selects. The loop terminates when the model produces a final answer, a leaf is reached, a cycle is detected, or the step cap is hit.
@@ -1032,7 +1055,7 @@ All types are exported from this package. Handler packages import them from here
 | `AiConfigRep` | The AI configuration object fetched from a LaunchDarkly flag variation |
 | `Tool` | A tool definition (name, description, JSON Schema parameters) |
 | `ProviderHandler` | The callable type that all handler packages produce |
-| `ProviderResponse` | The value returned to callers: `response`, `usage`, `track_data`, `judge_results?`, `judge_tasks?`. `judge_results` is populated when `skip_judges=False`; `judge_tasks` (a `list[JudgeTask]`) is populated when `skip_judges=True`. |
+| `ProviderResponse` | The value returned to callers: `response`, `usage`, `track_data`, `judge_results?`, `judge_tasks?`, `shadow_tasks?`. `judge_results` is populated when `skip_judges=False`; `judge_tasks` (a `list[JudgeTask]`) is populated when `skip_judges=True`; `shadow_tasks` (a `list[ShadowTask]`) is populated when `skip_shadows=True`. |
 | `ConfigArgs` | Arguments accepted by `config()` (key, handler, tool_handlers, registry) |
 | `NativeTool` | Marker class for provider built-in tools |
 | `LDContext` | Standard LaunchDarkly context dict. Import from `launchdarkly_ai_server`. |

packages/client/api-report/launchdarkly_ai_server.txt

@@ -10,10 +10,10 @@ any_multimodal(turns: list[dict[str, Any]]) -> bool
 build_judge_tasks(*, config: AiConfigRep, user_context: LDContext, handler: ProviderHandler, handlers: list[ProviderHandler] | None = None, llm_response: str, base_track_data: TrackData, user_input: str | None = None, trajectory: str = '') -> list[JudgeTask]  [async]
 compose(a: Registry, b: Registry) -> Registry
 compose_history(*, history: list[dict[str, Any]], user_input: str | None = None, config_messages: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]
-config(*, key: str, handler: ProviderHandler | list[ProviderHandler] | None = None, tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None = None, registry: Any = None, skip_judges: bool = False) -> ConfigInstance
+config(*, key: str, handler: ProviderHandler | list[ProviderHandler] | None = None, tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None = None, registry: Any = None, skip_judges: bool = False, skip_shadows: bool = False, shadow_timeout_s: float = 30.0) -> ConfigInstance
 config_skills(key: str, context: LDContext) -> SkillSource
 ConfigInstance  [class]
-  __init__(key: str, handler: ProviderHandler | list[ProviderHandler] | None, tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None, registry: Any, skip_judges: bool = False)
+  __init__(key: str, handler: ProviderHandler | list[ProviderHandler] | None, tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None, registry: Any, skip_judges: bool = False, skip_shadows: bool = False, shadow_timeout_s: float = 30.0)
   invoke(user_input: str | None, context: LDContext, variables: dict[str, Any] | None = None, history: list[dict[str, Any]] | None = None) -> ProviderResponse[Any]  [async]
   stream(user_input: str | None, context: LDContext, variables: dict[str, Any] | None = None, history: list[dict[str, Any]] | None = None) -> AsyncGenerator[StreamEvent, None]
 configure(*, options: SDKOptions) -> None
@@ -61,11 +61,11 @@ EvaluationsModule  [class]
   sdk_key: str | None  [property]
   tools: ToolsClient  [property]
   ui_base_uri: str  [property]
-execute_and_stream(*, config_key: str, config: AiConfigRep, meta: VariationMeta, user_context: LDContext, handler: ProviderHandler, user_input: str | None = None, tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None = None, variables: dict[str, Any] | None = None, graph_key: str | None = None, history: list[dict[str, Any]] | None = None) -> AsyncGenerator[ExecuteStreamEvent, None]  [async]
-execute_and_track(*, config_key: str, config: AiConfigRep, meta: VariationMeta, user_context: LDContext, handler: ProviderHandler, user_input: str | None = None, tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None = None, variables: dict[str, Any] | None = None, graph_key: str | None = None, history: list[dict[str, Any]] | None = None) -> dict[str, Any]  [async]
+execute_and_stream(*, config_key: str, config: AiConfigRep, meta: VariationMeta, user_context: LDContext, handler: ProviderHandler, user_input: str | None = None, tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None = None, variables: dict[str, Any] | None = None, graph_key: str | None = None, history: list[dict[str, Any]] | None = None, run_id: str | None = None) -> AsyncGenerator[ExecuteStreamEvent, None]  [async]
+execute_and_track(*, config_key: str, config: AiConfigRep, meta: VariationMeta, user_context: LDContext, handler: ProviderHandler, user_input: str | None = None, tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None = None, variables: dict[str, Any] | None = None, graph_key: str | None = None, history: list[dict[str, Any]] | None = None, run_id: str | None = None, extra_track_data: dict[str, Any] | None = None, span_attributes: dict[str, Any] | None = None) -> dict[str, Any]  [async]
 ExecuteStreamDoneEvent = dict[str, Any]
 ExecuteStreamEvent = dict[str, Any]
-extract_variation(config_key: str, context: dict[str, Any]) -> dict[str, Any]  [async]
+extract_variation(config_key: str, context: dict[str, Any], *, with_reason: bool = False) -> dict[str, Any]  [async]
 FDv2SkillStore = LaunchDarklySkillStore
 GenerationConfig(TypedDict, total=False)  [class]
   instructions: str
@@ -255,10 +255,11 @@ ProviderHandler  [class]
   provides_for: tuple[str, Literal['agent', 'messages']] | None = provides_for
   stream(config: AiConfigRep, user_input: str | None = None, tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None = None, variables: dict[str, Any] | None = None, history: list[dict[str, Any]] | None = None) -> AsyncGenerator[dict[str, Any], None]  [async]
 ProviderResponse(Generic[T])  [class, @dataclass]
-  __init__(response: T, usage: UsageDict, judge_results: dict[str, JudgeResult] | None = None, judge_tasks: list[JudgeTask] | None = None, track_data: TrackData | None = None)
+  __init__(response: T, usage: UsageDict, judge_results: dict[str, JudgeResult] | None = None, judge_tasks: list[JudgeTask] | None = None, track_data: TrackData | None = None, shadow_tasks: list[ShadowTask] | None = None)
   judge_results: dict[str, JudgeResult] | None = None
   judge_tasks: list[JudgeTask] | None = None
   response: T
+  shadow_tasks: list[ShadowTask] | None = None
   track_data: TrackData | None = None
   usage: UsageDict
 ReconcileAction  [class, @dataclass(frozen=True)]
@@ -286,6 +287,7 @@ resolve_handlers(registry: Registry | None, local_handlers: list[ProviderHandler
 resolve_tools(registry: Registry | None, local_tools: Mapping[str, Callable[..., Any] | NativeTool] | None) -> Mapping[str, Callable[..., Any] | NativeTool] | None
 run_judge(task: JudgeTask, handlers: list[ProviderHandler]) -> JudgeRunResult | None  [async]
 run_judges(*, config: AiConfigRep, user_context: LDContext, handler: ProviderHandler, handlers: list[ProviderHandler] | None = None, user_input: str | None, llm_response: str, base_track_data: TrackData, tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None = None, graph_key: str | None = None, trajectory: str = '') -> dict[str, JudgeResult]  [async]
+run_shadow(task: ShadowTask, handlers: list[ProviderHandler]) -> ShadowResult  [async]
 RunSummary  [class, @dataclass]
   __init__(total_rows: int = 0, passed_rows: int = 0, failed_rows: int = 0, error_rows: int = 0, pending_rows: int = 0)
   error_rows: int = 0
@@ -324,6 +326,24 @@ set_skill_store(store: SkillStore | None) -> None
 set_tool_call_content_attributes(span: Any, capture: bool, *, arguments: Any = None, result: Any = None) -> None
 set_tool_definition_attributes(span: Any, capture: bool, tools: list[ToolDefinitionInput]) -> None
 set_usage_span_attributes(span: Any, usage: SpanUsage) -> None
+ShadowResult  [class, @dataclass(frozen=True)]
+  __init__(usage: UsageDict, track_data: TrackData, error: str | None = None)
+  error: str | None = None
+  track_data: TrackData
+  usage: UsageDict
+ShadowTask  [class, @dataclass(frozen=True)]
+  __init__(config_key: str, shadow_request_id: str, shadow_config: AiConfigRep, shadow_meta: VariationMeta, user_input: str | None, history: list[dict[str, Any]] | None, variables: dict[str, Any] | None, user_context: LDContext, timeout_s: float, track_data: TrackData, span_attributes: dict[str, Any])
+  config_key: str
+  history: list[dict[str, Any]] | None
+  shadow_config: AiConfigRep
+  shadow_meta: VariationMeta
+  shadow_request_id: str
+  span_attributes: dict[str, Any]
+  timeout_s: float
+  track_data: TrackData
+  user_context: LDContext
+  user_input: str | None
+  variables: dict[str, Any] | None
 shutdown() -> None  [async]
 Skill  [class, @dataclass(frozen=True)]
   __init__(key: str, version: int, content: bytes, content_hash: str, name: str | None = None, description: str | None = None)

packages/client/src/launchdarkly_ai_server/__init__.py

@@ -71,6 +71,7 @@
 )
 from .sdk_info import SDK_INFO_CONTEXT, SDK_INFO_EVENT, register_ai_sdk_package
 from .sdk_usage import report_usage
+from .shadows import run_shadow
 from .skills import (
     InMemorySkillStore,
     all_skills,
@@ -128,6 +129,8 @@
     ReconcileActionKind,
     ReconcileReport,
     SDKOptions,
+    ShadowResult,
+    ShadowTask,
     Skill,
     SkillOutcome,
     SkillOutcomeReason,
@@ -200,6 +203,8 @@
     "ProviderGraphResponse",
     "ProviderHandler",
     "ProviderResponse",
+    "ShadowResult",
+    "ShadowTask",
     "StreamChunkEvent",
     "StreamDoneEvent",
     "StreamEvent",
@@ -276,6 +281,8 @@
     "build_judge_tasks",
     "run_judge",
     "run_judges",
+    # shadows
+    "run_shadow",
     # client
     "config",
     "ConfigInstance",

packages/client/src/launchdarkly_ai_server/client.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import math
+import uuid
 from collections.abc import AsyncGenerator, Callable, Mapping
 from contextlib import aclosing
 from typing import Any
@@ -11,13 +13,15 @@
 from .lifecycle import extract_variation
 from .registry import resolve_handlers, resolve_tools
 from .sdk_usage import report_usage
+from .shadows import build_shadow_tasks, start_shadows
 from .tracking import execute_and_stream, execute_and_track
 from .types import (
     AiConfigRep,
     LDContext,
     NativeTool,
     ProviderHandler,
     ProviderResponse,
+    ShadowTask,
     StreamEvent,
     VariationMeta,
 )
@@ -55,12 +59,18 @@ def __init__(
         tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None,
         registry: Any,  # Registry | None
         skip_judges: bool = False,
+        skip_shadows: bool = False,
+        shadow_timeout_s: float = 30.0,
     ) -> None:
+        if not (math.isfinite(shadow_timeout_s) and shadow_timeout_s > 0):
+            raise ValueError("shadow_timeout_s must be a finite number greater than 0")
         self._key = key
         self._handler = handler
         self._tool_handlers = tool_handlers
         self._registry = registry
         self._skip_judges = skip_judges
+        self._skip_shadows = skip_shadows
+        self._shadow_timeout_s = shadow_timeout_s
 
     def _normalize_handlers(self) -> list[ProviderHandler] | None:
         if self._handler is None:
@@ -69,6 +79,33 @@ def _normalize_handlers(self) -> list[ProviderHandler] | None:
             return self._handler
         return [self._handler]
 
+    def _shadows(
+        self,
+        variation: dict[str, Any],
+        context: LDContext,
+        user_input: str | None,
+        history: list[dict[str, Any]] | None,
+        variables: dict[str, Any] | None,
+        handlers: list[ProviderHandler],
+        run_id: str,
+    ) -> list[ShadowTask] | None:
+        """Starts the sampled shadows, or returns them when ``skip_shadows`` is set."""
+        tasks = build_shadow_tasks(
+            config_key=self._key,
+            variation=variation,
+            context=context,
+            user_input=user_input,
+            history=history,
+            variables=variables,
+            handlers=handlers,
+            run_id=run_id,
+            timeout_s=self._shadow_timeout_s,
+        )
+        if self._skip_shadows:
+            return tasks
+        start_shadows(tasks, handlers)
+        return None
+
     async def invoke(
         self,
         user_input: str | None,
@@ -91,13 +128,23 @@ async def _invoke(
         )
         resolved_tools = resolve_tools(self._registry, self._tool_handlers)
 
-        variation = await extract_variation(self._key, context)
+        variation = await extract_variation(self._key, context, with_reason=True)
         config: AiConfigRep = variation["config"]
         meta: VariationMeta = variation["meta"]
 
         if not resolved_handler_list:
             raise ValueError("No handlers provided to config()")
         handler = select_handler(config, meta, resolved_handler_list)
+        run_id = str(uuid.uuid4())
+        shadow_tasks = self._shadows(
+            variation,
+            context,
+            user_input,
+            history,
+            variables,
+            resolved_handler_list,
+            run_id,
+        )
 
         result = await execute_and_track(
             config_key=self._key,
@@ -109,6 +156,7 @@ async def _invoke(
             tool_handlers=resolved_tools,
             variables=variables,
             history=history,
+            run_id=run_id,
         )
 
         raw_response = result["response"]
@@ -140,6 +188,7 @@ async def _invoke(
                 response=parsed_response,
                 usage=usage_obj,
                 judge_tasks=judge_tasks,
+                shadow_tasks=shadow_tasks,
                 track_data=track_data,
             )
 
@@ -158,6 +207,7 @@ async def _invoke(
             response=parsed_response,
             usage=usage_obj,
             judge_results=judge_results if judge_results else None,
+            shadow_tasks=shadow_tasks,
             track_data=track_data,
         )
 
@@ -210,7 +260,7 @@ async def _execute_stream(
         )
         resolved_tools = resolve_tools(self._registry, self._tool_handlers)
 
-        variation = await extract_variation(self._key, context)
+        variation = await extract_variation(self._key, context, with_reason=True)
         config: AiConfigRep = variation["config"]
         meta: VariationMeta = variation["meta"]
 
@@ -228,6 +278,16 @@ async def _execute_stream(
         sink["tools"] = resolved_tools
         sink["provider"] = provider
         sink["model"] = model
+        run_id = str(uuid.uuid4())
+        sink["shadow_tasks"] = self._shadows(
+            variation,
+            context,
+            user_input,
+            history,
+            variables,
+            resolved_handler_list,
+            run_id,
+        )
 
         async with aclosing(
             execute_and_stream(
@@ -240,6 +300,7 @@ async def _execute_stream(
                 tool_handlers=resolved_tools,
                 variables=variables,
                 history=history,
+                run_id=run_id,
             )
         ) as source:
             async for event in source:
@@ -266,12 +327,15 @@ async def _stream_events(
             judge_results = await judge_completed_stream(
                 self, sink, done_event, user_input, context
             )
-            yield {
+            done: StreamEvent = {
                 "type": "done",
                 "response": done_event.get("response", ""),
                 "usage": done_event.get("usage"),
                 "judge_results": judge_results if judge_results else None,
             }
+            if sink.get("shadow_tasks") is not None:
+                done["shadow_tasks"] = sink["shadow_tasks"]
+            yield done
 
 
 class _InternalConfigInstance(ConfigInstance):
@@ -307,6 +371,8 @@ def config(
     tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None = None,
     registry: Any = None,
     skip_judges: bool = False,
+    skip_shadows: bool = False,
+    shadow_timeout_s: float = 30.0,
 ) -> ConfigInstance:
     """
     Creates a ``ConfigInstance`` bound to *key*. Accepts a single handler or a
@@ -319,13 +385,20 @@ def config(
     ``.invoke()`` / ``.stream()``. When set, ``invoke()`` returns
     ``judge_tasks: list[JudgeTask]`` — pre-packaged tasks ready for a background
     thread calling ``run_judge(task, handlers)``.
+
+    Shadow requests on the matched rule or fallthrough run in the background,
+    each limited to *shadow_timeout_s*. Pass ``skip_shadows=True`` to get them
+    back as ``shadow_tasks`` instead, and run each with
+    ``run_shadow(task, handlers)``.
     """
     return ConfigInstance(
         key=key,
         handler=handler,
         tool_handlers=tool_handlers,
         registry=registry,
         skip_judges=skip_judges,
+        skip_shadows=skip_shadows,
+        shadow_timeout_s=shadow_timeout_s,
     )
 
 
@@ -336,6 +409,8 @@ def _config(
     tool_handlers: Mapping[str, Callable[..., Any] | NativeTool] | None = None,
     registry: Any = None,
     skip_judges: bool = False,
+    skip_shadows: bool = False,
+    shadow_timeout_s: float = 30.0,
 ) -> ConfigInstance:
     """Non-reporting :func:`config` for SDK wrappers."""
     return _InternalConfigInstance(
@@ -344,4 +419,6 @@ def _config(
         tool_handlers=tool_handlers,
         registry=registry,
         skip_judges=skip_judges,
+        skip_shadows=skip_shadows,
+        shadow_timeout_s=shadow_timeout_s,
     )