fix(sdks): 五语言入站处理统一有界并发——读循环只投递，默认 CPU 核数

用户指正：只有 Go 是 worker 池不合理，且默认数应为 CPU 核数。
全量审计后统一各语言入站模型（同一定式：读循环只投递 + 固定
worker 池（默认 = CPU 核数，可配）+ 队列容量 workers×4 + 满
快速回错误响应让 Agent failover 接管）：

- Go：默认 8 → runtime.NumCPU()（下限 2），队列 32 → workers×4
- JS：读循环内 await handleInbound（头部阻塞）→ dispatchInbound
  投递 + 动态 worker 循环（默认 os.cpus().length）
- Python：reader 线程同步调 handler（头部阻塞）→
  ThreadPoolExecutor（默认 max(2, os.cpu_count())）+ 计数上限
- C#：fire-and-forget 无界 Task → SemaphoreSlim（默认
  ProcessorCount）+ Interlocked 计数上限 + 满时立即回空响应
- Java：readLoop 只路由响应、请求帧直接丢弃（handleLocalRequest
  生产零调用——Agent 调用根本到不了 handler）→ 补 InboundListener
  + ThreadPoolExecutor（默认 CPU 核数）+ CroupierClientImpl 接线
  attachInboundListener 路由 handleLocalRequest
- C++：同 Java 病（请求帧按 unknown req_id 丢弃）→ TCPTransport
  补 SetInboundHandler + 条件变量 worker 池（默认
  hardware_concurrency）+ CroupierClient 实现 handleAgentRequest
  （invoke + keepalive pong）并双接线（connect/reconnect）

Java/C++ 属于功能性缺陷修复：此前这两个 SDK 的 provider 模式
收不到任何 Agent 调用（帧被丢弃）。

本地验证：Go test/JS tsc/Python import/Java gradle/C# dotnet
全过；C++ 无本地 cmake 由 CI - C++ SDK 验证。

Author: cuihairu

sdks/cpp/include/croupier/sdk/tcp_transport.h

@@ -141,6 +141,17 @@ class TCPTransport {
         }
     };
 
+    // ---- Inbound (Agent -> Provider calls) ----
+    // Read loop only dispatches; handlers run on a bounded worker pool
+    // (default = hardware concurrency, queue = workers * 4, overflow
+    // fast-fails with an empty response so Agent failover takes over).
+    using InboundHandler = std::function<std::vector<uint8_t>(uint32_t msg_id, uint32_t req_id, const std::vector<uint8_t>& body)>;
+
+    void SetInboundHandler(InboundHandler handler);
+    void DispatchInbound(uint32_t msg_id, uint32_t req_id, std::vector<uint8_t> body);
+    void WriteResponseSilently(uint32_t resp_msg_id, uint32_t req_id, const std::vector<uint8_t>& body);
+    static int InboundWorkerCount();
+
     void ReadLoop();
     int ReadFully(void* buf, size_t count);
     static void PutMsgId(uint8_t* buf, uint32_t msg_id);
