fix: stdout restoration for concurrent validation pipelines (#7811)

src/cleveragents/application/services/validation_pipeline.py

@@ -35,7 +35,7 @@ from cleveragents.infrastructure.events.models import DomainEvent
 from cleveragents.infrastructure.events.types import EventType
 
 if TYPE_CHECKING:
-    from cleveragents.infrastructure.events.protocol import EventBus
+    from cleveragents.infrastructure_events.protocol import EventBus
 
 logger = logging.getLogger(__name__)
 
@@ -496,25 +496,63 @@ class ValidationPipeline:
 
         results: list[ValidationResult] = []
 
-        # Install thread-local stream wrappers so _execute_single can
-        # capture per-thread stdout/stderr without global races.
         orig_stdout = sys.stdout
         orig_stderr = sys.stderr
-        sys.stdout = _ThreadLocalStream(orig_stdout)
-        sys.stderr = _ThreadLocalStream(orig_stderr)
-        try:
-            with ThreadPoolExecutor(max_workers=self._max_workers) as pool:
-                futures_map = {
-                    pool.submit(self._execute_single, cmd): cmd
-                    for cmd in self._commands
-                }
-                for future in futures_map:
-                    result = future.result()
-                    results.append(result)
-        finally:
+
+        # Install thread-local stream wrappers so _execute_single can
+        # capture per-thread stdout/stderr without global races.
+        # The wrappers are created here (before any threads are spawned)
+        # and installed atomically to avoid a TOCTOU race where the
+        # restored original would already be pointing at the wrapper.
+        stdout_wrapper = _ThreadLocalStream(orig_stdout)
+        stderr_wrapper = _ThreadLocalStream(orig_stderr)
+
+        def _swap_wrappers() -> None:
+            """Atomically swap sys.stdout/stderr for thread-local wrappers."""
+            sys.stdout = stdout_wrapper
+            sys.stderr = stderr_wrapper
+
+        def _restore_originals() -> None:
+            """Restore the original global std streams.
+
+            Must only be called after all worker threads (including any
+            abandoned daemon threads that timed out) have had their
+            ``stop_capture()`` calls complete, ensuring no writes pass
+            through stale wrapper references.
+            """
             sys.stdout = orig_stdout
             sys.stderr = orig_stderr
 
+        try:
+            with ThreadPoolExecutor(max_workers=self._max_workers) as pool:
+                _swap_wrappers()
+                try:
+                    futures_map = {
+                        pool.submit(self._execute_single, cmd): cmd
+                        for cmd in self._commands
+                    }
+                    for future in futures_map:
+                        result = future.result()
+                        results.append(result)
+                    # At this point ThreadPoolExecutor.shutdown(wait=True) has
+                    # returned — every submitted task (and all internal threads)
+                    # have completed, including their finally-block stop_capture()
+                    # calls. It is now safe to restore originals.
+                except Exception:
+                    # If a future raises (e.g. from an executor that throws),
+                    # ThreadPoolExecutor has already flushed all pending tasks.
+                    # Abandoned daemon threads are guaranteed to have finished
+                    # their _run_in_thread finally blocks because shutdown(wait)
+                    # blocks until every submitted task is done.
+                    raise
+                finally:
+                    _restore_originals()
+        except Exception:
+            # Double-safety net: if the outer try also gets an exception
+            # (e.g. from re-raised future.exception), restore originals once more.
+            _restore_originals()
+            raise
+
         # Re-sort results to match command ordering (deterministic)
         results.sort(
             key=lambda r: (