fixups

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

Author: mudler

core/services/agentpool/agent_config_distributed.go

@@ -138,8 +138,8 @@ func (b *distributedAgentConfigBackend) GetObservables(userID, name string) (any
 }
 
 // ClearObservables clears observables from the PostgreSQL table.
-func (b *distributedAgentConfigBackend) ClearObservables(_, name string) error {
-	return b.store.ClearObservables(name)
+func (b *distributedAgentConfigBackend) ClearObservables(userID, name string) error {
+	return b.store.ClearObservables(agents.AgentKey(userID, name))
 }
 
 func (b *distributedAgentConfigBackend) ListAllGrouped() map[string][]UserAgentInfo {

core/services/agents/dispatcher.go

@@ -417,7 +417,7 @@ func (d *NATSDispatcher) buildNATSCallbacks(evt AgentChatEvent) Callbacks {
 					ActionResult: result,
 				}
 				if d.eventBridge != nil {
-					d.eventBridge.PersistObservable(evt.AgentName, "tool_result", obs)
+					d.eventBridge.PersistObservable(evt.AgentName, evt.UserID, "tool_result", obs)
 				}
 			}
 		},
@@ -446,7 +446,7 @@ func (d *NATSDispatcher) buildNATSCallbacks(evt AgentChatEvent) Callbacks {
 						},
 					}
 				}
-				d.eventBridge.PersistObservable(evt.AgentName, "chat", rootObs)
+				d.eventBridge.PersistObservable(evt.AgentName, evt.UserID, "chat", rootObs)
 			}
 		},
 	}

core/services/agents/events.go

@@ -60,19 +60,28 @@ func (b *EventBridge) PublishEvent(agentName, userID string, evt AgentEvent) err
 	return b.nats.Publish(subject, evt)
 }
 
-// PersistObservable writes a structured observable record to the database.
+// PersistObservable writes a structured observable record to the database
+// and publishes an observable_update SSE event for real-time UI updates.
 // The obs should be a coreTypes.Observable or compatible struct.
-func (b *EventBridge) PersistObservable(agentName, eventType string, obs any) {
+func (b *EventBridge) PersistObservable(agentName, userID, eventType string, obs any) {
 	if b.store == nil {
 		return
 	}
+	payload := mustJSON(obs)
 	b.store.AppendObservable(&AgentObservableRecord{
 		ID:          uuid.New().String(),
-		AgentName:   agentName,
+		AgentName:   AgentKey(userID, agentName),
 		EventType:   eventType,
-		PayloadJSON: mustJSON(obs),
+		PayloadJSON: payload,
 		CreatedAt:   time.Now(),
 	})
+	// Publish real-time SSE update (uses plain agentName for NATS subject routing)
+	b.PublishEvent(agentName, userID, AgentEvent{
+		AgentName: agentName,
+		UserID:    userID,
+		EventType: "observable_update",
+		Metadata:  payload,
+	})
 }
 
 // PublishMessage publishes a chat message event via NATS for SSE bridging.
@@ -240,7 +249,7 @@ func (b *EventBridge) handleSSEInternal(c echo.Context, agentName, userID string
 			if evt.Metadata != "" {
 				writeSSE(evt.EventType, evt.Metadata)
 			}
-		case "stream_event":
+		case "stream_event", "observable_update":
 			// Send the metadata JSON directly — React UI expects {type, content, ...}
 			if evt.Metadata != "" {
 				writeSSE(evt.EventType, evt.Metadata)