cleanups

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

Author: mudler

core/application/application.go

@@ -155,6 +155,11 @@ func (a *Application) FileManager() *storage.FileManager {
 	return a.fileManager
 }
 
+// HealthMonitor returns the health monitor, or nil if not in distributed mode.
+func (a *Application) HealthMonitor() *nodes.HealthMonitor {
+	return a.healthMon
+}
+
 // IsDistributed returns true if the application is running in distributed mode.
 func (a *Application) IsDistributed() bool {
 	return a.applicationConfig.Distributed.Enabled

core/cli/run.go

@@ -520,6 +520,12 @@ func (r *RunCMD) Run(ctx *cliContext.Context) error {
 			xlog.Error("error while stopping all grpc backends", "error", err)
 		}
 		// Clean up distributed services
+		if app.HealthMonitor() != nil {
+			app.HealthMonitor().Stop()
+		}
+		if app.JobDispatcher() != nil {
+			app.JobDispatcher().Stop()
+		}
 		if app.NatsClient() != nil {
 			app.NatsClient().Close()
 		}

core/http/endpoints/localai/nodes.go

@@ -316,9 +316,6 @@ func DeleteModelOnNodeEndpoint(unloader *nodes.RemoteUnloaderAdapter, registry *
 		if err := unloader.StopBackend(nodeID); err != nil {
 			// Non-fatal
 		}
-		// Send model.delete to remove files
-		subject := "nodes." + nodeID + ".model.delete"
-		_ = subject // delete is done via DeleteModelFiles which fans out
 		registry.RemoveNodeModel(nodeID, req.ModelName)
 		return c.JSON(http.StatusOK, map[string]string{"message": "model deleted from node"})
 	}

core/services/agent_pool.go

@@ -374,83 +374,11 @@ func (s *AgentPoolService) SetAgentStore(store AgentConfigStore) {
 
 // --- Agent CRUD ---
 
-func (s *AgentPoolService) ListAgents() map[string]bool {
-	statuses := map[string]bool{}
-	agents := s.pool.List()
-	for _, a := range agents {
-		ag := s.pool.GetAgent(a)
-		if ag == nil {
-			continue
-		}
-		statuses[a] = !ag.Paused()
-	}
-	return statuses
-}
-
-func (s *AgentPoolService) CreateAgent(config *state.AgentConfig) error {
-	if config.Name == "" {
-		return fmt.Errorf("name is required")
-	}
-	return s.pool.CreateAgent(config.Name, config)
-}
-
 func (s *AgentPoolService) GetAgent(name string) *agent.Agent {
-	return s.pool.GetAgent(name)
-}
-
-func (s *AgentPoolService) GetAgentConfig(name string) *state.AgentConfig {
-	return s.pool.GetConfig(name)
-}
-
-func (s *AgentPoolService) UpdateAgent(name string, config *state.AgentConfig) error {
-	old := s.pool.GetConfig(name)
-	if old == nil {
-		return fmt.Errorf("agent not found: %s", name)
-	}
-	return s.pool.RecreateAgent(name, config)
-}
-
-func (s *AgentPoolService) DeleteAgent(name string) error {
-	return s.pool.Remove(name)
-}
-
-func (s *AgentPoolService) PauseAgent(name string) error {
-	ag := s.pool.GetAgent(name)
-	if ag == nil {
-		return fmt.Errorf("agent not found: %s", name)
-	}
-	ag.Pause()
-	return nil
-}
-
-func (s *AgentPoolService) ResumeAgent(name string) error {
-	ag := s.pool.GetAgent(name)
-	if ag == nil {
-		return fmt.Errorf("agent not found: %s", name)
-	}
-	ag.Resume()
-	return nil
-}
-
-func (s *AgentPoolService) GetAgentStatus(name string) *state.Status {
-	return s.pool.GetStatusHistory(name)
-}
-
-func (s *AgentPoolService) GetAgentObservables(name string) ([]coreTypes.Observable, error) {
-	ag := s.pool.GetAgent(name)
-	if ag == nil {
-		return nil, fmt.Errorf("agent not found: %s", name)
-	}
-	return ag.Observer().History(), nil
-}
-
-func (s *AgentPoolService) ClearAgentObservables(name string) error {
-	ag := s.pool.GetAgent(name)
-	if ag == nil {
-		return fmt.Errorf("agent not found: %s", name)
+	if s.pool == nil {
+		return nil
 	}
-	ag.Observer().ClearHistory()
-	return nil
+	return s.pool.GetAgent(name)
 }
 
 // Chat sends a message to an agent and returns immediately. Responses come via SSE.
@@ -616,10 +544,6 @@ func (s *AgentPoolService) collectAndCopyMetadata(metadata map[string]any, userI
 	}
 }
 
-func (s *AgentPoolService) GetSSEManager(name string) sse.Manager {
-	return s.pool.GetManager(name)
-}
-
 func (s *AgentPoolService) GetConfigMeta() state.AgentConfigMeta {
 	return s.configMeta
 }
@@ -667,88 +591,6 @@ func (s *AgentPoolService) ExportAgent(name string) ([]byte, error) {
 	return nil, fmt.Errorf("agent not found: %s", name)
 }
 
-// ImportAgent creates an agent from JSON config data.
-func (s *AgentPoolService) ImportAgent(data []byte) error {
-	var cfg state.AgentConfig
-	if err := json.Unmarshal(data, &cfg); err != nil {
-		return fmt.Errorf("invalid agent config: %w", err)
-	}
-	if cfg.Name == "" {
-		return fmt.Errorf("agent name is required")
-	}
-	if s.pool == nil {
-		// Distributed mode — save to DB only
-		if s.agentStore != nil {
-			return s.agentStore.SaveConfig(&agents.AgentConfigRecord{
-				Name:       cfg.Name,
-				ConfigJSON: string(data),
-				Status:     "active",
-			})
-		}
-		return fmt.Errorf("cannot import agent: no pool or store available")
-	}
-	return s.pool.CreateAgent(cfg.Name, &cfg)
-}
-
-
-// --- Collections ---
-
-func (s *AgentPoolService) CollectionsBackend() collections.Backend {
-	return s.collectionsBackend
-}
-
-func (s *AgentPoolService) ListCollections() ([]string, error) {
-	return s.collectionsBackend.ListCollections()
-}
-
-func (s *AgentPoolService) CreateCollection(name string) error {
-	return s.collectionsBackend.CreateCollection(name)
-}
-
-func (s *AgentPoolService) UploadToCollection(collection, filename string, fileBody io.Reader) (string, error) {
-	return s.collectionsBackend.Upload(collection, filename, fileBody)
-}
-
-func (s *AgentPoolService) ListCollectionEntries(collection string) ([]string, error) {
-	return s.collectionsBackend.ListEntries(collection)
-}
-
-func (s *AgentPoolService) GetCollectionEntryContent(collection, entry string) (string, int, error) {
-	return s.collectionsBackend.GetEntryContent(collection, entry)
-}
-
-func (s *AgentPoolService) SearchCollection(collection, query string, maxResults int) ([]collections.SearchResult, error) {
-	return s.collectionsBackend.Search(collection, query, maxResults)
-}
-
-func (s *AgentPoolService) ResetCollection(collection string) error {
-	return s.collectionsBackend.Reset(collection)
-}
-
-func (s *AgentPoolService) DeleteCollectionEntry(collection, entry string) ([]string, error) {
-	return s.collectionsBackend.DeleteEntry(collection, entry)
-}
-
-func (s *AgentPoolService) AddCollectionSource(collection, sourceURL string, intervalMin int) error {
-	return s.collectionsBackend.AddSource(collection, sourceURL, intervalMin)
-}
-
-func (s *AgentPoolService) RemoveCollectionSource(collection, sourceURL string) error {
-	return s.collectionsBackend.RemoveSource(collection, sourceURL)
-}
-
-func (s *AgentPoolService) ListCollectionSources(collection string) ([]collections.SourceInfo, error) {
-	return s.collectionsBackend.ListSources(collection)
-}
-
-func (s *AgentPoolService) CollectionEntryExists(collection, entry string) bool {
-	return s.collectionsBackend.EntryExists(collection, entry)
-}
-
-func (s *AgentPoolService) GetCollectionEntryFilePath(collection, entry string) (string, error) {
-	return s.collectionsBackend.GetEntryFilePath(collection, entry)
-}
-
 // --- User Services ---
 
 // SetUserServicesManager sets the user services manager for per-user scoping.

core/services/agents/events.go

@@ -147,7 +147,7 @@ func (b *EventBridge) DeregisterCancel(agentName, userID string) {
 
 // StartCancelListener subscribes to NATS cancel events (broadcast to all instances).
 func (b *EventBridge) StartCancelListener() (*nats.Subscription, error) {
-	return b.nats.Subscribe("agent.*.cancel", func(data []byte) {
+	return b.nats.Subscribe(messaging.SubjectAgentCancelWildcard, func(data []byte) {
 		var evt AgentCancelEvent
 		if json.Unmarshal(data, &evt) != nil {
 			return

core/services/gallery.go

@@ -8,6 +8,7 @@ import (
 	"github.com/mudler/LocalAI/core/config"
 	"github.com/mudler/LocalAI/core/gallery"
 	"github.com/mudler/LocalAI/core/services/distributed"
+	"github.com/mudler/LocalAI/core/services/messaging"
 	"github.com/mudler/LocalAI/pkg/model"
 	"github.com/mudler/LocalAI/pkg/system"
 )
@@ -118,7 +119,7 @@ func (g *GalleryService) UpdateStatus(s string, op *OpStatus) {
 
 	// Publish progress to NATS in distributed mode
 	if g.natsClient != nil {
-		g.natsClient.Publish("gallery."+s+".progress", op)
+		g.natsClient.Publish(messaging.SubjectGalleryProgress(s), op)
 	}
 }
 
@@ -156,7 +157,7 @@ func (g *GalleryService) CancelOperation(id string) error {
 
 	// Publish cancellation to NATS in distributed mode
 	if g.natsClient != nil {
-		g.natsClient.Publish("gallery."+id+".cancel", map[string]string{"id": id})
+		g.natsClient.Publish(messaging.SubjectGalleryCancel(id), map[string]string{"id": id})
 	}
 
 	// Update status to reflect cancellation

core/services/jobs/dispatcher.go

@@ -11,6 +11,7 @@ import (
 	"github.com/mudler/LocalAI/core/services/messaging"
 	"github.com/mudler/xlog"
 	"github.com/nats-io/nats.go"
+	"github.com/robfig/cron/v3"
 	"gorm.io/gorm"
 )
 
@@ -72,8 +73,10 @@ type Dispatcher struct {
 	cancelRegistry sync.Map // jobID -> context.CancelFunc
 
 	// NATS subscriptions
-	jobSub    *nats.Subscription
-	cancelSub *nats.Subscription
+	jobSub      *nats.Subscription
+	cancelSub   *nats.Subscription
+	resultSub   *nats.Subscription
+	progressSub *nats.Subscription
 
 	// Lifecycle
 	ctx    context.Context
@@ -114,7 +117,7 @@ func (d *Dispatcher) Start(ctx context.Context) error {
 	}
 
 	// Subscribe to cancel events (broadcast to all — each instance checks its registry)
-	d.cancelSub, err = d.nats.Subscribe("jobs.*.cancel", func(data []byte) {
+	d.cancelSub, err = d.nats.Subscribe(messaging.SubjectJobCancelWildcard, func(data []byte) {
 		var evt CancelEvent
 		if json.Unmarshal(data, &evt) != nil {
 			return
@@ -130,7 +133,7 @@ func (d *Dispatcher) Start(ctx context.Context) error {
 
 	// Subscribe to job result events from workers (persist to DB)
 	if d.store != nil {
-		d.nats.Subscribe("jobs.*.result", func(data []byte) {
+		d.resultSub, _ = d.nats.Subscribe(messaging.SubjectJobResultWildcard, func(data []byte) {
 			var evt JobResultEvent
 			if json.Unmarshal(data, &evt) != nil {
 				return
@@ -139,7 +142,7 @@ func (d *Dispatcher) Start(ctx context.Context) error {
 		})
 
 		// Subscribe to trace events from workers (persist to DB)
-		d.nats.Subscribe("jobs.*.progress", func(data []byte) {
+		d.progressSub, _ = d.nats.Subscribe(messaging.SubjectJobProgressWildcard, func(data []byte) {
 			var evt ProgressEvent
 			if json.Unmarshal(data, &evt) != nil {
 				return
@@ -168,6 +171,12 @@ func (d *Dispatcher) Stop() {
 	if d.cancelSub != nil {
 		d.cancelSub.Unsubscribe()
 	}
+	if d.resultSub != nil {
+		d.resultSub.Unsubscribe()
+	}
+	if d.progressSub != nil {
+		d.progressSub.Unsubscribe()
+	}
 }
 
 // Enqueue publishes a job to the NATS queue for distributed processing.
@@ -389,23 +398,27 @@ func (d *Dispatcher) runDueCronTasks() {
 	}
 }
 
-// isCronDue checks if a cron task should run now.
-// Simple implementation: check if no job ran in the last interval.
+// cronParser supports standard 5-field cron expressions and descriptors like @every 5m.
+var cronParser = cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
+
+// isCronDue checks if a cron task should run now by parsing the cron expression
+// and comparing the next scheduled run against the last execution time.
 func (d *Dispatcher) isCronDue(task TaskRecord) bool {
+	schedule, err := cronParser.Parse(task.Cron)
+	if err != nil {
+		xlog.Warn("Invalid cron expression, skipping task", "taskID", task.ID, "cron", task.Cron, "error", err)
+		return false
+	}
+
 	// Find the most recent job for this task triggered by cron
 	var lastJob JobRecord
-	err := d.db.Where("task_id = ? AND triggered_by = ?", task.ID, "cron").
+	err = d.db.Where("task_id = ? AND triggered_by = ?", task.ID, "cron").
 		Order("created_at DESC").First(&lastJob).Error
 	if err != nil {
 		// No previous job — it's due
 		return true
 	}
 
-	// Parse cron to determine interval (simplified: use 15s minimum)
-	// A full cron parser would be better, but for MVP just check
-	// if enough time has passed since the last run
-	elapsed := time.Since(lastJob.CreatedAt)
-
-	// Minimum interval: don't run more than once per minute
-	return elapsed > 1*time.Minute
+	nextRun := schedule.Next(lastJob.CreatedAt)
+	return time.Now().After(nextRun)
 }