refactorings

Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

Author: mudler

Dockerfile

@@ -319,7 +319,6 @@ COPY ./.git ./.git
 # Some of the Go backends use libs from the main src, we could further optimize the caching by building the CPP backends before here
 COPY ./pkg/grpc ./pkg/grpc
 COPY ./pkg/utils ./pkg/utils
-COPY ./pkg/langchain ./pkg/langchain
 
 RUN ls -l ./
 RUN make protogen-go

core/application/agent_jobs.go

@@ -3,7 +3,7 @@ package application
 import (
 	"time"
 
-	"github.com/mudler/LocalAI/core/services"
+	"github.com/mudler/LocalAI/core/services/agentpool"
 	"github.com/mudler/xlog"
 )
 
@@ -22,13 +22,21 @@ func (a *Application) RestartAgentJobService() error {
 	}
 
 	// Create new service instance
-	agentJobService := services.NewAgentJobService(
+	agentJobService := agentpool.NewAgentJobService(
 		a.ApplicationConfig(),
 		a.ModelLoader(),
 		a.ModelConfigLoader(),
 		a.TemplatesEvaluator(),
 	)
 
+	// Re-apply distributed wiring if available (matches startup.go logic)
+	if a.jobDispatcher != nil {
+		agentJobService.SetDistributedBackends(a.jobDispatcher)
+	}
+	if a.jobStore != nil {
+		agentJobService.SetDistributedJobStore(a.jobStore)
+	}
+
 	// Start the service
 	err := agentJobService.Start(a.ApplicationConfig().Context)
 	if err != nil {

core/application/application.go

@@ -8,8 +8,9 @@ import (
 
 	"github.com/mudler/LocalAI/core/config"
 	mcpTools "github.com/mudler/LocalAI/core/http/endpoints/mcp"
-	"github.com/mudler/LocalAI/core/services"
 	"github.com/mudler/LocalAI/core/services/agents"
+	"github.com/mudler/LocalAI/core/services/agentpool"
+	"github.com/mudler/LocalAI/core/services/galleryop"
 	"github.com/mudler/LocalAI/core/services/distributed"
 	"github.com/mudler/LocalAI/core/services/jobs"
 	"github.com/mudler/LocalAI/core/services/messaging"
@@ -27,9 +28,9 @@ type Application struct {
 	applicationConfig  *config.ApplicationConfig
 	startupConfig      *config.ApplicationConfig // Stores original config from env vars (before file loading)
 	templatesEvaluator *templates.Evaluator
-	galleryService     *services.GalleryService
-	agentJobService    *services.AgentJobService
-	agentPoolService   atomic.Pointer[services.AgentPoolService]
+	galleryService     *galleryop.GalleryService
+	agentJobService    *agentpool.AgentJobService
+	agentPoolService   atomic.Pointer[agentpool.AgentPoolService]
 	authDB             *gorm.DB
 	watchdogMutex      sync.Mutex
 	watchdogStop       chan bool
@@ -84,15 +85,15 @@ func (a *Application) TemplatesEvaluator() *templates.Evaluator {
 	return a.templatesEvaluator
 }
 
-func (a *Application) GalleryService() *services.GalleryService {
+func (a *Application) GalleryService() *galleryop.GalleryService {
 	return a.galleryService
 }
 
-func (a *Application) AgentJobService() *services.AgentJobService {
+func (a *Application) AgentJobService() *agentpool.AgentJobService {
 	return a.agentJobService
 }
 
-func (a *Application) AgentPoolService() *services.AgentPoolService {
+func (a *Application) AgentPoolService() *agentpool.AgentPoolService {
 	return a.agentPoolService.Load()
 }
 
@@ -169,7 +170,7 @@ func (a *Application) IsDistributed() bool {
 // waitForHealthyWorker blocks until at least one healthy backend worker is registered.
 // This prevents the agent pool from failing during startup when workers haven't connected yet.
 func (a *Application) waitForHealthyWorker() {
-	const maxWait = 5 * time.Minute
+	maxWait := a.applicationConfig.Distributed.WorkerWaitTimeoutOrDefault()
 	const pollInterval = 2 * time.Second
 
 	xlog.Info("Waiting for at least one healthy backend worker before starting agent pool")
@@ -200,7 +201,7 @@ func (a *Application) InstanceID() string {
 }
 
 func (a *Application) start() error {
-	galleryService := services.NewGalleryService(a.ApplicationConfig(), a.ModelLoader())
+	galleryService := galleryop.NewGalleryService(a.ApplicationConfig(), a.ModelLoader())
 	err := galleryService.Start(a.ApplicationConfig().Context, a.ModelConfigLoader(), a.ApplicationConfig().SystemState)
 	if err != nil {
 		return err
@@ -209,7 +210,7 @@ func (a *Application) start() error {
 	a.galleryService = galleryService
 
 	// Initialize agent job service (Start() is deferred to after distributed wiring)
-	agentJobService := services.NewAgentJobService(
+	agentJobService := agentpool.NewAgentJobService(
 		a.ApplicationConfig(),
 		a.ModelLoader(),
 		a.ModelConfigLoader(),
@@ -228,7 +229,7 @@ func (a *Application) StartAgentPool() {
 	if !a.applicationConfig.AgentPool.Enabled {
 		return
 	}
-	aps, err := services.NewAgentPoolService(a.applicationConfig)
+	aps, err := agentpool.NewAgentPoolService(a.applicationConfig)
 	if err != nil {
 		xlog.Error("Failed to create agent pool service", "error", err)
 		return
@@ -262,7 +263,7 @@ func (a *Application) StartAgentPool() {
 	}
 
 	// Wire per-user scoped services so collections, skills, and jobs are isolated per user
-	usm := services.NewUserServicesManager(
+	usm := agentpool.NewUserServicesManager(
 		aps.UserStorage(),
 		a.applicationConfig,
 		a.modelLoader,

core/application/distributed.go

@@ -70,11 +70,18 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB) (*distribut
 	}
 	xlog.Info("Connected to NATS", "url", cfg.Distributed.NatsURL)
 
+	// Ensure NATS is closed if any subsequent initialization step fails.
+	success := false
+	defer func() {
+		if !success {
+			natsClient.Close()
+		}
+	}()
+
 	// Initialize object storage
 	var store storage.ObjectStore
 	if cfg.Distributed.StorageURL != "" {
 		if cfg.Distributed.StorageBucket == "" {
-			natsClient.Close()
 			return nil, fmt.Errorf("distributed storage bucket must be set when storage URL is configured")
 		}
 		s3Store, err := storage.NewS3Store(storage.S3Config{
@@ -86,7 +93,6 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB) (*distribut
 			ForcePathStyle:  true, // required for MinIO
 		})
 		if err != nil {
-			natsClient.Close()
 			return nil, fmt.Errorf("initializing S3 storage: %w", err)
 		}
 		xlog.Info("Object storage initialized (S3)", "endpoint", cfg.Distributed.StorageURL, "bucket", cfg.Distributed.StorageBucket)
@@ -95,7 +101,6 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB) (*distribut
 		// Fallback to filesystem storage in distributed mode (useful for single-node testing)
 		fsStore, err := storage.NewFilesystemStore(cfg.DataPath + "/objectstore")
 		if err != nil {
-			natsClient.Close()
 			return nil, fmt.Errorf("initializing filesystem storage: %w", err)
 		}
 		xlog.Info("Object storage initialized (filesystem fallback)", "path", cfg.DataPath+"/objectstore")
@@ -104,13 +109,11 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB) (*distribut
 
 	// Initialize node registry (requires the auth DB which is PostgreSQL)
 	if authDB == nil {
-		natsClient.Close()
 		return nil, fmt.Errorf("distributed mode requires auth database to be initialized first")
 	}
 
 	registry, err := nodes.NewNodeRegistry(authDB)
 	if err != nil {
-		natsClient.Close()
 		return nil, fmt.Errorf("initializing node registry: %w", err)
 	}
 	xlog.Info("Node registry initialized")
@@ -122,12 +125,14 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB) (*distribut
 	if galleriesJSON, err := json.Marshal(cfg.BackendGalleries); err == nil {
 		router.SetGalleriesJSON(string(galleriesJSON))
 	}
-	healthMon := nodes.NewHealthMonitor(registry, authDB)
+	healthMon := nodes.NewHealthMonitor(registry, authDB,
+		cfg.Distributed.HealthCheckIntervalOrDefault(),
+		cfg.Distributed.StaleNodeThresholdOrDefault(),
+	)
 
 	// Initialize job store
 	jobStore, err := jobs.NewJobStore(authDB)
 	if err != nil {
-		natsClient.Close()
 		return nil, fmt.Errorf("initializing job store: %w", err)
 	}
 	xlog.Info("Distributed job store initialized")
@@ -138,7 +143,6 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB) (*distribut
 	// Initialize agent store
 	agentStore, err := agents.NewAgentStore(authDB)
 	if err != nil {
-		natsClient.Close()
 		return nil, fmt.Errorf("initializing agent store: %w", err)
 	}
 	xlog.Info("Distributed agent store initialized")
@@ -149,15 +153,13 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB) (*distribut
 	// Initialize Phase 4 stores (MCP, Gallery, FineTune, Skills)
 	distStores, err := distributed.InitStores(authDB)
 	if err != nil {
-		natsClient.Close()
 		return nil, fmt.Errorf("initializing distributed stores: %w", err)
 	}
 
 	// Initialize file manager with local cache
 	cacheDir := cfg.DataPath + "/cache"
 	fileMgr, err := storage.NewFileManager(store, cacheDir)
 	if err != nil {
-		natsClient.Close()
 		return nil, fmt.Errorf("initializing file manager: %w", err)
 	}
 	xlog.Info("File manager initialized", "cacheDir", cacheDir)
@@ -185,6 +187,7 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB) (*distribut
 	// Create ModelRouterAdapter to wire into ModelLoader
 	modelAdapter := nodes.NewModelRouterAdapter(router)
 
+	success = true
 	return &distributedServices{
 		nats:         natsClient,
 		store:        store,

core/application/p2p.go

@@ -11,7 +11,7 @@ import (
 	"github.com/mudler/LocalAI/core/gallery"
 	"github.com/mudler/LocalAI/core/p2p"
 	"github.com/mudler/LocalAI/core/schema"
-	"github.com/mudler/LocalAI/core/services"
+	"github.com/mudler/LocalAI/core/services/galleryop"
 
 	"github.com/mudler/edgevpn/pkg/node"
 	"github.com/mudler/xlog"
@@ -146,22 +146,14 @@ func (a *Application) RestartP2P() error {
 		return fmt.Errorf("P2P token is not set")
 	}
 
-	// Create new context for P2P
-	ctx, cancel := context.WithCancel(appConfig.Context)
-	a.p2pCtx = ctx
-	a.p2pCancel = cancel
-
-	// Get API address from config
-	address := appConfig.APIAddress
-	if address == "" {
-		address = "127.0.0.1:8080" // default
-	}
-
 	// Start P2P stack in a goroutine
+	// Note: StartP2P creates its own context and assigns a.p2pCtx/a.p2pCancel
 	go func() {
 		if err := a.StartP2P(); err != nil {
 			xlog.Error("Failed to start P2P stack", "error", err)
-			cancel() // Cancel context on error
+			if a.p2pCancel != nil {
+				a.p2pCancel()
+			}
 		}
 	}()
 	xlog.Info("P2P stack restarted with new settings")
@@ -228,7 +220,7 @@ func syncState(ctx context.Context, n *node.Node, app *Application) error {
 			continue
 		}
 
-		app.GalleryService().ModelGalleryChannel <- services.ManagementOp[gallery.GalleryModel, gallery.ModelConfig]{
+		app.GalleryService().ModelGalleryChannel <- galleryop.ManagementOp[gallery.GalleryModel, gallery.ModelConfig]{
 			ID:                 uuid.String(),
 			GalleryElementName: model,
 			Galleries:          app.ApplicationConfig().Galleries,

core/application/startup.go

@@ -13,7 +13,7 @@ import (
 	"github.com/mudler/LocalAI/core/config"
 	"github.com/mudler/LocalAI/core/gallery"
 	"github.com/mudler/LocalAI/core/http/auth"
-	"github.com/mudler/LocalAI/core/services"
+	"github.com/mudler/LocalAI/core/services/galleryop"
 	"github.com/mudler/LocalAI/core/services/jobs"
 	"github.com/mudler/LocalAI/core/services/nodes"
 	"github.com/mudler/LocalAI/core/services/storage"
@@ -223,7 +223,7 @@ func New(opts ...config.AppOption) (*Application, error) {
 	}
 
 	for _, backend := range options.ExternalBackends {
-		if err := services.InstallExternalBackend(options.Context, options.BackendGalleries, options.SystemState, application.ModelLoader(), nil, backend, "", ""); err != nil {
+		if err := galleryop.InstallExternalBackend(options.Context, options.BackendGalleries, options.SystemState, application.ModelLoader(), nil, backend, "", ""); err != nil {
 			xlog.Error("error installing external backend", "error", err)
 		}
 	}
@@ -249,13 +249,13 @@ func New(opts ...config.AppOption) (*Application, error) {
 	}
 
 	if options.PreloadJSONModels != "" {
-		if err := services.ApplyGalleryFromString(options.SystemState, application.ModelLoader(), options.EnforcePredownloadScans, options.AutoloadBackendGalleries, options.Galleries, options.BackendGalleries, options.PreloadJSONModels); err != nil {
+		if err := galleryop.ApplyGalleryFromString(options.SystemState, application.ModelLoader(), options.EnforcePredownloadScans, options.AutoloadBackendGalleries, options.Galleries, options.BackendGalleries, options.PreloadJSONModels); err != nil {
 			return nil, err
 		}
 	}
 
 	if options.PreloadModelsFromPath != "" {
-		if err := services.ApplyGalleryFromFile(options.SystemState, application.ModelLoader(), options.EnforcePredownloadScans, options.AutoloadBackendGalleries, options.Galleries, options.BackendGalleries, options.PreloadModelsFromPath); err != nil {
+		if err := galleryop.ApplyGalleryFromFile(options.SystemState, application.ModelLoader(), options.EnforcePredownloadScans, options.AutoloadBackendGalleries, options.Galleries, options.BackendGalleries, options.PreloadModelsFromPath); err != nil {
 			return nil, err
 		}
 	}

core/backend/llm.go

@@ -15,7 +15,7 @@ import (
 	"github.com/mudler/LocalAI/core/config"
 	"github.com/mudler/LocalAI/core/trace"
 	"github.com/mudler/LocalAI/core/schema"
-	"github.com/mudler/LocalAI/core/services"
+	"github.com/mudler/LocalAI/core/services/galleryop"
 
 	"github.com/mudler/LocalAI/core/gallery"
 	"github.com/mudler/LocalAI/pkg/grpc/proto"
@@ -47,7 +47,7 @@ func ModelInference(ctx context.Context, s string, messages schema.Messages, ima
 
 	// Check if the modelFile exists, if it doesn't try to load it from the gallery
 	if o.AutoloadGalleries { // experimental
-		modelNames, err := services.ListModels(cl, loader, nil, services.SKIP_ALWAYS)
+		modelNames, err := galleryop.ListModels(cl, loader, nil, galleryop.SKIP_ALWAYS)
 		if err != nil {
 			return nil, err
 		}

core/cli/agent.go

@@ -10,7 +10,7 @@ import (
 
 	cliContext "github.com/mudler/LocalAI/core/cli/context"
 	"github.com/mudler/LocalAI/core/config"
-	"github.com/mudler/LocalAI/core/services"
+	"github.com/mudler/LocalAI/core/services/agentpool"
 	"github.com/mudler/LocalAGI/core/state"
 	coreTypes "github.com/mudler/LocalAGI/core/types"
 	"github.com/mudler/xlog"
@@ -59,7 +59,7 @@ func (r *AgentRunCMD) Run(ctx *cliContext.Context) error {
 
 	appConfig := r.buildAppConfig()
 
-	poolService, err := services.NewAgentPoolService(appConfig)
+	poolService, err := agentpool.NewAgentPoolService(appConfig)
 	if err != nil {
 		return fmt.Errorf("failed to create agent pool service: %w", err)
 	}