@@ -159,6 +170,13 @@ class TCPTransport {
     std::unordered_map<uint32_t, std::unique_ptr<ResponseLatch>> pending_responses_;
     std::mutex pending_mutex_;
     std::thread read_thread_;
+    InboundHandler inbound_handler_;
+    std::mutex inbound_pool_mutex_;
+    std::vector<std::thread> inbound_workers_;
+    std::queue<std::tuple<uint32_t, uint32_t, std::vector<uint8_t>>> inbound_queue_;
+    std::condition_variable inbound_cv_;
+    std::atomic<int> inbound_queued_{0};
+    bool inbound_pool_started_ = false;
 
     static constexpr size_t FRAME_HEADER_BYTES = 4;
     static constexpr size_t PROTOCOL_HEADER_SIZE = 8;

sdks/cpp/src/croupier_client.cpp

@@ -404,6 +404,9 @@ class CroupierClient::Impl {
             std::unique_ptr<TCPTransport> replacement =
                 std::make_unique<TCPTransport>(agent_address.host, agent_address.port, config_.timeout_seconds * 1000);
             replacement->SetConnectTimeout(config_.connect_timeout_seconds * 1000);
+            replacement->SetInboundHandler([this](uint32_t msg_id, uint32_t req_id, const std::vector<uint8_t>& body) {
+                return handleAgentRequest(msg_id, req_id, body);
+            });
             replacement->Connect();
             std::string session_id = registerWithAgent(*replacement);
 
@@ -420,6 +423,35 @@ class CroupierClient::Impl {
         }
     }
 
+    // handleAgentRequest 处理 Agent -> Provider 调用（invoke / start task），
+    // 由 TCPTransport 的有界 worker 池并发执行（读循环只投递）。
+    std::vector<uint8_t> handleAgentRequest(uint32_t msg_id, uint32_t /*req_id*/, const std::vector<uint8_t>& body) {
+        try {
+            if (msg_id == protocol::MSG_INVOKE_REQUEST) {
+                auto req = ParseMessage<::croupier::sdk::v1::InvokeRequest>(body, "InvokeRequest");
+                auto it = handlers_.find(req.function_id());
+                if (it == handlers_.end()) {
+                    SDK_LOG_ERROR("Agent invoke: function not found: " + req.function_id());
+                    return {};
+                }
+                std::string context = "{}";
+                std::string payload(req.payload().begin(), req.payload().end());
+                std::string result = it->second(context, payload);
+                ::croupier::sdk::v1::InvokeResponse resp;
+                resp.set_payload(result);
+                return SerializeMessage(resp);
+            }
+            if (msg_id == protocol::MSG_PROVIDER_HEARTBEAT_REQUEST) {
+                // keepalive pong（agent 侧探针）
+                ::croupier::sdk::v1::ProviderHeartbeatResponse resp;
+                return SerializeMessage(resp);
+            }
+        } catch (const std::exception& e) {
+            SDK_LOG_ERROR(std::string("Agent request handling failed: ") + e.what());
+        }
+        return {};
+    }
+
     bool Connect() {
         if (connected_)
             return true;
@@ -433,6 +465,9 @@ class CroupierClient::Impl {
             const auto agent_address = ParseTCPAddress(config_.agent_addr);
             auto transport =
                 std::make_unique<TCPTransport>(agent_address.host, agent_address.port, config_.timeout_seconds * 1000);
+            transport->SetInboundHandler([this](uint32_t msg_id, uint32_t req_id, const std::vector<uint8_t>& body) {
+                return handleAgentRequest(msg_id, req_id, body);
+            });
             transport->Connect();
             std::string session_id = registerWithAgent(*transport);
 

sdks/cpp/src/tcp_transport.cpp

@@ -505,15 +505,21 @@ void TCPTransport::ReadLoop() {
         std::vector<uint8_t> body(body_size);
         std::memcpy(body.data(), payload.data() + PROTOCOL_HEADER_SIZE, body_size);
 
-        // Route to pending request
-        std::lock_guard<std::mutex> lock(pending_mutex_);
-        auto it = pending_responses_.find(req_id);
-        if (it != pending_responses_.end()) {
-            it->second->Signal(std::move(body), msg_id);
+        if (protocol::IsRequest(msg_id)) {
+            // Agent -> Provider call: dispatch to bounded worker pool.
+            // (Read loop never executes handler logic — inline processing
+            // would head-of-line block every request on one slow handler.)
+            DispatchInbound(msg_id, req_id, std::move(body));
         } else {
-            // Debug: log unmatched response
-            std::cerr << "[DEBUG] TCPTransport: Received response for unknown req_id: " << req_id
-                      << ", msg_id: " << msg_id << ", body_size: " << body_size << '\n';
+            // Route response to pending request
+            std::lock_guard<std::mutex> lock(pending_mutex_);
+            auto it = pending_responses_.find(req_id);
+            if (it != pending_responses_.end()) {
+                it->second->Signal(std::move(body), msg_id);
+            } else {
+                std::cerr << "[DEBUG] TCPTransport: Received response for unknown req_id: " << req_id
+                          << ", msg_id: " << msg_id << ", body_size: " << body_size << '\n';
+            }
         }
     }
 
@@ -908,5 +914,83 @@ bool TCPServer::SendMessage(socket_t sock, uint32_t msg_type, uint32_t req_id, c
     return sent == static_cast<ssize_t>(frame.size());
 }
 
+// ---- Inbound dispatch ----
+
+int TCPTransport::InboundWorkerCount() {
+    unsigned int n = std::thread::hardware_concurrency();
+    return n == 0 ? 2 : static_cast<int>(std::max(2u, n));
+}
+
+void TCPTransport::SetInboundHandler(InboundHandler handler) {
+    inbound_handler_ = std::move(handler);
+}
+
+void TCPTransport::DispatchInbound(uint32_t msg_id, uint32_t req_id, std::vector<uint8_t> body) {
+    if (!inbound_handler_) {
+        return;
+    }
+    // 惰性启动固定 worker 池（默认 = 硬件并发数）
+    {
+        std::lock_guard<std::mutex> lock(inbound_pool_mutex_);
+        if (!inbound_pool_started_) {
+            int workers = InboundWorkerCount();
+            for (int i = 0; i < workers; ++i) {
+                inbound_workers_.emplace_back([this] {
+                    for (;;) {
+                        std::tuple<uint32_t, uint32_t, std::vector<uint8_t>> task;
+                        {
+                            std::unique_lock<std::mutex> lock(inbound_pool_mutex_);
+                            inbound_cv_.wait(lock, [this] { return !inbound_queue_.empty(); });
+                            task = std::move(inbound_queue_.front());
+                            inbound_queue_.pop();
+                        }
+                        auto [mid, rid, tbody] = std::move(task);
+                        std::vector<uint8_t> resp;
+                        try {
+                            resp = inbound_handler_(mid, rid, tbody);
+                        } catch (const std::exception& e) {
+                            std::cerr << "[ERROR] inbound handler: " << e.what() << '\n';
+                            resp.clear();
+                        }
+                        WriteResponseSilently(protocol::GetResponseMsgID(mid), rid, resp);
+                        inbound_queued_.fetch_sub(1);
+                    }
+                });
+            }
+            inbound_pool_started_ = true;
+        }
+    }
+    int workers = InboundWorkerCount();
+    if (inbound_queued_.load() >= workers * 4) {
+        // 队列满：立即回空响应，Agent 侧 failover 接管。
+        std::cerr << "[WARN] inbound queue full, fast-failing req_id=" << req_id << '\n';
+        WriteResponseSilently(protocol::GetResponseMsgID(msg_id), req_id, {});
+        return;
+    }
+    inbound_queued_.fetch_add(1);
+    {
+        std::lock_guard<std::mutex> lock(inbound_pool_mutex_);
+        inbound_queue_.emplace(msg_id, req_id, std::move(body));
+    }
+    inbound_cv_.notify_one();
+}
+
+void TCPTransport::WriteResponseSilently(uint32_t resp_msg_id, uint32_t req_id, const std::vector<uint8_t>& body) {
+    try {
+        auto frame = protocol::NewMessage(resp_msg_id, req_id, body);
+        std::vector<uint8_t> wrapped(4 + frame.size());
+        wrapped[0] = static_cast<uint8_t>((frame.size() >> 24) & 0xFF);
+        wrapped[1] = static_cast<uint8_t>((frame.size() >> 16) & 0xFF);
+        wrapped[2] = static_cast<uint8_t>((frame.size() >> 8) & 0xFF);
+        wrapped[3] = static_cast<uint8_t>(frame.size() & 0xFF);
+        std::memcpy(wrapped.data() + 4, frame.data(), frame.size());
+        ssize_t sent = send(socket_, reinterpret_cast<const char*>(wrapped.data()),
+                             static_cast<int>(wrapped.size()), 0);
+        (void)sent;
+    } catch (...) {
+        // best-effort
+    }
+}
+
 } // namespace sdk
 } // namespace croupier

sdks/csharp/src/Croupier.Sdk/Transport/TCPTransport.cs

@@ -14,6 +14,7 @@
 
 using System.Net.Sockets;
 using System.Collections.Concurrent;
+using System.Threading;
 using Croupier.Sdk.Logging;
 
 namespace Croupier.Sdk.Transport;
@@ -54,6 +55,14 @@ public sealed class TCPTransport : IClientTransport
     // Handler for inbound requests from Agent (e.g., InvokeRequest)
     private Func<int, int, byte[], Task<byte[]>>? _inboundRequestHandler;
 
+    // Inbound worker pool: bounded concurrency (default = CPU cores).
+    // Fire-and-forget per request is unbounded; a slow handler storm can
+    // exhaust memory. Queue capacity = workers * 4; overflow fast-fails
+    // with an empty response so the Agent-side failover takes over.
+    private readonly SemaphoreSlim _inboundLimiter = new(
+        Math.Max(2, Environment.ProcessorCount), Math.Max(2, Environment.ProcessorCount));
+    private int _inboundQueued;
+
     /// <summary>
     /// Gets whether the transport is connected.
     /// </summary>
@@ -334,7 +343,7 @@ private async Task ReadLoop(CancellationToken cancellationToken)
                     // awaiting a response to its own request. Do not await the
                     // callback in the sole read loop, otherwise its response
                     // cannot be read and both peers time out.
-                    _ = HandleInboundRequestAsync(parsed.MsgId, parsed.ReqId, parsed.Body, cancellationToken);
+                    DispatchInbound(parsed.MsgId, parsed.ReqId, parsed.Body, cancellationToken);
                 }
             }
             catch (OperationCanceledException)
