HomeAgent/internal/kernelboot/scheduler.go

package kernelboot

// 插件调度器：按 syncbus 状态决定**何时**拉起各插件（设计 §2.6.4）。
//
// ## 核心规则（三条，按优先级）
//
//	未声明依赖              ⇒ 等内核**完全**初始化（默认路径 = 现行为，§2.6.5）
//	声明了依赖且全就绪       ⇒ **立即**拉起
//	声明了依赖但未全就绪     ⇒ 挂起，等 syncbus 更新后再判断
//
// ## ★ 兼容性的落点就在第一条
//
// 存量 42 个插件都不声明依赖 ⇒ 全部走第一条 ⇒ 行为**逐字不变**。
// 这不是"顺便兼容"，而是本设计的硬约束：42 个插件里任何一个若因本设计
// 改变了启动时机，就可能在不该动的时候动（例如假设"Start 时 llm 已就绪"）。
// **静默改变启动时机 = 静默引入竞态。**
//
// ## 为什么"未声明"不能等价于"依赖集为空"
//
// 空依赖集在数学上"已满足"，若按此处理，未声明的插件会在**第一层完成后**
// 就被拉起 —— 比现状早得多，直接破坏兼容性。
// 所以"未声明"与"声明了空集"是**不同**语义，必须分开判断（见 Wants）。
//
// ## 为什么调度器不自带并发控制
//
// 拉起插件的实际动作（load/Start）由插件管理器做 —— 那里已有
// Registry 的锁与生命周期管理。本调度器只负责**判断时机**，
// 保持单一职责，也让判断逻辑可以脱离真实插件系统被单测（判据的关键）。

import (
	"fmt"
	"sort"
	"strings"
	"sync"
)

// PluginSpec 是调度器需要知道的插件声明（设计 §2.6.4/2.6.6）。
type PluginSpec struct {
	Name string

	// RequiresFacilities 是插件声明需要的设施名。
	//
	// nil/空 ⇒ **未声明**（走"等内核完全初始化"路径）
	// 非空   ⇒ 声明的依赖；全就绪即拉起
	RequiresFacilities []string

	// AutoStart 为 false 表示"不伴随启动"：即使依赖齐了也**不**自动拉起，
	// 等 agent 经 plugin_enable 显式启用（设计 §2.6.6）。
	//
	// 用指针语义的注意：这里用 bool + 一个显式标记区分"未声明"与"false"。
	// 但为了简洁，约定：**缺省（未声明）= true**（伴随启动，与现状一致），
	// 由 manifest 层负责把"未声明"翻译成 true。
	AutoStart bool
}

// Scheduler 按 syncbus 状态调度插件。
type Scheduler struct {
	mu       sync.Mutex
	bus      SyncState
	pending  []PluginSpec
	launched map[string]bool
	launch   func(PluginSpec) error

	// onLaunchErr 记录拉起失败（不中断其他插件）
	onLaunchErr func(PluginSpec, error)
}

// NewScheduler 创建调度器。launch 是实际拉起动作（由插件管理器提供）。
//
// ★ bus 参数类型是 **SyncState**（只读接口），不是 *SyncBus。
// 这就是设计 §2.6.3 的权限隔离：调度器**在编译期就没有**写 syncbus 的能力。
func NewScheduler(bus SyncState, launch func(PluginSpec) error) *Scheduler {
	s := &Scheduler{
		bus:      bus,
		launch:   launch,
		launched: map[string]bool{},
	}
	// ★ 订阅状态变化：syncbus 每次更新（层边界 / 内核完成）就重新评估挂起队列。
	//
	// 这是"自动拉起"的实现方式：调度器不需要轮询（空转）也不需要调用方
	// 记得在每处手动触发（漏了就会静默卡住）。订阅一次，永久生效。
	//
	// 若 bus 不支持订阅（只传了个只读快照实现），退化为"只在 Submit 时评估"——
	// 那种情况下调度器仍可用，只是不会自动跟进后续状态变化。
	if sub, ok := bus.(interface{ OnUpdate(func(SyncState)) }); ok {
		sub.OnUpdate(func(SyncState) { s.evaluate() })
	}
	return s
}

