internal/pluginhost/adapters_executors.go
Rama main · réplica de lectura del repositorio original.
package pluginhost
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"sort"
"strconv"
"strings"
"sync"
"github.com/MoeForge/moego/ojson"
"github.com/router-for-me/CLIProxyAPI/v8/internal/registry"
"github.com/router-for-me/CLIProxyAPI/v8/internal/runtime/executor/helps"
"github.com/router-for-me/CLIProxyAPI/v8/internal/thinking"
coreauth "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/auth"
coreexecutor "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/executor"
"github.com/router-for-me/CLIProxyAPI/v8/sdk/pluginapi"
sdktranslator "github.com/router-for-me/CLIProxyAPI/v8/sdk/translator"
)
type executorManager interface {
Executor(provider string) (coreauth.ProviderExecutor, bool)
RegisterExecutor(coreauth.ProviderExecutor)
UnregisterExecutor(provider string)
}
type executorRegistration struct {
provider string
adapter *executorAdapter
}
func (h *Host) RegisterExecutors(manager executorManager, modelRegistry modelProviderRegistry) {
if h == nil || manager == nil {
return
}
snap := h.Snapshot()
records := h.activeRecordsFromSnapshot(snap)
registrations := h.snapshotModelRegistrations()
selectedModels := make(map[string][]*registry.ModelInfo)
providerModels := make(map[string][]*registry.ModelInfo)
claimedModels := make(map[string]struct{})
claimedProviders := make(map[string]string)
for _, registration := range registrations {
if registration.hasExecutor {
continue
}
appendModelsForProvider(providerModels, registration.provider, registration.models)
// Reserve native catalog models before plugin arbitration, independently of
// credential publication or exclusions. This preserves native priority without
// publishing availability, which remains credential-bound.
if h.providerHasNativeExecutor(manager, registration.provider) {
for _, model := range registration.models {
if model != nil {
if modelID := strings.TrimSpace(model.ID); modelID != "" {
claimedModels[modelID] = struct{}{}
}
}
}
}
}
for _, record := range records {
executor := record.plugin.Capabilities.Executor
if executor == nil || h.isPluginFused(record.id) {
continue
}
provider, okProvider := h.executorProvider(record, executor)
if !okProvider {
continue
}
registration := h.modelRegistration(record.id)
if h.providerHasNativeExecutor(manager, provider) {
appendModelsForProvider(providerModels, provider, registration.models)
continue
}
if len(registration.models) == 0 {
continue
}
if owner := claimedProviders[provider]; owner != "" && owner != record.id {
continue
}
for _, model := range registration.models {
modelID := strings.TrimSpace(model.ID)
if modelID == "" {
continue
}
if _, claimed := claimedModels[modelID]; claimed {
continue
}
if h.modelHasNativeExecutor(manager, modelRegistry, modelID) {
continue
}
claimedModels[modelID] = struct{}{}
claimedProviders[provider] = record.id
selectedModels[record.id] = append(selectedModels[record.id], model)
}
}
seenProviders := make(map[string]struct{})
nextProviders := make(map[string]struct{})
nextModelClients := make(map[string]struct{})
executorRegistrations := make([]executorRegistration, 0)
modelClientRegistrations := make([]modelClientRegistration, 0)
for _, record := range records {
executor := record.plugin.Capabilities.Executor
if executor == nil || h.isPluginFused(record.id) {
continue
}
provider, okProvider := h.executorProvider(record, executor)
if !okProvider {
continue
}
registration := h.modelRegistration(record.id)
if len(registration.models) > 0 && len(selectedModels[record.id]) == 0 {
continue
}
if _, seenProvider := seenProviders[provider]; seenProvider {
continue
}
seenProviders[provider] = struct{}{}
if h.providerHasNativeExecutor(manager, provider) {
continue
}
nextProviders[provider] = struct{}{}
executorRegistrations = append(executorRegistrations, newExecutorAdapterRegistration(h, record, provider, executor))
appendModelsForProvider(providerModels, provider, selectedModels[record.id])
if len(selectedModels[record.id]) > 0 {
clientID := pluginExecutorModelClientID(record.id, provider)
modelClientRegistrations = append(modelClientRegistrations, modelClientRegistration{
clientID: clientID,
provider: provider,
models: selectedModels[record.id],
})
nextModelClients[clientID] = struct{}{}
}
if executorScopeAllowsOAuthModels(record.plugin.Capabilities) && hasRecordAuthProvider(record) {
if authIDs, okAuthIDs := h.callRecordAuthProviderIdentifiers(record); okAuthIDs {
for _, authID := range authIDs {
if authID == "" || authID == provider {
continue
}
if _, seenAuthProvider := seenProviders[authID]; seenAuthProvider {
continue
}
seenProviders[authID] = struct{}{}
if h.providerHasNativeExecutor(manager, authID) {
continue
}
nextProviders[authID] = struct{}{}
executorRegistrations = append(executorRegistrations, newExecutorAdapterRegistration(h, record, authID, executor))
}
}
}
}
h.commitExecutorState(snap, manager, modelRegistry, providerModels, executorRegistrations, nextProviders, modelClientRegistrations, nextModelClients)
}
func pluginExecutorModelClientID(pluginID, provider string) string {
return "plugin:" + pluginID + ":" + provider + ":executor"
}
func (h *Host) commitExecutorState(snap *Snapshot, manager executorManager, modelRegistry modelRegistry, providerModels map[string][]*registry.ModelInfo, registrations []executorRegistration, nextProviders map[string]struct{}, modelClientRegistrations []modelClientRegistration, nextModelClients map[string]struct{}) {
if h == nil || manager == nil {
return
}
h.mu.Lock()
if h.Snapshot() != snap {
h.mu.Unlock()
return
}
h.providerModels = make(map[string][]*registryModelInfo, len(providerModels))
for provider, models := range providerModels {
h.providerModels[provider] = cloneRegistryModels(models)
}
staleProviders := make([]string, 0)
for provider := range h.executorProviders {
if _, okProvider := nextProviders[provider]; !okProvider {
staleProviders = append(staleProviders, provider)
}
}
h.executorProviders = nextProviders
if nextModelClients == nil {
nextModelClients = make(map[string]struct{})
}
staleModelClients := make([]string, 0)
for clientID := range h.executorModelClientIDs {
if _, okClient := nextModelClients[clientID]; !okClient {
staleModelClients = append(staleModelClients, clientID)
}
}
h.executorModelClientIDs = nextModelClients
h.mu.Unlock()
for _, registration := range registrations {
if registration.adapter == nil || registration.provider == "" {
continue
}
manager.RegisterExecutor(registration.adapter)
}
for _, provider := range staleProviders {
existing, okExecutor := manager.Executor(provider)
if !okExecutor || !h.ownsExecutor(existing) {
continue
}
manager.UnregisterExecutor(provider)
}
if modelRegistry == nil {
return
}
for _, registration := range modelClientRegistrations {
modelRegistry.RegisterClient(registration.clientID, registration.provider, registration.models)
}
for _, clientID := range staleModelClients {
modelRegistry.UnregisterClient(clientID)
}
}
func newExecutorAdapterRegistration(h *Host, record capabilityRecord, provider string, executor pluginapi.ProviderExecutor) executorRegistration {
return executorRegistration{
provider: provider,
adapter: &executorAdapter{
host: h,
pluginID: record.id,
path: record.path,
version: record.version,
provider: provider,
executor: executor,
inputFormats: normalizeExecutorFormats(record.plugin.Capabilities.ExecutorInputFormats),
outputFormats: normalizeExecutorFormats(record.plugin.Capabilities.ExecutorOutputFormats),
},
}
}
func (h *Host) snapshotModelRegistrations() []pluginModelRegistration {
if h == nil {
return nil
}
h.mu.Lock()
defer h.mu.Unlock()
return h.snapshotModelRegistrationsLocked()
}
func (h *Host) snapshotModelRegistrationsLocked() []pluginModelRegistration {
if h == nil {
return nil
}
pluginIDs := make([]string, 0, len(h.modelRegistrations)+len(h.pluginModelRegistrations))
seenPlugins := make(map[string]struct{}, len(h.modelRegistrations)+len(h.pluginModelRegistrations))
for pluginID := range h.modelRegistrations {
seenPlugins[pluginID] = struct{}{}
pluginIDs = append(pluginIDs, pluginID)
}
for pluginID := range h.pluginModelRegistrations {
if _, exists := seenPlugins[pluginID]; !exists {
seenPlugins[pluginID] = struct{}{}
pluginIDs = append(pluginIDs, pluginID)
}
}
sort.Strings(pluginIDs)
registrations := make([]pluginModelRegistration, 0, len(pluginIDs))
for _, pluginID := range pluginIDs {
registrations = append(registrations, h.pluginRegistrationsLocked(pluginID)...)
}
sort.SliceStable(registrations, func(i, j int) bool {
if registrations[i].priority == registrations[j].priority {
return registrations[i].pluginID < registrations[j].pluginID
}
return registrations[i].priority > registrations[j].priority
})
return registrations
}
func (h *Host) pluginRegistrationsLocked(pluginID string) []pluginModelRegistration {
regs := h.pluginModelRegistrations[pluginID]
if len(regs) == 1 {
if single, ok := h.modelRegistrations[pluginID]; ok && (single.pluginID != "" || single.provider != "" || len(single.models) > 0) {
single.models = cloneRegistryModels(single.models)
return []pluginModelRegistration{single}
}
}
if len(regs) > 0 {
out := make([]pluginModelRegistration, 0, len(regs))
for _, reg := range regs {
reg.models = cloneRegistryModels(reg.models)
out = append(out, reg)
}
return out
}
if single, ok := h.modelRegistrations[pluginID]; ok && (single.pluginID != "" || single.provider != "" || len(single.models) > 0) {
single.models = cloneRegistryModels(single.models)
return []pluginModelRegistration{single}
}
return nil
}
func (h *Host) modelRegistration(pluginID string) pluginModelRegistration {
if h == nil {
return pluginModelRegistration{}
}
h.mu.Lock()
defer h.mu.Unlock()
registration := h.modelRegistrations[pluginID]
registration.models = cloneRegistryModels(registration.models)
return registration
}
func (h *Host) executorProvider(record capabilityRecord, executor pluginapi.ProviderExecutor) (string, bool) {
if h == nil || !h.recordCurrent(record) {
return "", false
}
provider := h.modelProvider(record.id)
if provider == "" {
identifier, okIdentifier := h.callExecutorIdentifier(record.id, executor)
if !okIdentifier {
return "", false
}
provider = identifier
}
provider = strings.ToLower(strings.TrimSpace(provider))
return provider, provider != ""
}
func (h *Host) callExecutorIdentifier(pluginID string, executor pluginapi.ProviderExecutor) (provider string, ok bool) {
if h == nil || executor == nil || h.isPluginFused(pluginID) {
return "", false
}
defer func() {
if recovered := recover(); recovered != nil {
h.fusePlugin(pluginID, "Executor.Identifier", recovered)
provider = ""
ok = false
}
}()
return executor.Identifier(), true
}
func (h *Host) providerHasNativeExecutor(manager executorManager, provider string) bool {
if h == nil || manager == nil {
return false
}
existing, okExecutor := manager.Executor(provider)
return okExecutor && existing != nil && !h.ownsExecutor(existing)
}
func (h *Host) modelHasNativeExecutor(manager executorManager, modelRegistry modelProviderRegistry, modelID string) bool {
if h == nil || manager == nil || modelRegistry == nil {
return false
}
for _, provider := range modelRegistry.GetModelProviders(modelID) {
if h.providerHasNativeExecutor(manager, provider) {
return true
}
}
return false
}
func appendModelsForProvider(out map[string][]*registry.ModelInfo, provider string, models []*registry.ModelInfo) {
provider = strings.ToLower(strings.TrimSpace(provider))
if provider == "" || len(models) == 0 {
return
}
seen := make(map[string]struct{}, len(out[provider])+len(models))
for _, model := range out[provider] {
if model != nil && strings.TrimSpace(model.ID) != "" {
seen[strings.TrimSpace(model.ID)] = struct{}{}
}
}
for _, model := range models {
if model == nil {
continue
}
modelID := strings.TrimSpace(model.ID)
if modelID == "" {
continue
}
if _, exists := seen[modelID]; exists {
continue
}
seen[modelID] = struct{}{}
out[provider] = append(out[provider], cloneRegistryModels([]*registry.ModelInfo{model})...)
}
}
func (h *Host) ModelsForProvider(provider string) []*registry.ModelInfo {
if h == nil {
return nil
}
provider = strings.ToLower(strings.TrimSpace(provider))
if provider == "" {
return nil
}
h.mu.Lock()
defer h.mu.Unlock()
return cloneRegistryModels(h.providerModels[provider])
}
func (h *Host) HasExecutorCandidateProvider(provider string) bool {
if h == nil {
return false
}
provider = strings.ToLower(strings.TrimSpace(provider))
if provider == "" {
return false
}
for _, record := range h.activeRecords() {
executor := record.plugin.Capabilities.Executor
if executor == nil || h.isPluginFused(record.id) {
continue
}
candidate, okCandidate := h.executorProvider(record, executor)
if okCandidate && candidate == provider {
return true
}
if executorScopeAllowsOAuthModels(record.plugin.Capabilities) && hasRecordAuthProvider(record) {
if authIDs, okAuthIDs := h.callRecordAuthProviderIdentifiers(record); okAuthIDs {
for _, authID := range authIDs {
if authID == provider {
return true
}
}
}
}
}
return false
}
// OwnsExecutor reports whether executor is an adapter managed by this host.
func (h *Host) OwnsExecutor(executor coreauth.ProviderExecutor) bool {
return h.ownsExecutor(executor)
}
func (h *Host) ownsExecutor(executor coreauth.ProviderExecutor) bool {
adapter, okAdapter := executor.(*executorAdapter)
return okAdapter && adapter != nil && adapter.host == h
}
func (h *Host) modelProvider(pluginID string) string {
if h == nil {
return ""
}
h.mu.Lock()
defer h.mu.Unlock()
return h.modelProviders[pluginID]
}
func (h *Host) modelProvidersForPlugin(pluginID string) []string {
if h == nil {
return nil
}
h.mu.Lock()
defer h.mu.Unlock()
list := h.modelProviderSets[pluginID]
primary := strings.TrimSpace(h.modelProviders[pluginID])
if len(list) == 1 && primary != "" {
return []string{primary}
}
out := make([]string, 0, len(list)+1)
seen := make(map[string]struct{}, len(list)+1)
if primary != "" {
seen[primary] = struct{}{}
out = append(out, primary)
}
for _, item := range list {
item = strings.TrimSpace(item)
if item == "" {
continue
}
if _, exists := seen[item]; exists {
continue
}
seen[item] = struct{}{}
out = append(out, item)
}
return out
}
type executorAdapter struct {
host *Host
pluginID string
path string
version string
provider string
executor pluginapi.ProviderExecutor
inputFormats []sdktranslator.Format
outputFormats []sdktranslator.Format
}
func (a *executorAdapter) Identifier() string {
if a == nil {
return ""
}
return a.provider
}
type preparedExecutorCall struct {
req coreexecutor.Request
opts coreexecutor.Options
inputRequested sdktranslator.Format
requestedFormat sdktranslator.Format
inputFormat sdktranslator.Format
outputFormat sdktranslator.Format
}
func (a *executorAdapter) prepareExecutorCall(req coreexecutor.Request, opts coreexecutor.Options) (preparedExecutorCall, error) {
inputRequested := executorInputFormat(req, opts)
requestedFormat := executorRequestedFormat(req, opts)
inputFormat, errInput := a.selectExecutorInputFormat(inputRequested)
if errInput != nil {
return preparedExecutorCall{}, errInput
}
outputFormat, errOutput := a.selectExecutorOutputFormat(requestedFormat, inputFormat)
if errOutput != nil {
return preparedExecutorCall{}, errOutput
}
baseModel := strings.TrimSpace(req.Model)
modelInfo := a.lookupPluginModelInfo(baseModel)
if parsed := thinking.ParseSuffix(req.Model); parsed.HasSuffix && modelInfo == nil {
baseModel = strings.TrimSpace(parsed.ModelName)
modelInfo = a.lookupPluginModelInfo(baseModel)
}
if baseModel == "" {
baseModel = req.Model
}
nativeReq := req
nativeOpts := opts
var updatesChanged bool
if inputRequested != "" && inputRequested != inputFormat {
translated := sdktranslator.TranslateRequestEnvelope(context.Background(), inputRequested, inputFormat, sdktranslator.RequestEnvelope{
Format: inputRequested,
Model: baseModel,
Stream: opts.Stream,
Body: req.Payload,
})
if translated.Err != nil {
return preparedExecutorCall{}, translated.Err
}
nativeReq.Payload = translated.Body
updatesChanged = translated.ConfigurationUpdatesChanged
}
hasPluginThinkingTarget := modelInfo != nil ||
registry.LookupModelInfo(baseModel, a.provider) != nil ||
thinking.GetProviderApplier(a.provider) != nil
originalTranslated := nativeReq.Payload
doc := ojson.NewDoc(nativeReq.Payload)
if hasPluginThinkingTarget && baseModel != "" && doc.Get("model").Exists() {
_ = helps.SetStringIfDifferentDoc(doc, "model", baseModel)
}
if doc.Len() > 0 {
if errThinking := helps.ApplyRequestThinkingWithModelInfoDoc(
doc,
req,
opts,
inputRequested.String(),
inputFormat.String(),
a.provider,
modelInfo,
updatesChanged,
); errThinking != nil {
return preparedExecutorCall{}, errThinking
}
}
// User payload rules are the last host-side mutation before the plugin sees
// the request, matching on the suffix-free model like the built-in executors
// and using the canonical client format as source protocol.
if a.host != nil {
if cfg := a.host.currentRuntimeConfig(); cfg != nil {
finalizeOpts := opts
if inputRequested != "" {
finalizeOpts.SourceFormat = inputRequested
}
if len(req.Metadata) > 0 {
finalizeOpts.Metadata = mergeExecutorMetadata(req.Metadata, opts.Metadata)
}
root := ""
if inputFormat == sdktranslator.FormatAntigravity {
root = "request"
}
finalize := helps.NewPayloadFinalizer(cfg, a.provider, baseModel, inputFormat.String(), root, originalTranslated, req, finalizeOpts)
helps.FinalizePayloadDoc(helps.WithPayloadFinalizer(context.Background(), finalize), doc)
}
}
if doc.Len() > 0 {
nativeReq.Payload = doc.Bytes()
}
if hasPluginThinkingTarget {
nativeReq.Model = baseModel
}
nativeReq.Format = outputFormat
nativeOpts.SourceFormat = inputFormat
nativeOpts.ResponseFormat = outputFormat
return preparedExecutorCall{
req: nativeReq,
opts: nativeOpts,
inputRequested: inputRequested,
requestedFormat: requestedFormat,
inputFormat: inputFormat,
outputFormat: outputFormat,
}, nil
}
func (a *executorAdapter) lookupPluginModelInfo(baseModel string) *registry.ModelInfo {
baseModel = strings.TrimSpace(baseModel)
if a == nil || a.host == nil || baseModel == "" {
return nil
}
for _, model := range a.host.ModelsForProvider(a.provider) {
if model != nil && strings.EqualFold(strings.TrimSpace(model.ID), baseModel) {
return model
}
}
for _, model := range a.host.modelRegistration(a.pluginID).models {
if model != nil && strings.EqualFold(strings.TrimSpace(model.ID), baseModel) {
return model
}
}
return nil
}
func (a *executorAdapter) RequestToFormat(req coreexecutor.Request, opts coreexecutor.Options) sdktranslator.Format {
if a == nil {
return ""
}
inputRequested := executorInputFormat(req, opts)
inputFormat, errInput := a.selectExecutorInputFormat(inputRequested)
if errInput != nil {
return ""
}
return inputFormat
}
func executorInputFormat(req coreexecutor.Request, opts coreexecutor.Options) sdktranslator.Format {
if opts.SourceFormat != "" {
return normalizeExecutorFormatName(opts.SourceFormat.String())
}
if req.Format != "" {
return normalizeExecutorFormatName(req.Format.String())
}
return sdktranslator.FormatOpenAI
}
func executorRequestedFormat(req coreexecutor.Request, opts coreexecutor.Options) sdktranslator.Format {
if format := coreexecutor.ResponseFormatOrSource(opts); format != "" {
return normalizeExecutorFormatName(format.String())
}
if req.Format != "" {
return normalizeExecutorFormatName(req.Format.String())
}
return sdktranslator.FormatOpenAI
}
func (a *executorAdapter) selectExecutorInputFormat(requested sdktranslator.Format) (sdktranslator.Format, error) {
if len(a.inputFormats) == 0 {
return "", fmt.Errorf("plugin executor %s declares no input formats", a.Identifier())
}
if executorFormatContains(a.inputFormats, requested) {
return requested, nil
}
for _, format := range a.inputFormats {
if requested == "" || sdktranslator.HasRequestTransformer(requested, format) {
return format, nil
}
}
return "", fmt.Errorf("plugin executor %s does not support input format %q", a.Identifier(), requested)
}
func (a *executorAdapter) selectExecutorOutputFormat(requested, inputFormat sdktranslator.Format) (sdktranslator.Format, error) {
if len(a.outputFormats) == 0 {
return "", fmt.Errorf("plugin executor %s declares no output formats", a.Identifier())
}
if executorFormatContains(a.outputFormats, requested) {
return requested, nil
}
if executorFormatContains(a.outputFormats, inputFormat) && a.executorResponseTranslationAvailable(inputFormat, requested) {
return inputFormat, nil
}
for _, format := range a.outputFormats {
if requested == "" || a.executorResponseTranslationAvailable(format, requested) {
return format, nil
}
}
return "", fmt.Errorf("plugin executor %s does not support output format %q", a.Identifier(), requested)
}
func (a *executorAdapter) executorResponseTranslationAvailable(from, to sdktranslator.Format) bool {
if from == "" || to == "" || from == to {
return true
}
if sdktranslator.HasResponseTransformer(to, from) {
return true
}
return a != nil && a.host.hasResponseTranslator()
}
func (h *Host) hasResponseTranslator() bool {
for _, record := range h.activeRecords() {
if h.isPluginFused(record.id) || record.plugin.Capabilities.ResponseTranslator == nil {
continue
}
return true
}
return false
}
// hasResponseTranslationHooks reports whether any active plugin can rewrite a
// streaming response through the translator registry hooks (before/after
// normalizers or a response translator). When none exists, a same-format
// stream without a registered native translator is forwarded unchanged instead
// of entering TranslateStream only to fall back to the original payload.
func (h *Host) hasResponseTranslationHooks() bool {
if h == nil {
return false
}
for _, record := range h.activeRecords() {
if h.isPluginFused(record.id) {
continue
}
caps := record.plugin.Capabilities
if caps.ResponseTranslator != nil || caps.ResponseBeforeTranslator != nil || caps.ResponseAfterTranslator != nil {
return true
}
}
return false
}
func executorNativeStreamResponseTranslatorExists(from, to sdktranslator.Format) bool {
if from == "" || to == "" || from == to {
return true
}
return sdktranslator.HasStreamResponseTransformer(to, from)
}
func (a *executorAdapter) translateExecutorResponse(ctx context.Context, prepared preparedExecutorCall, payload []byte, stream bool, param *any) []byte {
if prepared.requestedFormat == "" || prepared.outputFormat == prepared.requestedFormat {
out := bytes.Clone(payload)
if prepared.requestedFormat == sdktranslator.FormatOpenAIResponse {
out = helps.EnsureResponsesUsageDetails(out)
}
return out
}
originalRequest := prepared.opts.OriginalRequest
if len(originalRequest) == 0 {
originalRequest = prepared.req.Payload
}
if stream {
frames := a.translateExecutorStreamPayload(ctx, prepared, payload, param)
if len(frames) == 0 {
return nil
}
if len(frames) == 1 {
return bytes.Clone(frames[0])
}
return bytes.Join(frames, nil)
}
out := sdktranslator.TranslateNonStream(ctx, prepared.outputFormat, prepared.requestedFormat, prepared.req.Model, originalRequest, prepared.req.Payload, payload, param)
if prepared.requestedFormat == sdktranslator.FormatOpenAIResponse {
out = helps.EnsureResponsesUsageDetails(out)
}
return out
}
func (a *executorAdapter) translateExecutorStreamChunks(ctx context.Context, prepared preparedExecutorCall, in <-chan pluginapi.ExecutorStreamChunk) <-chan pluginapi.ExecutorStreamChunk {
// Same-format streams still go through stream translation when a registered
// same-format translator exists: it normalizes SSE framing (for example
// stripping the "data:" prefix and dropping "[DONE]" for chat-completions),
// matching what the built-in executors produce. Returning the channel
// unchanged for those formats let framed plugin chunks reach handlers that
// add their own "data:" prefix, emitting a doubled "data: data:" prefix.
// That case first passes through an SSE deframer so input may be split or
// coalesced at arbitrary byte boundaries; see deframeExecutorSSEChunks.
//
// Other same-format streams must not enter TranslateStream unconditionally:
// its plugin hooks (response before/after normalizers and, without a native
// translator, the plugin response translator) run even when no native
// translator exists, so the payload would no longer be forwarded verbatim.
// When no such hook capability is installed, those streams keep the
// previous untouched behavior; when one is installed, or for every
// cross-format stream, translation runs as before.
if prepared.requestedFormat == "" {
return in
}
if in == nil {
return nil
}
sameFormat := prepared.outputFormat == prepared.requestedFormat
// Responses always enters translation so its usage-details normalization
// in translateExecutorStreamPayload keeps running for same-format streams.
if sameFormat && prepared.requestedFormat != sdktranslator.FormatOpenAIResponse && !sdktranslator.HasStreamResponseTransformer(prepared.outputFormat, prepared.requestedFormat) && !a.host.hasResponseTranslationHooks() {
return in
}
if ctx == nil {
ctx = context.Background()
}
if sameFormat && sdktranslator.HasStreamResponseTransformer(prepared.outputFormat, prepared.requestedFormat) {
in = deframeExecutorSSEChunks(ctx, in)
}
out := make(chan pluginapi.ExecutorStreamChunk)
go func() {
defer close(out)
var param any
for {
select {
case <-ctx.Done():
return
case chunk, ok := <-in:
if !ok {
a.emitTranslatedExecutorStreamTail(ctx, prepared, out, ¶m)
return
}
if chunk.Err != nil {
_ = sendExecutorPluginStreamChunk(ctx, out, chunk)
continue
}
frames := a.translateExecutorStreamPayload(ctx, prepared, chunk.Payload, ¶m)
for _, frame := range frames {
if !sendExecutorPluginStreamChunk(ctx, out, pluginapi.ExecutorStreamChunk{Payload: frame}) {
return
}
}
}
}
}()
return out
}
func (a *executorAdapter) translateExecutorStreamPayload(ctx context.Context, prepared preparedExecutorCall, payload []byte, param *any) [][]byte {
originalRequest := prepared.opts.OriginalRequest
if len(originalRequest) == 0 {
originalRequest = prepared.req.Payload
}
frames := sdktranslator.TranslateStream(ctx, prepared.outputFormat, prepared.requestedFormat, prepared.req.Model, originalRequest, prepared.req.Payload, payload, param)
if executorStreamTranslationFellBack(prepared, payload, frames) {
return nil
}
if prepared.requestedFormat == sdktranslator.FormatOpenAIResponse {
for i, frame := range frames {
frames[i] = helps.EnsureResponsesUsageDetails(frame)
}
}
return frames
}
func executorStreamTranslationFellBack(prepared preparedExecutorCall, payload []byte, frames [][]byte) bool {
if prepared.requestedFormat == "" || prepared.outputFormat == "" || prepared.outputFormat == prepared.requestedFormat {
return false
}
if len(frames) != 1 || !bytes.Equal(frames[0], payload) {
return false
}
// A plugin executor only reaches this path after host-side response translation
// has been selected. An unchanged single frame is the SDK registry fallback,
// not a valid translated frame to send to the client.
return executorNativeStreamResponseTranslatorExists(prepared.outputFormat, prepared.requestedFormat)
}
func (a *executorAdapter) emitTranslatedExecutorStreamTail(ctx context.Context, prepared preparedExecutorCall, out chan<- pluginapi.ExecutorStreamChunk, param *any) {
tail := executorStreamDonePayload(prepared.outputFormat)
if len(tail) == 0 {
return
}
frames := a.translateExecutorStreamPayload(ctx, prepared, tail, param)
for _, frame := range frames {
if !sendExecutorPluginStreamChunk(ctx, out, pluginapi.ExecutorStreamChunk{Payload: frame}) {
return
}
}
}
func executorStreamDonePayload(format sdktranslator.Format) []byte {
switch format {
case sdktranslator.FormatOpenAI:
return []byte("data: [DONE]")
default:
return nil
}
}
func sendExecutorPluginStreamChunk(ctx context.Context, out chan<- pluginapi.ExecutorStreamChunk, chunk pluginapi.ExecutorStreamChunk) bool {
select {
case out <- pluginapi.ExecutorStreamChunk{Payload: bytes.Clone(chunk.Payload), Err: chunk.Err}:
return true
case <-ctx.Done():
return false
}
}
func (a *executorAdapter) Execute(ctx context.Context, auth *coreauth.Auth, req coreexecutor.Request, opts coreexecutor.Options) (resp coreexecutor.Response, err error) {
if a == nil || a.executor == nil || a.host.isPluginFused(a.pluginID) || !a.host.pluginIdentityCurrent(a.pluginID, a.path, a.version) {
return coreexecutor.Response{}, fmt.Errorf("plugin executor %s is unavailable", a.Identifier())
}
var reporter *helps.UsageReporter
if auth != nil {
modelName := strings.TrimSpace(thinking.ParseSuffix(req.Model).ModelName)
if modelName == "" {
modelName = req.Model
}
reporter = helps.NewExecutorUsageReporter(ctx, a, modelName, auth)
}
defer func() {
if recovered := recover(); recovered != nil {
a.host.fusePlugin(a.pluginID, "Executor.Execute", recovered)
resp = coreexecutor.Response{}
err = fmt.Errorf("plugin executor %s panic: %v", a.Identifier(), recovered)
if reporter != nil {
reporter.PublishFailure(ctx, err)
}
return
}
if err != nil && reporter != nil {
reporter.PublishFailure(ctx, err)
}
}()
prepared, errPrepare := a.prepareExecutorCall(req, opts)
if errPrepare != nil {
return coreexecutor.Response{}, errPrepare
}
if reporter != nil {
reporter.SetTranslatedReasoningEffort(prepared.req.Payload, prepared.inputFormat.String())
reporter.StartResponseTTFT()
}
pluginResp, errExecute := a.executor.Execute(ctx, buildExecutorRequest(a.host, a.provider, auth, prepared.req, prepared.opts))
if errExecute != nil {
return coreexecutor.Response{}, errExecute
}
if reporter != nil {
reporter.RecordFirstPacket()
detail := helps.ParsePluginExecutorResponseUsage(prepared.outputFormat.String(), pluginResp.Payload)
reporter.Publish(ctx, detail)
reporter.EnsurePublished(ctx)
}
return coreexecutor.Response{
Payload: a.translateExecutorResponse(ctx, prepared, pluginResp.Payload, false, nil),
Metadata: cloneAnyMap(pluginResp.Metadata),
Headers: cloneHeader(pluginResp.Headers),
}, nil
}
func (a *executorAdapter) ExecuteStream(ctx context.Context, auth *coreauth.Auth, req coreexecutor.Request, opts coreexecutor.Options) (result *coreexecutor.StreamResult, err error) {
if a == nil || a.executor == nil || a.host.isPluginFused(a.pluginID) || !a.host.pluginIdentityCurrent(a.pluginID, a.path, a.version) {
return nil, fmt.Errorf("plugin executor %s is unavailable", a.Identifier())
}
var reporter *helps.UsageReporter
if auth != nil {
modelName := strings.TrimSpace(thinking.ParseSuffix(req.Model).ModelName)
if modelName == "" {
modelName = req.Model
}
reporter = helps.NewExecutorUsageReporter(ctx, a, modelName, auth)
}
defer func() {
if recovered := recover(); recovered != nil {
a.host.fusePlugin(a.pluginID, "Executor.ExecuteStream", recovered)
result = nil
err = fmt.Errorf("plugin executor %s stream panic: %v", a.Identifier(), recovered)
if reporter != nil {
reporter.PublishFailure(ctx, err)
}
return
}
if err != nil && reporter != nil {
reporter.PublishFailure(ctx, err)
}
}()
prepared, errPrepare := a.prepareExecutorCall(req, opts)
if errPrepare != nil {
return nil, errPrepare
}
if reporter != nil {
reporter.SetTranslatedReasoningEffort(prepared.req.Payload, prepared.inputFormat.String())
reporter.StartResponseTTFT()
}
pluginResp, errExecuteStream := a.executor.ExecuteStream(ctx, buildExecutorRequest(a.host, a.provider, auth, prepared.req, prepared.opts))
if errExecuteStream != nil {
return nil, errExecuteStream
}
chunks := a.observeAndTranslateExecutorStream(ctx, prepared, pluginResp.Chunks, reporter)
return &coreexecutor.StreamResult{
Headers: cloneHeader(pluginResp.Headers),
Chunks: chunks,
}, nil
}
func (a *executorAdapter) observeAndTranslateExecutorStream(ctx context.Context, prepared preparedExecutorCall, in <-chan pluginapi.ExecutorStreamChunk, reporter *helps.UsageReporter) <-chan coreexecutor.StreamChunk {
if ctx == nil {
ctx = context.Background()
}
if in == nil {
out := make(chan coreexecutor.StreamChunk)
close(out)
if reporter != nil {
reporter.EnsurePublished(ctx)
}
return out
}
if reporter == nil {
return mapExecutorStreamChunks(ctx, a.translateExecutorStreamChunks(ctx, prepared, in))
}
observedIn := make(chan pluginapi.ExecutorStreamChunk)
var streamUsage helps.StreamUsageBuffer
var streamErr error
var publishOnce sync.Once
var lineBuffer []byte
publishResult := func() {
publishOnce.Do(func() {
if len(lineBuffer) > 0 {
helps.ObservePluginExecutorStreamUsage(prepared.outputFormat.String(), lineBuffer, &streamUsage)
lineBuffer = nil
}
if streamErr != nil {
if !streamUsage.PublishFailure(ctx, reporter, streamErr) {
reporter.PublishFailure(ctx, streamErr)
}
reporter.EnsurePublished(ctx)
return
}
if ctx.Err() != nil {
if !streamUsage.PublishFailure(ctx, reporter, ctx.Err()) {
reporter.PublishFailure(ctx, ctx.Err())
}
reporter.EnsurePublished(ctx)
return
}
if !streamUsage.Publish(ctx, reporter) {
reporter.EnsurePublished(ctx)
}
})
}
go func() {
defer close(observedIn)
defer publishResult()
for {
select {
case <-ctx.Done():
return
case chunk, ok := <-in:
if !ok {
return
}
if chunk.Err != nil {
if streamErr == nil {
streamErr = chunk.Err
}
publishResult()
select {
case observedIn <- chunk:
case <-ctx.Done():
return
}
continue
}
if len(chunk.Payload) > 0 {
helps.ObservePluginExecutorStreamTTFT(prepared.outputFormat.String(), reporter, chunk.Payload)
lineBuffer = append(lineBuffer, chunk.Payload...)
for {
idx := bytes.IndexByte(lineBuffer, '\n')
if idx < 0 {
break
}
line := lineBuffer[:idx+1]
lineBuffer = lineBuffer[idx+1:]
helps.ObservePluginExecutorStreamUsage(prepared.outputFormat.String(), line, &streamUsage)
}
if len(lineBuffer) > 0 {
if jsonBytes := helps.ExtractStreamJSONPayload(lineBuffer); len(jsonBytes) > 0 && json.Valid(jsonBytes) {
helps.ObservePluginExecutorStreamUsage(prepared.outputFormat.String(), lineBuffer, &streamUsage)
lineBuffer = nil
}
}
}
select {
case observedIn <- chunk:
case <-ctx.Done():
return
}
}
}
}()
translatedOut := a.translateExecutorStreamChunks(ctx, prepared, observedIn)
return mapExecutorStreamChunks(ctx, translatedOut)
}
func (a *executorAdapter) Refresh(ctx context.Context, auth *coreauth.Auth) (refreshed *coreauth.Auth, err error) {
if a == nil || a.executor == nil || a.host.isPluginFused(a.pluginID) || !a.host.pluginIdentityCurrent(a.pluginID, a.path, a.version) {
return nil, fmt.Errorf("plugin executor %s is unavailable", a.Identifier())
}
record, authProv := a.host.authProviderCredentialEntryForAuth(auth)
if record == nil || authProv == nil {
return auth.Clone(), nil
}
defer func() {
if recovered := recover(); recovered != nil {
a.host.fusePlugin(record.id, "AuthProvider.RefreshAuth", recovered)
refreshed = nil
err = fmt.Errorf("plugin executor %s refresh panic: %v", a.Identifier(), recovered)
}
}()
pluginResp, errRefresh := authProv.RefreshAuth(ctx, pluginapi.AuthRefreshRequest{
AuthID: authID(auth),
AuthProvider: authProvider(auth),
StorageJSON: storageJSONFromAuth(auth),
Metadata: cloneAnyMap(authMetadata(auth)),
Attributes: authAttributes(auth),
Host: a.host.hostConfigSummary(),
HTTPClient: a.host.newHTTPClient(auth),
})
if errRefresh != nil {
return nil, errRefresh
}
data := pluginResp.Auth
explicitPortalHint := authDataPortalHint(&data)
if strings.TrimSpace(data.Provider) == "" {
data.Provider = authProvider(auth)
}
if strings.TrimSpace(data.ID) == "" {
data.ID = authID(auth)
}
if strings.TrimSpace(data.FileName) == "" && auth != nil {
data.FileName = auth.FileName
}
if strings.TrimSpace(data.Label) == "" && auth != nil {
data.Label = auth.Label
}
if strings.TrimSpace(data.Prefix) == "" && auth != nil {
data.Prefix = auth.Prefix
}
if strings.TrimSpace(data.ProxyURL) == "" && auth != nil {
data.ProxyURL = auth.ProxyURL
}
if len(data.Metadata) == 0 && auth != nil {
data.Metadata = cloneAnyMap(auth.Metadata)
}
if len(data.Attributes) == 0 {
if auth != nil {
data.Attributes = cloneStringMap(auth.Attributes)
}
} else if auth != nil {
attributes := cloneStringMap(data.Attributes)
for key, value := range auth.Attributes {
if _, exists := attributes[key]; !exists {
attributes[key] = value
}
}
data.Attributes = attributes
}
preserveAuthOAuthProviderMetadata(&data, auth, explicitPortalHint)
preserveFileAuthPriority(&data, auth)
preserveAuthDisabledState(&data, auth)
if len(data.StorageJSON) == 0 {
data.StorageJSON = storageJSONFromAuth(auth)
}
if pluginResp.NextRefreshAfter.IsZero() && auth != nil {
data.NextRefreshAfter = auth.NextRefreshAfter
}
if !pluginResp.NextRefreshAfter.IsZero() {
data.NextRefreshAfter = pluginResp.NextRefreshAfter
}
path := ""
if auth != nil && auth.Attributes != nil {
path = auth.Attributes[coreauth.AttributePath]
}
next := a.host.AuthDataToCoreAuth(data, path, data.FileName)
if next == nil {
return nil, fmt.Errorf("plugin executor %s refresh returned invalid auth data", a.Identifier())
}
if auth != nil {
next.CreatedAt = auth.CreatedAt
next.UpdatedAt = auth.UpdatedAt
}
return next, nil
}
func (a *executorAdapter) CountTokens(ctx context.Context, auth *coreauth.Auth, req coreexecutor.Request, opts coreexecutor.Options) (resp coreexecutor.Response, err error) {
if a == nil || a.executor == nil || a.host.isPluginFused(a.pluginID) || !a.host.pluginIdentityCurrent(a.pluginID, a.path, a.version) {
return coreexecutor.Response{}, fmt.Errorf("plugin executor %s is unavailable", a.Identifier())
}
defer func() {
if recovered := recover(); recovered != nil {
a.host.fusePlugin(a.pluginID, "Executor.CountTokens", recovered)
resp = coreexecutor.Response{}
err = fmt.Errorf("plugin executor %s count tokens panic: %v", a.Identifier(), recovered)
}
}()
prepared, errPrepare := a.prepareExecutorCall(req, opts)
if errPrepare != nil {
return coreexecutor.Response{}, errPrepare
}
pluginResp, errCountTokens := a.executor.CountTokens(ctx, buildExecutorRequest(a.host, a.provider, auth, prepared.req, prepared.opts))
if errCountTokens != nil {
return coreexecutor.Response{}, errCountTokens
}
var payload []byte
translatedTokenCount := false
if prepared.outputFormat != prepared.requestedFormat {
// A token-count reply is not a model response; translating it as one
// wraps the count in a full completion message.
if count, okCount := pluginExecutorTokenCount(prepared.outputFormat, pluginResp.Payload); okCount {
if sdktranslator.HasTokenCountTransformer(prepared.outputFormat, prepared.requestedFormat) {
translated := sdktranslator.TranslateTokenCount(ctx, prepared.outputFormat, prepared.requestedFormat, count, pluginResp.Payload)
payload = translated
translatedTokenCount = true
}
}
}
if !translatedTokenCount {
payload = a.translateExecutorResponse(ctx, prepared, pluginResp.Payload, false, nil)
}
return coreexecutor.Response{
Payload: payload,
Metadata: cloneAnyMap(pluginResp.Metadata),
Headers: cloneHeader(pluginResp.Headers),
}, nil
}
func pluginExecutorTokenCount(format sdktranslator.Format, payload []byte) (int64, bool) {
for _, path := range []string{"choices", "content", "candidates", "output", "response.output"} {
if ojson.GetBytesNoCopy(payload, path).Exists() {
return 0, false
}
}
// response.usage.input_tokens is the native Codex count shape.
paths := []string{"usage.prompt_tokens", "input_tokens", "response.usage.input_tokens", "totalTokens", "total_tokens"}
if format == sdktranslator.FormatGemini {
paths = []string{"totalTokens", "usageMetadata.promptTokenCount", "input_tokens", "total_tokens"}
}
for _, path := range paths {
value := ojson.GetBytesNoCopy(payload, path)
if value.Exists() && value.Type == ojson.Number {
count, errParse := strconv.ParseInt(value.Raw, 10, 64)
if errParse == nil && count >= 0 {
return count, true
}
}
}
return 0, false
}
func (a *executorAdapter) HttpRequest(ctx context.Context, auth *coreauth.Auth, req *http.Request) (resp *http.Response, err error) {
if a == nil || a.executor == nil || a.host.isPluginFused(a.pluginID) || !a.host.pluginIdentityCurrent(a.pluginID, a.path, a.version) {
return nil, fmt.Errorf("plugin executor %s is unavailable", a.Identifier())
}
if req == nil {
return nil, fmt.Errorf("plugin executor %s received nil HTTP request", a.Identifier())
}
defer func() {
if recovered := recover(); recovered != nil {
a.host.fusePlugin(a.pluginID, "Executor.HttpRequest", recovered)
resp = nil
err = fmt.Errorf("plugin executor %s http request panic: %v", a.Identifier(), recovered)
}
}()
body, errReadAll := readAndRestoreRequestBody(req)
if errReadAll != nil {
return nil, fmt.Errorf("read plugin http request body: %w", errReadAll)
}
pluginResp, errHTTPRequest := a.executor.HttpRequest(ctx, pluginapi.ExecutorHTTPRequest{
AuthID: authID(auth),
AuthProvider: authProvider(auth),
Method: req.Method,
URL: req.URL.String(),
Headers: cloneHeader(req.Header),
Body: bytes.Clone(body),
StorageJSON: storageJSONFromAuth(auth),
Metadata: cloneAnyMap(authMetadata(auth)),
Attributes: authAttributes(auth),
HTTPClient: a.host.newHTTPClient(auth, a.provider),
})
if errHTTPRequest != nil {
return nil, errHTTPRequest
}
status := pluginResp.StatusCode
if status == 0 {
status = http.StatusOK
}
resp = &http.Response{
StatusCode: status,
Status: fmt.Sprintf("%d %s", status, http.StatusText(status)),
Header: cloneHeader(pluginResp.Headers),
Body: io.NopCloser(bytes.NewReader(bytes.Clone(pluginResp.Body))),
Request: req,
}
return resp, nil
}
func buildExecutorRequest(host *Host, provider string, auth *coreauth.Auth, req coreexecutor.Request, opts coreexecutor.Options) pluginapi.ExecutorRequest {
return pluginapi.ExecutorRequest{
AuthID: authID(auth),
AuthProvider: authProvider(auth),
Model: req.Model,
Format: req.Format.String(),
Stream: opts.Stream,
Alt: opts.Alt,
Headers: cloneHeader(opts.Headers),
Query: cloneValues(opts.Query),
OriginalRequest: bytes.Clone(opts.OriginalRequest),
SourceFormat: opts.SourceFormat.String(),
Payload: bytes.Clone(req.Payload),
Metadata: mergeExecutorMetadata(req.Metadata, opts.Metadata),
StorageJSON: storageJSONFromAuth(auth),
AuthMetadata: cloneAnyMap(authMetadata(auth)),
AuthAttributes: authAttributes(auth),
HTTPClient: host.newHTTPClientWithProxy(auth, opts.ProxyURL, provider),
}
}
func storageJSONFromAuth(auth *coreauth.Auth) []byte {
if auth == nil {
return nil
}
if rawProvider, okRaw := auth.Storage.(interface{ RawJSON() []byte }); okRaw {
return bytes.Clone(rawProvider.RawJSON())
}
if len(auth.Metadata) == 0 {
return nil
}
data, errMarshal := json.Marshal(auth.Metadata)
if errMarshal != nil {
return nil
}
return data
}
func authAttributes(auth *coreauth.Auth) map[string]string {
if auth == nil {
return nil
}
return cloneStringMap(auth.Attributes)
}
func mergeExecutorMetadata(reqMetadata, optsMetadata map[string]any) map[string]any {
if len(reqMetadata) == 0 && len(optsMetadata) == 0 {
return nil
}
merged := make(map[string]any, len(reqMetadata)+len(optsMetadata))
for key, value := range reqMetadata {
merged[key] = value
}
for key, value := range optsMetadata {
merged[key] = value
}
return merged
}
func mapExecutorStreamChunks(ctx context.Context, in <-chan pluginapi.ExecutorStreamChunk) <-chan coreexecutor.StreamChunk {
if ctx == nil {
ctx = context.Background()
}
out := make(chan coreexecutor.StreamChunk)
if in == nil {
close(out)
return out
}
go func() {
defer close(out)
for {
var mapped coreexecutor.StreamChunk
select {
case <-ctx.Done():
return
case chunk, ok := <-in:
if !ok {
return
}
mapped = coreexecutor.StreamChunk{
Payload: bytes.Clone(chunk.Payload),
Err: chunk.Err,
}
}
select {
case <-ctx.Done():
return
case out <- mapped:
}
}
}()
return out
}