v0.55.1 — Fix mcpclient nil interface bug + concurrent test race

Root cause: Classic Go nil interface gotcha — *rpcError nil pointer
stored in error interface became non-nil, causing ALL call()
results to be returned as errors with text '<nil>' instead of
successfully parsed responses.

Fix: Convert *rpcError to error interface with explicit nil check
before routing responses in readLoop.

Concurrent test: Mock server assumed request arrival order
determined which goroutine got which response. Fixed by
extracting caller identity from tool-call arguments (id:a vs id:b)
so responses are correctly routed regardless of goroutine schedule.

Author: molty3000

internal/mcpclient/client.go

@@ -140,6 +140,12 @@ type lineResult struct {
 	err  error
 }
 
+// callResponse carries a JSON-RPC response from readLoop to the waiting caller.
+type callResponse struct {
+	result json.RawMessage
+	err    error
+}
+
 // ── Client ─────────────────────────────────────────────────────────────
 
 // Client manages a connection to an external MCP server over stdio.
@@ -150,8 +156,10 @@ type Client struct {
 	stdout *bufio.Reader
 	lineCh chan lineResult // single-reader goroutine sends lines here
 	done   chan struct{}   // closed when process exits
-	mu     sync.Mutex
-	nextID int
+
+	mu      sync.Mutex
+	nextID  int
+	pending map[int]chan callResponse // routes responses to waiting callers
 }
 
 // New spawns an MCP server process and returns a client connected to it.
@@ -184,12 +192,13 @@ func New(name string, cfg ServerConfig) (*Client, error) {
 	}
 
 	c := &Client{
-		name:   name,
-		cmd:    cmd,
-		stdin:  stdin,
-		stdout: bufio.NewReader(stdout),
-		lineCh: make(chan lineResult, 10),
-		done:   make(chan struct{}),
+		name:    name,
+		cmd:     cmd,
+		stdin:   stdin,
+		stdout:  bufio.NewReader(stdout),
+		lineCh:  make(chan lineResult, 10),
+		done:    make(chan struct{}),
+		pending: make(map[int]chan callResponse),
 	}
 
 	// Start single-reader goroutine
@@ -333,19 +342,27 @@ func (c *Client) call(ctx context.Context, method string, params json.RawMessage
 		ctx = context.Background()
 	}
 
-	// Assign unique ID
+	// Assign unique ID and register a response channel.
+	respCh := make(chan callResponse, 1)
+
 	c.mu.Lock()
 	id := c.nextID
 	c.nextID++
-	c.mu.Unlock()
-
-	// Build request
 	req := request{
 		JSONRPC: "2.0",
 		ID:      id,
 		Method:  method,
 		Params:  params,
 	}
