Go SDK: emit JSON-RPC error on pending-buffer overflow + guard drop

Carries the Rust SDK PR #1394 follow-up review fixes into the Go port:

1. Cap the per-session parked-waiter list at 128. When exceeded, reject
   the oldest waiter with errPendingSessionBufferOverflow ('pending
   session buffer overflow'). The handler returns a *jsonrpc2.Error with
   code -32603 via the new pendingRoutingRPCError helper, so the runtime
   receives a proper error response instead of hanging on the request id
   until its own timeout. Mirrors Rust commit 491b442 and TS commit
   c167bc3.

2. When the last pending-routing guard drops without RegisterSession
   (e.g. session.create failed mid-RPC), signal all parked waiters with
   errPendingSessionRoutingEnded ('pending session routing ended before
   session was registered'). Distinct phrasing from the overflow path so
   debugging can tell the two cases apart. Mirrors Rust commit e0ff254
   and TS commit c167bc3.

Adds pendingRoutingRPCError helper that routes sentinel errors to
-32603 while unknown-session errors keep -32602.

Adds two tests:
- TestPendingRouting_OverflowEmitsError: 129 parked waiters, oldest
  gets -32603 overflow error, remaining 128 resolve normally after
  registration.
- TestPendingRouting_GuardDropDistinctMessage: parks a request, drops
  the guard without registration, verifies exact routing-ended message
  and -32603 code.

Updates TestPendingRouting_RejectsWaitersOnDispose to assert the new
exact message and code instead of the old 'dropped' substring check.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: tclem

go/client.go

@@ -88,6 +88,13 @@ func validateSessionFsConfig(config *SessionFsConfig) error {
 //	}
 //	defer client.Stop()
 //
+// Sentinel errors for the two pending-routing rejection paths. Using distinct
+// values lets callers (and debugging) tell overflow eviction from guard-drop.
+var (
+	errPendingSessionBufferOverflow = errors.New("pending session buffer overflow")
+	errPendingSessionRoutingEnded   = errors.New("pending session routing ended before session was registered")
+)
+
 // pendingResult carries the outcome of a parked inbound-request session lookup.
 type pendingResult struct {
 	session *Session
@@ -1021,7 +1028,7 @@ func (c *Client) beginPendingSessionRouting() func() {
 
 			for _, chs := range waiters {
 				for _, ch := range chs {
-					ch <- pendingResult{err: fmt.Errorf("request dropped: cloud session.create completed without registering this session id")}
+					ch <- pendingResult{err: errPendingSessionRoutingEnded}
 				}
 			}
 		})
@@ -1076,10 +1083,12 @@ func (c *Client) waitForSession(sessionID string) (*Session, error) {
 	ch := make(chan pendingResult, 1)
 	waiters := c.pending.waiters[sessionID]
 	if len(waiters) >= pendingSessionBufferLimit {
-		// Reject the oldest waiter to keep the queue bounded.
+		// Reject the oldest waiter to keep the queue bounded. Send a JSON-RPC
+		// error response via the handler return so the runtime doesn't hang on
+		// the request id waiting for a reply that would never come.
 		oldest := waiters[0]
 		waiters = waiters[1:]
-		oldest <- pendingResult{err: fmt.Errorf("request dropped: pending session waiter buffer full for %s", sessionID)}
+		oldest <- pendingResult{err: errPendingSessionBufferOverflow}
 	}
 	c.pending.waiters[sessionID] = append(waiters, ch)
 	c.pending.mu.Unlock()
@@ -2126,6 +2135,18 @@ func (c *Client) handleSessionEvent(req sessionEventRequest) {
 	c.pending.mu.Unlock()
 }
 
+// pendingRoutingRPCError maps an error from waitForSession to the appropriate
+// JSON-RPC error. Overflow and guard-drop rejections use -32603 (internal
+// error) so the runtime gets a proper error response instead of hanging on the
+// request id. All other waitForSession errors (e.g. unknown session) keep the
+// existing -32602 (invalid params) code.
+func pendingRoutingRPCError(err error) *jsonrpc2.Error {
+	if errors.Is(err, errPendingSessionBufferOverflow) || errors.Is(err, errPendingSessionRoutingEnded) {
+		return &jsonrpc2.Error{Code: -32603, Message: err.Error()}
+	}
+	return &jsonrpc2.Error{Code: -32602, Message: err.Error()}
+}
+
 // handleUserInputRequest handles a user input request from the CLI server.
 func (c *Client) handleUserInputRequest(req userInputRequest) (*userInputResponse, *jsonrpc2.Error) {
 	if req.SessionID == "" || req.Question == "" {
@@ -2134,7 +2155,7 @@ func (c *Client) handleUserInputRequest(req userInputRequest) (*userInputRespons
 
 	session, err := c.waitForSession(req.SessionID)
 	if err != nil {
-		return nil, &jsonrpc2.Error{Code: -32602, Message: err.Error()}
+		return nil, pendingRoutingRPCError(err)
 	}
 
 	response, err := session.handleUserInputRequest(UserInputRequest{
@@ -2161,7 +2182,7 @@ func (c *Client) handleExitPlanModeRequest(req exitPlanModeRequest) (*ExitPlanMo
 
 	session, err := c.waitForSession(req.SessionID)
 	if err != nil {
-		return nil, &jsonrpc2.Error{Code: -32602, Message: err.Error()}
+		return nil, pendingRoutingRPCError(err)
 	}
 
 	response, err := session.handleExitPlanModeRequest(ExitPlanModeRequest{
@@ -2185,7 +2206,7 @@ func (c *Client) handleAutoModeSwitchRequest(req autoModeSwitchRequest) (*autoMo
 
 	session, err := c.waitForSession(req.SessionID)
 	if err != nil {
-		return nil, &jsonrpc2.Error{Code: -32602, Message: err.Error()}
+		return nil, pendingRoutingRPCError(err)
 	}
 
 	response, err := session.handleAutoModeSwitchRequest(AutoModeSwitchRequest{
@@ -2207,7 +2228,7 @@ func (c *Client) handleHooksInvoke(req hooksInvokeRequest) (map[string]any, *jso
 
 	session, err := c.waitForSession(req.SessionID)
 	if err != nil {
-		return nil, &jsonrpc2.Error{Code: -32602, Message: err.Error()}
+		return nil, pendingRoutingRPCError(err)
 	}
 
 	output, err := session.handleHooksInvoke(req.Type, req.Input)
@@ -2230,7 +2251,7 @@ func (c *Client) handleSystemMessageTransform(req systemMessageTransformRequest)
 
 	session, err := c.waitForSession(req.SessionID)
 	if err != nil {
-		return systemMessageTransformResponse{}, &jsonrpc2.Error{Code: -32602, Message: err.Error()}
+		return systemMessageTransformResponse{}, pendingRoutingRPCError(err)
 	}
 
 	resp, err := session.handleSystemMessageTransform(req.Sections)

go/cloud_session_test.go

@@ -317,8 +317,133 @@ func TestPendingRouting_RejectsWaitersOnDispose(t *testing.T) {
 		if rpcErr == nil {
 			t.Fatal("expected an rpc error after dispose without registration")
 		}
-		if !strings.Contains(rpcErr.Message, "dropped") {
-			t.Errorf("expected 'dropped' in error message, got: %s", rpcErr.Message)
+		if !strings.Contains(rpcErr.Message, "routing ended before session was registered") {
+			t.Errorf("expected routing-ended message, got: %s", rpcErr.Message)
+		}
+		if rpcErr.Code != -32603 {
+			t.Errorf("expected code -32603, got: %d", rpcErr.Code)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for rejected waiter")
+	}
+}
+
+// TestPendingRouting_OverflowEmitsError verifies that when the parked-waiter
+// buffer reaches its cap, the oldest waiter receives the overflow error response
+// and the remaining 128 waiters resolve normally after registration.
+func TestPendingRouting_OverflowEmitsError(t *testing.T) {
+	client := newCloudTestClient()
+	dispose := client.beginPendingSessionRouting()
+
+	const pendingID = "overflow-request-session"
+	const total = pendingSessionBufferLimit + 1 // 129
+
+	type result struct {
+		resp *userInputResponse
+		err  *jsonrpcError
+	}
+
+	// Register a user-input handler so the session resolves successfully.
+	session, cleanup := newTestSession()
+	defer cleanup()
+	session.SessionID = pendingID
+	session.registerUserInputHandler(func(req UserInputRequest, _ UserInputInvocation) (UserInputResponse, error) {
+		return UserInputResponse{Answer: "yes"}, nil
+	})
+
+	results := make([]chan result, total)
+	for i := range total {
+		results[i] = make(chan result, 1)
+		go func(ch chan result) {
+			resp, rpcErr := client.handleUserInputRequest(userInputRequest{
+				SessionID: pendingID,
+				Question:  "Proceed?",
+			})
+			ch <- result{resp, rpcErr}
+		}(results[i])
+	}
+
+	// Give goroutines time to park.
+	time.Sleep(50 * time.Millisecond)
+
+	// Register the session and flush — this resolves the 128 remaining waiters.
+	client.sessionsMux.Lock()
+	client.sessions[pendingID] = session
+	client.sessionsMux.Unlock()
+	client.flushPendingForSession(pendingID, session)
+	dispose()
+
+	// Collect all results with a timeout.
+	var gotOverflow int
+	var gotSuccess int
+	deadline := time.After(2 * time.Second)
+	for _, ch := range results {
+		select {
+		case r := <-ch:
+			if r.err != nil {
+				if !strings.Contains(r.err.Message, "pending session buffer overflow") {
+					t.Errorf("unexpected error message: %s", r.err.Message)
+				}
+				if r.err.Code != -32603 {
+					t.Errorf("expected code -32603 for overflow, got: %d", r.err.Code)
+				}
+				gotOverflow++
+			} else {
+				gotSuccess++
+			}
+		case <-deadline:
+			t.Fatalf("timed out: overflow=%d success=%d", gotOverflow, gotSuccess)
+		}
+	}
+
+	if gotOverflow != 1 {
+		t.Errorf("expected exactly 1 overflow rejection, got %d", gotOverflow)
+	}
+	if gotSuccess != pendingSessionBufferLimit {
+		t.Errorf("expected %d successful resolutions, got %d", pendingSessionBufferLimit, gotSuccess)
+	}
+}
+
+// TestPendingRouting_GuardDropDistinctMessage verifies that when the last
+// pending-routing guard drops without registration, parked waiters receive the
+// distinct routing-ended error (not the overflow message) so the two paths are
+// distinguishable in logs and debugging.
+func TestPendingRouting_GuardDropDistinctMessage(t *testing.T) {
+	client := newCloudTestClient()
+	dispose := client.beginPendingSessionRouting()
+
+	const pendingID = "guard-drop-session"
+
+	resultCh := make(chan *jsonrpcError, 1)
+	go func() {
+		_, rpcErr := client.handleUserInputRequest(userInputRequest{
+			SessionID: pendingID,
+			Question:  "Proceed?",
+		})
+		resultCh <- rpcErr
+	}()
+
+	// Give the goroutine time to park.
+	time.Sleep(20 * time.Millisecond)
+
+	// Drop the guard without registering — simulates session.create failing.
+	dispose()
+
+	select {
+	case rpcErr := <-resultCh:
+		if rpcErr == nil {
+			t.Fatal("expected an rpc error after guard drop without registration")
+		}
+		const want = "pending session routing ended before session was registered"
+		if rpcErr.Message != want {
+			t.Errorf("expected exact message %q, got %q", want, rpcErr.Message)
+		}
+		if rpcErr.Code != -32603 {
+			t.Errorf("expected code -32603, got: %d", rpcErr.Code)
+		}
+		// Must NOT contain the overflow message.
+		if strings.Contains(rpcErr.Message, "buffer overflow") {
+			t.Errorf("guard-drop path must not use overflow message, got: %s", rpcErr.Message)
 		}
 	case <-time.After(2 * time.Second):
 		t.Fatal("timed out waiting for rejected waiter")