feat(worker): 实现基于进程的 Worker 调度与管理

- 新增 WorkerScheduler 接口替代原 WorkerRuntime
- 新增 WorkerEnvType 类型定义 (process/docker/kubevirt)
- 实现 ProcessScheduler 支持 Worker 进程启动/停止/健康检查
- Worker 启动时通过 WebSocket 从 Server 获取 DigitalAssistant 配置
- Server 新增 getConfig 消息处理，从数据库查询配置返回给 Worker
- DigitalAssistant Service 创建时自动启动对应的 Worker 进程
- 支持 --assistant-code 参数指定数字员工标识

架构流程:
用户创建 DigitalAssistant → 保存数据库 → 启动 Worker 进程 →
Worker 连接 Server 获取配置 → 初始化 Agent Runtime → 就绪

Author: ckeyer

.gitignore

@@ -40,3 +40,4 @@ example/
 
 # Prevent accidental Go source files in root directory (except go.mod/go.sum)
 # Note: Actual Go test files should be in backend/tests or appropriate package directories
+docs/superpowers

backend/cmd/singer/worker.go

@@ -3,274 +3,69 @@ package main
 import (
 	"context"
 	"fmt"
-	"net"
-	"net/http"
-	"strings"
-	"time"
 
-	"github.com/gin-gonic/gin"
 	"github.com/insmtx/SingerOS/backend/config"
-	"github.com/insmtx/SingerOS/backend/internal/agent"
-	"github.com/insmtx/SingerOS/backend/internal/agent/externalcli"
-	"github.com/insmtx/SingerOS/backend/internal/eventengine"
-	"github.com/insmtx/SingerOS/backend/internal/infra/mq"
-	singerMCP "github.com/insmtx/SingerOS/backend/mcp"
-	"github.com/insmtx/SingerOS/backend/runtime/engines"
-	"github.com/insmtx/SingerOS/backend/runtime/engines/builtin"
-	"github.com/insmtx/SingerOS/backend/tools"
-	skilltools "github.com/insmtx/SingerOS/backend/tools/skill"
+	"github.com/insmtx/SingerOS/backend/internal/worker/client"
 	"github.com/spf13/cobra"
-	"github.com/ygpkg/yg-go/lifecycle"
+	ygconfig "github.com/ygpkg/yg-go/config"
 	"github.com/ygpkg/yg-go/logs"
 )
 
 var (
-	workerConfigPath string
-	workerServerAddr string
+	workerConfigPath    string
+	workerServerAddr    string
+	workerAssistantCode string
 )
 
 var workerCmd = &cobra.Command{
 	Use:   "worker",
 	Short: "Start the SingerOS background worker",
 	Long:  `Start the background worker service for processing asynchronous tasks and events.`,
 	Run: func(cmd *cobra.Command, args []string) {
-		mcpServer, err := startWorkerMCPServer(workerServerAddr)
-		if err != nil {
-			logs.Fatalf("Failed to start worker MCP server: %v", err)
-			return
-		}
+		ctx := cmd.Context()
 
-		cfg, err := loadWorkerConfig(workerConfigPath, workerServerAddr)
+		worker, err := createWorker(ctx)
 		if err != nil {
-			logs.Fatalf("Failed to load config: %v", err)
+			logs.Fatalf("Failed to create worker: %v", err)
 			return
 		}
 
-		natsUrl := "nats://nats:4222"
-		if cfg.NATS != nil && cfg.NATS.URL != "" {
-			natsUrl = cfg.NATS.URL
-		}
-
-		subscriber, err := mq.NewPublisher(natsUrl)
-		if err != nil {
-			logs.Fatalf("Failed to create event subscriber: %v", err)
-			return
+		if err := worker.Start(ctx); err != nil {
+			logs.Fatalf("Failed to start worker: %v", err)
 		}
-
-		runtimeConfig, err := buildRuntimeConfig()
-		if err != nil {
-			logs.Fatalf("Failed to build runtime config: %v", err)
-			return
-		}
-
-		ctx, cancel := context.WithCancel(context.Background())
-		runner, err := buildRuntimeRunner(ctx, cfg, runtimeConfig)
-		if err != nil {
-			cancel()
-			logs.Fatalf("Failed to create agent runtime: %v", err)
-			return
-		}
-
-		orchestratorInstance := eventengine.NewOrchestrator(subscriber, runner)
-		if err := orchestratorInstance.Start(ctx); err != nil {
-			cancel()
-			logs.Fatalf("Failed to start orchestrator: %v", err)
-			return
-		}
-		logs.Info("Orchestrator started successfully")
-
-		lifecycle.Std().AddCloseFunc(func() error {
-			cancel()
-			return nil
-		})
-		lifecycle.Std().AddCloseFunc(func() error {
-			shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
-			defer shutdownCancel()
-			return mcpServer.Shutdown(shutdownCtx)
-		})
-		lifecycle.Std().AddCloseFunc(subscriber.Close)
-
-		logs.Info("Worker runtime initialized successfully")
-		logs.Info("Worker service started")
-
-		lifecycle.Std().WaitExit()
-
-		logs.Info("Worker exited")
 	},
 }
 
 func init() {
 	workerCmd.Flags().StringVar(&workerConfigPath, "config", "", "Configuration file path")
-	workerCmd.Flags().StringVar(&workerServerAddr, "server-addr", ":8081", "Worker MCP server listen address for runtime CLI bootstrap")
+	workerCmd.Flags().StringVar(&workerServerAddr, "server-addr", ":8080", "Server address for WebSocket connection")
+	workerCmd.Flags().StringVar(&workerAssistantCode, "assistant-code", "", "Assistant code for configuration retrieval")
 	rootCmd.AddCommand(workerCmd)
 }
 
-func startWorkerMCPServer(addr string) (*http.Server, error) {
-	if strings.TrimSpace(addr) == "" {
-		addr = ":8081"
-	}
-
-	listener, err := net.Listen("tcp", addr)
+func createWorker(ctx context.Context) (*client.Worker, error) {
+	_, err := loadWorkerConfig()
 	if err != nil {
-		return nil, fmt.Errorf("listen on %s: %w", addr, err)
-	}
-
-	r := gin.New()
-	v1 := r.Group("/v1")
-	singerMCP.RegisterRoutes(v1, singerMCP.NewServer())
-
-	server := &http.Server{
-		Addr:    addr,
-		Handler: r,
-	}
-
-	go func() {
-		logs.Infof("Worker MCP server listening on %s", listener.Addr().String())
-		if err := server.Serve(listener); err != nil && err != http.ErrServerClosed {
-			logs.Errorf("Worker MCP server stopped unexpectedly: %v", err)
-		}
-	}()
-
-	return server, nil
-}
-
-func loadWorkerConfig(configPath string, bootstrapAddr string) (*config.Config, error) {
-	cfg, err := loadConfig(configPath)
-	if err != nil {
-		return nil, err
-	}
-
-	bootstrapped, err := builtin.BootstrapCLIEngines(context.Background(), cfg.CLI, defaultCLIBootstrapOptions(bootstrapAddr))
-	if err != nil {
-		logs.Warnf("CLI bootstrap failed: %v", err)
-	}
-	if bootstrapped != nil {
-		cfg.CLI = bootstrapped
+		return nil, fmt.Errorf("load config: %w", err)
 	}
 
-	return cfg, nil
+	return client.NewWorker(ctx, &client.WorkerConfig{
+		ServerAddr:    workerServerAddr,
+		AssistantCode: workerAssistantCode,
+		SkillsDir:     "",
+		ToolsEnabled:  true,
+	})
 }
 
-func defaultCLIBootstrapOptions(addr string) builtin.BootstrapOptions {
-	return builtin.BootstrapOptions{
-		MCP: engines.MCPServerConfig{
-			Name:        "singeros",
-			URL:         mcpURLFromAddr(addr),
-			BearerToken: singerMCP.DefaultAuthToken(),
-		},
-	}
-}
-
-func mcpURLFromAddr(addr string) string {
-	host := "localhost"
-	port := "8081"
-
-	if strings.TrimSpace(addr) != "" {
-		if splitHost, splitPort, err := net.SplitHostPort(addr); err == nil {
-			if splitHost != "" && splitHost != "0.0.0.0" && splitHost != "::" && splitHost != "[::]" {
-				host = splitHost
-			}
-			if splitPort != "" {
-				port = splitPort
-			}
-		} else if strings.HasPrefix(addr, ":") {
-			port = strings.TrimPrefix(addr, ":")
-		} else {
-			host = addr
-		}
-	}
-
-	return fmt.Sprintf("http://%s:%s/v1/mcp", host, port)
-}
-
-func buildRuntimeConfig() (agent.Config, error) {
-	catalog, skillDir, err := skilltools.LoadDefaultCatalog()
-	if err != nil {
-		return agent.Config{}, fmt.Errorf("load skills: %w", err)
-	}
-
-	logs.Infof("Loaded %d skills from %s for runtime", len(catalog.List()), skillDir)
-
-	toolRegistry, err := buildTooling(catalog)
-	if err != nil {
-		return agent.Config{}, err
-	}
-
-	return agent.Config{
-		SkillsCatalog: catalog,
-		ToolRegistry:  toolRegistry,
-	}, nil
-}
-
-func buildTooling(catalog *skilltools.Catalog) (*tools.Registry, error) {
-	registry := tools.NewRegistry()
-
-	if err := skilltools.Register(registry, catalog); err != nil {
-		return nil, fmt.Errorf("register skill use tool: %w", err)
-	}
-
-	logs.Infof("Loaded %d tools for runtime", len(registry.List()))
-
-	return registry, nil
-}
-
-func buildRuntimeRunner(ctx context.Context, cfg *config.Config, runtimeConfig agent.Config) (agent.Runner, error) {
-	if cfg == nil {
-		return nil, fmt.Errorf("config is required")
-	}
-
-	router := agent.NewRuntimeRouter(agent.RuntimeKindSingerOS)
-	registered := 0
-
-	if cfg.LLM != nil && cfg.LLM.APIKey != "" {
-		switch cfg.LLM.Provider {
-		case "", "openai":
-			logs.Info("Registering SingerOS agent runtime")
-			singerRunner, err := agent.NewAgent(ctx, cfg.LLM, runtimeConfig)
-			if err != nil {
-				return nil, err
-			}
-			if err := router.Register(agent.RuntimeKindSingerOS, singerRunner); err != nil {
-				return nil, err
-			}
-			registered++
-		default:
-			logs.Warnf("Skipping SingerOS agent runtime for unsupported Eino chat model provider: %s", cfg.LLM.Provider)
-		}
-	}
-
-	cliRegistry, err := builtin.NewRegistryFromConfig(cfg.CLI)
-	if err != nil {
-		return nil, fmt.Errorf("create CLI engine registry: %w", err)
-	}
-	cliNames := cliRegistry.Names()
-	for _, name := range cliNames {
-		engine, ok := cliRegistry.Get(name)
-		if !ok {
-			continue
-		}
-		runner, err := externalcli.NewRunner(name, engine, cfg.LLM)
+func loadWorkerConfig() (*config.Config, error) {
+	if workerConfigPath != "" {
+		cfg := &config.Config{}
+		err := ygconfig.LoadYamlLocalFile(workerConfigPath, cfg)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to load config from %s: %w", workerConfigPath, err)
 		}
-		if err := router.Register(name, runner); err != nil {
-			return nil, err
-		}
-		registered++
-		logs.Infof("Registering external agent CLI runtime: %s", name)
+		return cfg, nil
 	}
 
-	if registered == 0 {
-		return nil, fmt.Errorf("no agent runtime is available")
-	}
-	if cfg.LLM == nil || cfg.LLM.APIKey == "" {
-		if cfg.CLI != nil && cfg.CLI.Default != "" {
-			router.SetDefault(cfg.CLI.Default)
-		} else if len(cliNames) > 0 {
-			router.SetDefault(cliNames[0])
-		}
-	} else {
-		router.SetDefault(agent.RuntimeKindSingerOS)
-	}
-	return router, nil
+	return &config.Config{}, nil
 }

backend/internal/api/router.go

@@ -16,6 +16,7 @@ import (
 	githubprovider "github.com/insmtx/SingerOS/backend/internal/infra/providers/github"
 	"github.com/insmtx/SingerOS/backend/internal/infra/websocket"
 	"github.com/insmtx/SingerOS/backend/internal/service"
+	"github.com/insmtx/SingerOS/backend/internal/worker/scheduler"
 	workerserver "github.com/insmtx/SingerOS/backend/internal/worker/server"
 	singerMCP "github.com/insmtx/SingerOS/backend/mcp"
 	ygmiddleware "github.com/ygpkg/yg-go/apis/runtime/middleware"
@@ -61,14 +62,18 @@ func SetupRouter(cfg config.Config, publisher eventbus.Publisher, db *gorm.DB) *
 	websocket.RegisterWebSocketRoutes(v1, publisher)
 	logs.Info("WebSocket connector registered successfully")
 
-	digitalAssistantService := service.NewDigitalAssistantService(db)
-	handler.RegisterDigitalAssistantRoutes(v1, digitalAssistantService)
-	logs.Info("Digital assistant routes registered successfully")
+	workerScheduler := scheduler.NewProcessScheduler(&scheduler.ProcessConfig{
+		ServerAddr: ":8080",
+	})
 
-	workerServer := workerserver.NewServer()
+	workerServer := workerserver.NewServer(workerScheduler, db)
 	workerServer.RegisterRoutes(v1)
 	logs.Info("Worker server routes registered successfully")
 
+	digitalAssistantService := service.NewDigitalAssistantService(db, workerScheduler)
+	handler.RegisterDigitalAssistantRoutes(v1, digitalAssistantService)
+	logs.Info("Digital assistant routes registered successfully")
+
 	singerMCP.RegisterRoutes(v1, singerMCP.NewServer())
 	logs.Info("MCP routes registered successfully")
 

backend/internal/service/digital_assistant_service.go

@@ -10,18 +10,22 @@ import (
 	"github.com/insmtx/SingerOS/backend/internal/api/auth"
 	"github.com/insmtx/SingerOS/backend/internal/api/contract"
 	"github.com/insmtx/SingerOS/backend/internal/infra/db"
+	"github.com/insmtx/SingerOS/backend/internal/worker"
 	"github.com/insmtx/SingerOS/backend/types"
+	"github.com/ygpkg/yg-go/logs"
 )
 
 var _ contract.DigitalAssistantService = (*digitalAssistantService)(nil)
 
 type digitalAssistantService struct {
-	db *gorm.DB
+	db              *gorm.DB
+	workerScheduler worker.WorkerScheduler
 }
 
-func NewDigitalAssistantService(db *gorm.DB) contract.DigitalAssistantService {
+func NewDigitalAssistantService(db *gorm.DB, workerScheduler worker.WorkerScheduler) contract.DigitalAssistantService {
 	return &digitalAssistantService{
-		db: db,
+		db:              db,
+		workerScheduler: workerScheduler,
 	}
 }
 
@@ -70,6 +74,17 @@ func (s *digitalAssistantService) CreateDigitalAssistant(ctx context.Context, re
 		return nil, err
 	}
 
+	if s.workerScheduler != nil && da.Status == string(contract.DigitalAssistantStatusActive) {
+		spec := &worker.WorkerSpec{
+			ID:      da.Code,
+			Name:    da.Name,
+			EnvType: worker.WorkerEnvProcess,
+		}
+		if _, err := s.workerScheduler.Start(ctx, spec); err != nil {
+			logs.Warnf("Failed to start worker for assistant %s: %v", da.Code, err)
+		}
+	}
+
 	return convertToContractDigitalAssistant(da), nil
 }