+	c.pending[id] = respCh
+	c.mu.Unlock()
+
+	// Unregister on exit to prevent map leak.
+	defer func() {
+		c.mu.Lock()
+		delete(c.pending, id)
+		c.mu.Unlock()
+	}()
 
 	reqRaw, err := json.Marshal(req)
 	if err != nil {
@@ -360,56 +377,68 @@ func (c *Client) call(ctx context.Context, method string, params json.RawMessage
 		return nil, fmt.Errorf("write: %w", err)
 	}
 
-	// Read responses until we find our ID
-	for {
-		select {
-		case <-ctx.Done():
-			return nil, ctx.Err()
-		default:
-		}
-
-		// Set a read deadline via context
-		line, err := c.readLine(ctx)
-		if err != nil {
-			return nil, fmt.Errorf("read: %w", err)
-		}
-
-		var resp response
-		if err := json.Unmarshal([]byte(line), &resp); err != nil {
-			continue // skip malformed lines
-		}
-
-		// Skip responses that aren't for our request
-		if resp.ID != id {
-			continue
+	// Wait for response via channel (dispatched by readLoop).
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case cr, ok := <-respCh:
+		if !ok {
+			return nil, fmt.Errorf("connection closed before response received")
 		}
-
-		if resp.Error != nil {
-			return nil, resp.Error
+		if cr.err != nil {
+			return nil, cr.err
 		}
-
-		return resp.Result, nil
+		return cr.result, nil
 	}
 }
 
 // readLoop is a single reader goroutine that reads lines from stdout and
-// sends them on lineCh. It exits when stdout returns an error (EOF on pipe close).
+// routes each response to the correct waiting caller via the pending map.
+// This prevents response misrouting when multiple concurrent call() instances
+// are reading from the same connection.
+// Exits when stdout returns an error (EOF on pipe close).
 func (c *Client) readLoop() {
-	defer close(c.lineCh)
 	for {
 		line, err := c.stdout.ReadString('\n')
-		select {
-		case c.lineCh <- lineResult{strings.TrimSpace(line), err}:
-		default:
-			// Channel full — drain the queue by consuming one stale entry,
-			// then retry. This prevents a stuck server from deadlocking the
-			// reader goroutine.
-			<-c.lineCh
-			c.lineCh <- lineResult{strings.TrimSpace(line), err}
-		}
 		if err != nil {
+			// Process exited or pipe closed — unblock all waiters.
+			c.mu.Lock()
+			for id, ch := range c.pending {
+				close(ch)
+				delete(c.pending, id)
+			}
+			c.mu.Unlock()
 			return
 		}
+		line = strings.TrimSpace(line)
+		if line == "" {
+			continue
+		}
+
+		var resp response
+		if err := json.Unmarshal([]byte(line), &resp); err != nil {
+			continue // skip malformed lines
+		}
+
+		// Route to the waiting caller, if any.
+		c.mu.Lock()
+		ch, ok := c.pending[resp.ID]
+		if ok {
+			delete(c.pending, resp.ID)
+		}
+		c.mu.Unlock()
+
+		if ok && ch != nil {
+			// Non-blocking send in case caller has already timed out.
+			var rpcErr error
+			if resp.Error != nil {
+				rpcErr = resp.Error
+			}
+			select {
+			case ch <- callResponse{result: resp.Result, err: rpcErr}:
+			default:
+			}
+		}
 	}
 }
 

internal/mcpclient/client_concurrent_test.go

@@ -0,0 +1,159 @@
+package mcpclient
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+// TestClient_ConcurrentCallsNoResponseLoss verifies that concurrent tool calls
+// to the same MCP client don't lose responses due to incorrect routing.
+// BUG: call() read from a shared lineCh and discarded responses with
+// non-matching IDs. With concurrent calls, responses for one goroutine
+// were consumed and silently dropped by another.
+func TestClient_ConcurrentCallsNoResponseLoss(t *testing.T) {
+	// Use io.Pipe for bidirectional communication with a simulated server.
+	serverRead, clientWrite := io.Pipe() // client writes to server
+	clientRead, serverWrite := io.Pipe() // server writes to client
+
+	client := &Client{
+		name:    "test",
+		nextID:  0,
+		stdin:   clientWrite,
+		stdout:  bufio.NewReader(clientRead),
+		pending: make(map[int]chan callResponse),
+		done:    make(chan struct{}),
+	}
+	go client.readLoop()
+
+	// Simulate a server that reads both requests, then sends responses
+	// out of order (pipelined), identifying each response by the
+	// "request_id" field in the tool call arguments rather than by
+	// arrival order — this avoids a race condition where goroutine
+	// scheduling determines which request arrives first.
+	type rawRequest struct {
+		ID      int                `json:"id"`
+		Method  string             `json:"method"`
+		Params  json.RawMessage    `json:"params,omitempty"`
+	}
+	go func() {
+		scanner := bufio.NewScanner(serverRead)
+		var reqs []rawRequest
+
+		for scanner.Scan() {
+			line := strings.TrimSpace(scanner.Text())
+			if line == "" {
+				continue
+			}
+			var req rawRequest
+			if err := json.Unmarshal([]byte(line), &req); err != nil {
+				t.Errorf("mock server: unmarshal request: %v", err)
+				continue
+			}
+			reqs = append(reqs, req)
+			if len(reqs) >= 2 {
+				break
+			}
+		}
+
+		if len(reqs) < 2 {
+			t.Errorf("mock server: only got %d requests", len(reqs))
+			return
+		}
+
+		// Build a map from request ID → caller's tool-call argument "id".
+		// This avoids a race condition: we don't know which goroutine's
+		// request arrived first, so we identify responses by content
+		// rather than by arrival order.
+		callerIDs := make(map[int]string, 2)
+		for _, req := range reqs {
+			var params struct {
+				Name      string         `json:"name"`
+				Arguments map[string]any `json:"arguments"`
+			}
+			if req.Params != nil {
+				json.Unmarshal(req.Params, &params)
+			}
+			callerID := "unknown"
+			if idVal, ok := params.Arguments["id"]; ok {
+				callerID, _ = idVal.(string)
+			}
+			callerIDs[req.ID] = callerID
+		}
+
+		// Send responses out of order: respond to reqs[1] first.
+		for _, idx := range []int{1, 0} {
+			req := reqs[idx]
+			callerID := callerIDs[req.ID]
+			resp, _ := json.Marshal(response{
+				JSONRPC: "2.0",
+				ID:      req.ID,
+				Result:  json.RawMessage(`{"content":[{"type":"text","text":"response-` + callerID + `"}]}`),
+			})
+			serverWrite.Write(append(resp, '\n'))
+			if idx == 1 {
+				time.Sleep(10 * time.Millisecond) // small delay between responses
+			}
+		}
+	}()
+
+	// Make two concurrent calls with a timeout.
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	var mu sync.Mutex
+	results := make(map[string]string)
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	go func() {
+		defer wg.Done()
+		text, err := client.CallTool(ctx, "test_tool", `{"id":"a"}`)
+		mu.Lock()
+		if err != nil {
+			results["a"] = fmt.Sprintf("err=%v", err)
+		} else {
+			results["a"] = fmt.Sprintf("ok=%s", text)
+		}
+		mu.Unlock()
+	}()
+	go func() {
+		defer wg.Done()
+		text, err := client.CallTool(ctx, "test_tool", `{"id":"b"}`)
+		mu.Lock()
+		if err != nil {
+			results["b"] = fmt.Sprintf("err=%v", err)
+		} else {
+			results["b"] = fmt.Sprintf("ok=%s", text)
+		}
+		mu.Unlock()
+	}()
+
+	wg.Wait()
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	if strings.HasPrefix(results["a"], "err=") && strings.Contains(results["a"], "context deadline") {
+		t.Errorf("call A timed out — response was stolen by call B")
+	}
+	if strings.HasPrefix(results["b"], "err=") && strings.Contains(results["b"], "context deadline") {
+		t.Errorf("call B timed out — response was stolen by call A")
+	}
+	if results["a"] == "ok=response-a" {
+		t.Logf("call A = %s ✓", results["a"])
+	} else {
+		t.Errorf("call A = %q, want 'ok=response-a'", results["a"])
+	}
+	if results["b"] == "ok=response-b" {
+		t.Logf("call B = %s ✓", results["b"])
+	} else {
+		t.Errorf("call B = %q, want 'ok=response-b'", results["b"])
+	}
+}