Cyrene_v2/internal/capability/plugin/manager.go

// Package plugin manages runtime-installed external tool providers. In v1 the
// only installable kind is an MCP server (DevDocs/design/09 插件生态 §1): a
// compiled Go plugin cannot be hot-loaded into a single static binary, so a
// runtime "install" is an external MCP server whose config is persisted and
// whose discovered tools are hot-registered into the shared capability registry.
//
// The manager owns the live *mcp.Registry of every installed server so the
// server can be refreshed (reinstall under the same name) or uninstalled, and
// the config can be restored on the next boot. A connect/discover failure never
// persists a half-installed plugin.
package plugin

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/mcp"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig"
)

const (
	// connectTimeout bounds one server's connect + discover handshake so an
	// unreachable server fails the install instead of hanging the request.
	connectTimeout = 15 * time.Second
	// mcpCapabilityPrefix names the KindMCP capability a server surfaces as, so
	// the admin plugin list enumerates the server itself as well as its tools.
	mcpCapabilityPrefix = "mcp:"
)

// Installed is one plugin as reported to the admin API: its persisted config,
// whether a live server currently backs it, and how many tools it registered.
type Installed struct {
	// Config is the persisted plugin config (the caller masks it before wire).
	Config runtimeconfig.PluginConfig
	// Live reports whether the MCP server is connected and its tools registered.
	Live bool
	// ToolCount is the number of tools the live server registered (0 when not
	// live).
	ToolCount int
}

// server is a live installed MCP server.
type server struct {
	cfg       runtimeconfig.PluginConfig
	reg       *mcp.Registry
	toolCount int
}

// Manager installs, refreshes, lists and uninstalls runtime plugins over one
// capability registry and one plugins.json path. It is safe for concurrent use.
type Manager struct {
	reg  *registry.Registry
	path string
	log  *slog.Logger

	mu      sync.Mutex
	servers map[string]*server
}

// NewManager returns a Manager backed by reg and the plugins.json at path.
func NewManager(reg *registry.Registry, path string, log *slog.Logger) *Manager {
	if log == nil {
		log = slog.New(slog.DiscardHandler)
	}
	return &Manager{reg: reg, path: path, log: log, servers: make(map[string]*server)}
}

// Install connects cfg's MCP server, registers its tools, persists the config
// and holds the live registry. Installing a name that already exists refreshes
// it: the previous server is unregistered first, and if the new one fails to
// connect the previous one is restored (the on-disk config is never changed on a
// failure).
func (m *Manager) Install(ctx context.Context, cfg runtimeconfig.PluginConfig) (Installed, error) {
	return m.install(ctx, cfg, true)
}

// install is Install with the persist step optional (Restore reuses it without
// rewriting the file it just read).
func (m *Manager) install(ctx context.Context, cfg runtimeconfig.PluginConfig, persist bool) (Installed, error) {
	cfg = cfg.Normalize()
	if err := cfg.Validate(); err != nil {
		return Installed{}, err
	}
	prev := m.take(cfg.Name)
	cfg = inheritIdentity(cfg, prev)
	if strings.TrimSpace(cfg.CreatedAt) == "" {
		cfg.CreatedAt = time.Now().UTC().Format(time.RFC3339)
	}
	if prev != nil {
		m.stop(ctx, prev)
	}
	srv, err := m.start(ctx, cfg)
	if err != nil {
		m.restore(prev)
		return Installed{}, err
	}
	if persist {
		if err := m.persistUpsert(cfg); err != nil {
			m.stop(ctx, srv)
			m.restore(prev)
			return Installed{}, err
		}
	}
	m.put(srv)
	m.log.Info("plugin installed", "name", cfg.Name, "kind", cfg.Kind, "transport", cfg.Transport, "tools", srv.toolCount)
	return Installed{Config: cfg, Live: true, ToolCount: srv.toolCount}, nil
}

// Uninstall removes the plugin matching name: it unregisters the live server's
// tools and closes it, then deletes the persisted config. It reports whether
// anything matched, so the caller can fall back to a plain capability removal.
func (m *Manager) Uninstall(ctx context.Context, name string) (bool, error) {
	srv := m.take(name)
	deleted, err := m.persistDelete(name)
	if srv != nil {
		m.stop(ctx, srv)
	}
	if err != nil {
		return srv != nil, fmt.Errorf("plugin %q: %w", name, err)
	}
	removed := srv != nil || deleted
	if removed {
		m.log.Info("plugin uninstalled", "name", name)
	}
	return removed, nil
}

// List returns every persisted plugin (in file order) followed by any live
// server not in the file, each annotated with its live state and tool count.
func (m *Manager) List() []Installed {
	cfgs, err := runtimeconfig.LoadPlugins(m.path)
	if err != nil {
		m.log.Warn("plugins.json load skipped", "path", m.path, "error", err)
		cfgs = nil
	}
	out := make([]Installed, 0, len(cfgs))
	seen := make(map[string]bool, len(cfgs))
	m.mu.Lock()
	defer m.mu.Unlock()
	for _, cfg := range cfgs {
		item := Installed{Config: cfg}
		if srv, ok := m.servers[cfg.Name]; ok {
			item.Live = true
			item.ToolCount = srv.toolCount
		}
		out = append(out, item)
		seen[cfg.Name] = true
	}
	for name, srv := range m.servers {
		if seen[name] {
			continue
		}
		out = append(out, Installed{Config: srv.cfg, Live: true, ToolCount: srv.toolCount})
	}
	return out
}

