test(sdk): 补齐新增功能的测试缺口——C++ manifest 帧回路 / Java 回包路径 / C# 派发接线

- C++：ManifestUploadTest.UploadsGzippedManifestToControlPlane——
  RawFakeAgent 握手 + 独立控制面监听，真实 TCP 帧收发 + gzip 解压
  断言 manifest 结构（此前仅编译验证的已知边界补齐）
- Java：InboundInvokeResponseTest——handleInvokeRequest 反射路径，
  违规 payload 回错误帧且 handler 不调用、合法调用、关闭兼容（3 例）
- C#：InboundValidationDispatchTests——HandleInboundRequestAsync
  反射派发路径，开关/错误响应/handler 调用断言（4 例）

Author: cuihairu

sdks/cpp/tests/test_provider_inbound.cpp

@@ -7,6 +7,7 @@
 #include <gtest/gtest.h>
 #include "croupier/sdk/croupier_client.h"
 #include "croupier/sdk/protocol.h"
+#include "croupier/agent/v1/register.pb.h"
 #include "croupier/sdk/v1/provider.pb.h"
 #include "croupier/sdk/v1/invocation.pb.h"
 
@@ -17,6 +18,7 @@
 #include <string>
 #include <thread>
 #include <vector>
+#include <zlib.h>
 
 #ifdef _WIN32
 #include <winsock2.h>
@@ -428,6 +430,143 @@ TEST(ProviderInboundTest, AgentDrainIsIdempotent) {
 }
 
 }  // namespace
