fix(mcp): 修复结果语义丢失、错误误判与生命周期回收

internal/app/bootstrap_test.go

@@ -271,6 +271,9 @@ func TestDefaultRegisterMCPStdioServerRefreshFailure(t *testing.T) {
 	if !strings.Contains(strings.ToLower(err.Error()), "list tools failed") {
 		t.Fatalf("unexpected error: %v", err)
 	}
+	if snapshots := registry.Snapshot(); len(snapshots) != 0 {
+		t.Fatalf("expected failed registration to rollback server, got %+v", snapshots)
+	}
 }
 
 func TestBuildToolRegistryIncludesMCPFromConfig(t *testing.T) {
@@ -434,6 +437,53 @@ func TestBuildMCPRegistryRegisterError(t *testing.T) {
 	}
 }
 
+func TestBuildMCPRegistryRollbackRegisteredServersOnFailure(t *testing.T) {
+	t.Parallel()
+
+	cfg := config.Default().Clone()
+	cfg.Workdir = t.TempDir()
+	cfg.Tools.MCP.Servers = []config.MCPServerConfig{
+		{ID: "docs", Enabled: true, Source: "stdio"},
+		{ID: "search", Enabled: true, Source: "stdio"},
+	}
+
+	closedByServer := map[string]*bool{
+		"docs":   new(bool),
+		"search": new(bool),
+	}
+
+	originalRegister := registerMCPStdioServer
+	t.Cleanup(func() { registerMCPStdioServer = originalRegister })
+	registerMCPStdioServer = func(registry *mcp.Registry, cfg config.Config, server config.MCPServerConfig) error {
+		client := &closeableStubMCPServerClient{closed: closedByServer[strings.TrimSpace(server.ID)]}
+		if err := registry.RegisterServer(server.ID, "stdio", server.Version, client); err != nil {
+			return err
+		}
+		if strings.EqualFold(strings.TrimSpace(server.ID), "search") {
+			return errors.New("search register failed")
+		}
+		return nil
+	}
+
+	registry, err := buildMCPRegistry(cfg)
+	if err == nil || !strings.Contains(err.Error(), "search register failed") {
+		t.Fatalf("expected wrapped register error, got %v", err)
+	}
+	if registry != nil {
+		t.Fatalf("expected nil registry on build failure")
+	}
+	if !*closedByServer["docs"] || !*closedByServer["search"] {
+		t.Fatalf("expected rollback to close all registered servers, got %+v", closedByServer)
+	}
+}
+
+func TestRollbackMCPServersBoundaries(t *testing.T) {
+	t.Parallel()
+
+	rollbackMCPServers(nil, []string{"docs"})
+	rollbackMCPServers(mcp.NewRegistry(), nil)
+}
+
 func TestInitialMCPRefreshTimeoutAndDurationConversion(t *testing.T) {
 	t.Parallel()
 
@@ -614,6 +664,29 @@ func (s *stubMCPServerClient) HealthCheck(ctx context.Context) error {
 	return nil
 }
 
+type closeableStubMCPServerClient struct {
+	closed *bool
+}
+
+func (s *closeableStubMCPServerClient) ListTools(ctx context.Context) ([]mcp.ToolDescriptor, error) {
+	return nil, nil
+}
+
+func (s *closeableStubMCPServerClient) CallTool(ctx context.Context, toolName string, arguments []byte) (mcp.CallResult, error) {
+	return mcp.CallResult{}, nil
+}
+
+func (s *closeableStubMCPServerClient) HealthCheck(ctx context.Context) error {
+	return nil
+}
+
+func (s *closeableStubMCPServerClient) Close() error {
+	if s.closed != nil {
+		*s.closed = true
+	}
+	return nil
+}
+
 func TestHelperProcessAppMCPStdioServer(t *testing.T) {
 	if os.Getenv("GO_WANT_APP_MCP_STDIO_HELPER") != "1" {
 		return

internal/app/mcp_bootstrap.go

@@ -23,6 +23,7 @@ func buildMCPRegistry(cfg config.Config) (*mcp.Registry, error) {
 
 	registry := mcp.NewRegistry()
 	enabledCount := 0
+	registeredServerIDs := make([]string, 0, len(cfg.Tools.MCP.Servers))
 	for index := range cfg.Tools.MCP.Servers {
 		server := cfg.Tools.MCP.Servers[index]
 		if !server.Enabled {
@@ -33,9 +34,12 @@ func buildMCPRegistry(cfg config.Config) (*mcp.Registry, error) {
 		switch strings.ToLower(strings.TrimSpace(server.Source)) {
 		case "", "stdio":
 			if err := registerMCPStdioServer(registry, cfg, server); err != nil {
+				rollbackMCPServers(registry, append(registeredServerIDs, strings.TrimSpace(server.ID)))
 				return nil, fmt.Errorf("app: register mcp server %q: %w", strings.TrimSpace(server.ID), err)
 			}
+			registeredServerIDs = append(registeredServerIDs, strings.TrimSpace(server.ID))
 		default:
+			rollbackMCPServers(registry, registeredServerIDs)
 			return nil, fmt.Errorf("app: unsupported mcp source %q", server.Source)
 		}
 	}
@@ -46,6 +50,16 @@ func buildMCPRegistry(cfg config.Config) (*mcp.Registry, error) {
 	return registry, nil
 }
 
+// rollbackMCPServers 在批量注册失败时回滚已注册 server，避免残留子进程或脏状态。
+func rollbackMCPServers(registry *mcp.Registry, serverIDs []string) {
+	if registry == nil || len(serverIDs) == 0 {
+		return
+	}
+	for index := len(serverIDs) - 1; index >= 0; index-- {
+		_ = registry.UnregisterServer(serverIDs[index])
+	}
+}
+
 // defaultRegisterMCPStdioServer 创建 stdio client 并完成 server 注册与 tools 快照初始化。
 func defaultRegisterMCPStdioServer(registry *mcp.Registry, cfg config.Config, server config.MCPServerConfig) error {
 	env, err := resolveMCPServerEnv(server)
@@ -79,6 +93,7 @@ func defaultRegisterMCPStdioServer(registry *mcp.Registry, cfg config.Config, se
 	refreshCtx, cancel := context.WithTimeout(context.Background(), initialMCPRefreshTimeout(cfg))
 	defer cancel()
 	if err := registry.RefreshServerTools(refreshCtx, serverID); err != nil {
+		_ = registry.UnregisterServer(serverID)
 		return err
 	}
 	return nil

internal/tools/mcp/adapter.go

@@ -132,10 +132,22 @@ func ensureObjectSchema(schema map[string]any) map[string]any {
 		}
 	}
 
-	if strings.TrimSpace(fmt.Sprintf("%v", cloned["type"])) == "" {
+	rawType, hasType := cloned["type"]
+	trimmedType := strings.TrimSpace(fmt.Sprintf("%v", rawType))
+	if !hasType || rawType == nil || trimmedType == "" || strings.EqualFold(trimmedType, "<nil>") {
 		cloned["type"] = "object"
+		if _, ok := cloned["properties"].(map[string]any); !ok {
+			cloned["properties"] = map[string]any{}
+		}
+		return cloned
+	}
+
+	if !strings.EqualFold(trimmedType, "object") {
+		cloned["type"] = "object"
+		cloned["properties"] = map[string]any{}
+		return cloned
 	}
-	if _, ok := cloned["properties"]; !ok {
+	if _, ok := cloned["properties"].(map[string]any); !ok {
 		cloned["properties"] = map[string]any{}
 	}
 	return cloned

internal/tools/mcp/adapter_test.go

@@ -227,3 +227,61 @@ func TestAdapterEnsureObjectSchemaDefaults(t *testing.T) {
 		t.Fatalf("expected properties object, got %+v", schema["properties"])
 	}
 }
+
+func TestAdapterEnsureObjectSchemaNormalizesInvalidType(t *testing.T) {
+	t.Parallel()
+
+	registry := NewRegistry()
+	adapter, err := NewAdapter(registry, "docs", ToolDescriptor{
+		Name: "search",
+		InputSchema: map[string]any{
+			"type": "array",
+			"items": map[string]any{
+				"type": "string",
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("NewAdapter() error = %v", err)
+	}
+	schema := adapter.Schema()
+	if schema["type"] != "object" {
+		t.Fatalf("expected normalized object type, got %v", schema["type"])
+	}
+	if _, ok := schema["properties"].(map[string]any); !ok {
+		t.Fatalf("expected normalized properties object, got %+v", schema["properties"])
+	}
+}
+
+func TestAdapterEnsureObjectSchemaKeepsPropertiesWhenTypeMissing(t *testing.T) {
+	t.Parallel()
+
+	registry := NewRegistry()
+	adapter, err := NewAdapter(registry, "docs", ToolDescriptor{
+		Name: "search",
+		InputSchema: map[string]any{
+			"properties": map[string]any{
+				"q": map[string]any{"type": "string"},
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("NewAdapter() error = %v", err)
+	}
+
+	schema := adapter.Schema()
+	if schema["type"] != "object" {
+		t.Fatalf("expected type object, got %v", schema["type"])
+	}
+	properties, ok := schema["properties"].(map[string]any)
+	if !ok {
+		t.Fatalf("expected properties object, got %+v", schema["properties"])
+	}
+	querySchema, ok := properties["q"].(map[string]any)
+	if !ok {
+		t.Fatalf("expected q schema map, got %+v", properties["q"])
+	}
+	if querySchema["type"] != "string" {
+		t.Fatalf("expected q type string, got %v", querySchema["type"])
+	}
+}

internal/tools/mcp/registry.go

@@ -55,6 +55,11 @@ type ServerClient interface {
 	HealthCheck(ctx context.Context) error
 }
 
+// closeableServerClient 描述支持主动关闭资源的 MCP client 扩展能力。
+type closeableServerClient interface {
+	Close() error
+}
+
 type serverEntry struct {
 	snapshot ServerSnapshot
 	client   ServerClient
@@ -116,11 +121,14 @@ func (r *Registry) UnregisterServer(serverID string) bool {
 	}
 
 	r.mu.Lock()
-	defer r.mu.Unlock()
-	if _, exists := r.servers[normalizedID]; !exists {
+	entry, exists := r.servers[normalizedID]
+	if !exists {
+		r.mu.Unlock()
 		return false
 	}
 	delete(r.servers, normalizedID)
+	r.mu.Unlock()
+	closeServerClient(entry.client)
 	return true
 }
 
@@ -337,3 +345,15 @@ func cloneAny(value any) any {
 		return value
 	}
 }
+
+// closeServerClient 在 server 注销时尽力释放 client 持有的底层资源。
+func closeServerClient(client ServerClient) {
+	if client == nil {
+		return
+	}
+	closeableClient, ok := client.(closeableServerClient)
+	if !ok {
+		return
+	}
+	_ = closeableClient.Close()
+}

internal/tools/mcp/registry_test.go

@@ -19,6 +19,16 @@ type stubServerClient struct {
 	lastArguments []byte
 }
 
+type closableStubServerClient struct {
+	stubServerClient
+	closed bool
+}
+
+func (s *closableStubServerClient) Close() error {
+	s.closed = true
+	return nil
+}
+
 func (s *stubServerClient) ListTools(ctx context.Context) ([]ToolDescriptor, error) {
 	if s.listErr != nil {
 		return nil, s.listErr
@@ -229,6 +239,35 @@ func TestRegistryRegisterAndUnregisterBoundaries(t *testing.T) {
 	}
 }
 
+func TestRegistryUnregisterServerClosesClient(t *testing.T) {
+	t.Parallel()
+
+	registry := NewRegistry()
+	client := &closableStubServerClient{}
+	if err := registry.RegisterServer("docs", "stdio", "v1", client); err != nil {
+		t.Fatalf("register server: %v", err)
+	}
+	if !registry.UnregisterServer("docs") {
+		t.Fatalf("expected unregister success")
+	}
+	if !client.closed {
+		t.Fatalf("expected client to be closed on unregister")
+	}
+}
+
+func TestCloseServerClientBoundaries(t *testing.T) {
+	t.Parallel()
+
+	closeServerClient(nil)
+	closeServerClient(&stubServerClient{})
+
+	client := &closableStubServerClient{}
+	closeServerClient(client)
+	if !client.closed {
+		t.Fatalf("expected closeable client to be closed")
+	}
+}
+
 func TestRegistrySetServerStatusValidation(t *testing.T) {
 	t.Parallel()