// SetLaunch 补上拉起动作（供"两段式"调度：装配前订阅、装配后接动作）。
//
// ## 为什么需要它（真实缺口）
//
// 插件调度器必须在**装配开始前**订阅 syncbus，否则"声明提前启动"不会生效
// （提交时全部设施早已就绪 ⇒ 永远退化成"立即拉起"）。
//
// 但**拉起动作**依赖插件注册表，而那是 stage-host 模块的产物
// ⇒ 装配前拿不到。所以分两段：
//
//	· 装配前：NewScheduler(bus, nil) 只订阅
//	· 装配后：SetLaunch(fn) 接上真实动作
//
// ★ 此时可能已有插件被标记 launched（分层过程中达到条件）。
// 调用方需自行补偿"已标记但未真正加载"的那些 —— 见 LaunchedFlag。
func (s *Scheduler) SetLaunch(fn func(PluginSpec) error) {
	s.mu.Lock()
	s.launch = fn
	s.mu.Unlock()
	// ★ 立刻重新评估一次：接上动作之前可能已有插件达到条件并被标记
	// （它们在 syncbus 更新时就评估过，但那时 launch==nil ⇒ 只标记未执行）。
	// 不补这一下，它们要等到**下一次** syncbus 更新才会被真正拉起 ——
	// 而若内核装配已结束（不再有更新），就永远不拉。
	s.evaluate()
}

// LaunchedFlag 报告某插件是否已被标记为"可拉起/已拉起"。
//
// ★ 与 Launched()（返回列表）分开：本方法用于**单个查询**，
// 供两段式调度的补偿逻辑判断"它到条件了但可能还没真正加载"。
func (s *Scheduler) LaunchedFlag(name string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.launched[name]
}

// SetLaunchErrorHandler 注册拉起失败的观察者（诊断用）。
func (s *Scheduler) SetLaunchErrorHandler(fn func(PluginSpec, error)) {
	s.mu.Lock()
	s.onLaunchErr = fn
	s.mu.Unlock()
}

// Submit 提交一批插件声明。返回**立即可拉起**的那些插件名。
//
// 未达条件的进挂起队列，等 OnUpdate 触发时再评估。
func (s *Scheduler) Submit(specs []PluginSpec) []string {
	// ★ 先校验：声明的设施名必须是**内核注册过的**。
	//
	// ## 为什么这条校验必须存在（修一个"静默永不启动"的缺口）
	//
	// 插件声明 `requires_facilities` 是字符串。拼错时原行为是：
	// 永远等不到 ⇒ **永不拉起**，且**无任何报错**、无超时兜底。
	// 那是本仓反复吃亏的"静默降级"形态（系统看着正常，某插件就是不工作）。
	//
	// 有了这条校验，拼错在**提交那一刻**就报出来，并列出全部已知设施名
	// 供对照（"拼错了"当场可修）。
	//
	// ★ 为什么在这里而不在 canLaunch：canLaunch 是**每轮状态变化**都会调的
	// 热路径，在那里报错会重复刷屏；而"名字不存在"是**一次性**的静态错误，
	// 提交时检查一次即可。
	s.mu.Lock()
	var rejected []PluginSpec
	var valid []PluginSpec
	for _, p := range specs {
		if bad := s.unknownFacilities(p); len(bad) > 0 {
			rejected = append(rejected, p)
			continue
		}
		valid = append(valid, p)
	}
	s.pending = append(s.pending, valid...)
	errFn := s.onLaunchErr
	s.mu.Unlock()

	// 报错在锁外（回调可能反查调度器状态 ⇒ 持锁会死锁）
	for _, p := range rejected {
		bad := s.unknownFacilities(p)
		err := fmt.Errorf("插件 %s 声明的设施名内核**从未注册过**: %v —— "+
			"这通常是拼写错误。内核已知的设施名: %v。"+
			"（若不校验，该插件会**永远等不到就绪**而静默不启动）",
			p.Name, bad, s.bus.KnownFacilities())
		if errFn != nil {
			errFn(p, err)
		}
	}
	return s.evaluate()
}

// unknownFacilities 返回插件声明里**内核从未注册过**的设施名（拼错检测）。
func (s *Scheduler) unknownFacilities(p PluginSpec) []string {
	var bad []string
	for _, f := range p.RequiresFacilities {
		if !s.bus.Known(f) {
			bad = append(bad, f)
		}
	}
	sort.Strings(bad)
	return bad
}

