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

In distributed mode the Open Responses store is process-local: a
sync.OnceValue over a map behind an RWMutex. With several frontend
replicas behind a round-robin load balancer, every request that lands on
a replica other than the creator misses.

Measured on a live 2-replica cluster (#10993): the same response id
returns 200 on the creating replica and 404 on its peer, and a cancel on
the peer returns 404 without ever invoking CancelFunc, so generation runs
to completion on the other replica while the caller is told the response
does not exist. previous_response_id chaining fails through the same
lookup.

Split the state by what can actually cross a process boundary:

- Replicated: response metadata (request, response resource, owner,
  expiry, stream/background flags) via syncstate.SyncedMap, the same
  component finetune, quantization and agent tasks already use. A local
  miss in Get/FindItem now falls back to it and returns a read-only
  remote view, so polling and chaining resolve on any replica.

- Delegated: cancellation. context.CancelFunc is a function pointer and
  exists only in the creating process, so a cancel that lands elsewhere
  is broadcast on responses.<id>.cancel and applied by whichever replica
  holds the function. The broadcast is fire-and-forget rather than
  request/reply: if the owner crashed or was scaled down nobody answers,
  and the handler must not block on a reply that will never come. The
  replicated status moves to cancelled either way, which is truthful,
  since a dead owner's generation died with its process.

- Refused: streaming resume. The resume buffer is a byte log plus a live
  notification channel and cannot be replicated without shipping every
  token over the bus. A resume that reaches the wrong replica now returns
  HTTP 409 naming the owning replica via the new ErrResponseNotLocal,
  instead of an empty event list that looks like a finished stream. It is
  deliberately distinct from ErrOffsetLost, which means the owner's
  buffer evicted the requested events.

Standalone deployments never call EnableDistributed and keep exactly the
previous process-local behaviour.

Fixes #10993


Assisted-by: Claude Code:claude-opus-4-8[1m] [Read] [Edit] [Bash]

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

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,10 +245,41 @@ 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) {
@@ -222,18 +287,19 @@ func (s *ResponseStore) FindItem(itemID string) (*schema.ORItemField, string, er
 		}
 
 		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

core/http/endpoints/openresponses/sync.go

@@ -0,0 +1,344 @@
+package openresponses
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"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"
+)
+
+// syncStateName is the syncstate namespace for replicated response metadata.
+// It becomes the NATS subject "state.responses.metadata.delta".
+const syncStateName = "responses.metadata"
+
+// ErrResponseNotLocal is returned by the stream-resume accessors when the
+// response exists in the cluster but its resume buffer lives in another
+// replica's process memory.
+//
+// The buffer is a byte log plus a live notification channel: neither can be
+// replicated without shipping every generated token over the message bus, so a
+// resume request that lands on the wrong replica cannot be served. Returning a
+// distinct error lets the HTTP layer say exactly that instead of handing the
+// client an empty event list that looks like a completed stream. It is
+// deliberately NOT ErrOffsetLost: that means "the buffer evicted the events you
+// asked for" (a retention decision on the owning replica) and a client may
+// reasonably give up on the stream. This one means "ask the replica that owns
+// it", which is a routing problem, and the two must stay distinguishable.
+var ErrResponseNotLocal = errors.New("response is owned by another replica: its stream resume buffer is not available here")
+
+// syncedResponse is the replicable projection of a StoredResponse.
+//
+// Only what a peer can act on is carried: the request (so previous_response_id
+// chaining can replay it), the response resource (so polling returns real
+// state), the expiry, and the owning replica. Everything process-local is
+// excluded by construction - the CancelFunc is a function pointer, EventsChan is
+// a live channel, and StreamEvents is an unbounded byte log. Cancel is therefore
+// delegated to the owner over the bus rather than replicated (see
+// delegateCancel), and stream resume is refused off-owner (ErrResponseNotLocal).
+type syncedResponse struct {
+	ID            string                       `json:"id"`
+	OwnerReplica  string                       `json:"owner_replica"`
+	Owner         string                       `json:"owner,omitempty"`
+	Request       *schema.OpenResponsesRequest `json:"request,omitempty"`
+	Response      *schema.ORResponseResource   `json:"response,omitempty"`
+	StoredAt      time.Time                    `json:"stored_at"`
+	ExpiresAt     *time.Time                   `json:"expires_at,omitempty"`
+	StreamEnabled bool                         `json:"stream_enabled,omitempty"`
+	IsBackground  bool                         `json:"is_background,omitempty"`
+}
+
+// responseCancelEvent is the wire envelope broadcast on SubjectResponseCancel.
+// Origin lets the publisher skip its own echo: it has already cancelled locally
+// (or knows it holds nothing to cancel) before publishing.
+type responseCancelEvent struct {
+	ResponseID string `json:"response_id"`
+	Origin     string `json:"origin"`
+}
+
+// EnableDistributed turns on cross-replica replication for this store.
+//
+// It is called once during route registration when the application is in
+// distributed mode, before any request is served. Standalone deployments never
+// call it and keep the store exactly as it was: a process-local map with no
+// broadcast, no subscription and no behavioural change.
+//
+// Two independent legs are wired:
+//
+//   - a syncstate.SyncedMap carrying response metadata, so a GET or a
+//     previous_response_id lookup that a round-robin load balancer sends to a
+//     replica which did not create the response still resolves;
+//   - a wildcard subscription on the response-cancel subject, so a cancel that
+//     lands on the wrong replica still reaches the context.CancelFunc.
+//
+// The SyncedMap has no durable Store: responses are ephemeral, TTL-bounded
+// state that today does not survive a process restart either, so peers converge
+// through deltas alone. A replica that joins later does not learn about
+// responses created before it started; that is the same visibility a client had
+// before this change and strictly better than the 404 it got from every peer.
+func (s *ResponseStore) EnableDistributed(ctx context.Context, nats messaging.MessagingClient, replicaID string) error {
+	if nats == nil {
+		return nil
+	}
+
+	// The subscription callbacks and the delegated-cancel publisher outlive the
+	// caller's ctx (which is startup-scoped), so hold a context that only Close
+	// cancels.
+	lifeCtx, lifeCancel := context.WithCancel(context.Background()) //#nosec G118 -- cancelled in Close()
+
+	synced := syncstate.New(syncstate.Config[string, *syncedResponse]{
+		Name: syncStateName,
+		Key:  func(v *syncedResponse) string { return v.ID },
+		Nats: nats,
+	})
+	if err := synced.Start(ctx); err != nil {
+		lifeCancel()
+		return fmt.Errorf("starting response metadata sync: %w", err)
+	}
+
+	// Publish the wiring before subscribing: the subscription handler re-enters
+	// the store, so everything it reads has to be in place first.
+	s.mu.Lock()
+	s.replicaID = replicaID
+	s.nats = nats
+	s.lifeCtx, s.lifeCancel = lifeCtx, lifeCancel
+	s.synced = synced
+	s.mu.Unlock()
+
+	sub, err := messaging.SubscribeJSON(nats, messaging.SubjectResponseCancelWildcard, s.applyRemoteCancel)
+	if err != nil {
+		if cerr := s.Close(); cerr != nil {
+			xlog.Warn("failed to tear down response metadata sync after subscribe error", "error", cerr)
+		}
+		return fmt.Errorf("subscribing to response cancel wildcard: %w", err)
+	}
+
+	s.mu.Lock()
+	s.cancelSub = sub
+	s.mu.Unlock()
+
+	xlog.Info("Open Responses store replicating across replicas", "replica_id", replicaID)
+	return nil
+}
+
+// Close tears down the distributed wiring. It is idempotent so a test (or a
+// double shutdown) can call it more than once, and is a no-op for a standalone
+// store.
+func (s *ResponseStore) Close() error {
+	s.mu.Lock()
+	sub := s.cancelSub
+	s.cancelSub = nil
+	synced := s.synced
+	s.synced = nil
+	cancel := s.lifeCancel
+	s.lifeCancel = nil
+	s.mu.Unlock()
+
+	if cancel != nil {
+		cancel()
+	}
+	if sub != nil {
+		if err := sub.Unsubscribe(); err != nil {
+			return err
+		}
+	}
+	if synced != nil {
+		return synced.Close()
+	}
+	return nil
+}
+
+// syncMap returns the replicated metadata map, or nil in standalone mode.
+func (s *ResponseStore) syncMap() *syncstate.SyncedMap[string, *syncedResponse] {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.synced
+}
+
+// distributed returns the replication handles as a consistent snapshot. Every
+// path that broadcasts reads them through here so a concurrent Close cannot be
+// observed half-applied. A nil map means standalone mode.
+func (s *ResponseStore) distributed() (*syncstate.SyncedMap[string, *syncedResponse], context.Context, messaging.MessagingClient, string) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	ctx := s.lifeCtx
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	return s.synced, ctx, s.nats, s.replicaID
+}
+
+// replicaIdentity returns this process's replica ID (empty in standalone mode).
+func (s *ResponseStore) replicaIdentity() string {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.replicaID
+}
+
+// mirror publishes the current metadata of a locally-owned response to peers.
+//
+// It must be called with s.mu released: the broadcast is delivered synchronously
+// by the in-memory bus used in tests, and a subscriber callback re-enters the
+// store to look up the response.
+func (s *ResponseStore) mirror(responseID string, stored *StoredResponse) {
+	m, ctx, _, replicaID := s.distributed()
+	if m == nil {
+		return
+	}
+
+	stored.mu.RLock()
+	v := &syncedResponse{
+		ID:            responseID,
+		OwnerReplica:  replicaID,
+		Owner:         stored.Owner,
+		Request:       stored.Request,
+		Response:      stored.Response,
+		StoredAt:      stored.StoredAt,
+		ExpiresAt:     stored.ExpiresAt,
+		StreamEnabled: stored.StreamEnabled,
+		IsBackground:  stored.IsBackground,
+	}
+	stored.mu.RUnlock()
+
+	if err := m.Set(ctx, v); err != nil {
+		xlog.Warn("failed to replicate Open Responses metadata", "response_id", responseID, "error", err)
+	}
+}
+
+// unmirror removes a response from the replicated metadata.
+func (s *ResponseStore) unmirror(responseID string) {
+	m, ctx, _, _ := s.distributed()
+	if m == nil {
+		return
+	}
+	if err := m.Delete(ctx, responseID); err != nil {
+		xlog.Warn("failed to remove replicated Open Responses metadata", "response_id", responseID, "error", err)
+	}
+}
+
+// remoteGet resolves a response from the replicated metadata and materialises a
+// read-only StoredResponse view of it.
+//
+// The view is rebuilt on every call rather than inserted into s.responses: it
+// carries no CancelFunc and no resume buffer, so caching it locally would make
+// the store unable to tell "mine" from "a peer's" and would silently break the
+// stream and cancel paths. Remote is the flag every process-local accessor keys
+// off.
+func (s *ResponseStore) remoteGet(responseID string) (*StoredResponse, bool) {
+	m := s.syncMap()
+	if m == nil {
+		return nil, false
+	}
+	v, ok := m.Get(responseID)
+	if !ok || v == nil || v.Response == nil {
+		return nil, false
+	}
+	if v.ExpiresAt != nil && time.Now().After(*v.ExpiresAt) {
+		return nil, false
+	}
+
+	// Rebuild the item index on this side so GetItem/FindItem behave the same
+	// as they do on the owner; the index is derived data and is not shipped.
+	items := make(map[string]*schema.ORItemField)
+	for i := range v.Response.Output {
+		item := &v.Response.Output[i]
+		if item.ID != "" {
+			items[item.ID] = item
+		}
+	}
+
+	return &StoredResponse{
+		Request:        v.Request,
+		Response:       v.Response,
+		Items:          items,
+		StoredAt:       v.StoredAt,
+		ExpiresAt:      v.ExpiresAt,
+		Owner:          v.Owner,
+		StreamEnabled:  v.StreamEnabled,
+		IsBackground:   v.IsBackground,
+		Remote:         true,
+		OwnerReplica:   v.OwnerReplica,
+		droppedThrough: -1,
+	}, true
+}
+
+// delegateCancel handles a cancel for a response this replica does not own.
+//
+// The CancelFunc lives in the owner's process, so the request is broadcast and
+// the owner applies it locally (applyRemoteCancel). The broadcast is
+// fire-and-forget: a request/reply would block this HTTP handler for a timeout
+// whenever the owner has crashed or been scaled down, which is exactly the case
+// where the caller most needs a prompt answer.
+//
+// The replicated status is moved to cancelled here rather than waiting for the
+// owner to confirm. If the owner is alive it will cancel and converge on the
+// same value; if it is gone its generation died with its process, so cancelled
+// is the truthful terminal state either way.
+func (s *ResponseStore) delegateCancel(v *syncedResponse) (*schema.ORResponseResource, error) {
+	if v.Response == nil {
+		return nil, fmt.Errorf("response not found: %s", v.ID)
+	}
+
+	status := v.Response.Status
+	if status == schema.ORStatusCompleted || status == schema.ORStatusFailed ||
+		status == schema.ORStatusIncomplete || status == schema.ORStatusCancelled {
+		xlog.Debug("Peer-owned response already in terminal state", "response_id", v.ID, "status", status)
+		return v.Response, nil
+	}
+
+	m, ctx, nats, replicaID := s.distributed()
+	if nats != nil {
+		if err := nats.Publish(messaging.SubjectResponseCancel(v.ID),
+			responseCancelEvent{ResponseID: v.ID, Origin: replicaID}); err != nil {
+			xlog.Warn("failed to broadcast Open Responses cancel", "response_id", v.ID, "error", err)
+		}
+	}
+
+	// Copy before mutating: the value in the SyncedMap is shared with any peer
+	// delta already applied, and the response resource is handed to the client.
+	updated := *v
+	response := *v.Response
+	response.Status = schema.ORStatusCancelled
+	now := time.Now().Unix()
+	response.CompletedAt = &now
+	updated.Response = &response
+
+	if m != nil {
+		if err := m.Set(ctx, &updated); err != nil {
+			xlog.Warn("failed to replicate cancelled Open Responses status", "response_id", v.ID, "error", err)
+		}
+	}
+
+	xlog.Debug("Delegated Open Responses cancel to owning replica", "response_id", v.ID, "owner_replica", v.OwnerReplica)
+	return &response, nil
+}
+
+// applyRemoteCancel runs a peer's delegated cancel against local state.
+//
+// Only the replica that actually holds the response in s.responses has anything
+// to do; every other subscriber drops the event. It deliberately does not
+// re-broadcast and does not re-write the replicated metadata - the delegating
+// replica already published the cancelled status - which is the same echo-loop
+// guard syncstate applies on its own apply path.
+func (s *ResponseStore) applyRemoteCancel(evt responseCancelEvent) {
+	if evt.ResponseID == "" || evt.Origin == s.replicaIdentity() {
+		return
+	}
+
+	s.mu.RLock()
+	stored, exists := s.responses[evt.ResponseID]
+	s.mu.RUnlock()
+	if !exists {
+		return
+	}
+
+	if _, err := s.cancelLocal(evt.ResponseID, stored); err != nil {
+		xlog.Warn("failed to apply delegated Open Responses cancel", "response_id", evt.ResponseID, "error", err)
+		return
+	}
+	xlog.Debug("Applied delegated Open Responses cancel", "response_id", evt.ResponseID, "origin", evt.Origin)
+}

