Something went wrong. Try again.
sloproxy
Something went wrong. Try again.
29 kB · 1064 lines
Go
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065package 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}) 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 h.mu.Unlock()
h.finishPluginLoadCleanup(id, request, loaded)}
func (h *Host) cleanupPluginLoadAndWait(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 h.mu.Unlock()
h.discardLoadedPlugin(loaded) h.clearLoadingRequest(id, request)}
func (h *Host) finishPluginLoadCleanup(id string, request *pluginLoadRequest, loaded *loadedPlugin) { go func() { h.discardLoadedPlugin(loaded) h.clearLoadingRequest(id, request) }()}
func (h *Host) clearLoadingRequest(id string, request *pluginLoadRequest) { if h == nil || request == nil { return } h.mu.Lock() if h.loading[id] == request { delete(h.loading, id) } h.mu.Unlock()}
func (h *Host) discardLoadedPlugin(loaded *loadedPlugin) { if loaded == nil || loaded.client == nil { return } shutdownPluginClient(context.Background(), loaded.client)}
func (h *Host) withLoadedPluginFallbacks(files []pluginFile, items map[string]runtimeItemConfig, desired map[string]string) []pluginFile { if h == nil || len(desired) == 0 { return files } selected := make(map[string]struct{}, len(files)) for _, file := range files { id := strings.TrimSpace(file.ID) if id != "" { selected[id] = struct{}{} } } ids := make([]string, 0, len(desired)) for id := range desired { ids = append(ids, id) } sort.Strings(ids)
h.mu.Lock() defer h.mu.Unlock() for _, id := range ids { if _, ok := selected[id]; ok { continue } if item, ok := items[id]; ok && !item.Enabled { continue } lp := h.loaded[id] if lp == nil || strings.TrimSpace(lp.path) == "" { continue } files = append(files, pluginFile{ ID: id, Path: lp.path, Version: strings.TrimSpace(lp.version), }) selected[id] = struct{}{} } return files}
// UnloadPlugin removes one plugin from the active runtime and closes its dynamic library.func (h *Host) UnloadPlugin(id string) bool { return h.UnloadPluginContext(context.Background(), id)}
// UnloadPluginContext detaches a plugin from the runtime before waiting for its// active calls. Physical client cleanup continues after cancellation if needed.func (h *Host) UnloadPluginContext(ctx context.Context, id string) bool { if h == nil { return false } id = strings.TrimSpace(id) if id == "" || !h.lockApply(ctx) { return false } defer h.unlockApply()
targets := make([]pluginUnloadTarget, 0) h.mu.Lock() lp := h.loaded[id] if lp != nil { targets = append(targets, pluginUnloadTarget{id: lp.id, name: lp.name, path: lp.path, version: lp.version, client: lp.client}) } for _, retired := range h.retired[id] { if retired == nil { continue } targets = append(targets, pluginUnloadTarget{id: retired.id, name: retired.name, path: retired.path, version: retired.version, client: retired.client}) } if len(targets) == 0 { h.mu.Unlock() return false } delete(h.loaded, id) delete(h.retired, id) delete(h.fused, id) delete(h.activePluginVersions, id) delete(h.activePluginPaths, id) for _, target := range targets { delete(h.pluginFileVersions, cleanPluginPath(target.path)) } records, enabled := h.snapshotWithoutPluginLocked(id) h.removePluginRuntimeStateLocked(id) h.snapshot.Store(&Snapshot{enabled: enabled, records: records}) h.mu.Unlock()
h.refreshThinkingProviders(records) h.RegisterFrontendAuthProviders() for _, target := range targets { if target.client != nil { shutdownPluginClient(ctx, target.client) } log.WithFields(pluginLogFields(target.id, target.name, target.version, target.path)).Info("pluginhost: plugin unloaded") } return true}
// ShutdownAll removes active plugin capabilities and closes all loaded dynamic libraries.func (h *Host) ShutdownAll() { h.ShutdownAllContext(context.Background())}
// ShutdownAllContext detaches all plugin runtime state without waiting beyond ctx// for active plugin calls to complete.func (h *Host) ShutdownAllContext(ctx context.Context) { if h == nil || !h.lockApply(ctx) { return } defer h.unlockApply()
targets := make([]pluginUnloadTarget, 0) var loading map[string]*pluginLoadRequest h.mu.Lock() loading = make(map[string]*pluginLoadRequest, len(h.loading)) for id, request := range h.loading { loading[id] = request } for _, lp := range h.loaded { if lp == nil || lp.client == nil { continue } targets = append(targets, pluginUnloadTarget{ id: lp.id, name: lp.name, path: lp.path, version: lp.version, client: lp.client, }) } for _, retiredPlugins := range h.retired { for _, lp := range retiredPlugins { if lp == nil || lp.client == nil { continue } targets = append(targets, pluginUnloadTarget{ id: lp.id, name: lp.name, path: lp.path, version: lp.version, client: lp.client, }) } } h.loaded = make(map[string]*loadedPlugin) h.retired = make(map[string][]*loadedPlugin) h.modelClientIDs = make(map[string]struct{}) h.executorModelClientIDs = make(map[string]struct{}) h.modelProviders = make(map[string]string) h.modelRegistrations = make(map[string]pluginModelRegistration) h.providerModels = make(map[string][]*registryModelInfo) h.executorProviders = make(map[string]struct{}) h.commandLineFlags = make(map[string]commandLineFlagRecord) h.commandLineHits = make(map[string]struct{}) h.managementRoutes = make(map[string]managementRouteRecord) h.resourceRoutes = make(map[string]resourceRouteRecord) h.pluginFileVersions = make(map[string]string) h.activePluginVersions = make(map[string]string) h.activePluginPaths = make(map[string]string) h.snapshot.Store(emptySnapshot()) h.mu.Unlock()
h.refreshThinkingProviders(nil) h.RegisterFrontendAuthProviders() for id, request := range loading { h.cleanupCanceledPluginLoad(id, request) } for _, target := range targets { shutdownPluginClient(ctx, target.client) log.WithFields(pluginLogFields(target.id, target.name, target.version, target.path)).Info("pluginhost: plugin unloaded") }}
func (h *Host) lockApply(ctx context.Context) bool { if h == nil { return false } if ctx == nil { ctx = context.Background() } select { case h.applyMu <- struct{}{}: return true default: } select { case h.applyMu <- struct{}{}: return true case <-ctx.Done(): return false }}
func (h *Host) unlockApply() { <-h.applyMu}
func shutdownPluginClient(ctx context.Context, client pluginClient) { if client == nil { return } if guarded, ok := client.(*guardedPluginClient); ok { guarded.ShutdownContext(ctx) return } client.Shutdown()}
func cleanPluginPath(path string) string { path = strings.TrimSpace(path) if path == "" { return "" } return filepath.Clean(path)}
func (h *Host) retireLoadedPluginLocked(lp *loadedPlugin) { if h == nil || lp == nil { return } h.retired[lp.id] = append(h.retired[lp.id], lp)}
func (h *Host) recordCurrent(record capabilityRecord) bool { return h.pluginIdentityCurrent(record.id, record.path, record.version)}
func (h *Host) pluginIdentityCurrent(id string, path string, version string) bool { if h == nil { return false } version = strings.TrimSpace(version) h.mu.Lock() defer h.mu.Unlock() id = strings.TrimSpace(id) if id == "" { return false } path = cleanPluginPath(path) if path == "" || h.activePluginPaths[id] != path { return false } activePathVersion, okVersion := h.pluginFileVersions[path] if !okVersion || activePathVersion != version { return false } return h.activePluginVersions[id] == version}
func (h *Host) snapshotWithoutPluginLocked(id string) ([]capabilityRecord, bool) { raw := h.snapshot.Load() snap, _ := raw.(*Snapshot) if snap == nil || len(snap.records) == 0 { return nil, snap != nil && snap.enabled } records := make([]capabilityRecord, 0, len(snap.records)) for _, record := range snap.records { if record.id == id { continue } records = append(records, record) } return records, snap.enabled}
func (h *Host) removePluginRuntimeStateLocked(id string) { for key, record := range h.managementRoutes { if record.pluginID == id { delete(h.managementRoutes, key) } } for key, record := range h.resourceRoutes { if record.pluginID == id { delete(h.resourceRoutes, key) } } for name, record := range h.commandLineFlags { if record.pluginID == id { delete(h.commandLineFlags, name) delete(h.commandLineHits, name) } } if registration, ok := h.modelRegistrations[id]; ok { delete(h.providerModels, registration.provider) } delete(h.modelProviders, id) delete(h.modelRegistrations, id)}
func (h *Host) rebuildActivePluginMapsLocked(records []capabilityRecord) { h.pluginFileVersions = make(map[string]string, len(records)) h.activePluginVersions = make(map[string]string, len(records)) h.activePluginPaths = make(map[string]string, len(records)) for _, record := range records { id := strings.TrimSpace(record.id) path := cleanPluginPath(record.path) if id == "" || path == "" { continue } h.pluginFileVersions[path] = strings.TrimSpace(record.version) h.activePluginVersions[id] = strings.TrimSpace(record.version) h.activePluginPaths[id] = path }}
func (h *Host) callQuiesce(ctx context.Context, lp *loadedPlugin) bool { if h == nil || lp == nil || lp.client == nil { return false } errQuiesce, okCall := h.safePluginAction(ctx, lp.id, pluginabi.MethodPluginQuiesce, func() error { _, errCall := callPlugin[rpcEmptyResponse](ctx, lp.client, pluginabi.MethodPluginQuiesce, rpcEmptyResponse{}) return errCall }) if errQuiesce != nil { logQuiesceError(lp.id, errQuiesce) } return okCall && errQuiesce == nil}
func logQuiesceError(id string, errQuiesce error) { entry := log.WithError(errQuiesce).WithField("plugin_id", id) switch { case quiesceUnsupported(errQuiesce): entry.Debug("pluginhost: plugin quiesce unsupported") case errors.Is(errQuiesce, context.Canceled), errors.Is(errQuiesce, context.DeadlineExceeded): entry.Debug("pluginhost: plugin quiesce canceled") default: entry.Warn("pluginhost: plugin quiesce failed") }}
func quiesceUnsupported(errQuiesce error) bool { if errQuiesce == nil { return false } var errRPC rpcError if errors.As(errQuiesce, &errRPC) { switch strings.ToLower(strings.TrimSpace(errRPC.Code)) { case "unknown_method", "method_not_found", "unsupported_method": return true } } if errors.Is(errQuiesce, errors.ErrUnsupported) { return true } message := strings.ToLower(strings.TrimSpace(errQuiesce.Error())) for _, indication := range []string{ "unknown method", "method not found", "unsupported method", "method unsupported", "method is not supported", "method not supported", } { if strings.Contains(message, indication) { return true } } return false}
func (h *Host) rollbackReplacement(lp *loadedPlugin, item runtimeItemConfig) (capabilityRecord, pluginFile, bool) { if h == nil || lp == nil { return capabilityRecord{}, pluginFile{}, false } h.mu.Lock() configYAML := bytes.Clone(lp.configYAML) previousPlugin := lp.plugin h.mu.Unlock() if len(configYAML) > 0 { item.ConfigYAML = configYAML }
plugin, okCall := h.callRegister(context.Background(), lp, item) if okCall { h.mu.Lock() delete(h.fused, lp.id) h.mu.Unlock() } else { plugin = previousPlugin } if !validPlugin(plugin) { return capabilityRecord{}, pluginFile{}, false } return capabilityRecord{ id: lp.id, path: lp.path, version: lp.version, priority: item.Priority, meta: plugin.Metadata, plugin: plugin, }, pluginFile{ ID: lp.id, Path: lp.path, Version: lp.version, }, true}
func (h *Host) callRegister(ctx context.Context, lp *loadedPlugin, item runtimeItemConfig) (pluginapi.Plugin, bool) { if lp == nil { return pluginapi.Plugin{}, false }
method := pluginabi.MethodPluginRegister h.mu.Lock() registered := lp.registered h.mu.Unlock() if registered { method = pluginabi.MethodPluginReconfigure }
plugin, okCall := h.safePluginCall(ctx, lp.id, method, func() pluginapi.Plugin { plugin, errRegister := registerRPCPlugin(ctx, h, lp.id, lp.client, method, item.ConfigYAML) if errRegister != nil { log.Warnf("pluginhost: plugin %s %s failed: %v", lp.id, method, errRegister) return pluginapi.Plugin{} } return plugin }) if !okCall { return pluginapi.Plugin{}, false } h.mu.Lock() lp.registered = true h.mu.Unlock() if !validPlugin(plugin) { log.Warnf("pluginhost: plugin %s returned invalid metadata or no capabilities", lp.id) return pluginapi.Plugin{}, false } plugin.Metadata = clonePluginMetadata(plugin.Metadata) h.mu.Lock() lp.name = strings.TrimSpace(plugin.Metadata.Name) if strings.TrimSpace(lp.version) == "" { lp.version = strings.TrimSpace(plugin.Metadata.Version) } lp.configYAML = bytes.Clone(item.ConfigYAML) lp.plugin = plugin h.mu.Unlock() return plugin, true}
func (h *Host) safePluginAction(ctx context.Context, id, method string, fn func() error) (err error, ok bool) { defer func() { if recovered := recover(); recovered != nil { h.fusePlugin(id, method, recovered) err = nil ok = false } }()
if ctx != nil { select { case <-ctx.Done(): return ctx.Err(), false default: } } return fn(), true}
func (h *Host) safePluginCall(ctx context.Context, id, method string, fn func() pluginapi.Plugin) (out pluginapi.Plugin, ok bool) { defer func() { if recovered := recover(); recovered != nil { h.fusePlugin(id, method, recovered) out = pluginapi.Plugin{} ok = false } }()
if ctx != nil { select { case <-ctx.Done(): return pluginapi.Plugin{}, false default: } } return fn(), true}
func validPlugin(plugin pluginapi.Plugin) bool { if strings.TrimSpace(plugin.Metadata.Name) == "" { return false } if strings.TrimSpace(plugin.Metadata.Version) == "" { return false } if strings.TrimSpace(plugin.Metadata.Author) == "" { return false } if strings.TrimSpace(plugin.Metadata.GitHubRepository) == "" { return false } caps := plugin.Capabilities return caps.ModelRegistrar != nil || caps.ModelProvider != nil || caps.AuthProvider != nil || caps.FrontendAuthProvider != nil || caps.Scheduler != nil || caps.ModelRouter != nil || caps.Executor != nil || caps.RequestTranslator != nil || caps.RequestNormalizer != nil || caps.RequestInterceptor != nil || caps.RequestLifecyclePlugin != nil || caps.ResponseTranslator != nil || caps.ResponseBeforeTranslator != nil || caps.ResponseAfterTranslator != nil || caps.ResponseInterceptor != nil || caps.StreamChunkInterceptor != nil || caps.WebSocketResponseObserver != nil || caps.ThinkingApplier != nil || caps.UsagePlugin != nil || caps.CommandLinePlugin != nil || caps.ManagementAPI != nil}
func typeName(v any) string { return fmt.Sprintf("%T", v)}