fix(openresponses): make responses visible and cancellable across replicas

core/http/endpoints/openresponses/responses.go

@@ -3047,11 +3047,26 @@ func GetResponseEndpoint() func(c echo.Context) error {
 
 // handleStreamResume handles resuming a streaming response from a specific sequence number
 func handleStreamResume(c echo.Context, store *ResponseStore, responseID string, stored *StoredResponse, startingAfter int) error {
+	// The resume buffer is process-local by design (see ErrResponseNotLocal), so
+	// a resume that the load balancer routed to a replica which is not
+	// generating this response cannot be served here. Say that plainly: a
+	// truthful error lets the caller retry against the right replica, whereas
+	// replaying an empty buffer would look like a stream that simply ended.
+	if stored.Remote {
+		return sendOpenResponsesError(c, 409, "invalid_request_error",
+			fmt.Sprintf("response %s is being generated on another replica (%s) and its stream cannot be resumed here; poll GET /v1/responses/%s instead, or route the resume with session affinity", responseID, stored.OwnerReplica, responseID),
+			"stream")
+	}
+
 	// Fetch buffered events before committing to an SSE response so an
 	// offset-lost gap can be reported as a clean HTTP status rather than a
 	// silently truncated event stream.
 	events, err := store.GetEventsAfter(responseID, startingAfter)
 	if err != nil {
+		if errors.Is(err, ErrResponseNotLocal) {
+			return sendOpenResponsesError(c, 409, "invalid_request_error",
+				fmt.Sprintf("response %s is being generated on another replica and its stream cannot be resumed here", responseID), "stream")
+		}
 		if errors.Is(err, ErrOffsetLost) {
 			return sendOpenResponsesError(c, 409, "invalid_request_error", fmt.Sprintf("starting_after=%d is older than the oldest retained event; the resume buffer evicted those events and the stream cannot be resumed from that point", startingAfter), "starting_after")
 		}

core/http/endpoints/openresponses/store.go

@@ -9,6 +9,8 @@ import (
 	"time"
 
 	"github.com/mudler/LocalAI/core/schema"
+	"github.com/mudler/LocalAI/core/services/messaging"
+	"github.com/mudler/LocalAI/core/services/syncstate"
 	"github.com/mudler/xlog"
 )
 
@@ -49,6 +51,16 @@ type ResponseStore struct {
 	// them. A value <= 0 disables that particular cap.
 	maxStreamEvents int
 	maxStreamBytes  int
+
+	// Cross-replica replication. All zero unless EnableDistributed was called
+	// (see sync.go), which is how a standalone deployment keeps exactly the
+	// previous process-local behaviour. Guarded by mu.
+	synced     *syncstate.SyncedMap[string, *syncedResponse]
+	nats       messaging.MessagingClient
+	cancelSub  messaging.Subscription
+	replicaID  string
+	lifeCtx    context.Context
+	lifeCancel context.CancelFunc
 }
 
 // StreamedEvent represents a buffered SSE event for streaming resume
@@ -80,6 +92,15 @@ type StoredResponse struct {
 	EventsChan    chan struct{}      // Signals new events for live subscribers
 	mu            sync.RWMutex       // Protect concurrent access to this response
 
+	// Remote marks a read-only view materialised from another replica's
+	// replicated metadata rather than from this process's own map. Such a view
+	// has no CancelFunc, no EventsChan and no resume buffer, so every path that
+	// needs one of those must check it and either delegate over the bus
+	// (cancel) or refuse with ErrResponseNotLocal (stream resume).
+	// OwnerReplica names the replica that does hold them.
+	Remote       bool
+	OwnerReplica string
+
 	// streamBytes tracks the total serialized size of the events currently
 	// retained in StreamEvents, used to enforce the byte cap. droppedThrough
 	// is the highest sequence number evicted from the front of the buffer
@@ -144,7 +165,6 @@ func NewResponseStore(ttl time.Duration) *ResponseStore {
 // Store stores a response with its request and items
 func (s *ResponseStore) Store(responseID string, request *schema.OpenResponsesRequest, response *schema.ORResponseResource) {
 	s.mu.Lock()
-	defer s.mu.Unlock()
 
 	// Build item index for quick lookup
 	items := make(map[string]*schema.ORItemField)
@@ -171,26 +191,40 @@ func (s *ResponseStore) Store(responseID string, request *schema.OpenResponsesRe
 	}
 
 	s.responses[responseID] = stored
+	s.mu.Unlock()
+
+	// Replicate outside the lock: the broadcast can be delivered synchronously
+	// and a subscriber re-enters the store.
+	s.mirror(responseID, stored)
 	xlog.Debug("Stored Open Responses response", "response_id", responseID, "items_count", len(items))
 }
 
-// Get retrieves a stored response by ID
+// Get retrieves a stored response by ID.
+//
+// In distributed mode a miss in this process's map is not proof the response
+// does not exist: a round-robin load balancer routes polls and
+// previous_response_id lookups to any replica, not the one that created it. So
+// a local miss falls back to the replicated metadata and returns a read-only
+// remote view (issue #10993). Only a miss in both is a real "not found".
 func (s *ResponseStore) Get(responseID string) (*StoredResponse, error) {
 	s.mu.RLock()
-	defer s.mu.RUnlock()
-
 	stored, exists := s.responses[responseID]
-	if !exists {
-		return nil, fmt.Errorf("response not found: %s", responseID)
+	s.mu.RUnlock()
+
+	if exists {
+		// Check expiration
+		if stored.ExpiresAt != nil && time.Now().After(*stored.ExpiresAt) {
+			// Expired, but we'll return it anyway and let caller handle cleanup
+			return nil, fmt.Errorf("response expired: %s", responseID)
+		}
+		return stored, nil
 	}
 
-	// Check expiration
-	if stored.ExpiresAt != nil && time.Now().After(*stored.ExpiresAt) {
-		// Expired, but we'll return it anyway and let caller handle cleanup
-		return nil, fmt.Errorf("response expired: %s", responseID)
+	if remote, ok := s.remoteGet(responseID); ok {
+		return remote, nil
 	}
 
-	return stored, nil
+	return nil, fmt.Errorf("response not found: %s", responseID)
 }
 
 // GetItem retrieves a specific item from a stored response
@@ -211,29 +245,61 @@ func (s *ResponseStore) GetItem(responseID, itemID string) (*schema.ORItemField,
 // FindItem searches for an item across all stored responses
 // Returns the item and the response ID it was found in
 func (s *ResponseStore) FindItem(itemID string) (*schema.ORItemField, string, error) {
+	now := time.Now()
+
+	if item, responseID, found := s.findItemLocal(itemID, now); found {
+		return item, responseID, nil
+	}
+
+	// An item referenced by ID can belong to a response created on any replica,
+	// so the local sweep alone reproduces the same coin-flip lookup #10993
+	// describes for GET /v1/responses/{id}.
+	if m := s.syncMap(); m != nil {
+		for responseID, v := range m.Snapshot() {
+			if v == nil || v.Response == nil {
+				continue
+			}
+			if v.ExpiresAt != nil && now.After(*v.ExpiresAt) {
+				continue
+			}
+			for i := range v.Response.Output {
+				if v.Response.Output[i].ID == itemID {
+					return &v.Response.Output[i], responseID, nil
+				}
+			}
+		}
+	}
+
+	return nil, "", fmt.Errorf("item not found in any stored response: %s", itemID)
+}
+
+// findItemLocal sweeps only this process's own responses. Split out so the
+// replicated sweep in FindItem runs with s.mu released - the SyncedMap is a
+// separate component with its own lock and must not be entered under ours.
+func (s *ResponseStore) findItemLocal(itemID string, now time.Time) (*schema.ORItemField, string, bool) {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
-	now := time.Now()
 	for responseID, stored := range s.responses {
 		// Skip expired responses
 		if stored.ExpiresAt != nil && now.After(*stored.ExpiresAt) {
 			continue
 		}
 
 		if item, exists := stored.Items[itemID]; exists {
-			return item, responseID, nil
+			return item, responseID, true
 		}
 	}
-
-	return nil, "", fmt.Errorf("item not found in any stored response: %s", itemID)
+	return nil, "", false
 }
 
 // Delete removes a response from storage
 func (s *ResponseStore) Delete(responseID string) {
 	s.mu.Lock()
-	defer s.mu.Unlock()
 	delete(s.responses, responseID)
+	s.mu.Unlock()
+
+	s.unmirror(responseID)
 	xlog.Debug("Deleted Open Responses response", "response_id", responseID)
 }
 
@@ -244,22 +310,33 @@ func (s *ResponseStore) Cleanup() int {
 	}
 
 	s.mu.Lock()
-	defer s.mu.Unlock()
-
 	now := time.Now()
-	count := 0
+	expired := []string{}
 	for id, stored := range s.responses {
 		if stored.ExpiresAt != nil && now.After(*stored.ExpiresAt) {
 			delete(s.responses, id)
-			count++
+			expired = append(expired, id)
+		}
+	}
+	s.mu.Unlock()
+
+	// Reap replicated entries too, including those whose owner never got to
+	// expire them because it was scaled down mid-flight. This covers the
+	// locally-expired IDs as well, since they were mirrored with the same
+	// ExpiresAt. Any replica may do this; the delete broadcast is idempotent.
+	if m := s.syncMap(); m != nil {
+		for id, v := range m.Snapshot() {
+			if v != nil && v.ExpiresAt != nil && now.After(*v.ExpiresAt) {
+				s.unmirror(id)
+			}
 		}
 	}
 
-	if count > 0 {
-		xlog.Debug("Cleaned up expired Open Responses", "count", count)
+	if len(expired) > 0 {
+		xlog.Debug("Cleaned up expired Open Responses", "count", len(expired))
 	}
 
-	return count
+	return len(expired)
 }
 
 // cleanupLoop runs periodic cleanup of expired responses
@@ -292,7 +369,6 @@ func (s *ResponseStore) Count() int {
 // StoreBackground stores a background response with cancel function and optional streaming support
 func (s *ResponseStore) StoreBackground(responseID string, request *schema.OpenResponsesRequest, response *schema.ORResponseResource, cancelFunc context.CancelFunc, streamEnabled bool) {
 	s.mu.Lock()
-	defer s.mu.Unlock()
 
 	// Build item index for quick lookup
 	items := make(map[string]*schema.ORItemField)
@@ -324,6 +400,11 @@ func (s *ResponseStore) StoreBackground(responseID string, request *schema.OpenR
 	}
 
 	s.responses[responseID] = stored
+	s.mu.Unlock()
+
+	// Only the metadata crosses the bus. CancelFunc and the resume buffer stay
+	// here, which is what makes this replica the owner for cancel and resume.
+	s.mirror(responseID, stored)
 	xlog.Debug("Stored background Open Responses response", "response_id", responseID, "stream_enabled", streamEnabled)
 }
 
@@ -338,10 +419,13 @@ func (s *ResponseStore) UpdateStatus(responseID string, status string, completed
 	}
 
 	stored.mu.Lock()
-	defer stored.mu.Unlock()
-
 	stored.Response.Status = status
 	stored.Response.CompletedAt = completedAt
+	stored.mu.Unlock()
+
+	// Peers poll this response too, so every status transition has to be
+	// republished or their view stays stuck at "queued" forever.
+	s.mirror(responseID, stored)
 
 	xlog.Debug("Updated response status", "response_id", responseID, "status", status)
 	return nil
@@ -358,7 +442,6 @@ func (s *ResponseStore) UpdateResponse(responseID string, response *schema.ORRes
 	}
 
 	stored.mu.Lock()
-	defer stored.mu.Unlock()
 
 	// Rebuild item index
 	items := make(map[string]*schema.ORItemField)
@@ -371,6 +454,10 @@ func (s *ResponseStore) UpdateResponse(responseID string, response *schema.ORRes
 
 	stored.Response = response
 	stored.Items = items
+	stored.mu.Unlock()
+
+	// The final output is what a peer's poll must return, so replicate it.
+	s.mirror(responseID, stored)
 
 	xlog.Debug("Updated response", "response_id", responseID, "status", response.Status, "items_count", len(items))
 	return nil
@@ -436,6 +523,13 @@ func (s *ResponseStore) GetEventsAfter(responseID string, startingAfter int) ([]
 	s.mu.RUnlock()
 
 	if !exists {
+		// The response may be alive on a peer, whose resume buffer is not
+		// replicated. Say so explicitly instead of reporting "not found" (which
+		// invites the client to give up on a live stream) or an empty slice
+		// (which looks like a finished one).
+		if _, remote := s.remoteGet(responseID); remote {
+			return nil, ErrResponseNotLocal
+		}
 		return nil, fmt.Errorf("response not found: %s", responseID)
 	}
 
@@ -460,16 +554,41 @@ func (s *ResponseStore) GetEventsAfter(responseID string, startingAfter int) ([]
 	return result, nil
 }
 
-// Cancel cancels a background response if it's still in progress
+// Cancel cancels a background response if it's still in progress.
+//
+// The context.CancelFunc that actually stops generation is a function pointer
+// and only exists in the process that created the response. When the cancel
+// lands on any other replica - which a round-robin load balancer makes roughly
+// as likely as landing on the right one - it is delegated to the owner over the
+// bus instead of being reported as a 404 while generation keeps running
+// (issue #10993).
 func (s *ResponseStore) Cancel(responseID string) (*schema.ORResponseResource, error) {
 	s.mu.RLock()
 	stored, exists := s.responses[responseID]
 	s.mu.RUnlock()
 
-	if !exists {
-		return nil, fmt.Errorf("response not found: %s", responseID)
+	if exists {
+		response, err := s.cancelLocal(responseID, stored)
+		if err != nil {
+			return nil, err
+		}
+		s.mirror(responseID, stored)
+		return response, nil
+	}
+
+	if m := s.syncMap(); m != nil {
+		if v, ok := m.Get(responseID); ok && v != nil {
+			return s.delegateCancel(v)
+		}
 	}
 
+	return nil, fmt.Errorf("response not found: %s", responseID)
+}
+
+// cancelLocal cancels a response this process owns. Shared by the direct HTTP
+// path and by the delegated path a peer triggers over the bus, so both go
+// through exactly the same terminal-state and CancelFunc handling.
+func (s *ResponseStore) cancelLocal(responseID string, stored *StoredResponse) (*schema.ORResponseResource, error) {
 	stored.mu.Lock()
 	defer stored.mu.Unlock()
 
@@ -502,6 +621,11 @@ func (s *ResponseStore) GetEventsChan(responseID string) (chan struct{}, error)
 	s.mu.RUnlock()
 
 	if !exists {
+		// A live subscriber channel cannot be handed across processes; see
+		// ErrResponseNotLocal.
+		if _, remote := s.remoteGet(responseID); remote {
+			return nil, ErrResponseNotLocal
+		}
 		return nil, fmt.Errorf("response not found: %s", responseID)
 	}
 
@@ -515,6 +639,11 @@ func (s *ResponseStore) IsStreamEnabled(responseID string) (bool, error) {
 	s.mu.RUnlock()
 
 	if !exists {
+		// Stream-enabled is replicated metadata, so a peer can answer this one
+		// even though it cannot serve the buffer itself.
+		if remote, ok := s.remoteGet(responseID); ok {
+			return remote.StreamEnabled, nil
+		}
 		return false, fmt.Errorf("response not found: %s", responseID)
 	}
 
@@ -541,6 +670,10 @@ func (s *ResponseStore) SetOwner(responseID, owner string) {
 	}
 
 	stored.Owner = owner
+
+	// Owner is part of the replicated metadata: without it a peer would treat
+	// the response as ownerless and skip the access check in accessAllowed.
+	s.mirror(responseID, stored)
 }
 
 // accessAllowed reports whether a caller identified by callerID may read or