core/http/endpoints/openresponses/sync_test.go

@@ -0,0 +1,208 @@
+package openresponses
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/mudler/LocalAI/core/schema"
+	"github.com/mudler/LocalAI/core/services/testutil"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+// These specs model the two-replica topology from issue #10993: two independent
+// ResponseStore instances (one per frontend process) sharing a single message
+// bus. Anything a client can observe through the HTTP API after a round-robin
+// load balancer sends it to the "wrong" replica must be asserted here.
+var _ = Describe("ResponseStore cross-replica", func() {
+	var (
+		bus      *testutil.FakeBus
+		replicaA *ResponseStore
+		replicaB *ResponseStore
+		ctx      context.Context
+	)
+
+	BeforeEach(func() {
+		ctx = context.Background()
+		bus = testutil.NewFakeBus()
+
+		replicaA = NewResponseStore(0)
+		replicaB = NewResponseStore(0)
+
+		Expect(replicaA.EnableDistributed(ctx, bus, "replica-a")).To(Succeed())
+		Expect(replicaB.EnableDistributed(ctx, bus, "replica-b")).To(Succeed())
+	})
+
+	AfterEach(func() {
+		Expect(replicaA.Close()).To(Succeed())
+		Expect(replicaB.Close()).To(Succeed())
+	})
+
+	newResponse := func(id, status string) *schema.ORResponseResource {
+		return &schema.ORResponseResource{
+			ID:        id,
+			Object:    "response",
+			CreatedAt: time.Now().Unix(),
+			Status:    status,
+			Model:     "test-model",
+			Output: []schema.ORItemField{
+				{Type: "message", ID: "msg_" + id, Role: "assistant"},
+			},
+		}
+	}
+
+	Describe("polling and previous_response_id chaining", func() {
+		It("makes a response created on one replica readable on its peer", func() {
+			const id = "resp_cross_replica"
+			request := &schema.OpenResponsesRequest{Model: "test-model", Input: "Hello"}
+
+			replicaA.Store(id, request, newResponse(id, schema.ORStatusCompleted))
+
+			// The peer never saw the POST that created this response; without
+			// replication this is the HTTP 404 reported in #10993.
+			stored, err := replicaB.Get(id)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(stored).ToNot(BeNil())
+			Expect(stored.Response.Status).To(Equal(schema.ORStatusCompleted))
+			// previous_response_id chaining replays the original request, so the
+			// request body has to survive the hop, not just the response.
+			Expect(stored.Request).ToNot(BeNil())
+			Expect(stored.Request.Model).To(Equal("test-model"))
+			// Item lookup is rebuilt on the peer so GetItem/FindItem work there too.
+			Expect(stored.Items).To(HaveKey("msg_" + id))
+		})
+
+		It("propagates status updates from the owner to the peer", func() {
+			const id = "resp_status_update"
+			replicaA.StoreBackground(id, &schema.OpenResponsesRequest{Model: "test-model"},
+				newResponse(id, schema.ORStatusQueued), func() {}, false)
+
+			completedAt := time.Now().Unix()
+			Expect(replicaA.UpdateStatus(id, schema.ORStatusCompleted, &completedAt)).To(Succeed())
+
+			stored, err := replicaB.Get(id)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(stored.Response.Status).To(Equal(schema.ORStatusCompleted))
+		})
+
+		It("removes a deleted response from the peer as well", func() {
+			const id = "resp_deleted"
+			replicaA.Store(id, &schema.OpenResponsesRequest{Model: "test-model"}, newResponse(id, schema.ORStatusCompleted))
+			Expect(replicaB.Get(id)).ToNot(BeNil())
+
+			replicaA.Delete(id)
+
+			_, err := replicaB.Get(id)
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("still reports a genuinely unknown response as not found", func() {
+			_, err := replicaB.Get("resp_never_existed")
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
+	Describe("cancel delegation", func() {
+		It("reaches the CancelFunc held by the owning replica", func() {
+			const id = "resp_cancel_delegated"
+			cancelled := make(chan struct{})
+			replicaA.StoreBackground(id, &schema.OpenResponsesRequest{Model: "test-model"},
+				newResponse(id, schema.ORStatusInProgress), func() { close(cancelled) }, false)
+
+			// Cancel lands on the replica that does NOT hold the CancelFunc. Today
+			// this 404s and generation keeps burning GPU on replica A.
+			resp, err := replicaB.Cancel(id)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(resp.Status).To(Equal(schema.ORStatusCancelled))
+
+			Eventually(cancelled).Should(BeClosed())
+
+			// The owner's own view must converge too, so a later poll on A does
+			// not report the response as still in progress.
+			ownerView, err := replicaA.Get(id)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(ownerView.Response.Status).To(Equal(schema.ORStatusCancelled))
+		})
+
+		It("does not hang when the owning replica is gone", func() {
+			const id = "resp_dead_owner"
+			replicaA.StoreBackground(id, &schema.OpenResponsesRequest{Model: "test-model"},
+				newResponse(id, schema.ORStatusInProgress), func() {}, false)
+
+			// Simulate the owner crashing / being scaled down: it stops consuming
+			// the bus, so nothing will ever answer a delegated cancel. The peer
+			// still holds the replicated metadata and must resolve on its own.
+			Expect(replicaA.Close()).To(Succeed())
+
+			done := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				defer close(done)
+				resp, err := replicaB.Cancel(id)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(resp.Status).To(Equal(schema.ORStatusCancelled))
+			}()
+			Eventually(done, 5*time.Second).Should(BeClosed())
+		})
+
+		It("is idempotent for a response already in a terminal state", func() {
+			const id = "resp_cancel_terminal"
+			replicaA.Store(id, &schema.OpenResponsesRequest{Model: "test-model"}, newResponse(id, schema.ORStatusCompleted))
+
+			resp, err := replicaB.Cancel(id)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(resp.Status).To(Equal(schema.ORStatusCompleted))
+		})
+	})
+
+	Describe("stream resume", func() {
+		It("reports a distinct error instead of a truncated stream when the buffer lives on a peer", func() {
+			const id = "resp_stream_remote"
+			replicaA.StoreBackground(id, &schema.OpenResponsesRequest{Model: "test-model"},
+				newResponse(id, schema.ORStatusInProgress), func() {}, true)
+			Expect(replicaA.AppendEvent(id, &schema.ORStreamEvent{SequenceNumber: 1, Type: "response.created"})).To(Succeed())
+
+			// The resume buffer is a byte buffer on replica A and is deliberately
+			// not replicated; asking B for it must be an explicit error, never an
+			// empty (silently truncated) event list.
+			events, err := replicaB.GetEventsAfter(id, 0)
+			Expect(events).To(BeEmpty())
+			Expect(errors.Is(err, ErrResponseNotLocal)).To(BeTrue())
+
+			// ErrOffsetLost means "the buffer evicted your events"; a peer lookup
+			// is a different condition and must not be conflated with it.
+			Expect(errors.Is(err, ErrOffsetLost)).To(BeFalse())
+
+			_, err = replicaB.GetEventsChan(id)
+			Expect(errors.Is(err, ErrResponseNotLocal)).To(BeTrue())
+		})
+
+		It("keeps serving the resume buffer on the owning replica", func() {
+			const id = "resp_stream_local"
+			replicaA.StoreBackground(id, &schema.OpenResponsesRequest{Model: "test-model"},
+				newResponse(id, schema.ORStatusInProgress), func() {}, true)
+			Expect(replicaA.AppendEvent(id, &schema.ORStreamEvent{SequenceNumber: 1, Type: "response.created"})).To(Succeed())
+
+			events, err := replicaA.GetEventsAfter(id, 0)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(events).To(HaveLen(1))
+		})
+	})
+
+	Describe("standalone mode", func() {
+		It("keeps a store with no messaging client purely local", func() {
+			standalone := NewResponseStore(0)
+			const id = "resp_standalone"
+			standalone.Store(id, &schema.OpenResponsesRequest{Model: "test-model"}, newResponse(id, schema.ORStatusCompleted))
+
+			stored, err := standalone.Get(id)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(stored).ToNot(BeNil())
+
+			// Nothing was broadcast, so peers on a shared bus stay unaware.
+			_, err = replicaB.Get(id)
+			Expect(err).To(HaveOccurred())
+		})
+	})
+})

core/http/routes/openresponses.go

@@ -9,6 +9,7 @@ import (
 	"github.com/mudler/LocalAI/core/http/endpoints/openresponses"
 	"github.com/mudler/LocalAI/core/http/middleware"
 	"github.com/mudler/LocalAI/core/schema"
+	"github.com/mudler/xlog"
 )
 
 func RegisterOpenResponsesRoutes(app *echo.Echo,
@@ -19,6 +20,16 @@ func RegisterOpenResponsesRoutes(app *echo.Echo,
 	var natsClient mcpTools.MCPNATSClient
 	if d := application.Distributed(); d != nil {
 		natsClient = d.Nats
+
+		// Replicate response metadata across frontend replicas and subscribe to
+		// delegated cancels. Without this a GET, a previous_response_id lookup or
+		// a cancel that the load balancer sends to a replica other than the
+		// creator 404s, and the cancel never reaches the CancelFunc (#10993).
+		// Standalone deployments skip this entirely and stay process-local.
+		if err := openresponses.GetGlobalStore().EnableDistributed(
+			application.ApplicationConfig().Context, d.Nats, application.InstanceID()); err != nil {
+			xlog.Error("Failed to enable cross-replica Open Responses store", "error", err)
+		}
 	}
 
 	// Open Responses API endpoint

core/services/messaging/subjects.go

@@ -96,6 +96,7 @@ const (
 	subjectAgentCancelPrefix    = "agent."
 	subjectFineTuneCancelPrefix = "finetune."
 	subjectGalleryCancelPrefix  = "gallery."
+	subjectResponseCancelPrefix = "responses."
 )
 
 // Wildcard subjects for NATS subscriptions that match all IDs.
@@ -106,6 +107,7 @@ const (
 	SubjectAgentCancelWildcard     = "agent.*.cancel"
 	SubjectGalleryCancelWildcard   = "gallery.*.cancel"
 	SubjectGalleryProgressWildcard = "gallery.*.progress"
+	SubjectResponseCancelWildcard  = "responses.*.cancel"
 )
 
 // SubjectJobCancel returns the NATS subject to cancel a running job.
@@ -128,6 +130,17 @@ func SubjectGalleryCancel(opID string) string {
 	return subjectGalleryCancelPrefix + sanitizeSubjectToken(opID) + ".cancel"
 }
 
+// SubjectResponseCancel returns the NATS subject used to cancel an in-flight
+// Open Responses generation. Only the replica that created the response holds
+// its context.CancelFunc, so a cancel that lands on any other replica is
+// broadcast here and applied by whichever replica actually owns the function.
+// Broadcast rather than request/reply on purpose: if the owner crashed or was
+// scaled down, nobody answers and the caller must not block waiting for a
+// reply that will never come.
+func SubjectResponseCancel(responseID string) string {
+	return subjectResponseCancelPrefix + sanitizeSubjectToken(responseID) + ".cancel"
+}
+
 // Node Backend Lifecycle (Pub/Sub — targeted to specific nodes)
 //
 // These subjects control the backend *process* lifecycle on a serve-backend node,

docs/content/features/text-generation.md

@@ -331,6 +331,22 @@ Cancel a background response that's still in progress:
 curl -X POST http://localhost:8080/v1/responses/resp_abc123/cancel
 ```
 
+#### Multiple Replicas (Distributed Mode)
+
+In distributed mode LocalAI replicates response metadata across frontend
+replicas, so retrieval, `previous_response_id` chaining and cancellation work
+regardless of which replica the load balancer picks:
+
+- `GET /v1/responses/{id}` returns the response from any replica.
+- `POST /v1/responses/{id}/cancel` is delegated over NATS to the replica that is
+  actually generating, so generation really stops. If that replica is gone, the
+  response is reported as `cancelled` without blocking.
+- **Streaming resume (`?stream=true`) is served only by the replica that created
+  the response.** The event buffer lives in that process's memory and is not
+  replicated. A resume request that reaches another replica returns HTTP 409
+  naming the owning replica instead of silently returning a truncated stream.
+  Poll the response instead, or route resume requests with session affinity.
+
 #### Tool Calling
 
 Open Responses API supports function calling with tools: