middleware/task_plugin.go
Rama main · réplica de lectura del repositorio original.
package middleware
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"mime"
"mime/multipart"
"net/http"
"net/url"
"slices"
"strconv"
"strings"
"time"
"unicode/utf8"
"github.com/MoeForge/moego/ojson"
"github.com/QuantumNous/new-api/common"
"github.com/QuantumNous/new-api/constant"
"github.com/QuantumNous/new-api/dto"
"github.com/QuantumNous/new-api/i18n"
"github.com/QuantumNous/new-api/logger"
"github.com/QuantumNous/new-api/model"
pluginruntime "github.com/QuantumNous/new-api/pkg/jsplugin"
taskjsplugin "github.com/QuantumNous/new-api/relay/channel/task/jsplugin"
relayconstant "github.com/QuantumNous/new-api/relay/constant"
"github.com/QuantumNous/new-api/relaykit/types"
"github.com/QuantumNous/new-api/service"
"github.com/gin-gonic/gin"
)
const contextKeyTaskPluginEndpointModel = "task_plugin_endpoint_model_request"
const contextKeyTaskPluginEndpointSelection = "task_plugin_endpoint_channel_selection"
type taskPluginEndpointSelection struct {
Generation *pluginruntime.RoutingGeneration
Channel *model.Channel
Model string
}
var errTaskPluginUnsupportedMediaType = errors.New("unsupported task plugin media type")
const taskPluginInvalidRouteResult = "plugin returned an invalid route result"
// Text fields have no limit of their own: like a JSON body, they are bounded
// by the request body limit, so a JSON request sent as a multipart field may
// be as large as the same request sent as the body.
const (
maxTaskPluginFormFields = 256
maxTaskPluginMultipartParts = 256
maxTaskPluginFiles = 32
maxTaskPluginFieldNameBytes = 256
maxTaskPluginFilenameBytes = 255
)
// NativeRouteCredential accepts, on the native routes of the built-in plugins
// listed here, the vendor's official API-key form as the gateway token and
// rewrites it into the Bearer form TokenAuth reads: Vidu's
// "Authorization: Token <key>", and the Gemini API's "x-goog-api-key: <key>"
// or "?key=<key>". A header the client already sent as Authorization wins.
// "?key=" is taken out of the URL in every case, so decoders, the stored
// request path and later handlers never see it. Every other plugin's routes
// accept the Bearer form only. pluginKeys are the plugins serving the route.
func NativeRouteCredential(pluginKeys ...string) gin.HandlerFunc {
viduToken := slices.Contains(pluginKeys, "vidu")
googleKey := slices.Contains(pluginKeys, "google")
return func(c *gin.Context) {
header := c.Request.Header
if viduToken {
if scheme, token, found := strings.Cut(header.Get("Authorization"), " "); found && strings.EqualFold(scheme, "Token") {
header.Set("Authorization", "Bearer "+strings.TrimSpace(token))
}
}
if googleKey {
queryKey, removed := "", false
kept := make([]string, 0)
for part := range strings.SplitSeq(c.Request.URL.RawQuery, "&") {
rawName, rawValue, _ := strings.Cut(part, "=")
if name, err := url.QueryUnescape(rawName); err != nil || name != "key" {
kept = append(kept, part)
continue
}
removed = true
if value, err := url.QueryUnescape(rawValue); err == nil && queryKey == "" {
queryKey = value
}
}
if removed {
c.Request.URL.RawQuery = strings.Join(kept, "&")
c.Request.RequestURI = c.Request.URL.RequestURI()
}
headerKey := header.Get("X-Goog-Api-Key")
header.Del("X-Goog-Api-Key")
if header.Get("Authorization") == "" {
if headerKey != "" {
header.Set("Authorization", "Bearer "+headerKey)
} else if queryKey != "" {
header.Set("Authorization", "Bearer "+queryKey)
}
}
}
c.Next()
}
}
// PrepareTaskPluginRoute resolves and executes the pinned declarative route.
// Query requests terminate here so channel distribution and billing are never
// entered; submit requests continue through the remaining route handlers.
func PrepareTaskPluginRoute() gin.HandlerFunc {
return func(c *gin.Context) {
pinnedValue, exists := c.Get(pluginruntime.ContextKeyPinnedRoute)
pinned, ok := pinnedValue.(pluginruntime.PinnedRoute)
if !exists || !ok || pinned.Plugin == nil {
abortTaskPluginRouteErrorDetail(c, http.StatusInternalServerError, "")
return
}
c.Set(pluginruntime.ContextKeyPinnedPlugin, pluginruntime.PinnedPlugin{
Generation: pinned.Generation,
Plugin: pinned.Plugin,
})
generation := uint64(0)
if pinned.Generation != nil {
generation = pinned.Generation.Number
}
candidates := pinned.Candidates
if len(candidates) == 0 {
candidates = []pluginruntime.RouteBinding{{Plugin: pinned.Plugin, Route: pinned.Route}}
}
logger.LogDebug(
c,
"task_plugin subsystem=route event=prepare_start generation=%d plugin=%q method=%q declared_type=%q candidates=%d",
generation,
pinned.Plugin.Meta.Key,
pinned.Route.Method,
pinned.Route.Type,
len(candidates),
)
requestContext, err := buildTaskPluginRouteRequest(c)
requestContext.Params = pluginruntime.RouteParams(c.Params, pinned.Route)
c.Set(pluginruntime.ContextKeyRouteRequest, requestContext)
if err != nil {
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=route event=prepare_rejected generation=%d plugin=%q stage=request_decode reason=invalid_request err=%q",
generation,
pinned.Plugin.Meta.Key,
err.Error()),
)
status := invalidRequestStatus(err)
if errors.Is(err, errTaskPluginUnsupportedMediaType) {
status = http.StatusUnsupportedMediaType
}
abortTaskPluginRouteErrorDetail(c, status, err.Error())
return
}
bodyObject, _ := requestContext.Body.(map[string]any)
bodyKind, _ := bodyObject["kind"].(string)
// Submit routes hand JSON and multipart bodies to the decoder, which
// rejects the kinds its vendor format does not use. A route scoped to
// models reads the model from a JSON body, so it takes JSON only.
// Query routes take no body.
eligible := make([]pluginruntime.RouteBinding, 0, len(candidates))
for _, candidate := range candidates {
submitBody := bodyKind == string(pluginruntime.BodyJSON) || bodyKind == string(pluginruntime.BodyMultipart) && len(candidate.Route.Models) == 0
if candidate.Route.Type == pluginruntime.RouteTypeQuery && bodyKind == string(pluginruntime.BodyNone) || candidate.Route.Type != pluginruntime.RouteTypeQuery && submitBody {
eligible = append(eligible, candidate)
}
}
if len(eligible) == 0 {
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=route event=prepare_rejected generation=%d plugin=%q stage=request_decode reason=body_kind_mismatch body_kind=%q",
generation,
pinned.Plugin.Meta.Key,
bodyKind),
)
detail := "this route requires a JSON or multipart body"
if len(pinned.Route.Models) > 0 {
detail = "this route requires a JSON body"
}
if pinned.Route.Type == pluginruntime.RouteTypeQuery {
detail = "unsupported request body for this operation"
}
abortTaskPluginRouteErrorDetail(c, http.StatusUnsupportedMediaType, detail)
return
}
if eligible[0].Route.Type == pluginruntime.RouteTypeQuery {
renderTaskPluginRouteQuery(c, pinned, requestContext, eligible)
return
}
// Every candidate decodes the request; the first that accepts it in
// key order names the model, and the candidates that accept the same
// model stay eligible for distribution, which picks the executing
// plugin by channel.
claimedModel := ""
if bodyValue, isObject := bodyObject["value"].(map[string]any); isObject {
claimedModel, _ = bodyValue["model"].(string)
}
hookStarted := time.Now()
var (
accepted []pluginruntime.RouteBinding
resolved map[string]any
resolvedIntent taskPluginRouteIntent
resolvedContext pluginruntime.RouteRequestContext
resolvedBodyText json.RawMessage
queryMissed bool
failures []string
)
rejectedPlugins := make(map[string][]string)
for _, candidate := range eligible {
candidateContext := requestContext
candidateContext.Params = pluginruntime.RouteParams(c.Params, candidate.Route)
var (
result map[string]any
bodyText json.RawMessage
intent taskPluginRouteIntent
reason string
detail string
)
// The list names declared models; a case variant or a channel
// alias of a listed model is allowed with it.
allowed := len(candidate.Route.Models) == 0
if !allowed {
_, _, allowed = model.ResolveTaskPluginModel(pinned.Generation, claimedModel, candidate.Route.Models)
}
if !allowed {
reason, detail = "model_not_allowed", fmt.Sprintf("model %q is not allowed on this route", claimedModel)
} else {
value, text, callErr := candidate.Plugin.Engine.CallPathWithMemberJSON(
c.Request.Context(), 0, candidate.Plugin.Meta.JSONTextMember("requestBody"),
"native", []string{candidate.Route.Decode}, candidateContext.JSValueFor(candidate.Plugin.Meta),
)
if callErr != nil {
reason, detail = "hook_failed", taskPluginHookDetail(callErr)
} else {
result, _ = value.(map[string]any)
bodyText = text
intent, reason, detail = checkTaskPluginRouteIntent(pinned.Generation, candidate, value)
}
}
if reason == "" && intent.kind == string(pluginruntime.RouteTypeQuery) {
if len(eligible) == 1 {
logger.LogDebug(
c,
"task_plugin subsystem=route event=resolved generation=%d plugin=%q kind=query renderer=%q task_count=%d distribute=false elapsed_ms=%d",
generation,
candidate.Plugin.Meta.Key,
candidate.Route.Render,
len(intent.taskIDs),
time.Since(hookStarted).Milliseconds(),
)
renderTaskPluginQuery(c, pinned, candidateContext, intent.taskIDs, candidate.Route.Render, true)
return
}
owned, lookupErr := taskPluginOwnsTasks(c, candidate.Plugin.Meta, intent.taskIDs)
if lookupErr != nil {
abortTaskPluginRouteError(c, http.StatusInternalServerError)
return
}
if owned {
pinned.Plugin, pinned.Route = candidate.Plugin, candidate.Route
c.Set(pluginruntime.ContextKeyPinnedRoute, pinned)
c.Set(pluginruntime.ContextKeyPinnedPlugin, pluginruntime.PinnedPlugin{Generation: pinned.Generation, Plugin: candidate.Plugin})
c.Set(pluginruntime.ContextKeyRouteRequest, candidateContext)
renderTaskPluginQuery(c, pinned, candidateContext, intent.taskIDs, candidate.Route.Render, true)
return
}
queryMissed = true
continue
}
if reason == "" && resolved != nil && intent.model != resolvedIntent.model {
logger.LogDebug(c, "task_plugin subsystem=route event=candidate_skipped generation=%d plugin=%q reason=other_model model=%q", generation, candidate.Plugin.Meta.Key, intent.model)
continue
}
if reason != "" {
if detail == "" {
detail = "Invalid request"
}
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=route event=prepare_rejected generation=%d plugin=%q stage=resolve_request reason=%s err=%q elapsed_ms=%d",
generation,
candidate.Plugin.Meta.Key,
reason,
detail,
time.Since(hookStarted).Milliseconds()),
)
if _, seen := rejectedPlugins[detail]; !seen {
failures = append(failures, detail)
}
rejectedPlugins[detail] = append(rejectedPlugins[detail], candidate.Plugin.Meta.Key)
continue
}
accepted = append(accepted, candidate)
if resolved == nil {
resolved, resolvedIntent, resolvedContext, resolvedBodyText = result, intent, candidateContext, bodyText
}
}
if len(accepted) == 0 {
if queryMissed {
abortTaskPluginRouteError(c, http.StatusNotFound)
return
}
if len(eligible) > 1 {
for index, detail := range failures {
failures[index] = strings.Join(rejectedPlugins[detail], ", ") + ": " + detail
}
}
abortTaskPluginRouteErrorDetail(c, http.StatusBadRequest, strings.Join(failures, "; "))
return
}
owner := accepted[0]
pinned.Plugin, pinned.Route, pinned.Candidates = owner.Plugin, owner.Route, accepted
c.Set(pluginruntime.ContextKeyPinnedRoute, pinned)
c.Set(pluginruntime.ContextKeyPinnedPlugin, pluginruntime.PinnedPlugin{Generation: pinned.Generation, Plugin: owner.Plugin})
_, bodyReplaced := resolved["requestBody"]
if bodyReplaced {
resolvedContext.RequestBody = resolved["requestBody"]
}
c.Set(pluginruntime.ContextKeyRouteRequest, resolvedContext)
c.Set("task_request", resolvedContext.RequestBody)
if resolvedBodyText != nil {
c.Set(pluginruntime.ContextKeyRequestBodyText, resolvedBodyText)
}
c.Set("resolved_task_model", resolvedIntent.model)
c.Set("expected_task_plugin_key", owner.Plugin.Meta.Key)
c.Set("task_plugin_key", owner.Plugin.Meta.Key)
c.Set("platform", owner.Plugin.Meta.Key)
service.AppendTaskPluginIdentityFilter(c, owner.Plugin.Meta.Key)
if resolvedIntent.action != "" {
c.Set("task_action", resolvedIntent.action)
}
c.Set("relay_mode", relayconstant.RelayModeVideoSubmit)
if intentErr := applyOriginTaskIntent(c, resolved, owner.Plugin.Meta); intentErr != nil {
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=route event=prepare_rejected generation=%d plugin=%q stage=origin_task reason=%s",
generation,
owner.Plugin.Meta.Key,
intentErr.Code),
)
abortTaskPluginRouteErrorDetail(c, intentErr.StatusCode, intentErr.Message)
return
}
logger.LogDebug(
c,
"task_plugin subsystem=route event=resolved generation=%d plugin=%q kind=submit model=%q action_present=%t request_body_replaced=%t candidates=%d distribute=true elapsed_ms=%d",
generation,
owner.Plugin.Meta.Key,
resolvedIntent.model,
resolvedIntent.action != "",
bodyReplaced,
len(accepted),
time.Since(hookStarted).Milliseconds(),
)
c.Next()
}
}
// taskPluginRouteIntent is what a native route's decoder resolved: a submit
// (model and action) or, on a dynamic route, a query (task ids).
type taskPluginRouteIntent struct {
kind string
model string
action string
taskIDs []string
}
// checkTaskPluginRouteIntent validates a native decoder's result against the
// candidate route. A rejection names a log reason and the client's detail.
func checkTaskPluginRouteIntent(generation *pluginruntime.RoutingGeneration, candidate pluginruntime.RouteBinding, value any) (intent taskPluginRouteIntent, reason, detail string) {
resolved, ok := value.(map[string]any)
if !ok {
return intent, "result_not_object", taskPluginInvalidRouteResult
}
kind, ok := resolved["kind"].(string)
if !ok {
return intent, "missing_kind", taskPluginInvalidRouteResult
}
if _, forbidden := resolved["renderer"]; forbidden {
return intent, "forbidden_renderer", taskPluginInvalidRouteResult
}
intent.kind = kind
switch kind {
case string(pluginruntime.RouteTypeSubmit):
modelName, valid := resolved["model"].(string)
if !valid || strings.TrimSpace(modelName) == "" {
return intent, "invalid_model", "decoded request is missing a model"
}
// The task is named as on the host protocols: a case variant by the
// declared spelling, a channel alias by the alias.
served, declared, owned := model.ResolveTaskPluginModel(generation, modelName, candidate.Plugin.Meta.Models)
if !owned {
return intent, "model_not_owned", fmt.Sprintf("model %q is not served by this plugin", modelName)
}
if len(candidate.Route.Models) > 0 && !slices.Contains(candidate.Route.Models, declared) {
return intent, "resolved_model_not_allowed", fmt.Sprintf("model %q is not allowed on this route", modelName)
}
intent.model = served
intent.action = candidate.Route.Action
if resolvedAction, present := resolved["action"]; present {
actionValue, actionOK := resolvedAction.(string)
if !actionOK {
return intent, "invalid_action", taskPluginInvalidRouteResult
}
if strings.TrimSpace(actionValue) != "" {
intent.action = actionValue
}
}
case string(pluginruntime.RouteTypeQuery):
if candidate.Route.Type != pluginruntime.RouteTypeDynamic {
return intent, "query_from_non_dynamic_route", taskPluginInvalidRouteResult
}
taskIDs, valid := resolvedTaskPluginIDs(resolved["taskIds"])
if !valid {
return intent, "invalid_query_result", taskPluginInvalidRouteResult
}
intent.taskIDs = taskIDs
default:
return intent, "unsupported_kind", taskPluginInvalidRouteResult
}
return intent, "", ""
}
// renderTaskPluginRouteQuery answers a native query route. A candidate reads
// the task id from its path parameter or, when its taskIdParam is not in the
// path, from the query string. On a path several plugins declare, the task's
// own plugin answers: the first candidate in key order that owns it. A task
// none of them owns is not found, like a missing one.
func renderTaskPluginRouteQuery(c *gin.Context, pinned pluginruntime.PinnedRoute, requestContext pluginruntime.RouteRequestContext, candidates []pluginruntime.RouteBinding) {
taskIDs := make([]string, len(candidates))
for index, candidate := range candidates {
if !candidate.Route.TaskIDInQuery() {
taskIDs[index] = pluginruntime.RouteParams(c.Params, candidate.Route)[candidate.Route.TaskIDParam]
} else if values := requestContext.Query[candidate.Route.TaskIDParam]; len(values) > 0 {
taskIDs[index] = strings.TrimSpace(values[0])
}
}
if !slices.ContainsFunc(taskIDs, func(taskID string) bool { return taskID != "" }) {
abortTaskPluginRouteErrorDetail(c, http.StatusBadRequest, candidates[0].Route.TaskIDParam+" is required")
return
}
ownerIndex := 0
if len(candidates) > 1 {
ownerIndex = -1
platforms := make([]constant.TaskPlatform, 0, len(candidates))
for index, candidate := range candidates {
if taskIDs[index] != "" {
platforms = append(platforms, taskPluginLegacyPlatforms(candidate.Plugin.Meta)...)
}
}
tasks, err := model.GetByTaskIdsForPlatforms(common.GetContextKeyInt(c, constant.ContextKeyUserId), platforms, slices.Compact(slices.Sorted(slices.Values(taskIDs))))
if err != nil {
abortTaskPluginRouteError(c, http.StatusInternalServerError)
return
}
for index, candidate := range candidates {
if slices.ContainsFunc(tasks, func(task *model.Task) bool {
return task.TaskID == taskIDs[index] && task.ResultRetrievable() && slices.Contains(taskPluginLegacyPlatforms(candidate.Plugin.Meta), task.Platform)
}) {
ownerIndex = index
break
}
}
if ownerIndex < 0 {
abortTaskPluginRouteError(c, http.StatusNotFound)
return
}
}
owner := candidates[ownerIndex]
if taskIDs[ownerIndex] == "" {
abortTaskPluginRouteErrorDetail(c, http.StatusBadRequest, owner.Route.TaskIDParam+" is required")
return
}
requestContext.Params = pluginruntime.RouteParams(c.Params, owner.Route)
pinned.Plugin, pinned.Route = owner.Plugin, owner.Route
c.Set(pluginruntime.ContextKeyPinnedRoute, pinned)
c.Set(pluginruntime.ContextKeyPinnedPlugin, pluginruntime.PinnedPlugin{Generation: pinned.Generation, Plugin: owner.Plugin})
c.Set(pluginruntime.ContextKeyRouteRequest, requestContext)
generation := uint64(0)
if pinned.Generation != nil {
generation = pinned.Generation.Number
}
logger.LogDebug(
c,
"task_plugin subsystem=route event=resolved generation=%d plugin=%q kind=query renderer=%q task_count=1 candidates=%d distribute=false",
generation,
owner.Plugin.Meta.Key,
owner.Route.Render,
len(candidates),
)
renderTaskPluginQuery(c, pinned, requestContext, []string{taskIDs[ownerIndex]}, owner.Route.Render, false)
}
// taskPluginOwnsTasks reports whether every task id names a retrievable task
// of the caller on one of the plugin's platforms.
func taskPluginOwnsTasks(c *gin.Context, meta pluginruntime.Meta, taskIDs []string) (bool, error) {
tasks, err := model.GetByTaskIdsForPlatforms(common.GetContextKeyInt(c, constant.ContextKeyUserId), taskPluginLegacyPlatforms(meta), taskIDs)
if err != nil {
return false, err
}
for _, taskID := range taskIDs {
if !slices.ContainsFunc(tasks, func(task *model.Task) bool { return task.TaskID == taskID && task.ResultRetrievable() }) {
return false, nil
}
}
return true, nil
}
// RouteTaskPluginEndpoint selects between ordinary relay and plugin channels
// before a plugin can validate or rewrite the client's request. Plugin calls
// still prepare their candidates and origin-task pins before final distribution.
func RouteTaskPluginEndpoint() gin.HandlerFunc {
pinEndpoint := PinTaskPluginEndpoint()
return func(c *gin.Context) {
generation := pluginruntime.DefaultRegistry.Generation()
request, err := getModelFromRequest(c)
if err != nil || generation == nil {
pinEndpoint(c)
return
}
endpoint, found := lookupTaskPluginEndpoint(generation, c.Request.Method, c.Request.URL.Path, request.Model)
if !found {
c.Set(contextKeyTaskPluginEndpointModel, *request)
c.Next()
return
}
filter := dto.ChannelFilter{Kind: dto.FilterTaskPluginEndpoint}
for _, candidate := range endpoint.Candidates {
filter.TaskPluginKeys = append(filter.TaskPluginKeys, candidate.Plugin.Meta.Key)
filter.TaskPluginChannelTypes = append(filter.TaskPluginChannelTypes, candidate.Plugin.Meta.ChannelTypes...)
}
constraints := service.GetChannelConstraints(c)
filterIndex := len(constraints.Filters)
constraints.AddFilter(filter)
constraints.AddFilter(dto.ChannelFilter{Kind: dto.FilterRequestPath, RequestPath: c.Request.URL.Path})
group := common.GetContextKeyString(c, constant.ContextKeyUsingGroup)
channel, _, selectErr := service.SelectChannelForRequest(c, request.Model, &service.RetryParam{
Ctx: c, ModelName: request.Model, TokenGroup: group,
RequestPath: c.Request.URL.Path, Retry: common.GetPointer(0),
})
c.Set(contextKeyTaskPluginEndpointSelection, taskPluginEndpointSelection{
Generation: generation, Channel: channel, Model: request.Model,
})
if selectErr != nil {
// Keep the existing plugin diagnostics and canonical-model lookup
// when selection cannot resolve the client's original spelling.
pinEndpoint(c)
return
}
filter.Kind = dto.FilterTaskPluginIdentity
filter.TaskPluginKey = endpoint.Plugin.Meta.Key
if pluginChannel, _ := model.ChannelSatisfiesFilters(channel, request.Model, []dto.ChannelFilter{filter}); !pluginChannel {
constraints.Filters[filterIndex].TaskPluginRelayOnly = true
c.Set(pluginruntime.ContextKeyRelayEndpoint, endpoint)
c.Set(contextKeyTaskPluginEndpointModel, *request)
c.Next()
return
}
pinEndpoint(c)
}
}
func lookupTaskPluginEndpoint(generation *pluginruntime.RoutingGeneration, method, path, modelName string) (pluginruntime.PinnedEndpoint, bool) {
lookupModel, pinModel, mappedModel := modelName, modelName, ""
if declared, ok := generation.CanonicalModel(modelName); ok {
lookupModel, pinModel = declared, declared
} else if target, ok := model.ResolveTaskModelAlias(generation, modelName); ok {
if target.Declared == "" {
return pluginruntime.PinnedEndpoint{}, false
}
lookupModel, pinModel, mappedModel = target.Declared, target.Alias, target.Declared
}
candidates := generation.LookupEndpointCandidates(method, path, lookupModel)
if len(candidates) == 0 {
return pluginruntime.PinnedEndpoint{}, false
}
binding := candidates[0]
return pluginruntime.PinnedEndpoint{
Generation: generation, Plugin: binding.Plugin, Protocol: binding.Protocol,
Operation: binding.Operation, Model: pinModel, MappedModel: mappedModel, Candidates: candidates,
}, true
}
// PinTaskPluginEndpoint decides shared-endpoint ownership without executing
// plugin code. Invalid or unidentifiable ordinary requests deliberately fall
// through so the existing endpoint remains responsible for its validation.
func PinTaskPluginEndpoint() gin.HandlerFunc {
return func(c *gin.Context) {
generation := pluginruntime.DefaultRegistry.Generation()
if value, exists := c.Get(contextKeyTaskPluginEndpointSelection); exists {
if selected, ok := value.(taskPluginEndpointSelection); ok {
generation = selected.Generation
}
}
if generation == nil {
c.Next()
return
}
modelRequest, err := getModelFromRequest(c)
if err != nil {
if _, _, protocolPath := pluginruntime.LookupHostProtocolOperation(c.Request.Method, c.Request.URL.Path); protocolPath {
abortWithOpenAiMessage(c, invalidRequestStatus(err), i18n.T(c, i18n.MsgDistributorInvalidRequest, map[string]any{"Error": err.Error()}))
return
}
c.Next()
return
}
claimedModel := modelRequest.Model
if strings.TrimSpace(claimedModel) == "" {
c.Set(contextKeyTaskPluginEndpointModel, *modelRequest)
c.Next()
return
}
pinnedEndpoint, found := lookupTaskPluginEndpoint(generation, c.Request.Method, c.Request.URL.Path, claimedModel)
if !found {
c.Set(contextKeyTaskPluginEndpointModel, *modelRequest)
c.Next()
return
}
if claimedModel != pinnedEndpoint.Model {
if rewriteErr := rewriteTaskPluginJSONModel(c, pinnedEndpoint.Model); rewriteErr != nil {
abortWithOpenAiMessage(c, http.StatusBadRequest, "Invalid task protocol request")
return
}
}
modelRequest.Model = pinnedEndpoint.Model
c.Set(contextKeyTaskPluginEndpointModel, *modelRequest)
candidates := pinnedEndpoint.Candidates
binding := candidates[0]
if definition, known := pluginruntime.HostProtocol(binding.Protocol); known && len(definition.DefinedModes()) > 0 {
stream, background := jsonBodyBoolFlags(c)
required := make([]string, 0, 2)
if stream {
required = append(required, "stream")
}
if background {
required = append(required, "background")
}
if !stream && !background {
required = append(required, "sync")
}
unfiltered := candidates
filtered := make([]pluginruntime.ProtocolBinding, 0, len(candidates))
for _, candidate := range candidates {
if candidate.Plugin == nil {
continue
}
supported := true
for _, mode := range required {
if !candidate.Plugin.Meta.ProtocolSupports(candidate.Protocol, mode) {
supported = false
break
}
}
if supported {
filtered = append(filtered, candidate)
}
}
if len(filtered) == 0 {
abortWithOpenAiMessage(c, http.StatusBadRequest, unsupportedProtocolFormMessage(unfiltered, binding.Protocol, stream, background))
return
}
candidates = filtered
binding = candidates[0]
}
pin := pluginruntime.PinnedPlugin{Generation: generation, Plugin: binding.Plugin}
pinnedEndpoint.Plugin = binding.Plugin
pinnedEndpoint.Protocol = binding.Protocol
pinnedEndpoint.Operation = binding.Operation
pinnedEndpoint.Candidates = candidates
c.Set(pluginruntime.ContextKeyPinnedPlugin, pin)
c.Set(pluginruntime.ContextKeyPinnedEndpoint, pinnedEndpoint)
logger.LogDebug(
c,
"task_plugin subsystem=endpoint event=claimed generation=%d plugin=%q version=%q method=%q protocol=%q model=%q",
generation.Number,
binding.Plugin.Meta.Key,
binding.Plugin.Meta.Version,
binding.Operation.Methods[0],
binding.Protocol,
pinnedEndpoint.Model,
)
c.Next()
}
}
func jsonBodyBoolFlags(c *gin.Context) (stream, background bool) {
storage, err := common.GetBodyStorage(c)
if err != nil {
return false, false
}
requestBody, err := storage.Bytes()
if err != nil {
return false, false
}
values := ojson.GetManyBytes(requestBody, "stream", "background")
return values[0].Type == ojson.True, values[1].Type == ojson.True
}
func unsupportedProtocolFormMessage(candidates []pluginruntime.ProtocolBinding, protocol string, stream, background bool) string {
supports := func(mode string) bool {
for _, candidate := range candidates {
if candidate.Plugin != nil && candidate.Plugin.Meta.ProtocolSupports(protocol, mode) {
return true
}
}
return false
}
if stream && !supports("stream") {
if supports("background") {
return `Streaming is not supported for this model. Set "stream": false, or use "background": true and retrieve the response later.`
}
return `Streaming is not supported for this model. Set "stream": false.`
}
if background && !supports("background") {
return `Background mode is not supported for this model. Remove "background": true.`
}
forms := make([]string, 0, 2)
if supports("stream") {
forms = append(forms, `"stream": true`)
}
if supports("background") {
forms = append(forms, `"background": true`)
}
message := "Synchronous non-streaming requests are not supported for this model."
if len(forms) == 0 {
return message
}
return message + " Set " + strings.Join(forms, " or ") + "."
}
// TaskPluginEndpointOnly applies middleware only after a shared endpoint has
// been claimed. This preserves the original middleware chain for unclaimed
// video requests while enforcing the plugin route protections on claimed ones.
func TaskPluginEndpointOnly(handler gin.HandlerFunc) gin.HandlerFunc {
return func(c *gin.Context) {
if _, exists := c.Get(pluginruntime.ContextKeyPinnedEndpoint); !exists {
c.Next()
return
}
handler(c)
}
}
// PrepareTaskPluginEndpoint normalizes a claimed shared request through the
// candidates pinned before distribution, retaining only plugins that accept it.
func PrepareTaskPluginEndpoint() gin.HandlerFunc {
return func(c *gin.Context) {
pinnedValue, exists := c.Get(pluginruntime.ContextKeyPinnedEndpoint)
pinned, ok := pinnedValue.(pluginruntime.PinnedEndpoint)
if !exists {
c.Next()
return
}
if !ok || pinned.Generation == nil || pinned.Plugin == nil {
abortWithOpenAiMessage(c, http.StatusInternalServerError, "Task protocol request failed")
return
}
logger.LogDebug(
c,
"task_plugin subsystem=endpoint event=prepare_start generation=%d plugin=%q protocol=%q claimed_model=%q",
pinned.Generation.Number,
pinned.Plugin.Meta.Key,
pinned.Protocol,
pinned.Model,
)
if !pluginruntime.SupportsHostProtocol(pinned.Protocol) {
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=endpoint event=prepare_rejected generation=%d plugin=%q stage=protocol_check reason=unsupported_protocol",
pinned.Generation.Number,
pinned.Plugin.Meta.Key),
)
abortWithOpenAiMessage(c, http.StatusNotImplemented, "Task protocol bridge is not available")
return
}
requestContext, err := buildTaskPluginRouteRequest(c)
if err != nil {
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=endpoint event=prepare_rejected generation=%d plugin=%q stage=request_decode reason=invalid_request",
pinned.Generation.Number,
pinned.Plugin.Meta.Key),
)
status := http.StatusBadRequest
if errors.Is(err, errTaskPluginUnsupportedMediaType) {
status = http.StatusUnsupportedMediaType
}
abortWithOpenAiMessage(c, status, err.Error())
return
}
if body, ok := requestContext.Body.(map[string]any); ok {
if fields, ok := body["fields"].(map[string][]string); ok {
if values := fields["model"]; len(values) > 0 && values[0] != pinned.Model {
fields["model"][0] = pinned.Model
}
}
}
bodyObject, _ := requestContext.Body.(map[string]any)
bodyKind, _ := bodyObject["kind"].(string)
allowedBody := false
for _, allowed := range pinned.Operation.BodyKinds {
if bodyKind == string(allowed) {
allowedBody = true
break
}
}
if !allowedBody {
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=endpoint event=prepare_rejected generation=%d plugin=%q stage=request_decode reason=body_kind_mismatch body_kind=%q",
pinned.Generation.Number,
pinned.Plugin.Meta.Key,
bodyKind),
)
detail := "unsupported request body for this operation"
if len(pinned.Operation.BodyKinds) == 1 && pinned.Operation.BodyKinds[0] == pluginruntime.BodyJSON {
detail = "this route requires a JSON body"
}
abortWithOpenAiMessage(c, http.StatusUnsupportedMediaType, detail)
return
}
stream := false
if body, bodyOK := requestContext.Body.(map[string]any); bodyOK && body["kind"] == string(pluginruntime.BodyJSON) {
requestBody, _ := body["value"].(map[string]any)
if streamValue, present := requestBody["stream"]; present {
stream, ok = streamValue.(bool)
if !ok {
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=endpoint event=prepare_rejected generation=%d plugin=%q stage=request_decode reason=invalid_stream_flag",
pinned.Generation.Number,
pinned.Plugin.Meta.Key),
)
abortWithOpenAiMessage(c, http.StatusBadRequest, "stream must be a boolean")
return
}
}
}
protocolContext := pluginruntime.ProtocolRequestContext{
RouteRequestContext: requestContext,
Protocol: pinned.Protocol,
Operation: pinned.Operation.Name,
Model: pinned.Model,
UpstreamModel: pinned.MappedModel,
Stream: stream,
}
hookStarted := time.Now()
candidates := pinned.Candidates
if len(candidates) == 0 {
candidates = []pluginruntime.ProtocolBinding{{Plugin: pinned.Plugin, Protocol: pinned.Protocol, Operation: pinned.Operation, Model: pinned.Model}}
}
accepted := make([]pluginruntime.ProtocolBinding, 0, len(candidates))
var resolved map[string]any
var resolvedBodyText json.RawMessage
var failures []string
rejectedPlugins := make(map[string][]string)
for _, candidate := range candidates {
candidateContext := protocolContext
candidateContext.Protocol = candidate.Protocol
candidateContext.Operation = candidate.Operation.Name
// Parsing belongs to durable submission; disconnecting only stops
// the later Responses observation.
resolvedValue, requestBodyText, callErr := candidate.Plugin.Engine.CallPathWithMemberJSON(
context.WithoutCancel(c.Request.Context()), pluginruntime.DefaultCallTimeout,
candidate.Plugin.Meta.JSONTextMember("requestBody"),
"protocols", []string{candidate.Protocol, "decodeRequest"}, candidateContext.JSValueFor(candidate.Plugin.Meta),
)
result, resultOK := resolvedValue.(map[string]any)
detail := ""
reason := ""
if callErr != nil {
reason = "hook_failed"
detail = taskPluginHookDetail(callErr)
if detail == "" {
detail = "Invalid task protocol request"
}
} else if !resultOK {
reason = "result_not_object"
detail = taskPluginInvalidRouteResult
} else if kind, _ := result["kind"].(string); kind != string(pluginruntime.RouteTypeSubmit) {
reason = "unsupported_kind"
detail = taskPluginInvalidRouteResult
} else if model, _ := result["model"].(string); strings.TrimSpace(model) == "" {
reason = "invalid_model"
detail = "decoded request is missing a model"
} else if model != pinned.Model || (pinned.MappedModel == "" && !slices.Contains(candidate.Plugin.Meta.Models, model)) {
reason = "resolved_model_not_owned"
detail = fmt.Sprintf("model %q is not served by this plugin", model)
}
if detail != "" {
logger.LogWarn(c, common.LogText("task_plugin subsystem=endpoint event=prepare_rejected generation=%d plugin=%q stage=parse_request reason=%s err=%q elapsed_ms=%d",
pinned.Generation.Number, candidate.Plugin.Meta.Key, reason, detail, time.Since(hookStarted).Milliseconds()))
if _, seen := rejectedPlugins[detail]; !seen {
failures = append(failures, detail)
}
rejectedPlugins[detail] = append(rejectedPlugins[detail], candidate.Plugin.Meta.Key)
continue
}
accepted = append(accepted, candidate)
if resolved == nil {
resolved = result
resolvedBodyText = requestBodyText
protocolContext = candidateContext
}
}
if len(accepted) == 0 {
if len(candidates) > 1 {
for index, detail := range failures {
failures[index] = strings.Join(rejectedPlugins[detail], ", ") + ": " + detail
}
}
abortWithOpenAiMessage(c, http.StatusBadRequest, strings.Join(failures, "; "))
return
}
pinned.Candidates = accepted
pinned.Plugin = accepted[0].Plugin
pinned.Protocol = accepted[0].Protocol
pinned.Operation = accepted[0].Operation
c.Set(pluginruntime.ContextKeyPinnedEndpoint, pinned)
c.Set(pluginruntime.ContextKeyPinnedPlugin, pluginruntime.PinnedPlugin{Generation: pinned.Generation, Plugin: pinned.Plugin})
c.Set(pluginruntime.ContextKeyProtocolRequest, protocolContext)
resolvedModel := pinned.Model
action := ""
if resolvedAction, present := resolved["action"]; present {
action, ok = resolvedAction.(string)
if !ok {
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=endpoint event=prepare_rejected generation=%d plugin=%q stage=parse_request reason=invalid_action",
pinned.Generation.Number,
pinned.Plugin.Meta.Key),
)
abortWithOpenAiMessage(c, http.StatusBadRequest, taskPluginInvalidRouteResult)
return
}
}
_, bodyReplaced := resolved["requestBody"]
if normalizedBody, present := resolved["requestBody"]; present {
requestContext.RequestBody = normalizedBody
}
c.Set(pluginruntime.ContextKeyRouteRequest, requestContext)
c.Set("task_request", requestContext.RequestBody)
if resolvedBodyText != nil {
c.Set(pluginruntime.ContextKeyRequestBodyText, resolvedBodyText)
}
c.Set("resolved_task_model", resolvedModel)
c.Set("expected_task_plugin_key", pinned.Plugin.Meta.Key)
c.Set("task_plugin_key", pinned.Plugin.Meta.Key)
c.Set("platform", pinned.Plugin.Meta.Key)
service.AppendTaskPluginIdentityFilter(c, pinned.Plugin.Meta.Key)
c.Set("relay_mode", relayconstant.RelayModeVideoSubmit)
if strings.TrimSpace(action) != "" {
c.Set("task_action", action)
}
if intentErr := applyOriginTaskIntent(c, resolved, pinned.Plugin.Meta); intentErr != nil {
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=endpoint event=prepare_rejected generation=%d plugin=%q stage=origin_task reason=%s",
pinned.Generation.Number,
pinned.Plugin.Meta.Key,
intentErr.Code),
)
abortWithOpenAiMessage(c, intentErr.StatusCode, intentErr.Message, types.ErrorCode(intentErr.Code))
return
}
logger.LogDebug(
c,
"task_plugin subsystem=endpoint event=prepared generation=%d plugin=%q protocol=%q claimed_model=%q resolved_model=%q action_present=%t stream=%t request_body_replaced=%t elapsed_ms=%d",
pinned.Generation.Number,
pinned.Plugin.Meta.Key,
pinned.Protocol,
pinned.Model,
resolvedModel,
action != "",
stream,
bodyReplaced,
time.Since(hookStarted).Milliseconds(),
)
c.Next()
}
}
func buildTaskPluginRouteRequest(c *gin.Context) (pluginruntime.RouteRequestContext, error) {
requestContext := pluginruntime.RouteRequestContext{
Path: c.Request.URL.Path,
Method: c.Request.Method,
Params: make(map[string]string, len(c.Params)),
Query: make(map[string][]string),
Headers: taskjsplugin.ClientRequestHeaders(c.Request.Header),
Body: map[string]any{"kind": string(pluginruntime.BodyNone)},
}
for _, param := range c.Params {
requestContext.Params[param.Key] = param.Value
}
for key, values := range c.Request.URL.Query() {
requestContext.Query[key] = append([]string(nil), values...)
}
contentTypes := c.Request.Header.Values("Content-Type")
contentType := strings.TrimSpace(c.GetHeader("Content-Type"))
if len(contentTypes) > 1 {
canonical := ""
for _, value := range contentTypes {
mediaType, params, parseErr := mime.ParseMediaType(value)
if parseErr != nil {
return requestContext, parseErr
}
current := mime.FormatMediaType(strings.ToLower(mediaType), params)
if canonical != "" && current != canonical {
return requestContext, fmt.Errorf("conflicting Content-Type headers")
}
canonical = current
}
contentType = canonical
}
if contentType == "" || c.Request.ContentLength == 0 {
return requestContext, nil
}
mediaType, mediaParams, err := mime.ParseMediaType(contentType)
if err != nil {
return requestContext, err
}
switch {
case mediaType == "application/json" || strings.HasSuffix(mediaType, "+json"):
storage, storageErr := common.GetBodyStorage(c)
if storageErr != nil {
return requestContext, storageErr
}
raw, bytesErr := storage.Bytes()
if bytesErr != nil {
return requestContext, bytesErr
}
if !utf8.Valid(raw) {
return requestContext, fmt.Errorf("JSON body must be valid UTF-8")
}
var value any
if err = common.Unmarshal(raw, &value); err != nil {
return requestContext, err
}
requestContext.Body = map[string]any{"kind": string(pluginruntime.BodyJSON), "value": value}
// Body storage never changes its bytes in place (a rewrite stores a
// new copy), so the text needs no copy of its own.
requestContext.BodyText = raw
case mediaType == "application/x-www-form-urlencoded":
storage, storageErr := common.GetBodyStorage(c)
if storageErr != nil {
return requestContext, storageErr
}
raw, bytesErr := storage.Bytes()
if bytesErr != nil {
return requestContext, bytesErr
}
if !utf8.Valid(raw) {
return requestContext, fmt.Errorf("form body must be valid UTF-8")
}
values, parseErr := url.ParseQuery(string(raw))
if parseErr != nil {
return requestContext, parseErr
}
if err = validateTaskPluginFields(values); err != nil {
return requestContext, err
}
fields := make(map[string][]string, len(values))
for field, values := range values {
fields[field] = append([]string(nil), values...)
}
requestContext.Body = map[string]any{"kind": string(pluginruntime.BodyForm), "fields": fields}
case mediaType == "multipart/form-data":
boundary := mediaParams["boundary"]
if boundary == "" {
return requestContext, fmt.Errorf("multipart boundary is required")
}
storage, storageErr := common.GetBodyStorage(c)
if storageErr != nil {
return requestContext, storageErr
}
raw, bytesErr := storage.Bytes()
if bytesErr != nil {
return requestContext, bytesErr
}
reader := multipart.NewReader(bytes.NewReader(raw), boundary)
partCount := 0
fileCount := 0
fieldCount := 0
fileLimitMB := constant.MaxFileDownloadMB
if fileLimitMB <= 0 {
fileLimitMB = 64
}
for {
part, nextErr := reader.NextPart()
if nextErr == io.EOF {
break
}
if nextErr != nil {
return requestContext, nextErr
}
partCount++
if partCount > maxTaskPluginMultipartParts {
part.Close()
return requestContext, fmt.Errorf("multipart body exceeds %d parts", maxTaskPluginMultipartParts)
}
name := part.FormName()
if !utf8.ValidString(name) || len(name) == 0 || len(name) > maxTaskPluginFieldNameBytes {
part.Close()
return requestContext, fmt.Errorf("invalid multipart field name")
}
partMediaType, _, partMediaErr := mime.ParseMediaType(part.Header.Get("Content-Type"))
if partMediaErr != nil && part.Header.Get("Content-Type") != "" {
part.Close()
return requestContext, fmt.Errorf("invalid multipart part Content-Type")
}
if strings.HasPrefix(strings.ToLower(partMediaType), "multipart/") {
part.Close()
return requestContext, fmt.Errorf("nested multipart is not supported")
}
filename := part.FileName()
if filename == "" {
fieldCount++
if fieldCount > maxTaskPluginFormFields {
part.Close()
return requestContext, fmt.Errorf("request body exceeds %d fields", maxTaskPluginFormFields)
}
value, readErr := io.ReadAll(part)
part.Close()
if readErr != nil {
return requestContext, readErr
}
if !utf8.Valid(value) {
return requestContext, fmt.Errorf("request field %q must be valid UTF-8", name)
}
continue
}
fileCount++
if fileCount > maxTaskPluginFiles {
part.Close()
return requestContext, fmt.Errorf("multipart body exceeds %d files", maxTaskPluginFiles)
}
if !utf8.ValidString(filename) || len(filename) > maxTaskPluginFilenameBytes {
part.Close()
return requestContext, fmt.Errorf("invalid multipart filename")
}
written, copyErr := io.Copy(io.Discard, io.LimitReader(part, (int64(fileLimitMB)<<20)+1))
part.Close()
if copyErr != nil {
return requestContext, copyErr
}
if written > int64(fileLimitMB)<<20 {
return requestContext, fmt.Errorf("multipart file exceeds %d MB", fileLimitMB)
}
}
form, parseErr := common.ParseMultipartFormReusable(c)
if parseErr != nil {
return requestContext, parseErr
}
defer form.RemoveAll()
if err = validateTaskPluginFields(form.Value); err != nil {
return requestContext, err
}
partCount = 0
fileCount = 0
for _, values := range form.Value {
partCount += len(values)
}
for _, headers := range form.File {
partCount += len(headers)
fileCount += len(headers)
}
if partCount > maxTaskPluginMultipartParts {
return requestContext, fmt.Errorf("multipart body exceeds %d parts", maxTaskPluginMultipartParts)
}
if fileCount > maxTaskPluginFiles {
return requestContext, fmt.Errorf("multipart body exceeds %d files", maxTaskPluginFiles)
}
textFields := make(map[string][]string, len(form.Value))
for field, values := range form.Value {
textFields[field] = append([]string(nil), values...)
}
files := make([]map[string]any, 0)
for field, headers := range form.File {
if !utf8.ValidString(field) || len(field) > maxTaskPluginFieldNameBytes {
return requestContext, fmt.Errorf("invalid multipart file field name")
}
for index, header := range headers {
if !utf8.ValidString(header.Filename) || len(header.Filename) > maxTaskPluginFilenameBytes {
return requestContext, fmt.Errorf("invalid multipart filename")
}
partMediaType, _, mediaErr := mime.ParseMediaType(header.Header.Get("Content-Type"))
if mediaErr != nil && header.Header.Get("Content-Type") != "" {
return requestContext, fmt.Errorf("invalid multipart part Content-Type")
}
if strings.HasPrefix(strings.ToLower(partMediaType), "multipart/") {
return requestContext, fmt.Errorf("nested multipart is not supported")
}
if header.Size < 0 || header.Size > int64(fileLimitMB)<<20 {
return requestContext, fmt.Errorf("multipart file exceeds %d MB", fileLimitMB)
}
ref := pluginruntime.FileReference(field, index)
files = append(files, map[string]any{"ref": ref, "field": field, "filename": header.Filename, "mimeType": header.Header.Get("Content-Type"), "size": header.Size})
}
}
requestContext.Files = files
requestContext.Body = map[string]any{"kind": string(pluginruntime.BodyMultipart), "fields": textFields, "files": files}
default:
return requestContext, fmt.Errorf("%w %q", errTaskPluginUnsupportedMediaType, mediaType)
}
return requestContext, nil
}
func validateTaskPluginFields(fields url.Values) error {
fieldCount := 0
for name, values := range fields {
if !utf8.ValidString(name) || len(name) == 0 || len(name) > maxTaskPluginFieldNameBytes {
return fmt.Errorf("invalid request field name")
}
for _, value := range values {
fieldCount++
if fieldCount > maxTaskPluginFormFields {
return fmt.Errorf("request body exceeds %d fields", maxTaskPluginFormFields)
}
if !utf8.ValidString(value) {
return fmt.Errorf("request field %q must be valid UTF-8", name)
}
}
}
return nil
}
const (
maxOriginTaskIDs = 16
maxOriginTaskIDLen = 128
)
type originTaskIntentError struct {
Code string
Message string
StatusCode int
}
// taskPluginLegacyPlatforms lists the Task.Platform values a plugin owns: its
// key (plugin-era tasks) plus every numeric legacy channel type its driver
// can drive (pre-plugin tasks, e.g. sora tasks submitted on OpenAI-type
// channels stored Platform "1").
func taskPluginLegacyPlatforms(meta pluginruntime.Meta) []constant.TaskPlatform {
platforms := []constant.TaskPlatform{constant.TaskPlatform(meta.Key)}
// The sunoapi plugin (adapter for the Suno-API proxy project) was renamed
// from "suno"; historical rows carry that named Platform value, which
// predates the numeric channel-type convention below.
if meta.Key == "sunoapi" {
platforms = append(platforms, constant.TaskPlatformSuno)
}
for _, channelType := range meta.ChannelTypes {
if channelType <= 0 || channelType == constant.ChannelTypeTaskPlugin {
continue
}
platform := constant.TaskPlatform(strconv.Itoa(channelType))
if slices.Contains(platforms, platform) {
continue
}
platforms = append(platforms, platform)
}
return platforms
}
func applyOriginTaskIntent(c *gin.Context, intent map[string]any, meta pluginruntime.Meta) *originTaskIntentError {
raw, present := intent["originTaskIds"]
if !present {
return nil
}
var values []any
switch typed := raw.(type) {
case []any:
values = typed
case []string:
values = make([]any, len(typed))
for i, id := range typed {
values[i] = id
}
default:
return &originTaskIntentError{Code: "invalid_origin_task_ids", Message: "origin task ids are invalid", StatusCode: http.StatusBadRequest}
}
if len(values) > maxOriginTaskIDs {
return &originTaskIntentError{Code: "invalid_origin_task_ids", Message: "origin task ids are invalid", StatusCode: http.StatusBadRequest}
}
ids := make([]string, 0, len(values))
seen := make(map[string]struct{}, len(values))
for _, value := range values {
id, ok := value.(string)
if !ok {
return &originTaskIntentError{Code: "invalid_origin_task_ids", Message: "origin task ids are invalid", StatusCode: http.StatusBadRequest}
}
id = strings.TrimSpace(id)
if id == "" || utf8.RuneCountInString(id) > maxOriginTaskIDLen {
return &originTaskIntentError{Code: "invalid_origin_task_ids", Message: "origin task ids are invalid", StatusCode: http.StatusBadRequest}
}
if _, exists := seen[id]; exists {
continue
}
seen[id] = struct{}{}
ids = append(ids, id)
}
if len(ids) == 0 {
return nil
}
userID := common.GetContextKeyInt(c, constant.ContextKeyUserId)
platforms := taskPluginLegacyPlatforms(meta)
allowedPlatform := make(map[constant.TaskPlatform]struct{}, len(platforms))
for _, platform := range platforms {
allowedPlatform[platform] = struct{}{}
}
tasks := make([]*model.Task, 0, len(ids))
channelID := 0
for _, id := range ids {
task, exist, err := model.GetByTaskId(userID, id)
if err != nil {
return &originTaskIntentError{Code: "origin_task_not_found", Message: "origin task not found or not owned by you", StatusCode: http.StatusInternalServerError}
}
if !exist || task == nil {
return &originTaskIntentError{Code: "origin_task_not_found", Message: "origin task not found or not owned by you", StatusCode: http.StatusBadRequest}
}
if _, allowed := allowedPlatform[task.Platform]; !allowed {
return &originTaskIntentError{Code: "origin_task_platform_mismatch", Message: "origin task does not belong to this plugin", StatusCode: http.StatusBadRequest}
}
if channelID == 0 {
channelID = task.ChannelId
} else if task.ChannelId != channelID {
return &originTaskIntentError{Code: "origin_task_channel_conflict", Message: "origin tasks must belong to the same channel", StatusCode: http.StatusBadRequest}
}
tasks = append(tasks, task)
}
channel, err := model.CacheGetChannel(channelID)
if err != nil || channel == nil || channel.Status != common.ChannelStatusEnabled {
return &originTaskIntentError{Code: "origin_task_channel_disabled", Message: "origin task channel is disabled", StatusCode: http.StatusBadRequest}
}
service.GetChannelConstraints(c).AddPin(dto.ChannelPin{
ChannelId: channel.Id,
Source: dto.PinSourceOriginTask,
Rank: dto.PinRankOriginTask,
RetryMode: dto.PinRetrySameChannel,
})
common.SetContextKey(c, constant.ContextKeyOriginTasks, tasks)
return nil
}
func resolvedTaskPluginIDs(value any) ([]string, bool) {
values, ok := value.([]any)
if !ok || len(values) > 100 {
return nil, false
}
taskIDs := make([]string, len(values))
for index, value := range values {
taskID, stringOK := value.(string)
if !stringOK || strings.TrimSpace(taskID) == "" {
return nil, false
}
taskIDs[index] = taskID
}
return taskIDs, true
}
func renderTaskPluginQuery(
c *gin.Context,
pinned pluginruntime.PinnedRoute,
requestContext pluginruntime.RouteRequestContext,
taskIDs []string,
renderer string,
multiple bool,
) {
generation := uint64(0)
if pinned.Generation != nil {
generation = pinned.Generation.Number
}
logger.LogDebug(
c,
"task_plugin subsystem=query event=lookup_start generation=%d plugin=%q renderer=%q requested=%d multiple=%t",
generation,
pinned.Plugin.Meta.Key,
renderer,
len(taskIDs),
multiple,
)
userID := common.GetContextKeyInt(c, constant.ContextKeyUserId)
platforms := taskPluginLegacyPlatforms(pinned.Plugin.Meta)
tasks, err := model.GetByTaskIdsForPlatforms(userID, platforms, taskIDs)
if err != nil {
logger.LogDebug(
c,
"task_plugin subsystem=query event=lookup_failed generation=%d plugin=%q reason=database_error requested=%d",
generation,
pinned.Plugin.Meta.Key,
len(taskIDs),
)
abortTaskPluginRouteError(c, http.StatusInternalServerError)
return
}
logger.LogDebug(
c,
"task_plugin subsystem=query event=lookup_complete generation=%d plugin=%q requested=%d found=%d",
generation,
pinned.Plugin.Meta.Key,
len(taskIDs),
len(tasks),
)
tasksByID := make(map[string]*model.Task, len(tasks))
for _, task := range tasks {
tasksByID[task.TaskID] = task
}
views := make([]dto.TaskView, 0, len(taskIDs))
for _, taskID := range taskIDs {
task := tasksByID[taskID]
if task == nil || !task.ResultRetrievable() {
logger.LogDebug(
c,
"task_plugin subsystem=query event=lookup_failed generation=%d plugin=%q reason=task_not_found requested=%d found=%d",
generation,
pinned.Plugin.Meta.Key,
len(taskIDs),
len(tasks),
)
abortTaskPluginRouteError(c, http.StatusNotFound)
return
}
view, viewErr := service.BuildTaskPluginView(task)
if viewErr != nil {
abortTaskPluginRouteError(c, http.StatusInternalServerError)
return
}
views = append(views, view)
}
var rendererInput any = views
if !multiple {
rendererInput = views[0]
}
encoded, err := common.Marshal(rendererInput)
if err != nil {
abortTaskPluginRouteError(c, http.StatusInternalServerError)
return
}
// The presenter of one finished task gets its artifacts as the gateway
// addresses host protocol renderers get.
presenterContext := requestContext.JSValue()
artifacts := map[string]any{}
if !multiple {
projected, projectErr := taskjsplugin.RendererArtifacts(pinned.Plugin, tasksByID[taskIDs[0]], service.BuildTaskArtifactContentURL)
if projectErr != nil {
logger.LogWarn(c, common.LogText("task_plugin subsystem=query event=artifacts_unavailable generation=%d plugin=%q err=%q", generation, pinned.Plugin.Meta.Key, projectErr.Error()))
} else {
artifacts = projected
}
}
presenterContext["artifacts"] = artifacts
renderStarted := time.Now()
result, err := pinned.Plugin.Engine.CallPath(c.Request.Context(), "native", []string{renderer}, presenterContext, pluginruntime.RawJSON(encoded))
if err != nil {
logger.LogDebug(
c,
"task_plugin subsystem=query event=render_failed generation=%d plugin=%q renderer=%q reason=hook_failed elapsed_ms=%d",
generation,
pinned.Plugin.Meta.Key,
renderer,
time.Since(renderStarted).Milliseconds(),
)
// A presenter with nothing to deliver for the task, such as a file
// it has no artifact for, answers like a missing task.
status := http.StatusInternalServerError
if taskPluginHookDetail(err) == "artifact_not_found" {
status = http.StatusNotFound
}
abortTaskPluginRouteError(c, status)
return
}
logger.LogDebug(
c,
"task_plugin subsystem=query event=render_complete generation=%d plugin=%q renderer=%q task_count=%d elapsed_ms=%d",
generation,
pinned.Plugin.Meta.Key,
renderer,
len(views),
time.Since(renderStarted).Milliseconds(),
)
c.Abort()
if replyErr := WriteNativeReply(c, http.StatusOK, result); replyErr != nil {
logger.LogWarn(c, common.LogText("task_plugin subsystem=query event=render_failed generation=%d plugin=%q renderer=%q err=%q", generation, pinned.Plugin.Meta.Key, renderer, replyErr.Error()))
abortTaskPluginRouteError(c, http.StatusInternalServerError)
}
}
// nativeReplyHeaders are the headers a reply wrapper may set: a redirect
// target and a retry hint.
var nativeReplyHeaders = []string{"Location", "Retry-After"}
// WriteNativeReply writes what a native presenter returned. A value is the
// JSON body of a reply with the given status (200 for submit and query
// presenters, the host's status for the error presenter), unless it is an
// object with a __status member: such a reply wrapper, {__status, __headers?,
// __body?}, answers with the vendor's own status, as xAI answers a pending
// video with 202. The status is an integer from 200 to 599, __headers may set
// only Location and Retry-After, and a wrapper without __body sends no body.
// An invalid wrapper is an error and writes nothing.
func WriteNativeReply(c *gin.Context, status int, value any) error {
wrapper, ok := value.(map[string]any)
if !ok {
c.JSON(status, value)
return nil
}
rawStatus, isWrapper := wrapper["__status"]
if !isWrapper {
c.JSON(status, value)
return nil
}
status = 0
switch typed := rawStatus.(type) {
case int64:
status = int(typed)
case int:
status = typed
case float64:
if typed == float64(int(typed)) {
status = int(typed)
}
}
if status < 200 || status > 599 {
return fmt.Errorf("reply wrapper __status must be an integer from 200 to 599")
}
headers := map[string]string{}
for key, member := range wrapper {
switch key {
case "__status", "__body":
case "__headers":
values, ok := member.(map[string]any)
if !ok {
return fmt.Errorf("reply wrapper __headers must be an object")
}
for name, headerValue := range values {
canonical := http.CanonicalHeaderKey(name)
text, isString := headerValue.(string)
if !slices.Contains(nativeReplyHeaders, canonical) || !isString {
return fmt.Errorf("reply wrapper may set only string Location and Retry-After headers")
}
headers[canonical] = text
}
default:
return fmt.Errorf("reply wrapper has unknown member %q", key)
}
}
for name, headerValue := range headers {
c.Header(name, headerValue)
}
if body, hasBody := wrapper["__body"]; hasBody {
c.JSON(status, body)
return nil
}
c.Status(status)
c.Writer.WriteHeaderNow()
return nil
}
// RespondTaskPluginError gives a pinned plugin a sanitized error DTO and writes
// its native error body with the host's status, unless the body is a reply
// wrapper that answers with the vendor's own status.
func RespondTaskPluginError(c *gin.Context, taskErr *dto.TaskError) bool {
if taskErr == nil {
return false
}
pinnedValue, exists := c.Get(pluginruntime.ContextKeyPinnedRoute)
pinned, ok := pinnedValue.(pluginruntime.PinnedRoute)
if !exists || !ok || pinned.Plugin == nil {
return false
}
sanitized := sanitizedTaskPluginError(taskErr.StatusCode, taskErr.Message)
requestID := c.GetString(common.RequestIdKey)
hasRenderer, err := pinned.Plugin.Engine.HasCallablePath(c.Request.Context(), "native", "error")
requestValue, exists := c.Get(pluginruntime.ContextKeyRouteRequest)
requestContext, ok := requestValue.(pluginruntime.RouteRequestContext)
if err == nil && hasRenderer && exists && ok {
body, callErr := pinned.Plugin.Engine.CallMember(c.Request.Context(), "native", "error", requestContext.JSValue(), map[string]any{
"code": sanitized.Code,
"message": sanitized.Message,
"httpStatus": sanitized.HTTPStatus,
"retryable": sanitized.Retryable,
"requestId": requestID,
})
if callErr == nil {
callErr = WriteNativeReply(c, sanitized.HTTPStatus, body)
}
if callErr == nil {
logger.LogDebug(
c,
"task_plugin subsystem=route event=error_rendered plugin=%q renderer=plugin status=%d code=%q",
pinned.Plugin.Meta.Key,
sanitized.HTTPStatus,
sanitized.Code,
)
return true
}
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=route event=error_renderer_failed plugin=%q reason=hook_failed status=%d err=%q",
pinned.Plugin.Meta.Key,
sanitized.HTTPStatus,
callErr.Error()),
)
}
logger.LogWarn(
c,
common.LogText("task_plugin subsystem=route event=error_rendered plugin=%q renderer=host_fallback status=%d code=%q",
pinned.Plugin.Meta.Key,
sanitized.HTTPStatus,
sanitized.Code),
)
message := sanitized.Message
if requestID != "" {
message = common.MessageWithRequestId(sanitized.Message, requestID)
}
c.JSON(sanitized.HTTPStatus, &dto.TaskError{
Code: sanitized.Code,
Message: message,
StatusCode: sanitized.HTTPStatus,
})
return true
}
func abortTaskPluginRouteError(c *gin.Context, status int) {
abortTaskPluginRouteErrorDetail(c, status, "")
}
func abortTaskPluginRouteErrorDetail(c *gin.Context, status int, detail string) {
taskErr := sanitizedTaskPluginError(status, detail)
c.Abort()
if RespondTaskPluginError(c, &dto.TaskError{Code: taskErr.Code, Message: detail, StatusCode: taskErr.HTTPStatus}) {
return
}
message := taskErr.Message
if requestID := c.GetString(common.RequestIdKey); requestID != "" {
message = common.MessageWithRequestId(taskErr.Message, requestID)
}
c.JSON(taskErr.HTTPStatus, &dto.TaskError{
Code: taskErr.Code,
Message: message,
StatusCode: taskErr.HTTPStatus,
})
}
func sanitizedTaskPluginError(status int, detail string) dto.TaskPluginError {
var taskErr dto.TaskPluginError
switch status {
case http.StatusBadRequest:
taskErr = dto.TaskPluginError{Code: "invalid_request", Message: "Invalid request", HTTPStatus: status}
case http.StatusUnauthorized:
taskErr = dto.TaskPluginError{Code: "authentication_error", Message: "Authentication failed", HTTPStatus: status}
case http.StatusForbidden:
taskErr = dto.TaskPluginError{Code: "permission_denied", Message: "Access denied", HTTPStatus: status}
case http.StatusNotFound:
taskErr = dto.TaskPluginError{Code: "task_not_found", Message: "Task not found", HTTPStatus: status}
case http.StatusConflict:
taskErr = dto.TaskPluginError{Code: "request_conflict", Message: "Request conflict", HTTPStatus: status}
case http.StatusTooManyRequests:
taskErr = dto.TaskPluginError{Code: "rate_limit_exceeded", Message: "Too many requests", HTTPStatus: status, Retryable: true}
default:
if status < 400 || status > 599 {
status = http.StatusInternalServerError
}
if status < 500 {
taskErr = dto.TaskPluginError{Code: "invalid_request", Message: "Invalid request", HTTPStatus: status}
} else {
taskErr = dto.TaskPluginError{Code: "server_error", Message: "Task request failed", HTTPStatus: status, Retryable: status >= 500}
}
}
if detail != "" && taskErr.HTTPStatus < 500 {
taskErr.Message = detail
}
return taskErr
}
func taskPluginHookDetail(err error) string {
var hookErr *pluginruntime.HookError
if errors.As(err, &hookErr) {
return hookErr.Message
}
return ""
}
func logTaskPluginChannelDecision(c *gin.Context, channel *model.Channel, modelName, event, reason string) {
expectedPlugin := c.GetString("expected_task_plugin_key")
if expectedPlugin == "" {
return
}
generation := uint64(0)
if pinnedValue, exists := c.Get(pluginruntime.ContextKeyPinnedPlugin); exists {
if pinned, ok := pinnedValue.(pluginruntime.PinnedPlugin); ok && pinned.Generation != nil {
generation = pinned.Generation.Number
}
}
if channel == nil {
logger.LogDebug(
c,
"task_plugin subsystem=distribution event=%s generation=%d plugin=%q model=%q reason=%q",
event,
generation,
expectedPlugin,
modelName,
reason,
)
return
}
identityMode := "legacy_channel_type"
switch channel.Type {
case constant.ChannelTypeTaskPlugin:
identityMode = "type59_setting"
case constant.ChannelTypeNewAPI:
identityMode = "type60_setting"
}
logger.LogDebug(
c,
"task_plugin subsystem=distribution event=%s generation=%d plugin=%q model=%q channel_id=%d channel_type=%d identity_mode=%q reason=%q",
event,
generation,
expectedPlugin,
modelName,
channel.Id,
channel.Type,
identityMode,
reason,
)
}
func PrepareTaskPluginSubmit() gin.HandlerFunc {
return func(c *gin.Context) {
pluginKey := strings.TrimSpace(c.Param("key"))
generation := pluginruntime.DefaultRegistry.Generation()
plugin, ok := generation.Get(pluginKey)
if !ok {
c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": gin.H{"message": "task plugin not found", "type": "invalid_request_error"}})
return
}
c.Set(pluginruntime.ContextKeyPinnedPlugin, pluginruntime.PinnedPlugin{
Generation: generation,
Plugin: plugin,
})
logger.LogDebug(
c,
"task_plugin subsystem=route event=legacy_entry_pinned generation=%d plugin=%q version=%q",
generation.Number,
plugin.Meta.Key,
plugin.Meta.Version,
)
var requestBody map[string]any
if err := common.UnmarshalBodyReusable(c, &requestBody); err != nil {
c.AbortWithStatusJSON(invalidRequestStatus(err), gin.H{"error": gin.H{"message": err.Error(), "type": "invalid_request_error"}})
return
}
modelName, _ := requestBody["model"].(string)
if strings.TrimSpace(modelName) == "" {
c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": gin.H{"message": "model is required", "type": "invalid_request_error"}})
return
}
exactOwned := slices.Contains(plugin.Meta.Models, modelName)
exactAlias := false
if target, resolved := model.ResolveTaskModelAlias(generation, modelName); resolved && target.Alias == modelName && target.PluginKey == plugin.Meta.Key {
exactAlias = true
}
if !exactOwned && !exactAlias {
folded := ""
if declared, ok := generation.CanonicalModel(modelName); ok && slices.Contains(plugin.Meta.Models, declared) && declared != modelName {
folded = declared
} else if target, resolved := model.ResolveTaskModelAlias(generation, modelName); resolved && target.PluginKey == plugin.Meta.Key && target.Alias != "" && target.Alias != modelName {
folded = target.Alias
}
if folded != "" {
if rewriteErr := rewriteTaskPluginJSONModel(c, folded); rewriteErr != nil {
c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": gin.H{"message": rewriteErr.Error(), "type": "invalid_request_error"}})
return
}
requestBody["model"] = folded
modelName = folded
}
}
c.Set("task_request", requestBody)
if plugin.Meta.PreservesJSONOrder() {
// The body as stored after any model rewrite: the text hooks get.
storage, err := common.GetBodyStorage(c)
var raw []byte
if err == nil {
raw, err = storage.Bytes()
}
if err != nil {
c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": gin.H{"message": err.Error(), "type": "invalid_request_error"}})
return
}
c.Set(pluginruntime.ContextKeyRequestBodyText, json.RawMessage(raw))
}
c.Set("resolved_task_model", modelName)
c.Set("expected_task_plugin_key", pluginKey)
service.AppendTaskPluginIdentityFilter(c, pluginKey)
c.Set("relay_mode", relayconstant.RelayModeVideoSubmit)
logger.LogDebug(
c,
"task_plugin subsystem=route event=resolved generation=%d plugin=%q kind=submit model=%q distribute=true entry=legacy",
generation.Number,
plugin.Meta.Key,
modelName,
)
c.Next()
}
}