refactoring and consolidation

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

Author: mudler

core/http/endpoints/localai/agents.go

@@ -251,7 +251,7 @@ func GetAgentObservablesEndpoint(app *application.Application) echo.HandlerFunc
 			return c.JSON(http.StatusNotFound, map[string]string{"error": err.Error()})
 		}
 		if history == nil {
-			history = []any{}
+			history = []json.RawMessage{}
 		}
 		return c.JSON(http.StatusOK, map[string]any{
 			"Name":    name,

core/services/advisorylock/advisorylock_test.go

@@ -1,6 +1,7 @@
 package advisorylock
 
 import (
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -18,6 +19,9 @@ var _ = Describe("AdvisoryLock", func() {
 		var db *gorm.DB
 
 		BeforeEach(func() {
+			if runtime.GOOS == "darwin" {
+				Skip("testcontainers requires Docker, not available on macOS CI")
+			}
 			db = testutil.SetupTestDB()
 		})
 

core/services/agentpool/agent_config_backend.go

@@ -1,6 +1,8 @@
 package agentpool
 
 import (
+	"encoding/json"
+
 	"github.com/mudler/LocalAGI/core/agent"
 	"github.com/mudler/LocalAGI/core/sse"
 	"github.com/mudler/LocalAGI/core/state"
@@ -26,7 +28,7 @@ type AgentConfigBackend interface {
 	GetAgent(userID, name string) *agent.Agent
 	GetSSEManager(userID, name string) sse.Manager
 	GetStatus(userID, name string) *state.Status
-	GetObservables(userID, name string) (any, error)
+	GetObservables(userID, name string) ([]json.RawMessage, error)
 	ClearObservables(userID, name string) error
 
 	// Admin aggregation

core/services/agentpool/agent_config_distributed.go

@@ -121,18 +121,14 @@ func (b *distributedAgentConfigBackend) GetStatus(_, _ string) *state.Status {
 }
 
 // GetObservables reads from the PostgreSQL observables table.
-func (b *distributedAgentConfigBackend) GetObservables(userID, name string) (any, error) {
+func (b *distributedAgentConfigBackend) GetObservables(userID, name string) ([]json.RawMessage, error) {
 	records, err := b.store.GetObservables(agents.AgentKey(userID, name), defaultObservableLimit)
 	if err != nil {
 		return nil, err
 	}
-	// Convert database records to generic maps for JSON serialization.
-	var result []map[string]any
+	result := make([]json.RawMessage, 0, len(records))
 	for _, rec := range records {
-		var obs map[string]any
-		if json.Unmarshal([]byte(rec.PayloadJSON), &obs) == nil {
-			result = append(result, obs)
-		}
+		result = append(result, json.RawMessage(rec.PayloadJSON))
 	}
 	return result, nil
 }

core/services/agentpool/agent_config_local.go

@@ -117,12 +117,21 @@ func (b *localAgentConfigBackend) GetStatus(userID, name string) *state.Status {
 	return b.svc.pool.GetStatusHistory(agents.AgentKey(userID, name))
 }
 
-func (b *localAgentConfigBackend) GetObservables(userID, name string) (any, error) {
+func (b *localAgentConfigBackend) GetObservables(userID, name string) ([]json.RawMessage, error) {
 	ag := b.svc.pool.GetAgent(agents.AgentKey(userID, name))
 	if ag == nil {
 		return nil, fmt.Errorf("%w: %s", ErrAgentNotFound, name)
 	}
-	return ag.Observer().History(), nil
+	history := ag.Observer().History()
+	result := make([]json.RawMessage, 0, len(history))
+	for _, obs := range history {
+		data, err := json.Marshal(obs)
+		if err != nil {
+			continue
+		}
+		result = append(result, data)
+	}
+	return result, nil
 }
 
 func (b *localAgentConfigBackend) ClearObservables(userID, name string) error {