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

@@ -305,6 +305,29 @@ result = await evals.run(
 
 **The list is checked before any network I/O.** A blank key, an uppercase key, a `schema` that is not a JSON object, a schema that is not JSON-serializable (including a `NaN` or `Infinity` value), and a non-callable implementation each fail with zero requests issued. A repeated key fails too, and keys are compared without case. A `NativeTool` is valid only for a library tool, because the provider supplies its schema.
 
+### Shadow requests
+
+A shadow request runs another variation of an AI Config in the background on the same input, and reports it on the normal AI metrics with `isShadow: true`. You set shadow requests up on 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 for you in the background, with a 30 second timeout. Shadows don't run for graphs or judges, and a shadow makes no tool calls and runs no judges.
+
+```python
+caller = config(key="my-ai-config", handler=handlers, shadow_timeout_s=10)
+```
+
+To run shadows on your own queue or worker instead, pass `skip_shadows=True`. Each response then carries `shadow_tasks`, which you can pickle and run later with `run_shadow`, the same way `skip_judges` and `run_judge` work:
+
+```python
+from launchdarkly_ai_server import config, run_shadow
+
+result = await config(key="my-ai-config", handler=handlers, skip_shadows=True).invoke(...)
+for task in result.shadow_tasks or []:
+    queue.put_nowait(task)
+
+# in the worker
+outcome = await run_shadow(task, handlers)
+```
+
 ### `config(**args)`
 
 The primary entry point for AI config invocations. Accepts either a single handler or a list of handlers and routes to the correct one at invoke-time based on the flag variation's provider and mode.

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 = DEFAULT_SHADOW_TIMEOUT_S) -> 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 = DEFAULT_SHADOW_TIMEOUT_S)
   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, extra_track_data: dict[str, Any] | None = None, span_attributes: 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, 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, shadow_tasks: list[ShadowTask] | None = None, track_data: TrackData | 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: dict[str, int], track_data: TrackData, error: str | None = None)
+  error: str | None = None
+  track_data: TrackData
+  usage: dict[str, int]
+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] = dict())
+  config_key: str
+  history: list[dict[str, Any]] | None
+  shadow_config: AiConfigRep
+  shadow_meta: VariationMeta
+  shadow_request_id: str
+  span_attributes: dict[str, Any] = field(default_factory=dict)
+  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",
@@ -275,6 +280,7 @@
     # judges
     "build_judge_tasks",
     "run_judge",
+    "run_shadow",
     "run_judges",
     # client
     "config",

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 DEFAULT_SHADOW_TIMEOUT_S, 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,
 )
@@ -45,6 +49,34 @@ def _resolve_output_format_response(
     return parsed if parsed is not None else raw_response
 
 
+def _shadow_tasks(
+    *,
+    config_key: str,
+    variation: dict[str, Any],
+    meta: VariationMeta,
+    context: LDContext,
+    user_input: str | None,
+    history: list[dict[str, Any]] | None,
+    variables: dict[str, Any] | None,
+    handlers: list[ProviderHandler],
+    run_id: str,
+    timeout_s: float,
+) -> list[ShadowTask]:
+    return build_shadow_tasks(
+        config_key=config_key,
+        raw_variation=variation.get("raw"),
+        reason=variation.get("reason"),
+        primary_variation_key=str(meta.get("variationKey", "")),
+        context=context,
+        user_input=user_input,
+        history=history,
+        variables=variables,
+        handlers=handlers,
+        run_id=run_id,
+        timeout_s=timeout_s,
+    )
+
+
 class ConfigInstance:
     """Return type of ``config()``."""
 
@@ -55,12 +87,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 = DEFAULT_SHADOW_TIMEOUT_S,
     ) -> 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:
@@ -91,13 +129,28 @@ 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 = _shadow_tasks(
+            config_key=self._key,
+            variation=variation,
+            meta=meta,
+            context=context,
+            user_input=user_input,
+            history=history,
+            variables=variables,
+            handlers=resolved_handler_list,
+            run_id=run_id,
+            timeout_s=self._shadow_timeout_s,
+        )
+        if not self._skip_shadows:
+            start_shadows(shadow_tasks, resolved_handler_list)
 
         result = await execute_and_track(
             config_key=self._key,
@@ -109,6 +162,7 @@ async def _invoke(
             tool_handlers=resolved_tools,
             variables=variables,
             history=history,
+            run_id=run_id,
         )
 
         raw_response = result["response"]
@@ -140,6 +194,7 @@ async def _invoke(
                 response=parsed_response,
                 usage=usage_obj,
                 judge_tasks=judge_tasks,
+                shadow_tasks=shadow_tasks if self._skip_shadows else None,
                 track_data=track_data,
             )
 
@@ -158,6 +213,7 @@ async def _invoke(
             response=parsed_response,
             usage=usage_obj,
             judge_results=judge_results if judge_results else None,
+            shadow_tasks=shadow_tasks if self._skip_shadows else None,
             track_data=track_data,
         )
 
@@ -210,7 +266,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 +284,22 @@ async def _execute_stream(
         sink["tools"] = resolved_tools
         sink["provider"] = provider
         sink["model"] = model
+        run_id = str(uuid.uuid4())
+        shadow_tasks = _shadow_tasks(
+            config_key=self._key,
+            variation=variation,
+            meta=meta,
+            context=context,
+            user_input=user_input,
+            history=history,
+            variables=variables,
+            handlers=resolved_handler_list,
+            run_id=run_id,
+            timeout_s=self._shadow_timeout_s,
+        )
+        if not self._skip_shadows:
+            start_shadows(shadow_tasks, resolved_handler_list)
+        sink["shadow_tasks"] = shadow_tasks if self._skip_shadows else None
 
         async with aclosing(
             execute_and_stream(
@@ -240,6 +312,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 +339,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 +383,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 = DEFAULT_SHADOW_TIMEOUT_S,
 ) -> ConfigInstance:
     """
     Creates a ``ConfigInstance`` bound to *key*. Accepts a single handler or a
@@ -319,13 +397,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 attached to the matched rule or fallthrough run in the
+    background with *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 +421,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 = DEFAULT_SHADOW_TIMEOUT_S,
 ) -> ConfigInstance:
     """Non-reporting :func:`config` for SDK wrappers."""
     return _InternalConfigInstance(
@@ -344,4 +431,6 @@ def _config(
         tool_handlers=tool_handlers,
         registry=registry,
         skip_judges=skip_judges,
+        skip_shadows=skip_shadows,
+        shadow_timeout_s=shadow_timeout_s,
     )