feat: add node reconciler, allow to schedule to group of nodes, min/max autoscaler (#9186)

* always enable parallel requests

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

* feat: add node reconciler, allow to schedule to group of nodes, min/max autoscaler

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

* chore: move tests to ginkgo

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

* chore(smart router): order by available vram

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

---------

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

Author: mudler

core/application/config_file_watcher.go

@@ -199,7 +199,6 @@ func readRuntimeSettingsJson(startupAppConfig config.ApplicationConfig) fileHand
 		envWatchdogBusyTimeout := appConfig.WatchDogBusyTimeout == startupAppConfig.WatchDogBusyTimeout
 		envSingleBackend := appConfig.SingleBackend == startupAppConfig.SingleBackend
 		envMaxActiveBackends := appConfig.MaxActiveBackends == startupAppConfig.MaxActiveBackends
-		envParallelRequests := appConfig.ParallelBackendRequests == startupAppConfig.ParallelBackendRequests
 		envMemoryReclaimerEnabled := appConfig.MemoryReclaimerEnabled == startupAppConfig.MemoryReclaimerEnabled
 		envMemoryReclaimerThreshold := appConfig.MemoryReclaimerThreshold == startupAppConfig.MemoryReclaimerThreshold
 		envThreads := appConfig.Threads == startupAppConfig.Threads
@@ -271,9 +270,6 @@ func readRuntimeSettingsJson(startupAppConfig config.ApplicationConfig) fileHand
 					appConfig.MaxActiveBackends = 0
 				}
 			}
-			if settings.ParallelBackendRequests != nil && !envParallelRequests {
-				appConfig.ParallelBackendRequests = *settings.ParallelBackendRequests
-			}
 			if settings.MemoryReclaimerEnabled != nil && !envMemoryReclaimerEnabled {
 				appConfig.MemoryReclaimerEnabled = *settings.MemoryReclaimerEnabled
 				if appConfig.MemoryReclaimerEnabled {

core/application/distributed.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/mudler/LocalAI/core/config"
@@ -28,6 +29,7 @@ type DistributedServices struct {
 	Registry     *nodes.NodeRegistry
 	Router       *nodes.SmartRouter
 	Health       *nodes.HealthMonitor
+	Reconciler   *nodes.ReplicaReconciler
 	JobStore     *jobs.JobStore
 	Dispatcher   *jobs.Dispatcher
 	AgentStore   *agents.AgentStore
@@ -240,6 +242,16 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB) (*Distribut
 		DB:            authDB,
 	})
 
+	// Create ReplicaReconciler for auto-scaling model replicas
+	reconciler := nodes.NewReplicaReconciler(nodes.ReplicaReconcilerOptions{
+		Registry:       registry,
+		Scheduler:      router,
+		Unloader:       remoteUnloader,
+		DB:             authDB,
+		Interval:       30 * time.Second,
+		ScaleDownDelay: 5 * time.Minute,
+	})
+
 	// Create ModelRouterAdapter to wire into ModelLoader
 	modelAdapter := nodes.NewModelRouterAdapter(router)
 
@@ -250,6 +262,7 @@ func initDistributed(cfg *config.ApplicationConfig, authDB *gorm.DB) (*Distribut
 		Registry:     registry,
 		Router:       router,
 		Health:       healthMon,
+		Reconciler:   reconciler,
 		JobStore:     jobStore,
 		Dispatcher:   dispatcher,
 		AgentStore:   agentStore,

core/application/startup.go

@@ -158,6 +158,10 @@ func New(opts ...config.AppOption) (*Application, error) {
 		application.modelLoader.SetModelStore(distStore)
 		// Start health monitor
 		distSvc.Health.Start(options.Context)
+		// Start replica reconciler for auto-scaling model replicas
+		if distSvc.Reconciler != nil {
+			go distSvc.Reconciler.Run(options.Context)
+		}
 		// In distributed mode, MCP CI jobs are executed by agent workers (not the frontend)
 		// because the frontend can't create MCP sessions (e.g., stdio servers using docker).
 		// The dispatcher still subscribes to jobs.new for persistence (result/progress subs)
@@ -439,11 +443,6 @@ func loadRuntimeSettingsFromFile(options *config.ApplicationConfig) {
 			}
 		}
 	}
-	if settings.ParallelBackendRequests != nil {
-		if !options.ParallelBackendRequests {
-			options.ParallelBackendRequests = *settings.ParallelBackendRequests
-		}
-	}
 	if settings.MemoryReclaimerEnabled != nil {
 		// Only apply if current value is default (false), suggesting it wasn't set from env var
 		if !options.MemoryReclaimerEnabled {

core/backend/options.go

@@ -59,9 +59,7 @@ func ModelOptions(c config.ModelConfig, so *config.ApplicationConfig, opts ...mo
 	grpcOpts := grpcModelOpts(c, so.SystemState.Model.ModelsPath)
 	defOpts = append(defOpts, model.WithLoadGRPCLoadModelOpts(grpcOpts))
 
-	if so.ParallelBackendRequests {
-		defOpts = append(defOpts, model.EnableParallelRequests)
-	}
+	defOpts = append(defOpts, model.EnableParallelRequests)
 
 	if c.GRPC.Attempts != 0 {
 		defOpts = append(defOpts, model.WithGRPCAttempts(c.GRPC.Attempts))