+
+// ===== F：控制面 manifest 上传——端到端帧回路 =====
+
+namespace {
+
+socket_t listen_tcp(unsigned short* out_port) {
+    socket_t fd = ::socket(AF_INET, SOCK_STREAM, 0);
+    if (fd == INVALID_SOCK) return INVALID_SOCK;
+    sockaddr_in addr{};
+    addr.sin_family = AF_INET;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+    addr.sin_port = 0;
+    if (::bind(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) != 0 ||
+        ::listen(fd, 1) != 0) {
+        closesocket(fd);
+        return INVALID_SOCK;
+    }
+    sockaddr_in bound{};
+    socklen_t blen = sizeof(bound);
+    ::getsockname(fd, reinterpret_cast<sockaddr*>(&bound), &blen);
+    *out_port = ntohs(bound.sin_port);
+    return fd;
+}
+
+std::vector<uint8_t> recv_exact(socket_t conn, size_t len) {
+    std::vector<uint8_t> out(len);
+    size_t got = 0;
+    while (got < len) {
+        int n = static_cast<int>(::recv(conn, reinterpret_cast<char*>(out.data()) + got,
+                                        static_cast<int>(len - got), 0));
+        if (n <= 0) return {};
+        got += static_cast<size_t>(n);
+    }
+    return out;
+}
+
+std::string gzip_decompress(const std::string& compressed) {
+    std::string out;
+    z_stream stream{};
+    if (inflateInit2(&stream, 15 + 16) != Z_OK) return out;
+    stream.next_in = reinterpret_cast<Bytef*>(const_cast<char*>(compressed.data()));
+    stream.avail_in = static_cast<uInt>(compressed.size());
+    char buffer[4096];
+    int status = Z_OK;
+    do {
+        stream.next_out = reinterpret_cast<Bytef*>(buffer);
+        stream.avail_out = sizeof(buffer);
+        status = inflate(&stream, Z_NO_FLUSH);
+        if (status != Z_OK && status != Z_STREAM_END && status != Z_BUF_ERROR) break;
+        out.append(buffer, sizeof(buffer) - stream.avail_out);
+    } while (status != Z_STREAM_END);
+    inflateEnd(&stream);
+    return out;
+}
+
+}  // namespace
+
+TEST(ManifestUploadTest, UploadsGzippedManifestToControlPlane) {
+    // Agent（复用 RawFakeAgent 完整握手）+ 控制面（独立监听）
+    RawFakeAgent agent;
+    std::thread agent_thread([&] { agent.AcceptAndHandshake(); });
+
+    unsigned short control_port = 0;
+    socket_t control_fd = listen_tcp(&control_port);
+    ASSERT_NE(control_fd, INVALID_SOCK);
+
+    std::atomic<bool> got_manifest{false};
+    std::thread control_thread([&] {
+        socket_t conn = ::accept(control_fd, nullptr, nullptr);
+        if (conn == INVALID_SOCK) return;
+        auto header = recv_exact(conn, 4);
+        if (header.empty()) return;
+        uint32_t len = (uint32_t(header[0]) << 24) | (uint32_t(header[1]) << 16) |
+                       (uint32_t(header[2]) << 8) | uint32_t(header[3]);
+        auto frame_body = recv_exact(conn, len);
+        constexpr size_t kHeaderSize = 8;  // version(1) + msg_id(3) + req_id(4)
+        if (frame_body.size() < kHeaderSize) return;
+        uint32_t msg_id = protocol::GetMsgID(frame_body.data() + 1);
+        ASSERT_EQ(msg_id, static_cast<unsigned>(protocol::MSG_REGISTER_CAPABILITIES_REQ));
+        uint32_t req_id = (uint32_t(frame_body[4]) << 24) | (uint32_t(frame_body[5]) << 16) |
+                          (uint32_t(frame_body[6]) << 8) | uint32_t(frame_body[7]);
+
+        ::croupier::agent::v1::RegisterCapabilitiesRequest req;
+        ASSERT_TRUE(req.ParseFromArray(frame_body.data() + kHeaderSize,
+                                       static_cast<int>(frame_body.size() - kHeaderSize)));
+        std::string decompressed =
+            gzip_decompress(std::string(req.manifest_json_gz().begin(),
+                                        req.manifest_json_gz().end()));
+        EXPECT_NE(decompressed.find("\"provider\""), std::string::npos);
+        EXPECT_NE(decompressed.find("player.ban"), std::string::npos);
+        got_manifest.store(true);
+
+        // 回确认帧
+        ::croupier::agent::v1::RegisterCapabilitiesResponse ack;
+        std::string ack_out;
+        ack.SerializeToString(&ack_out);
+        auto resp_frame = protocol::NewMessage(
+            protocol::GetResponseMsgID(msg_id), req_id,
+            std::vector<uint8_t>(ack_out.begin(), ack_out.end()));
+        std::vector<uint8_t> wrapped(4 + resp_frame.size());
+        wrapped[0] = static_cast<uint8_t>((resp_frame.size() >> 24) & 0xFF);
+        wrapped[1] = static_cast<uint8_t>((resp_frame.size() >> 16) & 0xFF);
+        wrapped[2] = static_cast<uint8_t>((resp_frame.size() >> 8) & 0xFF);
+        wrapped[3] = static_cast<uint8_t>(resp_frame.size() & 0xFF);
+        std::memcpy(wrapped.data() + 4, resp_frame.data(), resp_frame.size());
+        ::send(conn, reinterpret_cast<const char*>(wrapped.data()),
+               static_cast<int>(wrapped.size()), 0);
+        closesocket(conn);
+    });
+
+    ClientConfig config;
+    config.agent_addr = agent.address();
+    config.control_addr = "127.0.0.1:" + std::to_string(control_port);
+    config.service_id = "cpp-manifest-test";
+    config.game_id = "game-test";
+    config.env = "development";
+    config.timeout_seconds = 5;
+    config.disable_logging = true;
+
+    CroupierClient client(config);
+    FunctionDescriptor desc;
+    desc.id = "player.ban";
+    desc.version = "1.0.0";
+    desc.input_schema = R"({"type":"object","properties":{"id":{"type":"string"}}})";
+    ASSERT_TRUE(client.RegisterFunction(desc, [](const std::string&, const std::string&) {
+        return std::string("ok");
+    }));
+
+    // Connect 内部：agent 握手成功后向控制面上传 manifest（best-effort）
+    ASSERT_TRUE(client.Connect());
+    control_thread.join();
+    agent_thread.join();
+    client.Close();
+    closesocket(control_fd);
+    ASSERT_TRUE(got_manifest.load());
+}
+
 }  // namespace croupier::sdk::test
 
 namespace croupier::sdk::test {

sdks/csharp/src/Croupier.Sdk.Tests/InboundValidationDispatchTests.cs

@@ -0,0 +1,122 @@
+// Copyright 2025 Croupier Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System.Text.Json;
+using Croupier.Sdk.Models;
+using Croupier.Sdk.Transport;
+using Croupier.Sdk.V1;
+using Google.Protobuf;
+using Xunit;
+
+namespace Croupier.Sdk.Tests;
+
+/// <summary>
+/// F：Provider 侧入站校验的派发接线（HandleInboundRequestAsync 路径）。
+/// </summary>
+public class InboundValidationDispatchTests
+{
+    private sealed class CountingHandler : IFunctionHandler
+    {
+        public int Calls;
+        public Task<string> HandleAsync(FunctionContext context, string payload)
+        {
+            Calls++;
+            return Task.FromResult("ok");
+        }
+    }
+
+    private static CroupierClient NewClient(bool validate, out CountingHandler handler)
+    {
+        handler = new CountingHandler();
+        var config = new ClientConfig
+        {
+            ValidateInputPayloads = validate,
+        };
+        var client = new CroupierClient(config);
+        client.RegisterFunction(new FunctionDescriptor
+        {
+            Id = "player.ban",
+            Version = "1.0.0",
+            InputSchema = "{\"type\":\"object\",\"properties\":{\"id\":{\"type\":\"string\"}},\"required\":[\"id\"]}",
+        }, handler);
+        return client;
+    }
+
+    private static byte[] InvokeBody(string functionId, string payload)
+    {
+        var request = new InvokeRequest
+        {
+            FunctionId = functionId,
+            Payload = ByteString.CopyFromUtf8(payload),
+        };
+        return Google.Protobuf.MessageExtensions.ToByteArray(request);
+    }
+
+    private static async Task<byte[]> DispatchAsync(CroupierClient client, byte[] body)
+    {
+        var method = typeof(CroupierClient).GetMethod("HandleInboundRequestAsync",
+            System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance);
+        Assert.NotNull(method);
+        var task = (Task<byte[]>?)method!.Invoke(client, new object[] { Protocol.MsgInvokeRequest, 1, body });
+        Assert.NotNull(task);
+        return await task;
+    }
+
+    private static async Task<JsonElement?> PayloadAsJsonAsync(byte[] response)
+    {
+        var invokeResponse = InvokeResponse.Parser.ParseFrom(response);
+        var text = invokeResponse.Payload.ToStringUtf8();
+        if (text.Length == 0 || text[0] != '{') return null;
+        return JsonDocument.Parse(text).RootElement;
+    }
+
+    [Fact]
+    public async Task Config_Defaults_To_Off()
+    {
+        Assert.False(new ClientConfig().ValidateInputPayloads);
+    }
+
+    [Fact]
+    public async Task InvalidPayload_ReturnsErrorAndSkipsHandler()
+    {
+        var client = NewClient(validate: true, out var handler);
+        var response = await DispatchAsync(client, InvokeBody("player.ban", "{}"));
+        var payload = await PayloadAsJsonAsync(response);
+        Assert.NotNull(payload);
+        Assert.True((payload!.Value.TryGetProperty("error", out var error)
+                     && error.GetString()!.Contains("payload validation failed")),
+            "expected validation error, got: " + payload);
+        Assert.Equal(0, handler.Calls);
+    }
+
+    [Fact]
+    public async Task ValidPayload_InvokesHandler()
+    {
+        var client = NewClient(validate: true, out var handler);
+        var response = await DispatchAsync(client, InvokeBody("player.ban", "{\"id\":\"p1\"}"));
+        var payload = await PayloadAsJsonAsync(response);
+        Assert.Null(payload);
+        Assert.Equal(1, handler.Calls);
+    }
+
+    [Fact]
+    public async Task DisabledFlag_KeepsLegacyBehavior()
+    {
+        var client = NewClient(validate: false, out var handler);
+        var response = await DispatchAsync(client, InvokeBody("player.ban", "{}"));
+        var payload = await PayloadAsJsonAsync(response);
+        Assert.Null(payload);
+        Assert.Equal(1, handler.Calls);
+    }
+}

sdks/java/src/test/java/io/github/cuihairu/croupier/sdk/InboundInvokeResponseTest.java

@@ -0,0 +1,89 @@
+package io.github.cuihairu.croupier.sdk;
+
+import io.github.cuihairu.croupier.sdk.transport.Protocol;
+import io.github.cuihairu.croupier.sdk.wire.SdkWireMessages;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * F：入站校验的 invokeInbound 回包路径——违规 payload 回
+ * {"error":"payload validation failed: …"}，handler 不被调用。
+ */
+public class InboundInvokeResponseTest {
+
+    private CroupierClientImpl newClient(boolean validate, FunctionDescriptor descriptor,
+                                          FunctionHandler handler) throws CroupierException {
+        ClientConfig config = new ClientConfig();
+        config.setValidateInputPayloads(validate);
+        CroupierClientImpl client = new CroupierClientImpl(config);
+        if (descriptor != null) {
+            client.registerFunction(descriptor, handler);
+        }
+        return client;
+    }
+
+    private byte[] handleInvoke(CroupierClientImpl client, String functionId, String payload)
+            throws Exception {
+        Method method = CroupierClientImpl.class.getDeclaredMethod("handleInvokeRequest", byte[].class);
+        method.setAccessible(true);
+        return (byte[]) method.invoke(client, SdkWireMessages.encodeInvokeRequest(
+                new SdkWireMessages.InvokeRequest(functionId, "",
+                        payload.getBytes(StandardCharsets.UTF_8), Map.of())));
+    }
+
+    private FunctionDescriptor descriptorWithRequiredId() {
+        FunctionDescriptor descriptor = new FunctionDescriptor("player.ban", "1.0.0");
+        descriptor.setInputSchema(
+                "{\"type\":\"object\",\"properties\":{\"id\":{\"type\":\"string\"}},\"required\":[\"id\"]}");
+        return descriptor;
+    }
+
+    @Test
+    @Timeout(5)
+    public void invalidPayloadReturnsErrorAndSkipsHandler() throws Exception {
+        AtomicInteger calls = new AtomicInteger();
+        CroupierClientImpl client = newClient(true, descriptorWithRequiredId(),
+                (metadata, payload) -> {
+                    calls.incrementAndGet();
+                    return "ok";
+                });
+
+        byte[] response = handleInvoke(client, "player.ban", "{}");
+        String body = new String(response, StandardCharsets.UTF_8);
+        assertTrue(body.contains("payload validation failed"), "body=" + body);
+        assertFalse(body.contains("ok"));
+        assertEquals(0, calls.get(), "handler must not be invoked");
+    }
+
+    @Test
+    @Timeout(5)
+    public void validPayloadInvokesHandler() throws Exception {
+        CroupierClientImpl client = newClient(true, descriptorWithRequiredId(),
+                (metadata, payload) -> "done");
+        byte[] response = handleInvoke(client, "player.ban", "{\"id\":\"p1\"}");
+        assertEquals("done", new String(SdkWireMessages.decodeInvokeResponse(response).payload, StandardCharsets.UTF_8));
+    }
+
+    @Test
+    @Timeout(5)
+    public void disabledFlagKeepsLegacyBehavior() throws Exception {
+        AtomicInteger calls = new AtomicInteger();
+        CroupierClientImpl client = newClient(false, descriptorWithRequiredId(),
+                (metadata, payload) -> {
+                    calls.incrementAndGet();
+                    return "ok";
+                });
+        byte[] response = handleInvoke(client, "player.ban", "{}");
+        assertEquals("ok", new String(SdkWireMessages.decodeInvokeResponse(response).payload, StandardCharsets.UTF_8));
+        assertEquals(1, calls.get());
+    }
+}