feat!: Add per-execution runId, at-most-once tracking, and cross-process tracker resumption

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: jsonbailey

packages/sdk/server-ai/src/ldai/client.py

@@ -1,4 +1,6 @@
-from typing import Any, Dict, List, Optional, Tuple
+import base64
+import json
+from typing import Any, Callable, Dict, List, Optional, Tuple
 
 import chevron
 from ldclient import Context
@@ -61,14 +63,53 @@ def __init__(self, client: LDClient):
             1,
         )
 
+    def create_tracker(self, token: str, context: Context) -> LDAIConfigTracker:
+        """
+        Reconstruct a tracker from a resumption token.
+
+        This is used for cross-process scenarios such as deferred feedback,
+        where a different service needs to associate tracking events with the
+        original execution's ``runId``.
+
+        :param token: A URL-safe Base64-encoded resumption token obtained from
+            :attr:`LDAIConfigTracker.resumption_token`.
+        :param context: The context to use for track events.
+        :return: A new :class:`LDAIConfigTracker` bound to the original
+            ``runId`` from the token.
+        :raises ValueError: If the token is invalid or missing required fields.
+        """
+        try:
+            # Add padding back before decoding
+            padded = token + "=" * (-len(token) % 4)
+            payload = json.loads(
+                base64.urlsafe_b64decode(padded.encode("utf-8")).decode("utf-8")
+            )
+        except (json.JSONDecodeError, Exception) as e:
+            raise ValueError(f"Invalid resumption token: {e}") from e
+
+        for field in ("runId", "configKey", "version"):
+            if field not in payload:
+                raise ValueError(f"Invalid resumption token: missing required field '{field}'")
+
+        return LDAIConfigTracker(
+            ld_client=self._client,
+            variation_key=payload.get("variationKey", ""),
+            config_key=payload["configKey"],
+            version=payload["version"],
+            model_name="",
+            provider_name="",
+            context=context,
+            run_id=payload["runId"],
+        )
+
     def _completion_config(
         self,
         key: str,
         context: Context,
         default: AICompletionConfigDefault,
         variables: Optional[Dict[str, Any]] = None,
     ) -> AICompletionConfig:
