LocalAI/core/http/routes/openresponses.go

package routes

import (
	"context"

	"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"
	"github.com/mudler/LocalAI/core/http/endpoints/openresponses"
	"github.com/mudler/LocalAI/core/http/middleware"
	"github.com/mudler/LocalAI/core/schema"
	"github.com/mudler/LocalAI/core/services/distributed"
	"github.com/mudler/xlog"
)

func RegisterOpenResponsesRoutes(app *echo.Echo,
	re *middleware.RequestExtractor,
	application *application.Application) {

	// How the MCP endpoints reach an agent worker; nil outside distributed mode.
	agentControl := mcpAgentControl(application)
	if d := application.Distributed(); d != nil {
		if err := enableDistributedResponses(application.ApplicationConfig().Context, d,
			openresponses.GetGlobalStore(), 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(),
		agentControl,
	)

	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)
		}
	}
}

// enableDistributedResponses replicates response metadata across frontend
// replicas and subscribes to delegated cancels. Without it 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 never reach here and stay
// process-local.
//
// The durable store is what a replica re-hydrates from after its subscription
// missed a delta; without it the same response_id answers 404 here and 200 on
// the peer that created it, forever.
//
// A named function rather than a block inside route registration, and that is
// the point of it. EnableDistributed takes a messaging.Broadcaster, as it must:
// its own specs publish through a double. So handing it any carrier other than
// the deployment's COMPILES and reddens nothing anywhere, and the only symptom
// is a cancel that answers 404 on every replica but one. The broker client that
// used to be the second carrier in scope is gone; what pins the choice is the
// spec beside this file, which drives it from the OTHER carrier. Registering
// routes needs a whole Application and therefore has no spec; this needs a
// DistributedServices and a store, and therefore has one.
func enableDistributedResponses(ctx context.Context, d *application.DistributedServices,
	store *openresponses.ResponseStore, replicaID string) error {
	var responseStore *distributed.ResponseMetadataStore
	if d.DistStores != nil {
		responseStore = d.DistStores.Responses
	}
	return store.EnableDistributed(ctx, d.Broadcast(), replicaID, responseStore)
}