fix(sdk): 审查发现修复——schemadiff 枚举方向按数据流判定 + manifest 上传 fire-and-forget

审查发现 #1（设计缺陷）：schemadiff 枚举方向标反
- input_schema：收窄 = breaking（旧调用发被删值会被拒）、扩张 = compatible
- output_schema：扩张 = breaking（消费方见到新值）、收窄 = compatible
- 原实现只按「新增 = breaking」单向判定，测试固化了错误语义；按 source
  分方向修正，测试改为四象限断言（input/output × 收窄/扩张）

审查发现 #2（健壮性）：manifest 上传阻塞注册主路径
- Go：Register 拆 maybeRegisterCapabilitiesAsync（快照 functions/
  serviceID 后台 goroutine，避免与重注册竞态），注册主路径不再被
  控制面连接/调用阻塞最长 10s
- JS：fire-and-forget + .catch 兜底；Python：守护线程；Java：守护线程；
  C#：discard await，并把 transport.Connect 移入 fail-open try（顺带
  修复 CroupierClientLifecycleTests 文档化的死控制面中止连接 bug，
  测试改为断言 fail-open 新语义）

Author: cuihairu

internal/function/schemadiff/diff.go

@@ -187,31 +187,61 @@ func diffRequired(source, path string, oldMap, newMap map[string]interface{}, fi
 	}
 }
 
+// diffEnum 枚举方向性判定（审查修正）：破坏方向取决于数据流向——
+//   - input_schema：收窄 = breaking（旧调用方发被删的值会被服务端拒绝），
+//     扩张 = compatible（旧调用方不受影响）；
+//   - output_schema：扩张 = breaking（消费方会见到新值），
+//     收窄 = compatible（消费方只会见到更少的值）。
 func diffEnum(source, path string, oldMap, newMap map[string]interface{}, findings *[]Finding) {
 	oldEnum, okOld := oldMap["enum"].([]interface{})
 	newEnum, okNew := newMap["enum"].([]interface{})
 	if !okOld || !okNew {
 		return
 	}
-	// 枚举收窄 = breaking：新取值域必须是旧取值域的子集
+	isInput := source == "input_schema"
 	oldValues := make(map[string]bool, len(oldEnum))
 	for _, item := range oldEnum {
 		oldValues[fmt.Sprint(item)] = true
 	}
-	newValues := make([]string, 0, len(newEnum))
+	newValues := make(map[string]bool, len(newEnum))
+	changed := make([]string, 0)
 	for _, item := range newEnum {
 		key := fmt.Sprint(item)
-		newValues = append(newValues, key)
+		newValues[key] = true
 		if !oldValues[key] {
-			*findings = append(*findings, Finding{
-				Severity: SeverityBreaking,
-				Source:   source,
-				Path:     path + "/enum/" + key,
-				Reason:   fmt.Sprintf("枚举新增取值 %q", item),
-			})
+			changed = append(changed, key)
+		}
+	}
+	for _, item := range oldEnum {
+		key := fmt.Sprint(item)
+		if !newValues[key] {
+			changed = append(changed, key)
+		}
+	}
+	sort.Strings(changed)
+	for _, key := range changed {
+		added := newValues[key]
+		breaking := added != isInput // input：删旧值破坏；output：增新值破坏
+		severity := SeverityCompatible
+		reason := fmt.Sprintf("枚举新增取值 %q", key)
+		if !added {
+			reason = fmt.Sprintf("枚举删除取值 %q", key)
+		}
+		if breaking {
+			severity = SeverityBreaking
+			if added {
+				reason = fmt.Sprintf("枚举新增取值 %q（消费方会见到新值）", key)
+			} else {
+				reason = fmt.Sprintf("枚举删除取值 %q（旧调用方发被删值将被拒绝）", key)
+			}
 		}
+		*findings = append(*findings, Finding{
+			Severity: severity,
+			Source:   source,
+			Path:     path + "/enum/" + key,
+			Reason:   reason,
+		})
 	}
-	sort.Strings(newValues)
 }
 
 func requiredSet(node map[string]interface{}) map[string]bool {

internal/function/schemadiff/diff_test.go

@@ -58,21 +58,28 @@ func TestDiffTypeChanged(t *testing.T) {
 	}
 }
 