// Restore installs every persisted plugin. A server that is unreachable at boot
// is logged and left in plugins.json (so the owner can retry or uninstall it);
// boot never fails because an external server is down.
func (m *Manager) Restore(ctx context.Context) {
	cfgs, err := runtimeconfig.LoadPlugins(m.path)
	if err != nil {
		m.log.Warn("plugins.json load skipped", "path", m.path, "error", err)
		return
	}
	for _, cfg := range cfgs {
		if _, err := m.install(ctx, cfg, false); err != nil {
			m.log.Warn("plugin restore skipped", "name", cfg.Name, "error", err)
		}
	}
}

// Close unregisters every live server without touching plugins.json.
func (m *Manager) Close(ctx context.Context) {
	m.mu.Lock()
	live := make([]*server, 0, len(m.servers))
	for _, srv := range m.servers {
		live = append(live, srv)
	}
	m.servers = make(map[string]*server)
	m.mu.Unlock()
	for _, srv := range live {
		m.stop(ctx, srv)
	}
}

// start connects cfg, registers its tools and surfaces its KindMCP capability.
// On a registration failure it cleans up and returns the error.
func (m *Manager) start(ctx context.Context, cfg runtimeconfig.PluginConfig) (*server, error) {
	mreg := mcp.NewRegistry(m.reg, mcpConfigOf(cfg), mcp.WithLogger(m.log))
	connectCtx, cancel := context.WithTimeout(ctx, connectTimeout)
	defer cancel()
	if err := mreg.Register(connectCtx); err != nil {
		_ = mreg.Unregister(context.Background())
		return nil, fmt.Errorf("plugin %q: %w", cfg.Name, err)
	}
	names := mreg.ToolNames()
	err := m.reg.Register(registry.Capability{
		Name:        mcpCapabilityPrefix + cfg.Name,
		Kind:        registry.KindMCP,
		Description: "MCP server " + cfg.Name,
		Value:       mreg,
	})
	if err != nil && !errors.Is(err, registry.ErrDuplicate) {
		m.log.Warn("plugin capability register skipped", "name", cfg.Name, "error", err)
	}
	return &server{cfg: cfg, reg: mreg, toolCount: len(names)}, nil
}

// stop unregisters a live server's tools and closes it, then drops its KindMCP
// capability. Errors are logged, never fatal: uninstall must still remove the
// config.
func (m *Manager) stop(ctx context.Context, srv *server) {
	if srv == nil {
		return
	}
	if err := srv.reg.Unregister(ctx); err != nil {
		m.log.Warn("plugin unregister", "name", srv.cfg.Name, "error", err)
	}
	if err := m.reg.Remove(mcpCapabilityPrefix + srv.cfg.Name); err != nil && !errors.Is(err, registry.ErrNotFound) {
		m.log.Warn("plugin capability remove", "name", srv.cfg.Name, "error", err)
	}
}

// restore re-installs a server that a failed refresh displaced, so a failed
// refresh never leaves the registry without a previously working server.
func (m *Manager) restore(prev *server) {
	if prev == nil {
		return
	}
	srv, err := m.start(context.Background(), prev.cfg)
	if err != nil {
		m.log.Warn("plugin rollback failed", "name", prev.cfg.Name, "error", err)
		return
	}
	m.put(srv)
}

// take removes and returns the live server named name, or nil.
func (m *Manager) take(name string) *server {
	m.mu.Lock()
	defer m.mu.Unlock()
	srv := m.servers[name]
	delete(m.servers, name)
	return srv
}

// put records srv as the live server for its name.
func (m *Manager) put(srv *server) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.servers[srv.cfg.Name] = srv
}

// persistUpsert writes cfg into plugins.json (upsert by name).
func (m *Manager) persistUpsert(cfg runtimeconfig.PluginConfig) error {
	cur, err := runtimeconfig.LoadPlugins(m.path)
	if err != nil {
		return fmt.Errorf("load plugins.json: %w", err)
	}
	if err := runtimeconfig.SavePlugins(m.path, runtimeconfig.UpsertPlugin(cur, cfg)); err != nil {
		return fmt.Errorf("save plugins.json: %w", err)
	}
	return nil
}

// persistDelete removes name (by name or id) from plugins.json, reporting
// whether an entry was removed. A missing file is not an error.
func (m *Manager) persistDelete(name string) (bool, error) {
	cur, err := runtimeconfig.LoadPlugins(m.path)
	if err != nil {
		return false, fmt.Errorf("load plugins.json: %w", err)
	}
	out, removed := runtimeconfig.DeletePlugin(cur, name)
	if !removed {
		return false, nil
	}
	if err := runtimeconfig.SavePlugins(m.path, out); err != nil {
		return false, fmt.Errorf("save plugins.json: %w", err)
	}
	return true, nil
}

// mcpConfigOf maps a plugin config to the MCP client config.
func mcpConfigOf(cfg runtimeconfig.PluginConfig) mcp.MCPConfig {
	return mcp.MCPConfig{
		Name:          cfg.Name,
		Command:       cfg.Command,
		Args:          cfg.Args,
		URL:           cfg.Endpoint,
		TransportType: cfg.Transport,
		APIKey:        cfg.APIKey,
		Env:           cfg.Env,
		Enabled:       true,
	}
}

// inheritIdentity keeps a refresh's id and install time when the request omits
// them, so a reinstall updates a plugin rather than minting a new identity.
func inheritIdentity(cfg runtimeconfig.PluginConfig, prev *server) runtimeconfig.PluginConfig {
	if prev == nil {
		return cfg
	}
	if strings.TrimSpace(cfg.ID) == "" {
		cfg.ID = prev.cfg.ID
	}
	if strings.TrimSpace(cfg.CreatedAt) == "" {
		cfg.CreatedAt = prev.cfg.CreatedAt
	}
	return cfg
}