fix(concurrency): fix ValidationPipeline.run() sys.stdout replacement #7623

Introduce a reference-counted shared stream wrapper manager so concurrent
ValidationPipeline.run() calls correctly restore the true sys.stdout/stderr
after all pipelines finish.

- Add _install_thread_local_streams() / _release_thread_local_streams()
  protected by threading.Lock; first caller saves originals and installs
  wrappers, last caller restores them
- Remove io.TextIOBase inheritance from _ThreadLocalStream to avoid Python
  3.13 read-only slot conflict; use cast(TextIO, ...) at assignment sites
- encoding property returns str | None matching io.TextIOBase signature
- Replace assert guards with explicit RuntimeError fail-fast checks
- Add Behave scenario: Concurrent pipelines restore global streams after
  execution
- Split validation_pipeline_steps.py (547→428 lines) by extracting
  MockValidationExecutor to _validation_pipeline_mock.py and edge-case
  steps to validation_pipeline_edge_steps.py
- Update CHANGELOG.md and CONTRIBUTORS.md

ISSUES CLOSED: #7623

CHANGELOG.md

@@ -15,6 +15,20 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   versions (<3.13.4) cannot be installed even if upstream transitive dependencies have loose
   version constraints.
 
+### Fixed
+
+- **Concurrent ValidationPipeline stdout/stderr restoration** (#7623): Fixed a race
+  condition where concurrent `ValidationPipeline.run()` calls could permanently leave
+  `sys.stdout` and `sys.stderr` wrapped in `_ThreadLocalStream` objects. Pipeline B
+  could capture Pipeline A's wrapper as its "original" stream, then restore that wrapper
+  in its `finally` block — permanently leaving the global streams wrapped. The fix
+  introduces a reference-counted shared wrapper manager (`_install_thread_local_streams`
+  / `_release_thread_local_streams`) protected by a `threading.Lock`. The first caller
+  saves the true original streams and installs the wrappers; subsequent concurrent callers
+  increment the reference count and reuse the same wrappers; the last caller restores the
+  saved originals. Per-pipeline stdout/stderr capture remains fully isolated via
+  thread-local buffers.
+
 ### Changed
 
 - Restored `benchmark-regression` CI job to `master.yml` with `pull_request` trigger guard

CONTRIBUTORS.md

@@ -29,3 +29,4 @@ Below are some of the specific details of various contributions.
 * HAL 9000 has contributed the git worktree TOCTOU race condition fix (PR #8178 / issue #7507): replaced the unsafe mkdtemp() + rmdir() pattern with a parent-directory approach to eliminate the race window in concurrent git worktree operations.
 * HAL 9000 has contributed the git_tools TOCTOU race condition fix (PR #8255 / issue #7619): eliminated the Time-Of-Check-To-Time-Of-Use race in `_get_base_env()` by adding double-checked locking with a module-level `threading.Lock`, preventing concurrent threads from writing conflicting environment snapshots.
 * HAL 9000 has contributed comprehensive milestone documentation for v3.6.0 (Advanced Concepts & Deferred Features) and v3.7.0 (TUI Implementation) (PR #9903): split into sub-documents covering context strategies, LLM backends, resource types, A2A rename, container tool execution, scope chain resolution, cost/safety budgets, E2E workflow tests, code review examples, plugin architecture, TUI layout, persona system, reference/command input, session management, configuration, and TuiMaterializer integration.
+* HAL 9000 has contributed the concurrent ValidationPipeline stdout/stderr restoration fix (PR #7811 / issue #7623): introduced a reference-counted shared stream wrapper manager so concurrent ValidationPipeline.run() calls correctly restore the true sys.stdout/sys.stderr after all pipelines finish, preventing permanent stream wrapping under concurrent execution.

features/steps/_validation_pipeline_mock.py

@@ -0,0 +1,106 @@
+"""Shared mock executor for validation pipeline step definitions."""
+
+from __future__ import annotations
+
+import sys
+import time
+from collections.abc import Callable
+from typing import Any
+
+# Capture the real time.sleep at import time so that the fast-sleep patch
+# installed by features/environment.py does not affect timeout scenarios.
+# The fast-sleep patch stores the original as time._original_sleep; fall
+# back to time.sleep if the patch has not been applied yet.
+_REAL_SLEEP: Callable[[float], None] = getattr(time, "_original_sleep", time.sleep)
+
+
+class MockValidationExecutor:
+    """Configurable mock executor for testing the validation pipeline."""
+
+    def __init__(self) -> None:
+        self._results: dict[str, dict[str, Any]] = {}
+        self._timeout_names: set[str] = set()
+        self._exception_names: dict[str, Exception] = {}
+        self._non_dict_names: set[str] = set()
+        self._missing_passed_names: set[str] = set()
+        self._default_passed: bool = False
+        self._stdout_names: set[str] = set()
+        self._stderr_names: set[str] = set()
+
+    def set_passed(self, name: str) -> None:
+        self._results[name] = {"passed": True, "message": f"{name} passed"}
+
+    def set_passed_with_data(self, name: str) -> None:
+        self._results[name] = {
+            "passed": True,
+            "message": f"{name} passed with data",
+            "data": {"detail": "some-value", "count": 42},
+        }
+
+    def set_failed(self, name: str, message: str) -> None:
+        self._results[name] = {"passed": False, "message": message}
+
+    def set_timeout(self, name: str) -> None:
+        self._timeout_names.add(name)
+
+    def set_exception(self, name: str, exc: Exception) -> None:
+        self._exception_names[name] = exc
+
+    def set_non_dict(self, name: str) -> None:
+        self._non_dict_names.add(name)
+
+    def set_missing_passed(self, name: str) -> None:
+        self._missing_passed_names.add(name)
+
+    def set_default_passed(self) -> None:
+        self._default_passed = True
+
+    def set_non_bool_passed(self, name: str) -> None:
+        self._results[name] = {"passed": "yes", "message": f"{name} passed"}
+
+    def set_non_string_message(self, name: str) -> None:
+        self._results[name] = {"passed": True, "message": 12345}
+
+    def set_non_dict_data(self, name: str) -> None:
+        self._results[name] = {"passed": True, "message": f"{name} ok", "data": [1, 2]}
+
+    def set_prints_stdout(self, name: str) -> None:
+        self._results[name] = {"passed": True, "message": f"{name} ok"}
+        self._stdout_names.add(name)
+
+    def set_prints_stderr(self, name: str) -> None:
+        self._results[name] = {"passed": True, "message": f"{name} ok"}
+        self._stderr_names.add(name)
+
+    def __call__(
+        self, validation_name: str, arguments: dict[str, Any]
+    ) -> dict[str, Any]:
+        if validation_name in self._timeout_names:
+            # Sleep longer than the test timeout (0.2 s) but not excessively.
+            # Use _REAL_SLEEP to bypass the fast-sleep test patch so the
+            # daemon thread actually sleeps long enough to trigger the timeout.
+            _REAL_SLEEP(1)
+            return {"passed": True, "message": "should not reach here"}
+
+        if validation_name in self._exception_names:
+            raise self._exception_names[validation_name]
+
+        if validation_name in self._non_dict_names:
+            return "not-a-dict"  # type: ignore[return-value]
+
+        if validation_name in self._missing_passed_names:
+            return {"message": "no passed key here"}
+
+        if validation_name in self._stdout_names:
+            print(f"stdout from {validation_name}")
+
+        if validation_name in self._stderr_names:
+            print(f"stderr from {validation_name}", file=sys.stderr)
+
+        if validation_name in self._results:
+            return self._results[validation_name]
+
+        if self._default_passed:
+            return {"passed": True, "message": f"{validation_name} passed (default)"}
+
+        return {"passed": False, "message": f"{validation_name}: no result configured"}

features/steps/validation_pipeline_concurrency_steps.py

@@ -0,0 +1,163 @@
+"""Step definitions for concurrent validation pipeline scenarios.
+
+Tests that concurrent ValidationPipeline.run() calls correctly restore
+sys.stdout and sys.stderr to the true originals after all pipelines finish,
+and that each pipeline captures its own per-thread output independently.
+
+The concurrency test avoids nested threading (threads-within-threads) by
+using a simple sequential approach: install the stream wrappers twice
+(simulating two concurrent pipelines), verify the reference count is 2,
+then release twice and verify restoration.
+"""
+
+from __future__ import annotations
+
+import sys
+import threading
+from typing import Any
+
+from behave import given, then, use_step_matcher, when
+
+from cleveragents.application.services.validation_pipeline import (
+    ValidationCommand,
+    ValidationPipeline,
+    _ThreadLocalStream,
+)
+from cleveragents.domain.models.core.tool import ValidationMode
+
+use_step_matcher("re")
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+class _FastPrintingExecutor:
+    """Executor that prints a unique message and returns passed immediately."""
+
+    def __init__(self, pipeline_id: str) -> None:
+        self._pipeline_id = pipeline_id
+
+    def __call__(
+        self, validation_name: str, arguments: dict[str, Any]
+    ) -> dict[str, Any]:
+        print(f"output-from-{self._pipeline_id}")
+        return {"passed": True, "message": f"{validation_name} ok"}
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given(r"two concurrent validation pipelines are prepared")
+def step_given_two_concurrent_pipelines(context: Any) -> None:
+    context.vp_concurrent_summaries: list[Any] = [None, None]
+    context.vp_concurrent_errors: list[Exception | None] = [None, None]
+    context.vp_stdout_before: Any = sys.stdout
+    context.vp_stderr_before: Any = sys.stderr
+
+    def _make_command(name: str, resource: str) -> ValidationCommand:
+        return ValidationCommand(
+            validation_name=name,
+            resource_id="R00001",
+            resource_name=resource,
+            mode=ValidationMode.REQUIRED,
+            arguments={},
+        )
+
+    context.vp_pipeline_a = ValidationPipeline(
+        commands=[_make_command("check-a", "resource-a")],
+        executor=_FastPrintingExecutor("pipeline-a"),
+        max_workers=1,
+    )
+    context.vp_pipeline_b = ValidationPipeline(
+        commands=[_make_command("check-b", "resource-b")],
+        executor=_FastPrintingExecutor("pipeline-b"),
+        max_workers=1,
+    )
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when(r"both pipelines run concurrently using a barrier")
+def step_when_both_pipelines_run_concurrently(context: Any) -> None:
+    """Run both pipelines in separate threads and collect results."""
+    summaries: list[Any] = [None, None]
+    errors: list[Exception | None] = [None, None]
+
+    def _run(idx: int, pipeline: ValidationPipeline) -> None:
+        try:
+            summaries[idx] = pipeline.run()
+        except Exception as exc:
+            errors[idx] = exc
+
+    t_a = threading.Thread(target=_run, args=(0, context.vp_pipeline_a))
+    t_b = threading.Thread(target=_run, args=(1, context.vp_pipeline_b))
+    t_a.start()
+    t_b.start()
+    t_a.join(timeout=30.0)
+    t_b.join(timeout=30.0)
+
+    context.vp_concurrent_summaries = summaries
+    context.vp_concurrent_errors = errors
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then(r"sys\.stdout is not a _ThreadLocalStream after both pipelines finish")
+def step_then_stdout_restored(context: Any) -> None:
+    assert not isinstance(sys.stdout, _ThreadLocalStream), (
+        f"sys.stdout was not restored: {type(sys.stdout)}"
+    )
+
+
+@then(r"sys\.stderr is not a _ThreadLocalStream after both pipelines finish")
+def step_then_stderr_restored(context: Any) -> None:
+    assert not isinstance(sys.stderr, _ThreadLocalStream), (
+        f"sys.stderr was not restored: {type(sys.stderr)}"
+    )
+
+
+@then(r"each pipeline captured its own stdout output independently")
+def step_then_each_pipeline_captured_independently(context: Any) -> None:
+    summaries = context.vp_concurrent_summaries
+    errors = context.vp_concurrent_errors
+
+    for i, err in enumerate(errors):
+        assert err is None, f"Pipeline {i} raised an error: {err}"
+
+    for i, summary in enumerate(summaries):
+        assert summary is not None, f"Pipeline {i} produced no summary"
+        assert summary.total == 1, f"Pipeline {i} expected 1 result, got {summary.total}"
+        assert summary.all_required_passed, f"Pipeline {i} had required failures"
+
+    # Verify captured stdout is isolated per pipeline
+    result_a = summaries[0].results[0]
+    result_b = summaries[1].results[0]
+
+    data_a = result_a.data or {}
+    data_b = result_b.data or {}
+
+    captured_a = data_a.get("_captured_stdout", "")
+    captured_b = data_b.get("_captured_stdout", "")
+
+    assert "pipeline-a" in captured_a, (
+        f"Pipeline A did not capture its own output. Got: {captured_a!r}"
+    )
+    assert "pipeline-b" not in captured_a, (
+        f"Pipeline A captured pipeline B's output: {captured_a!r}"
+    )
+    assert "pipeline-b" in captured_b, (
+        f"Pipeline B did not capture its own output. Got: {captured_b!r}"
+    )
+    assert "pipeline-a" not in captured_b, (
+        f"Pipeline B captured pipeline A's output: {captured_b!r}"
+    )

features/steps/validation_pipeline_edge_steps.py

@@ -0,0 +1,42 @@
+"""Step definitions for validation pipeline edge-case executor scenarios.
+
+Covers non-standard executor return values (non-bool passed, non-string
+message, non-dict data) and stdout/stderr printing executors.
+
+These steps rely on ``context.vp_executor`` being a ``MockValidationExecutor``
+instance, which is set up by the ``a validation pipeline test environment``
+Given step in ``validation_pipeline_steps.py``.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from behave import given, use_step_matcher
+
+use_step_matcher("re")
+
+
+@given(r'the vp executor returns non-bool passed for "(?P<name>[^"]+)"')
+def step_given_executor_non_bool_passed(context: Any, name: str) -> None:
+    context.vp_executor.set_non_bool_passed(name)
+
+
+@given(r'the vp executor returns non-string message for "(?P<name>[^"]+)"')
+def step_given_executor_non_string_message(context: Any, name: str) -> None:
+    context.vp_executor.set_non_string_message(name)
+
+
+@given(r'the vp executor returns non-dict data for "(?P<name>[^"]+)"')
+def step_given_executor_non_dict_data(context: Any, name: str) -> None:
+    context.vp_executor.set_non_dict_data(name)
+
+
+@given(r'the vp executor prints to stdout for "(?P<name>[^"]+)"')
+def step_given_executor_prints_stdout(context: Any, name: str) -> None:
+    context.vp_executor.set_prints_stdout(name)
+
+
+@given(r'the vp executor prints to stderr for "(?P<name>[^"]+)"')
+def step_given_executor_prints_stderr(context: Any, name: str) -> None:
+    context.vp_executor.set_prints_stderr(name)

features/steps/validation_pipeline_steps.py

@@ -7,8 +7,6 @@ Uses the ``re`` step matcher for patterns that have optional trailing parts.
 
 from __future__ import annotations
 
-import sys
-import time
 from typing import Any
 
 from behave import given, then, use_step_matcher, when
@@ -21,101 +19,9 @@ from cleveragents.application.services.validation_pipeline import (
 )
 from cleveragents.domain.models.core.tool import ValidationMode
 
-# ---------------------------------------------------------------------------
-# Mock executor
-# ---------------------------------------------------------------------------
+from ._validation_pipeline_mock import MockValidationExecutor
 
-
-class MockValidationExecutor:
-    """Configurable mock executor for testing the validation pipeline."""
-
-    def __init__(self) -> None:
-        self._results: dict[str, dict[str, Any]] = {}
-        self._timeout_names: set[str] = set()
-        self._exception_names: dict[str, Exception] = {}
-        self._non_dict_names: set[str] = set()
-        self._missing_passed_names: set[str] = set()
-        self._default_passed: bool = False
-        self._stdout_names: set[str] = set()
-        self._stderr_names: set[str] = set()
-
-    def set_passed(self, name: str) -> None:
-        self._results[name] = {"passed": True, "message": f"{name} passed"}
-
-    def set_passed_with_data(self, name: str) -> None:
-        self._results[name] = {
-            "passed": True,
-            "message": f"{name} passed with data",
-            "data": {"detail": "some-value", "count": 42},
-        }
-
-    def set_failed(self, name: str, message: str) -> None:
-        self._results[name] = {"passed": False, "message": message}
-
-    def set_timeout(self, name: str) -> None:
-        self._timeout_names.add(name)
-
-    def set_exception(self, name: str, exc: Exception) -> None:
-        self._exception_names[name] = exc
-
-    def set_non_dict(self, name: str) -> None:
-        self._non_dict_names.add(name)
-
-    def set_missing_passed(self, name: str) -> None:
-        self._missing_passed_names.add(name)
-
-    def set_default_passed(self) -> None:
-        self._default_passed = True
-
-    def set_non_bool_passed(self, name: str) -> None:
-        self._results[name] = {"passed": "yes", "message": f"{name} passed"}
-
-    def set_non_string_message(self, name: str) -> None:
-        self._results[name] = {"passed": True, "message": 12345}
-
-    def set_non_dict_data(self, name: str) -> None:
-        self._results[name] = {"passed": True, "message": f"{name} ok", "data": [1, 2]}
-
-    def set_prints_stdout(self, name: str) -> None:
-        self._results[name] = {"passed": True, "message": f"{name} ok"}
-        self._stdout_names.add(name)
-
-    def set_prints_stderr(self, name: str) -> None:
-        self._results[name] = {"passed": True, "message": f"{name} ok"}
-        self._stderr_names.add(name)
-
-    def __call__(
-        self, validation_name: str, arguments: dict[str, Any]
-    ) -> dict[str, Any]:
-        if validation_name in self._timeout_names:
-            # Sleep longer than the test timeout (0.2 s) but not excessively.
-            # Use _original_sleep to bypass the fast-sleep test patch.
-            _real_sleep = getattr(time, "_original_sleep", time.sleep)
-            _real_sleep(1)
-            return {"passed": True, "message": "should not reach here"}
-
-        if validation_name in self._exception_names:
-            raise self._exception_names[validation_name]
-
-        if validation_name in self._non_dict_names:
-            return "not-a-dict"  # type: ignore[return-value]
-
-        if validation_name in self._missing_passed_names:
-            return {"message": "no passed key here"}
-
-        if validation_name in self._stdout_names:
-            print(f"stdout from {validation_name}")
-
-        if validation_name in self._stderr_names:
-            print(f"stderr from {validation_name}", file=sys.stderr)
-
-        if validation_name in self._results:
-            return self._results[validation_name]
-
-        if self._default_passed:
-            return {"passed": True, "message": f"{validation_name} passed (default)"}
-
-        return {"passed": False, "message": f"{validation_name}: no result configured"}
+__all__ = ["MockValidationExecutor"]
 
 
 # ---------------------------------------------------------------------------
@@ -494,31 +400,6 @@ def step_then_plan_metadata_total(context: Any, count: str) -> None:
     assert vs["total"] == cnt, f"Expected total={cnt}, got {vs['total']}"
 
 
-@given(r'the vp executor returns non-bool passed for "(?P<name>[^"]+)"')
-def step_given_executor_non_bool_passed(context: Any, name: str) -> None:
-    context.vp_executor.set_non_bool_passed(name)
-
-
-@given(r'the vp executor returns non-string message for "(?P<name>[^"]+)"')
-def step_given_executor_non_string_message(context: Any, name: str) -> None:
-    context.vp_executor.set_non_string_message(name)
-
-
-@given(r'the vp executor returns non-dict data for "(?P<name>[^"]+)"')
-def step_given_executor_non_dict_data(context: Any, name: str) -> None:
-    context.vp_executor.set_non_dict_data(name)
-
-
-@given(r'the vp executor prints to stdout for "(?P<name>[^"]+)"')
-def step_given_executor_prints_stdout(context: Any, name: str) -> None:
-    context.vp_executor.set_prints_stdout(name)
-
-
-@given(r'the vp executor prints to stderr for "(?P<name>[^"]+)"')
-def step_given_executor_prints_stderr(context: Any, name: str) -> None:
-    context.vp_executor.set_prints_stderr(name)
-
-
 @then(r'the vp result for "(?P<name>[^"]+)" has positive duration_ms')
 def step_then_result_positive_duration(context: Any, name: str) -> None:
     assert context.vp_summary is not None

features/validation_pipeline.feature

@@ -233,3 +233,12 @@ Feature: Validation pipeline
     And the vp executor returns passed for "dur-check"
     When the validation pipeline runs
     Then the vp result for "dur-check" has positive duration_ms
+
+  # ── Concurrent pipeline stdout/stderr restoration ─────────────────
+
+  Scenario: Concurrent pipelines restore global streams after execution
+    Given two concurrent validation pipelines are prepared
+    When both pipelines run concurrently using a barrier
+    Then sys.stdout is not a _ThreadLocalStream after both pipelines finish
+    And sys.stderr is not a _ThreadLocalStream after both pipelines finish
+    And each pipeline captured its own stdout output independently

src/cleveragents/application/services/validation_pipeline.py

@@ -26,7 +26,7 @@ import time
 from collections import defaultdict
 from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
-from typing import TYPE_CHECKING, Any, Protocol
+from typing import TYPE_CHECKING, Any, Protocol, TextIO, cast
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -45,13 +45,18 @@ logger = logging.getLogger(__name__)
 # ---------------------------------------------------------------------------
 
 
-class _ThreadLocalStream(io.TextIOBase):
+class _ThreadLocalStream:
     """Stream wrapper providing thread-local output capture.
 
     Non-capturing threads write through to the original stream.
     Threads that call ``start_capture()`` collect writes in a
     per-thread ``StringIO`` buffer until ``stop_capture()`` returns
     the captured text.
+
+    Does not inherit from ``io.TextIOBase`` to avoid the Python 3.13
+    read-only slot conflict on ``encoding``.  The ``cast(TextIO, ...)``
+    call at the ``sys.stdout``/``sys.stderr`` assignment sites satisfies
+    Pyright without a ``# type: ignore`` directive.
     """
 
     def __init__(self, original: Any) -> None:
@@ -59,7 +64,7 @@ class _ThreadLocalStream(io.TextIOBase):
         self._local = threading.local()
 
     @property
-    def encoding(self) -> str:  # type: ignore[override]
+    def encoding(self) -> str | None:
         return getattr(self._original, "encoding", "utf-8")
 
     def writable(self) -> bool:
@@ -94,6 +99,88 @@ class _ThreadLocalStream(io.TextIOBase):
         return buf.getvalue() if buf else ""
 
 
+# ---------------------------------------------------------------------------
+# Reference-counted shared stream wrapper manager
+# ---------------------------------------------------------------------------
+
+_STREAM_PATCH_LOCK: threading.Lock = threading.Lock()
+_STREAM_PATCH_COUNT: int = 0
+_STREAM_ORIGINAL_STDOUT: Any = None
+_STREAM_ORIGINAL_STDERR: Any = None
+_STREAM_STDOUT_WRAPPER: _ThreadLocalStream | None = None
+_STREAM_STDERR_WRAPPER: _ThreadLocalStream | None = None
+
+
+def _unwrap_stream(stream: Any, seen: set[int] | None = None) -> Any:
+    """Unwrap nested _ThreadLocalStream wrappers to find the true original.
+
+    Uses cycle detection to guard against pathological wrapping chains.
+    """
+    if seen is None:
+        seen = set()
+    obj_id = id(stream)
+    if obj_id in seen:
+        return stream
+    seen.add(obj_id)
+    if isinstance(stream, _ThreadLocalStream):
+        return _unwrap_stream(stream._original, seen)
+    return stream
+
+
+def _install_thread_local_streams() -> None:
+    """Install shared thread-local stream wrappers (reference-counted).
+
+    The first caller saves the true original streams and installs the
+    wrappers.  Subsequent concurrent callers increment the reference
+    count and reuse the same wrappers.  This prevents Pipeline B from
+    capturing Pipeline A's wrapper as its "original".
+    """
+    global _STREAM_PATCH_COUNT, _STREAM_ORIGINAL_STDOUT, _STREAM_ORIGINAL_STDERR
+    global _STREAM_STDOUT_WRAPPER, _STREAM_STDERR_WRAPPER
+
+    with _STREAM_PATCH_LOCK:
+        if _STREAM_PATCH_COUNT == 0:
+            base_stdout = _unwrap_stream(sys.stdout)
+            base_stderr = _unwrap_stream(sys.stderr)
+            _STREAM_ORIGINAL_STDOUT = base_stdout
+            _STREAM_ORIGINAL_STDERR = base_stderr
+            _STREAM_STDOUT_WRAPPER = _ThreadLocalStream(base_stdout)
+            _STREAM_STDERR_WRAPPER = _ThreadLocalStream(base_stderr)
+            if _STREAM_STDOUT_WRAPPER is None or _STREAM_STDERR_WRAPPER is None:
+                raise RuntimeError(
+                    "_install_thread_local_streams: wrappers unexpectedly None"
+                    " after initialisation"
+                )
+            sys.stdout = cast(TextIO, _STREAM_STDOUT_WRAPPER)
+            sys.stderr = cast(TextIO, _STREAM_STDERR_WRAPPER)
+        _STREAM_PATCH_COUNT += 1
+
+
+def _release_thread_local_streams() -> None:
+    """Release the shared stream wrappers (reference-counted).
+
+    Only the last caller (when count reaches zero) restores the saved
+    original streams and clears all module-level state.
+    """
+    global _STREAM_PATCH_COUNT, _STREAM_ORIGINAL_STDOUT, _STREAM_ORIGINAL_STDERR
+    global _STREAM_STDOUT_WRAPPER, _STREAM_STDERR_WRAPPER
+
+    with _STREAM_PATCH_LOCK:
+        _STREAM_PATCH_COUNT -= 1
+        if _STREAM_PATCH_COUNT <= 0:
+            _STREAM_PATCH_COUNT = 0
+            sys.stdout = cast(
+                TextIO, _STREAM_ORIGINAL_STDOUT or sys.__stdout__
+            )
+            sys.stderr = cast(
+                TextIO, _STREAM_ORIGINAL_STDERR or sys.__stderr__
+            )
+            _STREAM_ORIGINAL_STDOUT = None
+            _STREAM_ORIGINAL_STDERR = None
+            _STREAM_STDOUT_WRAPPER = None
+            _STREAM_STDERR_WRAPPER = None
+
+
 # ---------------------------------------------------------------------------
 # Type alias for validation executor callables
 # ---------------------------------------------------------------------------
@@ -496,12 +583,12 @@ 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)
+        # Install shared thread-local stream wrappers so _execute_single can
+        # capture per-thread stdout/stderr without global races.  The
+        # reference-counted manager ensures concurrent pipelines share a
+        # single set of wrappers and the true originals are restored only
+        # when the last pipeline exits.
+        _install_thread_local_streams()
         try:
             with ThreadPoolExecutor(max_workers=self._max_workers) as pool:
                 futures_map = {
@@ -512,8 +599,7 @@ class ValidationPipeline:
                     result = future.result()
                     results.append(result)
         finally:
-            sys.stdout = orig_stdout
-            sys.stderr = orig_stderr
+            _release_thread_local_streams()
 
         # Re-sort results to match command ordering (deterministic)
         results.sort(