fix: resolve stdio transport race condition for concurrent tool calls (#529)

* fix: resolve stdio transport race condition for concurrent tool calls

Fixes #528 by replacing goroutine-per-request pattern with a worker pool
to prevent multiple goroutines from accessing the non-thread-safe bufio.Reader.

- Add worker pool pattern with configurable size (default: 5 workers)
- Add buffered queue for tool calls (default: 100 capacity)
- Protect concurrent writes with mutex
- Maintain backward compatibility while fixing the race condition
- Add comprehensive test for concurrent tool calls

🤖 Generated with [opencode](https://opencode.ai)

Co-Authored-By: opencode <[email protected]>

* fix: address golangci-lint errcheck warnings

- Add error checking for Write operations in concurrent tool calls test
- Ensures all error return values are properly handled

🤖 Generated with [opencode](https://opencode.ai)

Co-Authored-By: opencode <[email protected]>

* refactor: remove redundant mutex around thread-safe sync.Map

sync.Map is already thread-safe and doesn't require additional mutex protection.
This simplifies the test code while maintaining the same functionality.

🤖 Generated with [opencode](https://opencode.ai)

Co-Authored-By: opencode <[email protected]>

* refactor: improve test control flow and add configuration bounds validation

- Replace goto statements with labeled break for more idiomatic Go
- Reduce test timeout from 5s to 2s for faster test execution
- Add upper bounds validation for worker pool size (max 100) and queue size (max 10000)
- Add tests to verify configuration bounds are respected
- Log warnings when configuration values exceed maximum limits

Addresses remaining CodeRabbit review feedback.

🤖 Generated with [opencode](https://opencode.ai)

Co-Authored-By: opencode <[email protected]>

* fix: use maximum values instead of defaults when limits exceeded

When WithWorkerPoolSize or WithQueueSize receive values exceeding their
maximum limits, use the maximum allowed value instead of falling back
to the default. This better aligns with user expectations when they
request large pool/queue sizes.

🤖 Generated with [opencode](https://opencode.ai)

Co-Authored-By: opencode <[email protected]>

* test: update stdio tests to expect maximum values when limits exceeded

Update test expectations to match the new behavior where WithWorkerPoolSize
and WithQueueSize use maximum values instead of defaults when the requested
size exceeds limits.

🤖 Generated with [opencode](https://opencode.ai)

Co-Authored-By: opencode <[email protected]>

---------

Co-authored-by: opencode <[email protected]>

Author: ezynda3

server/stdio.go

@@ -29,6 +29,20 @@ type StdioServer struct {
 	server      *MCPServer
 	errLogger   *log.Logger
 	contextFunc StdioContextFunc
+
+	// Thread-safe tool call processing
+	toolCallQueue  chan *toolCallWork
+	workerWg       sync.WaitGroup
+	workerPoolSize int
+	queueSize      int
+	writeMu        sync.Mutex // Protects concurrent writes
+}
+
+// toolCallWork represents a queued tool call request
+type toolCallWork struct {
+	ctx     context.Context
+	message json.RawMessage
+	writer  io.Writer
 }
 
 // StdioOption defines a function type for configuring StdioServer
@@ -50,6 +64,32 @@ func WithStdioContextFunc(fn StdioContextFunc) StdioOption {
 	}
 }
 
+// WithWorkerPoolSize sets the number of workers for processing tool calls
+func WithWorkerPoolSize(size int) StdioOption {
+	return func(s *StdioServer) {
+		const maxWorkerPoolSize = 100
+		if size > 0 && size <= maxWorkerPoolSize {
+			s.workerPoolSize = size
+		} else if size > maxWorkerPoolSize {
+			s.errLogger.Printf("Worker pool size %d exceeds maximum (%d), using maximum", size, maxWorkerPoolSize)
+			s.workerPoolSize = maxWorkerPoolSize
+		}
+	}
+}
+
+// WithQueueSize sets the size of the tool call queue
+func WithQueueSize(size int) StdioOption {
+	return func(s *StdioServer) {
+		const maxQueueSize = 10000
+		if size > 0 && size <= maxQueueSize {
+			s.queueSize = size
+		} else if size > maxQueueSize {
+			s.errLogger.Printf("Queue size %d exceeds maximum (%d), using maximum", size, maxQueueSize)
+			s.queueSize = maxQueueSize
+		}
+	}
+}
+
 // stdioSession is a static client session, since stdio has only one client.
 type stdioSession struct {
 	notifications      chan mcp.JSONRPCNotification
@@ -218,6 +258,8 @@ func NewStdioServer(server *MCPServer) *StdioServer {
 			"",
 			log.LstdFlags,
 		), // Default to discarding logs
+		workerPoolSize: 5,   // Default worker pool size
+		queueSize:      100, // Default queue size
 	}
 }
 
@@ -281,6 +323,30 @@ func (s *StdioServer) processInputStream(ctx context.Context, reader *bufio.Read
 	}
 }
 
+// toolCallWorker processes tool calls from the queue
+func (s *StdioServer) toolCallWorker(ctx context.Context) {
+	defer s.workerWg.Done()
+
+	for {
+		select {
+		case work, ok := <-s.toolCallQueue:
+			if !ok {
+				// Channel closed, exit worker
+				return
+			}
+			// Process the tool call
+			response := s.server.HandleMessage(work.ctx, work.message)
+			if response != nil {
+				if err := s.writeResponse(response, work.writer); err != nil {
+					s.errLogger.Printf("Error writing tool response: %v", err)
+				}
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
 // readNextLine reads a single line from the input reader in a context-aware manner.
 // It uses channels to make the read operation cancellable via context.
 // Returns the read line and any error encountered. If the context is cancelled,
@@ -315,6 +381,9 @@ func (s *StdioServer) Listen(
 	stdin io.Reader,
 	stdout io.Writer,
 ) error {
+	// Initialize the tool call queue
+	s.toolCallQueue = make(chan *toolCallWork, s.queueSize)
+
 	// Set a static client context since stdio only has one client
 	if err := s.server.RegisterSession(ctx, &stdioSessionInstance); err != nil {
 		return fmt.Errorf("register session: %w", err)
@@ -332,9 +401,23 @@ func (s *StdioServer) Listen(
 
 	reader := bufio.NewReader(stdin)
 
+	// Start worker pool for tool calls
+	for i := 0; i < s.workerPoolSize; i++ {
+		s.workerWg.Add(1)
+		go s.toolCallWorker(ctx)
+	}
+
 	// Start notification handler
 	go s.handleNotifications(ctx, stdout)
-	return s.processInputStream(ctx, reader, stdout)
+
+	// Process input stream
+	err := s.processInputStream(ctx, reader, stdout)
+
+	// Shutdown workers gracefully
+	close(s.toolCallQueue)
+	s.workerWg.Wait()
+
+	return err
 }
 
 // processMessage handles a single JSON-RPC message and writes the response.
@@ -367,16 +450,25 @@ func (s *StdioServer) processMessage(
 		Method string `json:"method"`
 	}
 	if json.Unmarshal(rawMessage, &baseMessage) == nil && baseMessage.Method == "tools/call" {
-		// Process tool calls concurrently to avoid blocking on sampling requests
-		go func() {
+		// Queue tool calls for processing by workers
+		select {
+		case s.toolCallQueue <- &toolCallWork{
+			ctx:     ctx,
+			message: rawMessage,
+			writer:  writer,
+		}:
+			return nil
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			// Queue is full, process synchronously as fallback
+			s.errLogger.Printf("Tool call queue full, processing synchronously")
 			response := s.server.HandleMessage(ctx, rawMessage)
 			if response != nil {
-				if err := s.writeResponse(response, writer); err != nil {
-					s.errLogger.Printf("Error writing tool response: %v", err)
-				}
+				return s.writeResponse(response, writer)
 			}
-		}()
-		return nil
+			return nil
+		}
 	}
 
 	// Handle other messages synchronously
@@ -462,6 +554,10 @@ func (s *StdioServer) writeResponse(
 		return err
 	}
 
+	// Protect concurrent writes
+	s.writeMu.Lock()
+	defer s.writeMu.Unlock()
+
 	// Write response followed by newline
 	if _, err := fmt.Fprintf(writer, "%s\n", responseBytes); err != nil {
 		return err