package pluginhost import (	"bytes"	"context"	"errors"	"fmt"	"path/filepath"	"sort"	"strings"	"sync"	"sync/atomic" 	"github.com/router-for-me/CLIProxyAPI/v7/internal/config"	"github.com/router-for-me/CLIProxyAPI/v7/internal/interfaces"	"github.com/router-for-me/CLIProxyAPI/v7/sdk/api/handlers"	coreauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth"	"github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginabi"	"github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginapi"	log "github.com/sirupsen/logrus") type loadedPlugin struct {	id         string	path       string	version    string	name       string	configYAML []byte	plugin     pluginapi.Plugin	registered bool	client     pluginClient} type modelExecutor interface {	ExecuteModel(context.Context, handlers.ModelExecutionRequest) (handlers.ModelExecutionResponse, *interfaces.ErrorMessage)	ExecuteModelStream(context.Context, handlers.ModelExecutionRequest) (handlers.ModelExecutionStream, *interfaces.ErrorMessage)} type pluginUnloadTarget struct {	id      string	name    string	path    string	version string	client  pluginClient} type pluginLoadRequest struct {	result         chan pluginLoadResult	cleanupStarted bool} type pluginLoadResult struct {	loaded      *loadedPlugin	plugin      pluginapi.Plugin	initialized bool	err         error} type Host struct {	applyMu                chan struct{}	mu                     sync.Mutex	loader                 pluginLoader	loaded                 map[string]*loadedPlugin	retired                map[string][]*loadedPlugin	loading                map[string]*pluginLoadRequest	fused                  map[string]string	pluginFileVersions     map[string]string	activePluginVersions   map[string]string	activePluginPaths      map[string]string	cleanupFilesPending    bool	runtimeConfig          *config.Config	authManager            *coreauth.Manager	modelExecutor          modelExecutor	modelClientIDs         map[string]struct{}	executorModelClientIDs map[string]struct{}	modelProviders         map[string]string	modelRegistrations     map[string]pluginModelRegistration	providerModels         map[string][]*registryModelInfo	executorProviders      map[string]struct{}	accessProviderKeys     map[string]struct{}	commandLineFlags       map[string]commandLineFlagRecord	commandLineHits        map[string]struct{}	managementRoutes       map[string]managementRouteRecord	resourceRoutes         map[string]resourceRouteRecord	streams                *streamBridge	httpStreams            *hostHTTPStreamBridge	modelStreams           *modelStreamBridge	callbackContexts       *callbackContextRegistry	snapshot               atomic.Value} func New() *Host {	h := &Host{		applyMu:                make(chan struct{}, 1),		loader:                 defaultPluginLoader(),		loaded:                 make(map[string]*loadedPlugin),		retired:                make(map[string][]*loadedPlugin),		loading:                make(map[string]*pluginLoadRequest),		fused:                  make(map[string]string),		pluginFileVersions:     make(map[string]string),		activePluginVersions:   make(map[string]string),		activePluginPaths:      make(map[string]string),		cleanupFilesPending:    true,		modelClientIDs:         make(map[string]struct{}),		executorModelClientIDs: make(map[string]struct{}),		modelProviders:         make(map[string]string),		modelRegistrations:     make(map[string]pluginModelRegistration),		providerModels:         make(map[string][]*registryModelInfo),		executorProviders:      make(map[string]struct{}),		accessProviderKeys:     make(map[string]struct{}),		commandLineFlags:       make(map[string]commandLineFlagRecord),		commandLineHits:        make(map[string]struct{}),		managementRoutes:       make(map[string]managementRouteRecord),		resourceRoutes:         make(map[string]resourceRouteRecord),		streams:                newStreamBridge(),		httpStreams:            newHostHTTPStreamBridge(),		modelStreams:           newModelStreamBridge(),		callbackContexts:       newCallbackContextRegistry(),	}	h.snapshot.Store(emptySnapshot())	return h} func NewForTest(loader pluginLoader) *Host {	h := New()	h.loader = loader	return h} func (h *Host) SetModelExecutor(executor modelExecutor) {	if h == nil {		return	}	h.mu.Lock()	h.modelExecutor = executor	h.mu.Unlock()} func (h *Host) currentModelExecutor() modelExecutor {	if h == nil {		return nil	}	h.mu.Lock()	executor := h.modelExecutor	h.mu.Unlock()	return executor} func (h *Host) Snapshot() *Snapshot {	if h == nil {		return emptySnapshot()	}	raw := h.snapshot.Load()	if snap, ok := raw.(*Snapshot); ok && snap != nil {		return snap	}	return emptySnapshot()} // PluginLoaded reports whether a plugin dynamic library is still loaded by the host.func (h *Host) PluginLoaded(id string) bool {	if h == nil {		return false	}	id = strings.TrimSpace(id)	if id == "" {		return false	}	h.mu.Lock()	defer h.mu.Unlock()	_, ok := h.loaded[id]	if ok {		return true	}	return len(h.retired[id]) > 0} // PluginBusy reports whether a plugin dynamic library is loaded or being loaded.func (h *Host) PluginBusy(id string) bool {	if h == nil {		return false	}	id = strings.TrimSpace(id)	if id == "" {		return false	}	h.mu.Lock()	defer h.mu.Unlock()	if _, ok := h.loaded[id]; ok {		return true	}	if len(h.retired[id]) > 0 {		return true	}	_, ok := h.loading[id]	return ok} func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) {	if h == nil || !h.lockApply(ctx) {		return	}	defer h.unlockApply()	if ctx == nil {		ctx = context.Background()	}	if errContext := ctx.Err(); errContext != nil {		return	} 	rc, errRuntimeConfig := runtimeConfigFromConfig(cfg)	if errRuntimeConfig != nil {		log.WithError(errRuntimeConfig).Error("failed to apply plugin runtime config")		return	}	h.mu.Lock()	h.runtimeConfig = cfg	h.mu.Unlock() 	if !rc.Enabled {		h.mu.Lock()		h.managementRoutes = make(map[string]managementRouteRecord)		h.resourceRoutes = make(map[string]resourceRouteRecord)		h.rebuildActivePluginMapsLocked(nil)		h.snapshot.Store(emptySnapshot())		h.mu.Unlock()		h.refreshThinkingProviders(nil)		return	} 	desiredVersions := desiredPluginVersions(rc.Items)	files, errSelect := selectPluginFiles(rc.Dir, desiredVersions)	if errSelect != nil {		log.Warnf("pluginhost: failed to select plugin files: %v", errSelect)		h.mu.Lock()		h.managementRoutes = make(map[string]managementRouteRecord)		h.resourceRoutes = make(map[string]resourceRouteRecord)		h.rebuildActivePluginMapsLocked(nil)		h.snapshot.Store(emptySnapshot())		h.mu.Unlock()		h.refreshThinkingProviders(nil)		return	}	files = h.withLoadedPluginFallbacks(files, rc.Items, desiredVersions) 	records := make([]capabilityRecord, 0, len(files))	loadedFiles := make([]pluginFile, 0, len(files))	hotReloadLogs := make([]log.Fields, 0)	for _, file := range files {		item, ok := rc.Items[file.ID]		if !ok {			item = defaultRuntimeItemConfig(file.ID)		}		if !item.Enabled {			continue		}		h.mu.Lock()		lp := h.loaded[file.ID]		var replaced *loadedPlugin		if lp != nil && cleanPluginPath(lp.path) != cleanPluginPath(file.Path) {			replaced = lp			lp = nil		}		_, disabled := h.fused[file.ID]		h.mu.Unlock()		if disabled && replaced == nil {			continue		} 		loadedNow := false		var hotReloadFields log.Fields		var plugin pluginapi.Plugin		registeredNow := false		if lp == nil {			request := &pluginLoadRequest{result: make(chan pluginLoadResult, 1)}			h.mu.Lock()			if _, loading := h.loading[file.ID]; loading {				h.mu.Unlock()				continue			}			h.loading[file.ID] = request			h.mu.Unlock() 			if replaced != nil {				h.callQuiesce(ctx, replaced)				if errContext := ctx.Err(); errContext != nil {					h.clearLoadingRequest(file.ID, request)					_, _, _ = h.rollbackReplacement(replaced, item)					return				}			}			h.startPluginLoad(ctx, file, item, request) 			loadResult, completed := h.waitForPluginLoad(ctx, request)			if !completed {				if replaced == nil {					h.cleanupCanceledPluginLoad(file.ID, request)				} else {					h.cleanupCanceledPluginLoadAndWait(file.ID, request)					_, _, _ = h.rollbackReplacement(replaced, item)				}				return			}			if loadResult.err != nil || (replaced != nil && !loadResult.initialized) {				if replaced == nil {					h.cleanupPluginLoad(file.ID, request, loadResult.loaded)				} else {					h.cleanupPluginLoadAndWait(file.ID, request, loadResult.loaded)					if rollbackRecord, rollbackFile, okRollback := h.rollbackReplacement(replaced, item); okRollback {						records = append(records, rollbackRecord)						loadedFiles = append(loadedFiles, rollbackFile)					}				}				if loadResult.err != nil {					log.Warnf("pluginhost: failed to load plugin %s from %s: %v", file.ID, file.Path, loadResult.err)				}				continue			} 			h.mu.Lock()			if h.loading[file.ID] != request {				h.mu.Unlock()				h.discardLoadedPlugin(loadResult.loaded)				if replaced != nil {					_, _, _ = h.rollbackReplacement(replaced, item)				}				return			}			if errContext := ctx.Err(); errContext != nil {				h.mu.Unlock()				if replaced == nil {					h.cleanupPluginLoad(file.ID, request, loadResult.loaded)				} else {					h.cleanupPluginLoadAndWait(file.ID, request, loadResult.loaded)					_, _, _ = h.rollbackReplacement(replaced, item)				}				return			}			delete(h.loading, file.ID)			lp = loadResult.loaded			if replaced != nil {				hotReloadFields = pluginHotReloadLogFields(file.ID, file.Version, file.Path, replaced.version, replaced.path)				h.retireLoadedPluginLocked(replaced)				delete(h.fused, file.ID)				h.removePluginRuntimeStateLocked(file.ID)			}			h.loaded[file.ID] = lp			loadedNow = true			plugin = loadResult.plugin			registeredNow = loadResult.initialized			h.mu.Unlock()			log.WithFields(pluginLogFields(file.ID, "", file.Version, file.Path)).Info("pluginhost: plugin loaded")		} 		if !registeredNow {			if loadedNow {				continue			}			var okCall bool			plugin, okCall = h.callRegister(ctx, lp, item)			if !okCall {				continue			}		}		plugin.Metadata = clonePluginMetadata(plugin.Metadata)		h.mu.Lock()		if lp != nil {			lp.name = strings.TrimSpace(plugin.Metadata.Name)			if strings.TrimSpace(lp.version) == "" {				lp.version = strings.TrimSpace(plugin.Metadata.Version)			}		}		h.mu.Unlock()		if loadedNow {			log.WithFields(pluginLogFieldsFromMetadata(file.ID, plugin.Metadata, file.Path)).Info("pluginhost: plugin registered")		}		if hotReloadFields != nil {			hotReloadLogs = append(hotReloadLogs, hotReloadFields)		}		records = append(records, capabilityRecord{			id:       file.ID,			path:     file.Path,			version:  file.Version,			priority: item.Priority,			meta:     plugin.Metadata,			plugin:   plugin,		})		loadedFiles = append(loadedFiles, file)	} 	sortRecords(records)	h.mu.Lock()	cleanupFiles := h.cleanupFilesPending	if len(loadedFiles) > 0 {		h.cleanupFilesPending = false	}	h.rebuildActivePluginMapsLocked(records)	h.snapshot.Store(&Snapshot{enabled: true, records: records, quotaSupportedProviders: make(map[string][]string)})	h.mu.Unlock()	h.refreshThinkingProviders(records)	for _, fields := range hotReloadLogs {		log.WithFields(fields).Info("pluginhost: plugin hot reloaded")	}	if cleanupFiles && len(loadedFiles) > 0 {		if errCleanup := cleanupUnselectedPluginFiles(rc.Dir, loadedFiles); errCleanup != nil {			log.Warnf("pluginhost: failed to clean old plugin files: %v", errCleanup)		}	}} func (h *Host) startPluginLoad(ctx context.Context, file pluginFile, item runtimeItemConfig, request *pluginLoadRequest) {	if h == nil || request == nil || request.result == nil {		return	}	if ctx == nil {		ctx = context.Background()	}	go func() {		client, errOpen := h.loader.Open(file, h)		if errOpen != nil {			request.result <- pluginLoadResult{err: errOpen}			return		}		if client == nil {			request.result <- pluginLoadResult{err: fmt.Errorf("plugin loader returned nil client")}			return		}		loaded := &loadedPlugin{			id:      file.ID,			path:    file.Path,			version: file.Version,			client:  newGuardedPluginClient(client),		}		plugin, okCall := h.callRegister(ctx, loaded, item)		request.result <- pluginLoadResult{loaded: loaded, plugin: plugin, initialized: okCall}	}()} func (h *Host) waitForPluginLoad(ctx context.Context, request *pluginLoadRequest) (pluginLoadResult, bool) {	if h == nil || request == nil || request.result == nil {		return pluginLoadResult{}, false	}	if ctx == nil {		ctx = context.Background()	}	select {	case result := <-request.result:		return result, true	case <-ctx.Done():		return pluginLoadResult{}, false	}} func (h *Host) cleanupCanceledPluginLoad(id string, request *pluginLoadRequest) {	if h == nil || request == nil || request.result == nil {		return	}	h.mu.Lock()	if h.loading[id] != request || request.cleanupStarted {		h.mu.Unlock()		return	}	request.cleanupStarted = true	h.mu.Unlock() 	go func() {		result := <-request.result		h.finishPluginLoadCleanup(id, request, result.loaded)	}()} func (h *Host) cleanupCanceledPluginLoadAndWait(id string, request *pluginLoadRequest) {	if h == nil || request == nil || request.result == nil {		return	}	h.mu.Lock()	if h.loading[id] != request || request.cleanupStarted {		h.mu.Unlock()		return	}	request.cleanupStarted = true	h.mu.Unlock() 	result := <-request.result	h.discardLoadedPlugin(result.loaded)	h.clearLoadingRequest(id, request)} // cleanupPluginLoad retains the matching load token until the client has physically// shut down, preventing a replacement ApplyConfig from opening a second client.func (h *Host) cleanupPluginLoad(id string, request *pluginLoadRequest, loaded *loadedPlugin) {	if h == nil || request == nil {		return	}	h.mu.Lock()	if h.loading[id] != request || request.cleanupStarted {		h.mu.Unlock()		return	}	request.cleanupStarted = true

Lines 1–500 of 1065. Load more