Test/uv 16309 span flush race + fix

packages/uipath/src/uipath/eval/evaluators/legacy_trajectory_evaluator.py

@@ -101,6 +101,12 @@ async def evaluate(
         evaluation_prompt = self._create_evaluation_prompt(
             expected_agent_behavior=workload_execution.expected_agent_behavior,
             agent_run_history=workload_execution.workload_trace,
+            # Route through the base class's resolution (target_output_key
+            # extraction, job-attachment URI download) instead of the raw
+            # field - otherwise a trajectory evaluator configured with a
+            # specific target_output_key would get the whole output object
+            # (or an unresolved attachment URI) in AgentRunHistory.
+            workload_output=self._get_actual_output(workload_execution),
         )
         llm_response = await self._get_llm_response(evaluation_prompt)
 
@@ -113,6 +119,7 @@ def _create_evaluation_prompt(
         self,
         expected_agent_behavior: Any,
         agent_run_history: Any,
+        workload_output: Any = None,
     ) -> str:
         """Create the evaluation prompt for the LLM."""
         # Validate that expected agent behavior is not empty
@@ -143,6 +150,18 @@ def _create_evaluation_prompt(
         else:
             agent_run_history = str(agent_run_history)
 
+        # trace_to_str only ever renders tool-call spans, so a tool-free
+        # text-only response leaves agent_run_history empty even though the
+        # agent genuinely answered. Append the actual final output so the
+        # judge always sees what the agent answered (UV-16309).
+        if workload_output:
+            final_output_section = f"Agent Final Response:\n{workload_output}"
+            agent_run_history = (
+                f"{agent_run_history}\n\n{final_output_section}"
+                if agent_run_history
+                else final_output_section
+            )
+
         formatted_prompt = formatted_prompt.replace(
             self.agent_run_history_placeholder,
             agent_run_history,

packages/uipath/src/uipath/eval/evaluators/llm_judge_trajectory_evaluator.py

@@ -79,8 +79,25 @@ async def evaluate(
         return await super().evaluate(workload_execution, evaluation_criteria)
 
     def _get_actual_output(self, workload_execution: WorkloadExecution) -> Any:
-        """Get the actual output from the workload execution."""
-        return trace_to_str(workload_execution.workload_trace)
+        """Get the actual output from the workload execution.
+
+        `trace_to_str` only ever renders tool-call spans, so a run where the
+        agent responded with plain text and made no tool calls at all
+        produces an empty string here - not because nothing happened, but
+        because there was nothing for `trace_to_str` to render. Append the
+        agent's actual final output so the judge always sees what the agent
+        answered, not just which tools it used (UV-16309).
+        """
+        history = trace_to_str(workload_execution.workload_trace)
+        final_output = workload_execution.workload_output
+        if final_output:
+            final_output_section = f"Agent Final Response:\n{final_output}"
+            history = (
+                f"{history}\n\n{final_output_section}"
+                if history
+                else final_output_section
+            )
+        return history
 
     def _get_expected_output(
         self, evaluation_criteria: TrajectoryEvaluationCriteria

packages/uipath/src/uipath/eval/runtime/runtime.py

@@ -970,6 +970,17 @@ async def _generate_input_for_eval(
     def _get_and_clear_execution_data(
         self, execution_id: str
     ) -> tuple[list[ReadableSpan], list[logging.LogRecord]]:
+        # Force a flush right before reading: the flush triggered by the root
+        # execution span ending only exports spans whose on_end() had already
+        # fired by that moment. A tool-call span that finishes a beat later
+        # (e.g. a short-lived background task) would otherwise still be
+        # sitting in the batch processor's queue and be silently missing from
+        # AgentRunHistory (UV-16309). This is a snapshot barrier, not a full
+        # closure: a span belonging to a task that is still running (hasn't
+        # called on_end() yet) when this flush runs is still missed - closing
+        # that would require the runtime to track and await such tasks before
+        # this point, which is out of scope for this fix.
+        self.trace_manager.flush_spans()
         spans = self.span_exporter.get_spans(execution_id)
         self.span_exporter.clear(execution_id)
         self.span_collector.clear(execution_id)

packages/uipath/tests/cli/eval/test_execution_span_race.py

@@ -0,0 +1,240 @@
+"""Reproduces the UV-16309 intermittent-empty-AgentRunHistory race.
+
+`ExecutionSpanProcessor` (eval/runtime/_exporters.py) is a real OTel
+`BatchSpanProcessor`: `on_end()` only queues a span, it does not export it.
+`ExecutionSpanExporter.get_spans()` (read by `trace_to_str()` to build
+`AgentRunHistory`) only sees spans that have already reached `export()`.
+
+Production flushes the queue via `UiPathTraceManager.flush_spans()` (called
+from `uipath-runtime`'s `UiPathExecutionRuntime.execute()` finally block, and
+from `start_execution_span`'s finally block) right as the root execution span
+ends. That flush only catches spans whose `on_end()` already fired by that
+moment. A tool-call span that legitimately ends *after* the root span (e.g. a
+background/detached task finishing a beat late) is queued but not yet
+exported, so a read of `get_spans()` taken between "root span ended" and
+"next flush" sees a real, completed tool call as if it never happened -
+exactly the "history genuinely exists but AgentRunHistory is empty,
+intermittently" symptom.
+
+The fix in `_get_and_clear_execution_data()` adds one more flush right before
+the read, which closes the race for any span that has *already ended* by
+read time - the common case, e.g. a tool call wrapped in a short-lived
+background task. It does NOT close the race for a span belonging to a truly
+detached task that is still running (has no `on_end()` yet) when the read
+happens - there is no handle for the runtime to await in that case, and
+closing it would require tracking/awaiting such tasks, which is a separate,
+larger change than this one-line flush.
+"""
+
+import json
+
+from opentelemetry import trace
+from opentelemetry.sdk.trace import TracerProvider
+
+from uipath.core.tracing import UiPathTraceManager
+from uipath.eval._execution_context import ExecutionSpanCollector
+from uipath.eval._helpers.evaluators_helpers import trace_to_str
+from uipath.eval.runtime._exporters import (
+    ExecutionLogsExporter,
+    ExecutionSpanExporter,
+    ExecutionSpanProcessor,
+)
+from uipath.eval.runtime.runtime import UiPathEvalRuntime
+
+EXECUTION_ID = "exec-race-1"
+
+
+def _pin_batch_schedule(processor: ExecutionSpanProcessor) -> None:
+    """Stop the processor's own background worker from auto-exporting.
+
+    `ExecutionSpanProcessor` is a real `BatchSpanProcessor` with its own
+    background worker thread, which auto-flushes every 5s by default (or
+    when its queue fills). Tests asserting "not yet exported at this exact
+    moment" would otherwise flake if that worker thread woke up and exported
+    on its own between a span ending and the assertion running. Pin the
+    schedule delay far beyond any test's runtime so only an explicit
+    `force_flush()`/`flush_spans()` call ever exports anything.
+    """
+    batch_processor = processor._batch_processor
+    batch_processor._schedule_delay_millis = 3600_000
+    batch_processor._schedule_delay = 3600.0
+
+
+def _make_processor() -> tuple[
+    ExecutionSpanProcessor, ExecutionSpanExporter, trace.Tracer
+]:
+    exporter = ExecutionSpanExporter()
+    collector = ExecutionSpanCollector()
+    processor = ExecutionSpanProcessor(exporter, collector)
+    _pin_batch_schedule(processor)
+
+    provider = TracerProvider()
+    provider.add_span_processor(processor)
+    tracer = provider.get_tracer("test")
+
+    return processor, exporter, tracer
+
+
+def test_flush_only_exports_spans_ended_before_it_runs() -> None:
+    """A late-ending tool span is invisible to get_spans() until the *next* flush."""
+    processor, exporter, tracer = _make_processor()
+
+    # Root execution span starts and ends (mirrors start_execution_span's `with` block).
+    with tracer.start_as_current_span(
+        "root", attributes={"execution.id": EXECUTION_ID}
+    ):
+        pass
+
+    # Mirrors flush_spans() firing right as the root span's context manager exits.
+    processor.force_flush()
+
+    assert [s.name for s in exporter.get_spans(EXECUTION_ID)] == ["root"]
+
+    # A tool-call span that genuinely happened, but whose on_end() only fires
+    # *after* the flush above - e.g. a detached/background task that outlives
+    # the awaited delegate.execute() call.
+    with tracer.start_as_current_span(
+        "tool_call", attributes={"execution.id": EXECUTION_ID, "tool.name": "search"}
+    ):
+        pass
+
+    # This is the moment _get_and_clear_execution_data() reads spans in
+    # eval/runtime/runtime.py: no flush has happened since the tool span ended.
+    spans_at_read_time = exporter.get_spans(EXECUTION_ID)
+
+    assert [s.name for s in spans_at_read_time] == ["root"], (
+        "the tool-call span genuinely ended but is not yet exported - "
+        "trace_to_str() would build an AgentRunHistory missing this real tool call"
+    )
+
+    # A subsequent flush (e.g. one added right before the read) makes it visible.
+    processor.force_flush()
+    assert {s.name for s in exporter.get_spans(EXECUTION_ID)} == {"root", "tool_call"}
+
+
+def test_flush_immediately_before_read_closes_the_race() -> None:
+    """Proposed fix: force_flush() right before get_spans() sees every ended span."""
+    processor, exporter, tracer = _make_processor()
+
+    with tracer.start_as_current_span(
+        "root", attributes={"execution.id": EXECUTION_ID}
+    ):
+        with tracer.start_as_current_span(
+            "tool_call",
+            attributes={"execution.id": EXECUTION_ID, "tool.name": "search"},
+        ):
+            pass
+
+    # A flush right before the read (unlike production today) picks up
+    # everything that has ended by then, including the tool call.
+    processor.force_flush()
+
+    spans_at_read_time = exporter.get_spans(EXECUTION_ID)
+    assert {s.name for s in spans_at_read_time} == {"root", "tool_call"}
+
+
+def test_race_produces_empty_agent_run_history_for_a_real_tool_call() -> None:
+    """Ties the exporter-level race to the actual UV-16309 symptom.
+
+    trace_to_str() (used to build AgentRunHistory for the trajectory/LLM-judge
+    evaluators) is handed exactly what get_spans() returns. If the read happens
+    in the window between the root-span flush and the late tool span's own
+    flush, the genuinely-completed tool call is silently dropped from the
+    evaluator prompt - not because it didn't happen, but because it wasn't
+    exported yet.
+    """
+    processor, exporter, tracer = _make_processor()
+
+    with tracer.start_as_current_span(
+        "root", attributes={"execution.id": EXECUTION_ID}
+    ):
+        pass
+    processor.force_flush()
+
+    with tracer.start_as_current_span(
+        "tool_call",
+        attributes={
+            "execution.id": EXECUTION_ID,
+            "tool.name": "search",
+            # input.value/output.value are OTel span attributes: they must be
+            # primitive/sequence-of-primitive values, so real spans always
+            # carry a JSON-encoded string here, never a raw dict.
+            "input.value": json.dumps({"query": "uipath"}),
+            "output.value": "42 results",
+        },
+    ):
+        pass
+
+    # No flush here - mirrors _get_and_clear_execution_data() reading
+    # immediately after the delegate returns, with nothing forcing the
+    # late-ending tool span's export first.
+    agent_run_history = trace_to_str(exporter.get_spans(EXECUTION_ID))
+    assert agent_run_history == "", (
+        "expected the unflushed tool call to be missing from AgentRunHistory, "
+        "reproducing UV-16309's 'history genuinely exists but comes back empty' case"
+    )
+
+    # With the extra flush (the proposed fix) in place, the same real tool
+    # call is captured correctly.
+    processor.force_flush()
+    agent_run_history_after_flush = trace_to_str(exporter.get_spans(EXECUTION_ID))
+    assert "Tool: search" in agent_run_history_after_flush
+    assert "42 results" in agent_run_history_after_flush
+
+
+def test_get_and_clear_execution_data_flushes_before_reading() -> None:
+    """Exercises the actual production method, not just a standalone processor.
+
+    Builds the same trace_manager/span_exporter/span_collector/logs_exporter
+    wiring UiPathEvalRuntime.__init__ sets up, then calls the real
+    `_get_and_clear_execution_data` (unbound, via the class) against it. This
+    fails without the `flush_spans()` line in that method - removing that
+    line reproduces the empty-AgentRunHistory bug here directly, not just in
+    the lower-level exporter tests above.
+    """
+    # UiPathTraceManager() registers a default batch processor on the global
+    # delegating processor and doesn't unregister it on its own - shut it
+    # down in `finally` so its processors (and their background worker
+    # threads) don't leak into other tests/jobs.
+    trace_manager = UiPathTraceManager()
+    try:
+        span_exporter = ExecutionSpanExporter()
+        span_collector = ExecutionSpanCollector()
+        span_processor = ExecutionSpanProcessor(span_exporter, span_collector)
+        _pin_batch_schedule(span_processor)
+        trace_manager.add_span_processor(span_processor)
+        logs_exporter = ExecutionLogsExporter()
+
+        fake_runtime = type(
+            "FakeEvalRuntime",
+            (),
+            {
+                "trace_manager": trace_manager,
+                "span_exporter": span_exporter,
+                "span_collector": span_collector,
+                "logs_exporter": logs_exporter,
+            },
+        )()
+
+        tracer = trace_manager.tracer_provider.get_tracer("test")
+        with tracer.start_as_current_span(
+            "root", attributes={"execution.id": EXECUTION_ID}
+        ):
+            pass
+        # Root span's own flush (mirrors start_execution_span's finally block)
+        # happens before the late tool call below - only the tool call is at risk.
+        trace_manager.flush_spans()
+
+        with tracer.start_as_current_span(
+            "tool_call",
+            attributes={"execution.id": EXECUTION_ID, "tool.name": "search"},
+        ):
+            pass
+
+        spans, _logs = UiPathEvalRuntime._get_and_clear_execution_data(
+            fake_runtime, EXECUTION_ID
+        )
+
+        assert {s.name for s in spans} == {"root", "tool_call"}
+    finally:
+        trace_manager.shutdown()