fix: isolate per-execution log buffers for parallel eval runs

When UIPATH_JOB_KEY is set and eval runs execute in parallel, two bugs
caused jumbled and truncated log output in the execution log file:

1. **Interleaved output**: Each child interceptor replaced the global
   sys.stdout/sys.stderr with a new LoggerWriter, stomping the previous
   one. All concurrent async tasks wrote to the last-installed writer's
   single shared buffer, merging partial lines from different executions.

2. **Truncated output**: LoggerWriter.buffer was never flushed before
   teardown. Partial lines (without a trailing newline) were silently
   discarded when the handler was closed.

Fix:
- LoggerWriter now maintains per-context buffers keyed by
  current_execution_id, so concurrent tasks never share a buffer.
- Child interceptors no longer replace sys.stdout/sys.stderr — only the
  master interceptor owns the global streams. Children register their
  handler on the stdout/stderr loggers; the existing filter system
  routes records to the correct handler.
- Teardown flushes buffers before clearing context and removing handlers,
  so partial lines are never lost.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: smflorentino

src/uipath/runtime/logging/_interceptor.py

@@ -148,9 +148,16 @@ def setup(self) -> None:
                 # logger.propagate remains True (default)
                 self.patched_loggers.add(logger_name)
 
-            # Child executions should redirect stdout/stderr to their own handler
-            # This ensures print statements are captured per execution
-            self._redirect_stdout_stderr()
+            # Register our handler on stdout/stderr loggers so that
+            # print() output routed through the master's LoggerWriter
+            # is captured per-execution via filters.
+            # We do NOT replace sys.stdout/sys.stderr — the master owns those.
+            if not isinstance(sys.stdout, LoggerWriter):
+                self.logger.warning(
+                    "Child interceptor set up without a master LoggerWriter on sys.stdout. "
+                    "print() output will not be captured for this execution context."
+                )
+            self._register_stdout_stderr_handlers()
         else:
             # Master execution mode: remove all handlers and add only ours
             self._clean_all_handlers(self.root_logger)
@@ -165,28 +172,33 @@ def setup(self) -> None:
             # Master redirects stdout/stderr
             self._redirect_stdout_stderr()
 
+    def _register_stdout_stderr_handlers(self) -> None:
+        """Register our handler on stdout/stderr loggers without replacing the streams."""
+        stdout_logger = logging.getLogger("stdout")
+        stderr_logger = logging.getLogger("stderr")
+
+        stdout_logger.propagate = False
+        stderr_logger.propagate = False
+
+        if self.log_handler not in stdout_logger.handlers:
+            stdout_logger.addHandler(self.log_handler)
+        if self.log_handler not in stderr_logger.handlers:
+            stderr_logger.addHandler(self.log_handler)
+
     def _redirect_stdout_stderr(self) -> None:
-        """Redirect stdout and stderr to the logging system."""
-        # Set up stdout and stderr loggers
+        """Redirect stdout and stderr to the logging system.
+
+        Only called by master execution mode. Replaces sys.stdout/sys.stderr
+        with LoggerWriter instances that route output through the logging system.
+        """
         stdout_logger = logging.getLogger("stdout")
         stderr_logger = logging.getLogger("stderr")
 
-        if self.execution_id:
-            # Child execution: add our handler to stdout/stderr loggers
-            stdout_logger.propagate = False
-            stderr_logger.propagate = False
-
-            if self.log_handler not in stdout_logger.handlers:
-                stdout_logger.addHandler(self.log_handler)
-            if self.log_handler not in stderr_logger.handlers:
-                stderr_logger.addHandler(self.log_handler)
-        else:
-            # Master execution: clean and set up handlers
-            stdout_logger.propagate = False
-            stderr_logger.propagate = False
+        stdout_logger.propagate = False
+        stderr_logger.propagate = False
 
