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
	// observed 是每台 server 最近一次被观察到的运行期事实（成功连上的时刻、最近一次失败
	// 的文本与时刻）。它与 servers（活没活）是两个不同的事实：一台连不上的 server 不在
	// servers 里，但它的失败记录必须留着 —— 否则「为什么挂了」在快照里就是一个空。
	observed map[string]observed
	// panelKeys 记住每个已安装插件**当时用过的**面板目录键（插件名 → 目录键）。
	// 它是 memo 而不是第二份真源：目录键由 PanelPluginID 从 cfg 推出，这里只把
	// 结果留下，好让 Uninstall 在活 server 已不在时仍能精确清掉该插件的面板。
	panelKeys map[string]string
	// panels 是可选的面板声明中枢（§5.2.1）。为 nil 时管理器完全不知道面板，
	// 既有行为逐字不变（未接线的进程/测试就是这样）。
	panels *Panels
}

// 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),
		observed:  make(map[string]observed),
		panelKeys: make(map[string]string),
	}
}

// AttachPanels 把面板声明中枢接到管理器上（design/09 §5.2.1「插件管理器 = 唯一
// 中枢」）。必须在插件被安装**之前**接好，否则启动期恢复的那些插件不会留下目录键。
// 传 nil 等于不接线（合法：未接线的进程行为与接线前逐字一致）。
func (m *Manager) AttachPanels(p *Panels) {
	m.mu.Lock()
	m.panels = p
	m.mu.Unlock()
}

// Panels 返回已接线的面板中枢；未接线时返回 nil（网关据此诚实作答 501，
// 而不是回一份空面板列表假装「没有插件的页面」）。
func (m *Manager) Panels() *Panels {
	if m == nil {
		return nil
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	return m.panels
}

// 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, PanelChangeInstalled)
}

// install is Install with the persist step optional (Restore reuses it without
// rewriting the file it just read). changeReason labels the panel change pushed
// when the plugin becomes live (install vs enable).
//
// It respects cfg.Enabled: a config that says enabled=false is **persisted but not
// connected** (no live server, no tools), which is how "installed but switched off"
// survives a restart. Restore goes through this same path, so startup honours the
// flag too.
func (m *Manager) install(ctx context.Context, cfg runtimeconfig.PluginConfig, persist bool, changeReason string) (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)
	}
	if !cfg.IsEnabled() {
		return m.installDisabled(cfg, persist, prev)
	}
	srv, err := m.start(ctx, cfg)
	if err != nil {
		// 失败也要留痕：状态快照要能说出「为什么挂了」，而不是把失败悄悄丢进日志。
		m.noteFailure(cfg, err)
		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.noteSeen(cfg)
	m.log.Info("plugin installed", "name", cfg.Name, "kind", cfg.Kind, "transport", cfg.Transport, "tools", srv.toolCount)
	m.markInstalled(cfg, changeReason)
	return Installed{Config: cfg, Live: true, ToolCount: srv.toolCount}, nil
}

// installDisabled 走「已安装但停用」这条路：只落配置，不连接、不发现、不注册任何工具，
// 并把面板入口隐藏（`when:{enabled:true}` 的入口不出现）。
//
// 调用前 prev（同名活 server）已经被停掉：配置现在说的是「停用」，让旧 server 继续挂着
// 就成了两份互相矛盾的事实。落盘失败时把 prev 恢复回来（回滚到「启用且活着」那个状态）。
func (m *Manager) installDisabled(cfg runtimeconfig.PluginConfig, persist bool, prev *server) (Installed, error) {
	if persist {
		if err := m.persistUpsert(cfg); err != nil {
			m.restore(prev)
			return Installed{}, err
		}
	}
	m.noteDisabled(cfg)
	m.markDisabled(cfg)
	m.log.Info("plugin installed disabled", "name", cfg.Name, "kind", cfg.Kind, "transport", cfg.Transport)
	return Installed{Config: cfg}, 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.takeAny(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 {
		panelName := name
		if srv != nil {
			panelName = srv.cfg.Name
		}
		m.forgetPanels(panelName)
		m.forgetObserved(panelName)
		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, PanelChangeInstalled); 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)
	// 回滚成功也是一次成功连接：快照的 lastSeenAt 不该停留在这之前的时刻。
	m.noteSeen(prev.cfg)
}

// 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
}