From 2240a028c8f4cf0bfa570db903e05085d1cd06f7 Mon Sep 17 00:00:00 2001
From: CleverThis <hal9000@cleverthis.com>
Date: Thu, 23 Apr 2026 23:13:29 +0000
Subject: [PATCH] fix(tests): replace ThreadPoolExecutor with threading.Thread
 in RLock concurrency steps

ThreadPoolExecutor holds internal locks (e.g. _global_shutdown_lock) that
can be in a locked state after fork(). The behave-parallel runner uses
multiprocessing.Pool with the fork start method, so forked worker processes
inherit these locked states and deadlock when trying to create or shut down
a ThreadPoolExecutor.

Replace both concurrent step implementations with direct threading.Thread
usage, following the pattern established in context_tier_thread_safety_steps.py
and other step files throughout the codebase. This eliminates the fork+lock
deadlock while preserving the deterministic concurrency-counter approach for
proving the RLock is released during transport calls.
---
 ...tdd_mcp_adapter_rlock_concurrency_steps.py | 61 +++++++++++++------
 1 file changed, 43 insertions(+), 18 deletions(-)

diff --git a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py
index cd49feb30..b4dca4027 100644
--- a/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py
+++ b/features/steps/tdd_mcp_adapter_rlock_concurrency_steps.py
@@ -11,7 +11,6 @@ from __future__ import annotations
 
 import threading
 import time
-from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Any
 
 from behave import given, then, when
@@ -261,34 +260,60 @@ def step_lock_checking_discovery_transport(context: Context) -> None:
 
 @when('I invoke "{tool_name}" concurrently from {count:d} threads')
 def step_concurrent_invoke(context: Context, tool_name: str, count: int) -> None:
-    results: list[MCPToolResult] = []
+    # Use threading.Thread directly instead of ThreadPoolExecutor to avoid
+    # deadlocks in forked worker processes (behave-parallel uses fork).
+    # ThreadPoolExecutor holds internal locks that can be in a locked state
+    # after fork(), causing child processes to deadlock on acquisition.
+    results: list[MCPToolResult | None] = [None] * count
+    errors: list[BaseException | None] = [None] * count
+    errors_lock = threading.Lock()
 
-    with ThreadPoolExecutor(max_workers=count) as executor:
-        futures = [
-            executor.submit(context.mcp_adapter.invoke, tool_name, {})
-            for _ in range(count)
-        ]
-        for future in as_completed(futures):
-            results.append(future.result())
+    def _invoke(idx: int) -> None:
+        try:
+            results[idx] = context.mcp_adapter.invoke(tool_name, {})
+        except BaseException as exc:
+            with errors_lock:
+                errors[idx] = exc
 
-    context.concurrent_invoke_results = results
+    threads = [threading.Thread(target=_invoke, args=(i,)) for i in range(count)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    # Re-raise the first error encountered, if any
+    for err in errors:
+        if err is not None:
+            raise err
+
+    context.concurrent_invoke_results = [r for r in results if r is not None]
 
 
 @when("I call discover_tools concurrently from {count:d} threads")
 def step_concurrent_discover(context: Context, count: int) -> None:
+    # Use threading.Thread directly instead of ThreadPoolExecutor to avoid
+    # deadlocks in forked worker processes (behave-parallel uses fork).
+    # ThreadPoolExecutor holds internal locks that can be in a locked state
+    # after fork(), causing child processes to deadlock on acquisition.
     results: list[Any] = []
     errors: list[Exception] = []
+    collect_lock = threading.Lock()
 
-    with ThreadPoolExecutor(max_workers=count) as executor:
-        futures = [
-            executor.submit(context.mcp_adapter.discover_tools) for _ in range(count)
-        ]
-        for future in as_completed(futures):
-            try:
-                results.append(future.result())
-            except Exception as exc:
+    def _discover() -> None:
+        try:
+            result = context.mcp_adapter.discover_tools()
+            with collect_lock:
+                results.append(result)
+        except Exception as exc:
+            with collect_lock:
                 errors.append(exc)
 
+    threads = [threading.Thread(target=_discover) for _ in range(count)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
     context.concurrent_discover_results = results
     context.concurrent_discover_errors = errors