@@ -372,6 +381,47 @@ private async Task ReadLoop(CancellationToken cancellationToken)
         _pending.Clear();
     }
 
+    private void DispatchInbound(int msgId, int reqId, byte[] body, CancellationToken cancellationToken)
+    {
+        int capacity = Math.Max(2, Environment.ProcessorCount) * 4;
+        if (Interlocked.CompareExchange(ref _inboundQueued, 0, 0) >= capacity)
+        {
+            // Queue full: respond immediately (empty) so the Agent fails over.
+            _ = WriteFrameAsync(
+                Protocol.NewMessage(Protocol.GetResponseMsgId(msgId), reqId, new byte[0]),
+                cancellationToken);
+            return;
+        }
+        Interlocked.Increment(ref _inboundQueued);
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await _inboundLimiter.WaitAsync(cancellationToken).ConfigureAwait(false);
+                try
+                {
+                    await HandleInboundRequest(msgId, reqId, body, cancellationToken).ConfigureAwait(false);
+                }
+                finally
+                {
+                    _ = _inboundLimiter.Release();
+                }
+            }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                // Normal shutdown.
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError("TCPTransport", $"Inbound request processing failed: {ex.Message}", ex);
+            }
+            finally
+            {
+                Interlocked.Decrement(ref _inboundQueued);
+            }
+        });
+    }
+
     private async Task HandleInboundRequestAsync(int msgId, int reqId, byte[] body, CancellationToken cancellationToken)
     {
         try

sdks/go/pkg/croupier/transport/config.go

@@ -59,8 +59,10 @@ type Config struct {
 	// requests sent by the Agent over this connection are handled in place.
 	InboundHandler InboundHandler
 
-	// InboundWorkers 是处理 Agent 入站请求的固定 worker 数（0=默认 8）。
-	// 显式有界——每请求一个 goroutine 的无界模型在并发下会爆内存。
+	// InboundWorkers 是处理 Agent 入站请求的固定 worker 数
+	//（0=默认 runtime.NumCPU()——CPU 核数即并发处理上限，handler
+	// 通常含 IO 等待时可在连接配置里显式调高）。显式有界——每请求
+	// 一个 goroutine 的无界模型在并发下会爆内存。
 	InboundWorkers int
 	// InboundQLen 是入站请求队列长度（0=默认 32）。队列满时新请求
 	// 立即回错误响应（Agent 侧 failover 接管），不排队积累内存。

sdks/go/pkg/croupier/transport/tcp_client.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"runtime"
 	"strconv"
 	"strings"
 	"sync"
@@ -124,11 +125,15 @@ func NewTCPClient(config *Config) (*TCPClient, error) {
 	if config.InboundHandler != nil {
 		workers := config.InboundWorkers
 		if workers <= 0 {
-			workers = 8
+			workers = runtime.NumCPU()
+			if workers < 2 {
+				workers = 2
+			}
 		}
 		qlen := config.InboundQLen
 		if qlen <= 0 {
-			qlen = 32
+			// 队列容量与 worker 数成比例：突发吸收 4 轮 worker 满载
+			qlen = workers * 4
 		}
 		client.inbox = make(chan inboundTask, qlen)
 		client.inboxWg.Add(workers)

sdks/java/src/main/java/io/github/cuihairu/croupier/sdk/CroupierClientImpl.java

@@ -485,6 +485,7 @@ private void reconnectOnce() throws Exception {
             TransportClient old = transport;
             transport = nextTransport;
             sessionId = response.sessionId;
+            attachInboundListener(nextTransport);
             if (old != null) {
                 old.close();
             }
@@ -496,6 +497,13 @@ private void reconnectOnce() throws Exception {
         }
     }
 
+    /** 把 transport 的入站请求路由到本地 handler（Agent -> Provider 调用）。 */
+    private void attachInboundListener(TransportClient client) {
+        if (client instanceof io.github.cuihairu.croupier.sdk.transport.TCPTransport tcp) {
+            tcp.setInboundListener((msgId, requestId, body) -> handleLocalRequest(msgId, requestId, body));
+        }
+    }
+
     private byte[] handleLocalRequest(int msgType, int requestId, byte[] body) throws Exception {
         return switch (msgType) {
             case Protocol.MSG_INVOKE_REQUEST -> handleInvokeRequest(body);