mirror of
https://github.com/mudler/LocalAI.git
synced 2026-07-31 10:28:43 -04:00
d7e04dcc32
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>
88 lines
3.8 KiB
Go
88 lines
3.8 KiB
Go
package routes
|
|
|
|
import (
|
|
"github.com/labstack/echo/v4"
|
|
"github.com/mudler/LocalAI/core/application"
|
|
"github.com/mudler/LocalAI/core/config"
|
|
localai "github.com/mudler/LocalAI/core/http/endpoints/localai"
|
|
mcpTools "github.com/mudler/LocalAI/core/http/endpoints/mcp"
|
|
"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,
|
|
re *middleware.RequestExtractor,
|
|
application *application.Application) {
|
|
|
|
// NATS client for distributed MCP tool routing (nil when not in distributed mode)
|
|
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
|
|
responsesHandler := openresponses.ResponsesEndpoint(
|
|
application.ModelConfigLoader(),
|
|
application.ModelLoader(),
|
|
application.TemplatesEvaluator(),
|
|
application.ApplicationConfig(),
|
|
natsClient,
|
|
)
|
|
|
|
responsesMiddleware := []echo.MiddlewareFunc{
|
|
// Intercept requests where the model name matches an agent — route directly
|
|
// to the agent pool without going through the model config resolution pipeline.
|
|
localai.AgentResponsesInterceptor(application),
|
|
middleware.UsageMiddleware(application.StatsRecorder(), application.FallbackUser()),
|
|
middleware.TraceMiddleware(application),
|
|
re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_CHAT)),
|
|
re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.OpenResponsesRequest) }),
|
|
setOpenResponsesRequestContext(re),
|
|
}
|
|
|
|
// Main Open Responses endpoint
|
|
app.POST("/v1/responses", responsesHandler, responsesMiddleware...)
|
|
|
|
// Also support without version prefix for compatibility
|
|
app.POST("/responses", responsesHandler, responsesMiddleware...)
|
|
|
|
// WebSocket mode for Responses API
|
|
wsHandler := openresponses.WebSocketEndpoint(application)
|
|
app.GET("/v1/responses", wsHandler, middleware.UsageMiddleware(application.StatsRecorder(), application.FallbackUser()), middleware.TraceMiddleware(application))
|
|
app.GET("/responses", wsHandler, middleware.UsageMiddleware(application.StatsRecorder(), application.FallbackUser()), middleware.TraceMiddleware(application))
|
|
|
|
// GET /responses/:id - Retrieve a response (for polling background requests)
|
|
getResponseHandler := openresponses.GetResponseEndpoint()
|
|
app.GET("/v1/responses/:id", getResponseHandler, middleware.TraceMiddleware(application))
|
|
app.GET("/responses/:id", getResponseHandler, middleware.TraceMiddleware(application))
|
|
|
|
// POST /responses/:id/cancel - Cancel a background response
|
|
cancelResponseHandler := openresponses.CancelResponseEndpoint()
|
|
app.POST("/v1/responses/:id/cancel", cancelResponseHandler, middleware.TraceMiddleware(application))
|
|
app.POST("/responses/:id/cancel", cancelResponseHandler, middleware.TraceMiddleware(application))
|
|
}
|
|
|
|
// setOpenResponsesRequestContext sets up the context and cancel function for Open Responses requests
|
|
func setOpenResponsesRequestContext(re *middleware.RequestExtractor) echo.MiddlewareFunc {
|
|
return func(next echo.HandlerFunc) echo.HandlerFunc {
|
|
return func(c echo.Context) error {
|
|
if err := re.SetOpenResponsesRequest(c); err != nil {
|
|
return err
|
|
}
|
|
return next(c)
|
|
}
|
|
}
|
|
}
|