-// 4. enum 收窄/扩张 = breaking（新取值不在旧取值域）
-func TestDiffEnumNarrowed(t *testing.T) {
+//  4. enum 方向性（审查修正）：input 收窄 = breaking、扩张 = compatible；
+//     output 恰好相反。
+func TestDiffEnumDirectional(t *testing.T) {
 	oldRaw := mustRaw(t, `{"type":"object","properties":{"level":{"type":"string","enum":["low","high"]}}}`)
-	// 收窄（去掉 high）兼容；扩张（加 critical）breaking
-	compatible := mustRaw(t, `{"type":"object","properties":{"level":{"type":"string","enum":["low"]}}}`)
-	if findings := DiffSchemas("input_schema", oldRaw, compatible); HasBreaking(findings) {
-		t.Fatalf("enum narrowing (subset) should be compatible, got %+v", findings)
-	}
+	narrowed := mustRaw(t, `{"type":"object","properties":{"level":{"type":"string","enum":["low"]}}}`)
 	expanded := mustRaw(t, `{"type":"object","properties":{"level":{"type":"string","enum":["low","high","critical"]}}}`)
-	findings := DiffSchemas("input_schema", oldRaw, expanded)
-	if !HasBreaking(findings) {
-		t.Fatalf("expected breaking finding for enum expansion, got %+v", findings)
+
+	// input_schema：收窄 = breaking（旧调用发被删值会被拒）
+	if findings := DiffSchemas("input_schema", oldRaw, narrowed); !HasBreaking(findings) {
+		t.Fatalf("input enum narrowing should be breaking, got %+v", findings)
+	}
+	// input_schema：扩张 = compatible（旧调用方不受影响）
+	if findings := DiffSchemas("input_schema", oldRaw, expanded); HasBreaking(findings) {
+		t.Fatalf("input enum expansion should be compatible, got %+v", findings)
+	}
+	// output_schema：扩张 = breaking（消费方会见到新值）
+	if findings := DiffSchemas("output_schema", oldRaw, expanded); !HasBreaking(findings) {
+		t.Fatalf("output enum expansion should be breaking, got %+v", findings)
 	}
-	if _, ok := findByPath(findings, "$/level/enum/critical"); !ok {
-		t.Fatalf("expected enum/critical finding, got %+v", findings)
+	// output_schema：收窄 = compatible
+	if findings := DiffSchemas("output_schema", oldRaw, narrowed); HasBreaking(findings) {
+		t.Fatalf("output enum narrowing should be compatible, got %+v", findings)
 	}
 }
 

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

@@ -223,12 +223,10 @@ public async Task ConnectAsync_WithControlAddr_RegistersCapabilitiesWithGzippedM
     }
 
     [Fact]
-    public async Task ConnectAsync_WithDeadControlAddr_FailsProviderConnect_DocumentsBug()
+    public async Task ConnectAsync_WithDeadControlAddr_IsFailOpen()
     {
-        // Documents a bug: in RegisterCapabilitiesAsync, transport.Connect()
-        // runs OUTSIDE the try/catch that guards CallAsync. A dead ControlAddr
-        // therefore aborts the whole provider connect, even though capability
-        // registration is supposed to be best-effort (warning + continue).
+        // 审查发现 #2 修复后：manifest 上传 fire-and-forget 且整体 fail-open
+        // （原实现在 try 外 Connect，死控制面会中止整个 provider connect）。
         var probe = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0);
         probe.Start();
         var deadPort = ((System.Net.IPEndPoint)probe.LocalEndpoint).Port;
@@ -241,10 +239,8 @@ public async Task ConnectAsync_WithDeadControlAddr_FailsProviderConnect_Document
         }));
         client.RegisterFunction(Descriptor("fn.cap-dead"), (ctx, payload) => Task.FromResult("{}"));
 
-        var action = () => client.ConnectAsync();
-
-        await action.Should().ThrowAsync<Exception>();
-        client.IsConnected.Should().BeFalse();
+        await client.ConnectAsync();
+        client.IsConnected.Should().BeTrue();
     }
 
     [Fact]
