Something went wrong. Try again.
sloproxy
Something went wrong. Try again.
13 kB · 392 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393package pluginhost
import ( "bytes" "context" "fmt" "runtime/debug" "strings"
"github.com/router-for-me/CLIProxyAPI/v7/internal/logging" "github.com/router-for-me/CLIProxyAPI/v7/internal/registry" "github.com/router-for-me/CLIProxyAPI/v7/internal/thinking" coreusage "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/usage" "github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginapi" sdktranslator "github.com/router-for-me/CLIProxyAPI/v7/sdk/translator" log "github.com/sirupsen/logrus")
func (h *Host) RegisterUsagePlugins() { if h == nil { return }
for _, record := range h.activeRecords() { plugin := record.plugin.Capabilities.UsagePlugin if plugin == nil || h.isPluginFused(record.id) { continue } coreusage.RegisterNamedPlugin("plugin:"+record.id, &usageAdapter{ host: h, pluginID: record.id, plugin: plugin, }) }}
func (h *Host) refreshThinkingProviders(records []capabilityRecord) { thinking.ClearPluginProviders() if h == nil { return } for _, record := range records { applier := record.plugin.Capabilities.ThinkingApplier if applier == nil || h.isPluginFused(record.id) { continue } provider, okProvider := h.callThinkingIdentifier(record, applier) if !okProvider { continue } thinking.RegisterPluginProvider(record.id, provider, record.priority, &thinkingAdapter{ host: h, pluginID: record.id, path: record.path, version: record.version, provider: provider, applier: applier, }) }}
func (h *Host) callThinkingIdentifier(record capabilityRecord, applier pluginapi.ThinkingApplier) (provider string, ok bool) { if h == nil || applier == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) { return "", false } defer func() { if recovered := recover(); recovered != nil { h.fusePlugin(record.id, "ThinkingApplier.Identifier", recovered) provider = "" ok = false } }() provider = strings.ToLower(strings.TrimSpace(applier.Identifier())) if provider == "" { return "", false } return provider, true}
func (h *Host) currentUsagePlugin(pluginID string) pluginapi.UsagePlugin { if h == nil || strings.TrimSpace(pluginID) == "" { return nil } for _, record := range h.activeRecords() { if record.id != pluginID { continue } if h.isPluginFused(record.id) { return nil } return record.plugin.Capabilities.UsagePlugin } return nil}
func (h *Host) fusePlugin(id, method string, recovered any) { if h == nil { return } h.mu.Lock() h.fused[id] = fmt.Sprintf("%s panic: %v", method, recovered) h.mu.Unlock() thinking.UnregisterPluginProviders(id) log.WithField("plugin_id", id).WithField("method", method).Errorf("pluginhost: plugin panic recovered: %v\n%s", recovered, debug.Stack())}
func (h *Host) isPluginFused(id string) bool { if h == nil { return false } h.mu.Lock() _, fused := h.fused[id] h.mu.Unlock() return fused}
type usageAdapter struct { host *Host pluginID string plugin pluginapi.UsagePlugin}
type thinkingAdapter struct { host *Host pluginID string path string version string provider string applier pluginapi.ThinkingApplier}
func (a *usageAdapter) HandleUsage(ctx context.Context, record coreusage.Record) { if a == nil { return } plugin := a.host.currentUsagePlugin(a.pluginID) if plugin == nil { return } if ctx == nil { ctx = context.Background() } else { ctx = context.WithoutCancel(ctx) } defer func() { if recovered := recover(); recovered != nil { a.host.fusePlugin(a.pluginID, "UsagePlugin.HandleUsage", recovered) } }() sessionID := strings.TrimSpace(record.SessionID) parentSessionID := strings.TrimSpace(record.ParentSessionID) if sessionID == "" { clientMeta := logging.GetClientRequestMetadata(ctx) sessionID = strings.TrimSpace(clientMeta.SessionID) parentSessionID = strings.TrimSpace(clientMeta.ParentSessionID) } else if parentSessionID == "" { clientMeta := logging.GetClientRequestMetadata(ctx) if sessionID == strings.TrimSpace(clientMeta.SessionID) { parentSessionID = strings.TrimSpace(clientMeta.ParentSessionID) } } if sessionID == "" || sessionID == parentSessionID { parentSessionID = "" } plugin.HandleUsage(ctx, pluginapi.UsageRecord{ Provider: record.Provider, ExecutorType: record.ExecutorType, Model: record.Model, Alias: record.Alias, APIKey: record.APIKey, SessionID: sessionID, ParentSessionID: parentSessionID, AuthID: record.AuthID, AuthIndex: record.AuthIndex, AuthType: record.AuthType, Source: record.Source, ReasoningEffort: record.ReasoningEffort, ServiceTier: record.ServiceTier, Generate: coreusage.GenerateEnabled(record.Generate), RequestedAt: record.RequestedAt, Latency: record.Latency, TTFT: record.TTFT, Failed: record.Failed, Failure: pluginapi.UsageFailure{ StatusCode: record.Fail.StatusCode, Body: record.Fail.Body, }, Detail: pluginapi.UsageDetail{ InputTokens: record.Detail.InputTokens, OutputTokens: record.Detail.OutputTokens, ReasoningTokens: record.Detail.ReasoningTokens, CachedTokens: record.Detail.CachedTokens, CacheReadTokens: record.Detail.CacheReadTokens, CacheCreationTokens: record.Detail.CacheCreationTokens, TotalTokens: record.Detail.TotalTokens, }, ResponseHeaders: cloneHeader(record.ResponseHeaders), })}
func (a *thinkingAdapter) Apply(body []byte, config thinking.ThinkingConfig, modelInfo *registry.ModelInfo) (out []byte, err error) { if a == nil || a.applier == nil || a.host == nil || a.host.isPluginFused(a.pluginID) || !a.host.pluginIdentityCurrent(a.pluginID, a.path, a.version) { return bytes.Clone(body), nil } defer func() { if recovered := recover(); recovered != nil { a.host.fusePlugin(a.pluginID, "ThinkingApplier.ApplyThinking", recovered) out = bytes.Clone(body) err = nil } }() resp, errApply := a.applier.ApplyThinking(context.Background(), pluginapi.ThinkingApplyRequest{ Provider: a.provider, Model: registryModelInfoToPluginModelInfo(modelInfo), Config: pluginapi.ThinkingConfig{ Mode: config.Mode.String(), Budget: config.Budget, Level: string(config.Level), }, Body: bytes.Clone(body), }) if errApply != nil || len(resp.Body) == 0 { return bytes.Clone(body), nil } return bytes.Clone(resp.Body), nil}
func (h *Host) NormalizeRequest(ctx context.Context, from, to sdktranslator.Format, model string, body []byte, stream bool) []byte { current := bytes.Clone(body) for _, record := range h.activeRecords() { if h.isPluginFused(record.id) || record.plugin.Capabilities.RequestNormalizer == nil { continue } if normalized, ok := h.callRequestNormalizer(ctx, record, from, to, model, current, stream); ok { current = normalized } } return current}
func (h *Host) TranslateRequest(ctx context.Context, from, to sdktranslator.Format, model string, body []byte, stream bool) ([]byte, bool) { for _, record := range h.activeRecords() { if h.isPluginFused(record.id) || record.plugin.Capabilities.RequestTranslator == nil { continue } if translated, ok := h.callRequestTranslator(ctx, record, from, to, model, body, stream); ok { return translated, true } } return bytes.Clone(body), false}
func (h *Host) NormalizeResponseBefore(ctx context.Context, from, to sdktranslator.Format, model string, originalRequestRawJSON, requestRawJSON, body []byte, stream bool) []byte { current := bytes.Clone(body) for _, record := range h.activeRecords() { normalizer := record.plugin.Capabilities.ResponseBeforeTranslator if h.isPluginFused(record.id) || normalizer == nil { continue } if normalized, ok := h.callResponseNormalizer(ctx, record, "ResponseBeforeTranslator.NormalizeResponse", normalizer, from, to, model, originalRequestRawJSON, requestRawJSON, current, stream); ok { current = normalized } } return current}
func (h *Host) TranslateResponse(ctx context.Context, from, to sdktranslator.Format, model string, originalRequestRawJSON, requestRawJSON, body []byte, stream bool) ([]byte, bool) { for _, record := range h.activeRecords() { translator := record.plugin.Capabilities.ResponseTranslator if h.isPluginFused(record.id) || translator == nil { continue } if translated, ok := h.callResponseTranslator(ctx, record, translator, from, to, model, originalRequestRawJSON, requestRawJSON, body, stream); ok { return translated, true } } return bytes.Clone(body), false}
func (h *Host) NormalizeResponseAfter(ctx context.Context, from, to sdktranslator.Format, model string, originalRequestRawJSON, requestRawJSON, body []byte, stream bool) []byte { current := bytes.Clone(body) for _, record := range h.activeRecords() { normalizer := record.plugin.Capabilities.ResponseAfterTranslator if h.isPluginFused(record.id) || normalizer == nil { continue } if normalized, ok := h.callResponseNormalizer(ctx, record, "ResponseAfterTranslator.NormalizeResponse", normalizer, from, to, model, originalRequestRawJSON, requestRawJSON, current, stream); ok { current = normalized } } return current}
func (h *Host) callRequestNormalizer(ctx context.Context, record capabilityRecord, from, to sdktranslator.Format, model string, body []byte, stream bool) (out []byte, ok bool) { if h == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) || record.plugin.Capabilities.RequestNormalizer == nil { return nil, false } defer func() { if recovered := recover(); recovered != nil { h.fusePlugin(record.id, "RequestNormalizer.NormalizeRequest", recovered) out = nil ok = false } }() resp, errNormalizeRequest := record.plugin.Capabilities.RequestNormalizer.NormalizeRequest(ctx, pluginapi.RequestTransformRequest{ FromFormat: from.String(), ToFormat: to.String(), Model: model, Stream: stream, Body: bytes.Clone(body), }) if errNormalizeRequest != nil || len(resp.Body) == 0 { return nil, false } return bytes.Clone(resp.Body), true}
func (h *Host) callRequestTranslator(ctx context.Context, record capabilityRecord, from, to sdktranslator.Format, model string, body []byte, stream bool) (out []byte, ok bool) { if h == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) || record.plugin.Capabilities.RequestTranslator == nil { return nil, false } defer func() { if recovered := recover(); recovered != nil { h.fusePlugin(record.id, "RequestTranslator.TranslateRequest", recovered) out = nil ok = false } }() resp, errTranslateRequest := record.plugin.Capabilities.RequestTranslator.TranslateRequest(ctx, pluginapi.RequestTransformRequest{ FromFormat: from.String(), ToFormat: to.String(), Model: model, Stream: stream, Body: bytes.Clone(body), }) if errTranslateRequest != nil || len(resp.Body) == 0 { return nil, false } return bytes.Clone(resp.Body), true}
func (h *Host) callResponseNormalizer(ctx context.Context, record capabilityRecord, method string, normalizer pluginapi.ResponseNormalizer, from, to sdktranslator.Format, model string, originalRequestRawJSON, requestRawJSON, body []byte, stream bool) (out []byte, ok bool) { if h == nil || normalizer == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) { return nil, false } defer func() { if recovered := recover(); recovered != nil { h.fusePlugin(record.id, method, recovered) out = nil ok = false } }() resp, errNormalizeResponse := normalizer.NormalizeResponse(ctx, pluginapi.ResponseTransformRequest{ FromFormat: from.String(), ToFormat: to.String(), Model: model, Stream: stream, OriginalRequest: bytes.Clone(originalRequestRawJSON), TranslatedRequest: bytes.Clone(requestRawJSON), Body: bytes.Clone(body), }) if errNormalizeResponse != nil || len(resp.Body) == 0 { return nil, false } return bytes.Clone(resp.Body), true}
func (h *Host) callResponseTranslator(ctx context.Context, record capabilityRecord, translator pluginapi.ResponseTranslator, from, to sdktranslator.Format, model string, originalRequestRawJSON, requestRawJSON, body []byte, stream bool) (out []byte, ok bool) { if h == nil || translator == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) { return nil, false } defer func() { if recovered := recover(); recovered != nil { h.fusePlugin(record.id, "ResponseTranslator.TranslateResponse", recovered) out = nil ok = false } }() resp, errTranslateResponse := translator.TranslateResponse(ctx, pluginapi.ResponseTransformRequest{ FromFormat: from.String(), ToFormat: to.String(), Model: model, Stream: stream, OriginalRequest: bytes.Clone(originalRequestRawJSON), TranslatedRequest: bytes.Clone(requestRawJSON), Body: bytes.Clone(body), }) if errTranslateResponse != nil || len(resp.Body) == 0 { return nil, false } return bytes.Clone(resp.Body), true}