-        model, provider, messages, instructions, tracker, enabled, judge_configuration, _ = self.__evaluate(
+        model, provider, messages, instructions, tracker, tracker_factory, enabled, judge_configuration, _ = self.__evaluate(
             key, context, default.to_dict(), variables
         )
 
@@ -79,6 +120,7 @@ def _completion_config(
             messages=messages,
             provider=provider,
             tracker=tracker,
+            create_tracker=tracker_factory if enabled else None,
             judge_configuration=judge_configuration,
         )
 
@@ -134,7 +176,7 @@ def _judge_config(
         default: AIJudgeConfigDefault,
         variables: Optional[Dict[str, Any]] = None,
     ) -> AIJudgeConfig:
-        model, provider, messages, instructions, tracker, enabled, judge_configuration, variation = self.__evaluate(
+        model, provider, messages, instructions, tracker, tracker_factory, enabled, judge_configuration, variation = self.__evaluate(
             key, context, default.to_dict(), variables
         )
 
@@ -163,6 +205,7 @@ def _extract_evaluation_metric_key(variation: Dict[str, Any]) -> Optional[str]:
             messages=messages,
             provider=provider,
             tracker=tracker,
+            create_tracker=tracker_factory if enabled else None,
         )
 
         return config
@@ -248,14 +291,14 @@ async def create_judge(
                 key, context, default or AIJudgeConfigDefault.disabled(), extended_variables
             )
 
-            if not judge_config.enabled or not judge_config.tracker:
+            if not judge_config.enabled or not judge_config.create_tracker:
                 return None
 
             provider = RunnerFactory.create_model(judge_config, default_ai_provider)
             if not provider:
                 return None
 
-            return Judge(judge_config, judge_config.tracker, provider)
+            return Judge(judge_config, judge_config.create_tracker(), provider)
         except Exception as error:
             return None
 
@@ -345,7 +388,7 @@ async def create_model(
         log.debug(f"Creating managed model for key: {key}")
         config = self._completion_config(key, context, default or AICompletionConfigDefault.disabled(), variables)
 
-        if not config.enabled or not config.tracker:
+        if not config.enabled or not config.create_tracker:
             return None
 
         runner = RunnerFactory.create_model(config, default_ai_provider)
@@ -361,7 +404,7 @@ async def create_model(
                 default_ai_provider,
             )
 
-        return ManagedModel(config, config.tracker, runner, judges)
+        return ManagedModel(config, config.create_tracker(), runner, judges)
 
     async def create_chat(
         self,
@@ -428,14 +471,14 @@ async def create_agent(
         log.debug(f"Creating managed agent for key: {key}")
         config = self.__evaluate_agent(key, context, default or AIAgentConfigDefault.disabled(), variables)
 
-        if not config.enabled or not config.tracker:
+        if not config.enabled or not config.create_tracker:
             return None
 
         runner = RunnerFactory.create_agent(config, tools or {}, default_ai_provider)
         if not runner:
             return None
 
-        return ManagedAgent(config, config.tracker, runner)
+        return ManagedAgent(config, config.create_tracker(), runner)
 
     def agent_config(
         self,
@@ -750,7 +793,7 @@ def __evaluate(
         variables: Optional[Dict[str, Any]] = None,
     ) -> Tuple[
         Optional[ModelConfig], Optional[ProviderConfig], Optional[List[LDMessage]],
-        Optional[str], LDAIConfigTracker, bool, Optional[Any], Dict[str, Any]
+        Optional[str], LDAIConfigTracker, Callable[[], LDAIConfigTracker], bool, Optional[Any], Dict[str, Any]
     ]:
         """
         Internal method to evaluate a configuration and extract components.
@@ -801,15 +844,23 @@ def __evaluate(
                 custom=custom
             )
 
-        tracker = LDAIConfigTracker(
-            self._client,
-            variation.get('_ldMeta', {}).get('variationKey', ''),
-            key,
-            int(variation.get('_ldMeta', {}).get('version', 1)),
-            model.name if model else '',
-            provider_config.name if provider_config else '',
-            context,
-        )
+        variation_key = variation.get('_ldMeta', {}).get('variationKey', '')
+        version = int(variation.get('_ldMeta', {}).get('version', 1))
+        model_name = model.name if model else ''
+        provider_name = provider_config.name if provider_config else ''
+
+        def tracker_factory() -> LDAIConfigTracker:
+            return LDAIConfigTracker(
+                self._client,
+                variation_key,
+                key,
+                version,
+                model_name,
+                provider_name,
+                context,
+            )
+
+        tracker = tracker_factory()
 
         enabled = variation.get('_ldMeta', {}).get('enabled', False)
 
@@ -828,7 +879,7 @@ def __evaluate(
                 if judges:
                     judge_configuration = JudgeConfiguration(judges=judges)
 
-        return model, provider_config, messages, instructions, tracker, enabled, judge_configuration, variation
+        return model, provider_config, messages, instructions, tracker, tracker_factory, enabled, judge_configuration, variation
 
     def __evaluate_agent(
         self,
@@ -846,7 +897,7 @@ def __evaluate_agent(
         :param variables: Variables for interpolation.
         :return: Configured AIAgentConfig instance.
         """
-        model, provider, messages, instructions, tracker, enabled, judge_configuration, _ = self.__evaluate(
+        model, provider, messages, instructions, tracker, tracker_factory, enabled, judge_configuration, _ = self.__evaluate(
             key, context, default.to_dict(), variables
         )
 
@@ -860,6 +911,7 @@ def __evaluate_agent(
             provider=provider or default.provider,
             instructions=final_instructions,
             tracker=tracker,
+            create_tracker=tracker_factory if enabled else None,
             judge_configuration=judge_configuration or default.judge_configuration,
         )
 

packages/sdk/server-ai/src/ldai/judge/__init__.py

@@ -68,10 +68,12 @@ async def evaluate(
                 log.debug(f'Judge evaluation skipped due to sampling rate: {sampling_rate}')
                 return None
 
+            tracker = self._ai_config.create_tracker() if self._ai_config.create_tracker else self._ai_config_tracker
+
             messages = self._construct_evaluation_messages(input_text, output_text)
             assert self._evaluation_response_structure is not None
 
-            response = await self._ai_config_tracker.track_metrics_of_async(
+            response = await tracker.track_metrics_of_async(
                 lambda: self._model_runner.invoke_structured_model(messages, self._evaluation_response_structure),
                 lambda result: result.metrics,
             )

packages/sdk/server-ai/src/ldai/managed_agent.py

@@ -30,7 +30,8 @@ async def run(self, input: str) -> AgentResult:
         :param input: The user prompt or input to the agent
         :return: AgentResult containing the agent's output and metrics
         """
-        return await self._tracker.track_metrics_of_async(
+        tracker = self._ai_config.create_tracker() if self._ai_config.create_tracker else self._tracker
+        return await tracker.track_metrics_of_async(
             lambda: self._agent_runner.run(input),
             lambda result: result.metrics,
         )

packages/sdk/server-ai/src/ldai/managed_model.py

@@ -42,13 +42,15 @@ async def invoke(self, prompt: str) -> ModelResponse:
         :param prompt: The user prompt to send to the model
         :return: ModelResponse containing the model's response and metrics
         """
+        tracker = self._ai_config.create_tracker() if self._ai_config.create_tracker else self._tracker
+
         user_message = LDMessage(role='user', content=prompt)
         self._messages.append(user_message)
 
         config_messages = self._ai_config.messages or []
         all_messages = config_messages + self._messages
 
-        response = await self._tracker.track_metrics_of_async(
+        response = await tracker.track_metrics_of_async(
             lambda: self._model_runner.invoke_model(all_messages),
             lambda result: result.metrics,
         )
@@ -57,13 +59,14 @@ async def invoke(self, prompt: str) -> ModelResponse:
             self._ai_config.judge_configuration
             and self._ai_config.judge_configuration.judges
         ):
-            response.evaluations = self._start_judge_evaluations(self._messages, response)
+            response.evaluations = self._start_judge_evaluations(tracker, self._messages, response)
 
         self._messages.append(response.message)
         return response
 
     def _start_judge_evaluations(
         self,
+        tracker: LDAIConfigTracker,
         messages: List[LDMessage],
         response: ModelResponse,
     ) -> List[asyncio.Task[Optional[JudgeResponse]]]:
@@ -77,7 +80,7 @@ async def evaluate_judge(judge_config: Any) -> Optional[JudgeResponse]:
                 return None
             eval_result = await judge.evaluate_messages(messages, response, judge_config.sampling_rate)
             if eval_result and eval_result.success:
-                self._tracker.track_judge_response(eval_result)
+                tracker.track_judge_response(eval_result)
             return eval_result
 
         return [

packages/sdk/server-ai/src/ldai/models.py

@@ -1,6 +1,6 @@
 import warnings
 from dataclasses import dataclass, field
-from typing import Any, Dict, List, Literal, Optional, Union
+from typing import Any, Callable, Dict, List, Literal, Optional, Union
 
 from ldai.tracker import LDAIConfigTracker
 
@@ -183,6 +183,7 @@ class AIConfig:
     model: Optional[ModelConfig] = None
     provider: Optional[ProviderConfig] = None
     tracker: Optional[LDAIConfigTracker] = None
+    create_tracker: Optional[Callable[[], LDAIConfigTracker]] = None
 
     def _base_to_dict(self) -> Dict[str, Any]:
         """

packages/sdk/server-ai/src/ldai/tracker.py

@@ -1,3 +1,5 @@
+import base64
+import json
 import logging
 import time
 import uuid
@@ -81,6 +83,7 @@ def __init__(
         model_name: str,
         provider_name: str,
         context: Context,
+        run_id: Optional[str] = None,
     ):
         """
         Initialize an AI Config tracker.
@@ -92,6 +95,7 @@ def __init__(
         :param model_name: Name of the model used.
         :param provider_name: Name of the provider used.
         :param context: Context for evaluation.
+        :param run_id: Optional run ID. When not provided, a new UUID is generated.
         """
         self._ld_client = ld_client
         self._variation_key = variation_key
@@ -101,9 +105,26 @@ def __init__(
         self._provider_name = provider_name
         self._context = context
         self._summary = LDAIMetricSummary()
-        self._run_id = str(uuid.uuid4())
+        self._run_id = run_id or str(uuid.uuid4())
         self._tracked: Dict[str, bool] = {}
 
+    @property
+    def resumption_token(self) -> str:
+        """
+        A URL-safe Base64-encoded JSON string that can be used to reconstruct
+        a tracker in a different process (e.g. for deferred feedback).
+
+        The token contains ``runId``, ``configKey``, ``variationKey``, and
+        ``version``.  ``modelName`` and ``providerName`` are **not** included.
+        """
+        payload = json.dumps({
+            "runId": self._run_id,
+            "configKey": self._config_key,
+            "variationKey": self._variation_key,
+            "version": self._version,
+        })
+        return base64.urlsafe_b64encode(payload.encode("utf-8")).rstrip(b"=").decode("utf-8")
+
     def __get_track_data(self, graph_key: Optional[str] = None) -> dict:
         """
         Get tracking data for events.

packages/sdk/server-ai/tests/test_managed_agent.py

@@ -55,6 +55,7 @@ class TestManagedAgentRun:
     async def test_run_delegates_to_agent_runner(self):
         """Should delegate run() to the underlying AgentRunner."""
         mock_config = MagicMock(spec=AIAgentConfig)
+        mock_config.create_tracker = None  # fall back to passed tracker
         mock_tracker = MagicMock()
         mock_tracker.track_metrics_of_async = AsyncMock(
             return_value=AgentResult(
@@ -79,6 +80,31 @@ async def test_run_delegates_to_agent_runner(self):
         assert result.metrics.success is True
         mock_tracker.track_metrics_of_async.assert_called_once()
 
+    @pytest.mark.asyncio
+    async def test_run_uses_create_tracker_when_available(self):
+        """Should use create_tracker() factory for a fresh tracker per invocation."""
+        mock_config = MagicMock(spec=AIAgentConfig)
+        fresh_tracker = MagicMock()
+        fresh_tracker.track_metrics_of_async = AsyncMock(
+            return_value=AgentResult(
+                output="Fresh tracker response",
+                raw=None,
+                metrics=LDAIMetrics(success=True, usage=None),
+            )
+        )
+        mock_config.create_tracker = MagicMock(return_value=fresh_tracker)
+
+        old_tracker = MagicMock()
+        mock_runner = MagicMock()
+
+        agent = ManagedAgent(mock_config, old_tracker, mock_runner)
+        result = await agent.run("Hello")
+
+        assert result.output == "Fresh tracker response"
+        mock_config.create_tracker.assert_called_once()
+        fresh_tracker.track_metrics_of_async.assert_called_once()
+        old_tracker.track_metrics_of_async.assert_not_called()
+
     def test_get_agent_runner_returns_runner(self):
         """Should return the underlying AgentRunner."""
         mock_runner = MagicMock()