@@ -259,6 +255,12 @@ public async Task ConnectAsync_WithAliveControlAddr_ConnectSucceeds()
         await client.ConnectAsync();
 
         client.IsConnected.Should().BeTrue();
+        // fire-and-forget：能力帧异步到达，轮询等待
+        var deadline = DateTime.UtcNow.AddSeconds(5);
+        while (control.CapabilityRequests.Count == 0 && DateTime.UtcNow < deadline)
+        {
+            await Task.Delay(50);
+        }
         control.CapabilityRequests.Should().ContainSingle();
     }
 

sdks/csharp/src/Croupier.Sdk/CroupierClient.cs

@@ -677,7 +677,9 @@ private async Task ConnectAndRegisterAsync(CancellationToken cancellationToken)
             // Register inbound request handler for InvokeRequest from Agent
             transport.SetInboundRequestHandler(HandleInboundRequestAsync);
 
-            await RegisterCapabilitiesAsync(cancellationToken);
+            // 审查发现 #2：manifest 上传 fire-and-forget——控制面慢/不可达
+            // 不得拖慢注册主路径（方法内部已 fail-open）。
+            _ = RegisterCapabilitiesAsync(cancellationToken);
         }
         catch
         {
@@ -739,10 +741,12 @@ private async Task RegisterCapabilitiesAsync(CancellationToken cancellationToken
             ? _config.ControlAddr["tcp://".Length..]
             : _config.ControlAddr;
         using var transport = _transportFactory(address, _config.TimeoutSeconds * 1000, _config.ConnectTimeoutSeconds * 1000, _logger);
-        transport.Connect();
 
         try
         {
+            // 审查发现 #2：连接也在 fail-open 范围内（原实现在 try 外，
+            // 死控制面会中止整个 provider connect）
+            transport.Connect();
             await transport.CallAsync(
                 Protocol.MsgRegisterCapabilitiesReq,
                 BuildRegisterCapabilitiesRequestData(),

sdks/go/pkg/croupier/manifest_upload_test.go

@@ -44,7 +44,7 @@ func TestTCPManager_BuildManifest(t *testing.T) {
 		{Id: "  "}, // 空 id 跳过
 	}
 
-	manifest := m.buildManifest()
+	manifest := m.buildManifest("go-1", "2.0.0", m.functions)
 	provider, ok := manifest["provider"].(map[string]interface{})
 	if !ok {
 		t.Fatalf("provider missing: %+v", manifest)
@@ -135,7 +135,7 @@ func TestTCPManager_MaybeRegisterCapabilities(t *testing.T) {
 	m.serviceID = "go-1"
 	m.functions = []*sdkv1.ProviderFunctionDescriptor{{Id: "f1"}}
 
-	m.maybeRegisterCapabilities()
+	m.maybeRegisterCapabilities("go-1", "2.0.0", m.functions)
 	if !gotCapabilities {
 		t.Fatal("control plane did not receive a valid manifest")
 	}
@@ -148,7 +148,7 @@ func TestTCPManager_MaybeRegisterCapabilities_NoControlAddr(t *testing.T) {
 	}
 	m := manager.(*TCPManager)
 	// 未配置 ControlAddr：应直接返回（无连接、无 panic）
-	m.maybeRegisterCapabilities()
+	m.maybeRegisterCapabilities("go-1", "2.0.0", m.functions)
 	if strings.TrimSpace(m.config.ControlAddr) != "" {
 		t.Fatal("expected empty ControlAddr")
 	}

sdks/go/pkg/croupier/tcp_manager.go

@@ -294,23 +294,26 @@ func (m *TCPManager) RegisterWithAgent(ctx context.Context, serviceID, serviceVe
 	m.heartbeatStop = cancel
 	go m.heartbeatLoop(ctx)
 
-	// 控制面 manifest 上传（best-effort，不阻断注册结果）
-	m.maybeRegisterCapabilities()
+	// 控制面 manifest 上传（审查发现 #2）：fire-and-forget——控制面慢/
+	// 不可达时不得拖慢注册主路径（同步执行最长 10s）。参数全部取
+	// 快照传值，避免与后续重注册竞态。
+	m.maybeRegisterCapabilitiesAsync()
 
 	return m.sessionID, nil
 }
 
 // buildManifest 构建供控制面注册的能力清单（provider 元数据 + 函数摘要），
 // 与 JS/C# 参考实现同构。
-func (m *TCPManager) buildManifest() map[string]interface{} {
+func (m *TCPManager) buildManifest(serviceID, serviceVersion string,
+	functions []*sdkv1.ProviderFunctionDescriptor) map[string]interface{} {
 	provider := map[string]interface{}{
-		"id":      orDefault(m.serviceID, "go-service"),
-		"version": orDefault(m.serviceVersion, "1.0.0"),
+		"id":      orDefault(serviceID, "go-service"),
+		"version": orDefault(serviceVersion, "1.0.0"),
 		"lang":    orDefault(m.config.ProviderLang, "go"),
 		"sdk":     orDefault(m.config.ProviderSDK, "croupier-go-sdk"),
 	}
-	functions := make([]map[string]interface{}, 0, len(m.functions))
-	for _, descriptor := range m.functions {
+	entries := make([]map[string]interface{}, 0, len(functions))
+	for _, descriptor := range functions {
 		if descriptor == nil || strings.TrimSpace(descriptor.Id) == "" {
 			continue
 		}
@@ -339,9 +342,9 @@ func (m *TCPManager) buildManifest() map[string]interface{} {
 		if descriptor.OutputSchema != "" {
 			entry["outputSchema"] = json.RawMessage(descriptor.OutputSchema)
 		}
-		functions = append(functions, entry)
+		entries = append(entries, entry)
 	}
-	return map[string]interface{}{"provider": provider, "functions": functions}
+	return map[string]interface{}{"provider": provider, "functions": entries}
 }
 
 func orDefault(value, fallback string) string {
@@ -364,14 +367,31 @@ func gzipBytes(data []byte) ([]byte, error) {
 	return buf.Bytes(), nil
 }
 
+// maybeRegisterCapabilitiesAsync 异步 fire-and-forget：先快照上传所需
+// 状态（functions 会在重注册时整体替换），再交后台执行——注册主路径
+// 不再被控制面连接/调用阻塞（审查发现 #2）。
+func (m *TCPManager) maybeRegisterCapabilitiesAsync() {
+	if strings.TrimSpace(m.config.ControlAddr) == "" {
+		return
+	}
+	m.mu.RLock()
+	serviceID, serviceVersion := m.serviceID, m.serviceVersion
+	functions := make([]*sdkv1.ProviderFunctionDescriptor, len(m.functions))
+	copy(functions, m.functions)
+	m.mu.RUnlock()
+
+	go m.maybeRegisterCapabilities(serviceID, serviceVersion, functions)
+}
+
 // maybeRegisterCapabilities 向控制面（config.ControlAddr）上传能力清单。
 // 独立短连接 + 5s 超时；任何失败仅告警，不影响已完成的函数注册。
-func (m *TCPManager) maybeRegisterCapabilities() {
+func (m *TCPManager) maybeRegisterCapabilities(serviceID, serviceVersion string,
+	functions []*sdkv1.ProviderFunctionDescriptor) {
 	if strings.TrimSpace(m.config.ControlAddr) == "" {
 		return
 	}
 
-	manifestJSON, err := json.Marshal(m.buildManifest())
+	manifestJSON, err := json.Marshal(m.buildManifest(serviceID, serviceVersion, functions))
 	if err != nil {
 		logWarnf("register capabilities: marshal manifest: %v", err)
 		return

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

@@ -146,8 +146,11 @@ public CompletableFuture<Void> connect() {
 
                 logger.info("Successfully connected");
 
-                // F：控制面 manifest 上传（best-effort，不阻断连接结果）
-                maybeRegisterCapabilities();
+                // F/审查发现 #2：控制面 manifest 上传 fire-and-forget——
+                // 控制面慢/不可达不得阻塞注册主路径（方法内部已 fail-open）
+                Thread capabilitiesThread = new Thread(this::maybeRegisterCapabilities);
+                capabilitiesThread.setDaemon(true);
+                capabilitiesThread.start();
             } catch (Exception e) {
                 connected.set(false);
                 sessionId = "";