// Pending 返回仍在挂起的插件名（诊断/判据用）。
func (s *Scheduler) Pending() []string {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]string, 0, len(s.pending))
	for _, p := range s.pending {
		out = append(out, p.Name)
	}
	sort.Strings(out)
	return out
}

// Launched 返回已拉起的插件名（判据用）。
func (s *Scheduler) Launched() []string {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]string, 0, len(s.launched))
	for k := range s.launched {
		out = append(out, k)
	}
	sort.Strings(out)
	return out
}

// evaluate 挑出当前可拉起的插件并拉起它们。
//
// 返回本次真正拉起的插件名。
func (s *Scheduler) evaluate() []string {
	s.mu.Lock()
	// 挑出可拉起的 + 留下的。
	//
	// ★★ 必须**就地去重**：pending 里可能有同一插件的多份副本，
	// 因为并发 Submit 各自把同一批插件追加进来。
	//
	// 实测（压力测试 TestStressSchedulerConcurrentStateChange）：
	// 8 个 goroutine 并发提交 200 个插件 + 内核并发标记就绪
	// ⇒ 同一插件在**同一次** evaluate 的 ready 列表里出现多次
	// ⇒ `launched` 查重救不了（它只在**跨轮**生效）⇒ 被拉起 203 次。
	//
	// 后果：插件被重复加载（spawn 两个进程 / 工具注册冲突）。
	//
	// 去重用 seen 集合（同一次 evaluate 内），并把它记进 launched
	// 以确保跨轮也幂等 —— 两者都需要。
	var ready []PluginSpec
	var still []PluginSpec
	seen := map[string]bool{}
	for _, p := range s.pending {
		if s.launched[p.Name] || seen[p.Name] {
			continue // 已拉起（跨轮）或本批已选中（同批）⇒ 丢弃
		}
		if s.canLaunch(p) {
			seen[p.Name] = true
			ready = append(ready, p)
		} else {
			still = append(still, p)
		}
	}
	s.pending = still
	for _, p := range ready {
		s.launched[p.Name] = true
	}
	launchFn := s.launch
	errFn := s.onLaunchErr
	s.mu.Unlock()

	// ★ 在**不持锁**的情况下执行拉起动作：
	//   拉起可能很慢（spawn 进程、握手），持锁会让其他插件全卡住；
	//   且拉起过程可能回调调度器（如插件注册新插件）⇒ 死锁。
	var names []string
	for _, p := range ready {
		names = append(names, p.Name)
		if launchFn == nil {
			continue
		}
		if err := launchFn(p); err != nil && errFn != nil {
			errFn(p, err)
		}
	}
	sort.Strings(names)
	return names
}

// canLaunch 判断单个插件当前是否可拉起。
func (s *Scheduler) canLaunch(p PluginSpec) bool {
	// ① 不伴随启动：依赖齐了也**不**自动拉起（§2.6.6）
	if !p.AutoStart {
		return false
	}
	// ② 未声明依赖 ⇒ 等内核**完全**初始化（§2.6.5，存量插件走这条）
	if len(p.RequiresFacilities) == 0 {
		return s.bus.KernelComplete()
	}
	// ③ 声明了依赖 ⇒ 全就绪即拉起
	return s.bus.AllReady(p.RequiresFacilities)
}

// WaitString 渲染"还差什么"（供日志：让等待可解释，而不是静默卡住）。
func (s *Scheduler) WaitString(p PluginSpec) string {
	if !p.AutoStart {
		return "不伴随启动（等显式启用）"
	}
	if len(p.RequiresFacilities) == 0 {
		return "等内核完全初始化"
	}
	// AllReady 需要 *SyncBus；这里只做只读渲染，用 Snapshot 判断
	snap := s.bus.Snapshot()
	var missing []string
	for _, f := range p.RequiresFacilities {
		if !snap[f] {
			missing = append(missing, f)
		}
	}
	if len(missing) == 0 {
		return "依赖已就绪"
	}
	sort.Strings(missing)
	return fmt.Sprintf("还差: %s", strings.Join(missing, ","))
}