fix: eliminate fire-and-forget tasks and harden eval task tracking

- Redesign ManagedModel._track_judge_results to call evaluator.evaluate()
  internally and attach tracking via add_done_callback, returning the task
  so the reference is held by ModelResponse.evaluations — no GC risk
- Warn instead of silently dropping eval tasks when the LangGraph ContextVar
  is unexpectedly unset in a node's execution context
- Make AgentGraphDefinition.create_tracker a required parameter; all
  production and test call sites already supply it, and this matches the
  invariant that runners only execute on enabled (always-tracked) graphs

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

Author: jsonbailey

packages/ai-providers/server-ai-langchain/src/ldai_langchain/langgraph_agent_graph_runner.py

@@ -192,7 +192,14 @@ async def invoke(state: WorkflowState) -> dict:
                             response.content if hasattr(response, 'content') else str(response)
                         )
                         task = node_obj.get_config().evaluator.evaluate(input_text, output_text)
-                        _run_eval_tasks.get({}).setdefault(nk, []).append(task)
+                        run_tasks = _run_eval_tasks.get(None)
+                        if run_tasks is not None:
+                            run_tasks.setdefault(nk, []).append(task)
+                        else:
+                            log.warning(
+                                f"LangGraphAgentGraphRunner: eval task for node '{nk}' "
+                                "has no run context; judge results will not be tracked"
+                            )
 
                     return {'messages': [response]}
 

packages/ai-providers/server-ai-langchain/tests/test_langgraph_callback_handler.py

@@ -433,7 +433,7 @@ async def test_flush_with_no_graph_key_on_node_tracker():
         nodes=nodes,
         context=context,
         enabled=True,
-        create_tracker=lambda: None,
+        create_tracker=lambda: AIGraphTracker(mock_ld_client, 'v1', 'test-graph', 1, context),
     )
 
     handler = LDMetricsCallbackHandler({'root-agent'}, {})

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

@@ -52,7 +52,7 @@ def __init__(
         nodes: Dict[str, AgentGraphNode],
         context: Context,
         enabled: bool,
-        create_tracker: Optional[Callable[[], AIGraphTracker]] = None,
+        create_tracker: Callable[[], AIGraphTracker],
     ):
         self._agent_graph = agent_graph
         self._context = context

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

@@ -49,22 +49,32 @@ async def invoke(self, prompt: str) -> ModelResponse:
             lambda result: result.metrics,
         )
 
-        evaluator = self._ai_config.evaluator
         input_text = '\r\n'.join(m.content for m in self._messages) if self._messages else ''
         output_text = response.message.content
-        response.evaluations = evaluator.evaluate(input_text, output_text)
-        self._track_judge_results(tracker, response.evaluations)
+        response.evaluations = self._track_judge_results(tracker, input_text, output_text)
 
         self._messages.append(response.message)
         return response
 
-    def _track_judge_results(self, tracker: LDAIConfigTracker, eval_task: asyncio.Task[List[JudgeResult]]) -> None:
-        async def _run() -> None:
-            results = await eval_task
-            for r in results:
+    def _track_judge_results(
+        self,
+        tracker: LDAIConfigTracker,
+        input_text: str,
+        output_text: str,
+    ) -> asyncio.Task[List[JudgeResult]]:
+        eval_task = self._ai_config.evaluator.evaluate(input_text, output_text)
+
+        def _on_done(task: asyncio.Task) -> None:
+            if task.cancelled():
+                return
+            if task.exception() is not None:
+                return
+            for r in task.result():
                 if r.success:
                     tracker.track_judge_result(r)
-        asyncio.create_task(_run())
+
+        eval_task.add_done_callback(_on_done)
+        return eval_task
 
     def get_messages(self, include_config_messages: bool = False) -> List[LDMessage]:
         """