-            self._clean_all_handlers(stdout_logger)
-            self._clean_all_handlers(stderr_logger)
+        self._clean_all_handlers(stdout_logger)
+        self._clean_all_handlers(stderr_logger)
 
         # Use the min_level in the LoggerWriter to filter messages
         sys.stdout = LoggerWriter(
@@ -198,15 +210,29 @@ def _redirect_stdout_stderr(self) -> None:
 
     def teardown(self) -> None:
         """Restore original logging configuration."""
-        # Clear the context variable
+        # Step 1: Flush LoggerWriter buffers while context and handlers are still active.
+        # Child mode: flush only this context's buffer from the shared LoggerWriter.
+        # Master mode: flush ALL remaining buffers before restoring streams.
+        if self.execution_id:
+            if isinstance(sys.stdout, LoggerWriter):
+                sys.stdout.flush()
+            if isinstance(sys.stderr, LoggerWriter):
+                sys.stderr.flush()
+        else:
+            if isinstance(sys.stdout, LoggerWriter):
+                sys.stdout.flush_all()
+            if isinstance(sys.stderr, LoggerWriter):
+                sys.stderr.flush_all()
+
+        # Step 2: Clear the context variable (after flush used it)
         if self.execution_id:
             current_execution_id.set(None)
 
-        # Restore the original disable level
+        # Step 3: Restore the original disable level
         if not self.execution_id:
             logging.disable(self.original_disable_level)
 
-        # Remove our handler and filter
+        # Step 4: Remove our handler and filter
         if self.execution_filter:
             self.log_handler.removeFilter(self.execution_filter)
 
@@ -240,8 +266,8 @@ def teardown(self) -> None:
         if self._owns_handler:
             self.log_handler.close()
 
-        # Only restore streams if we redirected them
-        if self.original_stdout and self.original_stderr:
+        # Step 5: Only master restores streams (children never replaced them)
+        if not self.execution_id and self.original_stdout and self.original_stderr:
             sys.stdout = self.original_stdout
             sys.stderr = self.original_stderr
 

src/uipath/runtime/logging/_writers.py

@@ -3,9 +3,15 @@
 import logging
 from typing import TextIO
 
+from uipath.runtime.logging._context import current_execution_id
+
 
 class LoggerWriter:
-    """Redirect stdout/stderr to logging system."""
+    """Redirect stdout/stderr to logging system.
+
+    Maintains per-execution-context buffers so that concurrent async tasks
+    (e.g. parallel eval runs) do not interleave partial lines.
+    """
 
     def __init__(
         self,
@@ -18,7 +24,7 @@ def __init__(
         self.logger = logger
         self.level = level
         self.min_level = min_level
-        self.buffer = ""
+        self._buffers: dict[str | None, str] = {}
         self.sys_file = sys_file
         self._in_logging = False  # Recursion guard
 
@@ -35,17 +41,22 @@ def write(self, message: str) -> None:
 
         try:
             self._in_logging = True
-            self.buffer += message
-            while "\n" in self.buffer:
-                line, self.buffer = self.buffer.split("\n", 1)
+            ctx = current_execution_id.get()
+            buf = self._buffers.get(ctx, "") + message
+            while "\n" in buf:
+                line, buf = buf.split("\n", 1)
                 # Only log if the message is not empty and the level is sufficient
                 if line and self.level >= self.min_level:
                     self.logger._log(self.level, line, ())
+            if buf:
+                self._buffers[ctx] = buf
+            else:
+                self._buffers.pop(ctx, None)
         finally:
             self._in_logging = False
 
     def flush(self) -> None:
-        """Flush any remaining buffered messages to the logger."""
+        """Flush the current execution context's buffered messages to the logger."""
         if self._in_logging:
             if self.sys_file:
                 try:
@@ -56,10 +67,24 @@ def flush(self) -> None:
 
         try:
             self._in_logging = True
-            # Log any remaining content in the buffer on flush
-            if self.buffer and self.level >= self.min_level:
-                self.logger._log(self.level, self.buffer, ())
-            self.buffer = ""
+            ctx = current_execution_id.get()
+            buf = self._buffers.pop(ctx, "")
+            if buf and self.level >= self.min_level:
+                self.logger._log(self.level, buf, ())
+        finally:
+            self._in_logging = False
+
+    def flush_all(self) -> None:
+        """Flush all execution contexts' buffered messages. Called by master teardown."""
+        if self._in_logging:
+            return
+
+        try:
+            self._in_logging = True
+            for buf in self._buffers.values():
+                if buf and self.level >= self.min_level:
+                    self.logger._log(self.level, buf, ())
+            self._buffers.clear()
         finally:
             self._in_logging = False
 

tests/test_executor.py

@@ -1,5 +1,7 @@
 """Simple test for runtime factory and executor span capture."""
 
+import logging
+import sys
 from typing import Any, AsyncGenerator, TypeVar
 
 import pytest
@@ -13,6 +15,7 @@
     UiPathRuntimeProtocol,
 )
 from uipath.runtime.base import UiPathStreamOptions
+from uipath.runtime.logging._interceptor import UiPathRuntimeLogsInterceptor
 from uipath.runtime.result import UiPathRuntimeResult, UiPathRuntimeStatus
 from uipath.runtime.schema import UiPathRuntimeSchema
 
@@ -119,9 +122,32 @@ async def new_runtime(
         return self.runtime_class()
 
 
+@pytest.fixture(autouse=True)
+def _isolate_logging():
+    """Save and restore logging state so tests don't leak into each other."""
+    root = logging.getLogger()
+    original_level = root.level
+    original_handlers = list(root.handlers)
+    original_stdout = sys.stdout
+    original_stderr = sys.stderr
+    yield
+    root.setLevel(original_level)
+    root.handlers = original_handlers
+    sys.stdout = original_stdout
+    sys.stderr = original_stderr
+    logging.disable(logging.NOTSET)
+
+
 @pytest.mark.asyncio
-async def test_multiple_factories_same_executor():
+async def test_multiple_factories_same_executor(tmp_path):
     """Test factories using same trace manager, verify spans are captured correctly."""
+    # Set up a master interceptor so that sys.stdout is a LoggerWriter,
+    # matching real production usage where UiPathRuntimeContext provides one.
+    master = UiPathRuntimeLogsInterceptor(
+        job_id="test-job", dir=str(tmp_path), file="test.log"
+    )
+    master.setup()
+
     trace_manager = UiPathTraceManager()
 
     # Create factories for different runtimes
@@ -228,3 +254,5 @@ async def test_multiple_factories_same_executor():
     assert execution_runtime_c.log_handler
     assert len(execution_runtime_c.log_handler.buffer) > 0
     assert execution_runtime_c.log_handler.buffer[0].msg == "executing {'input': 'c'}"
+
+    master.teardown()