jianf/HomeAgent
1
0
Fork 0
mirror of https://gitcode.com/JianFeeeee/HomeAgent.git synced 2026-11-01 06:47:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge commit 'bf44654' into ci/boot-graph-tests

Browse Source
This commit is contained in:
JianFeeeee
2026-10-25 07:57:21 +08:00
parent 599252f9b0 bf44654926
commit f9ebdd5c74
4 changed files with 631 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
internal/plugin/dynamic_proc.go
View File

@ -125,8 +125,8 @@ func (r *Registry) ensureProcHost() (*proc.Host, error) {
r.procHost = host
// 事件环适配层：Bus 发布 → 写 EvtRing slot → eventfd 通知子进程
if r.evBus != nil {
er := NewEventRing(host.EvtRing(), int(host.Evtfd().Fd()), r.evBus)
if bus := r.eventBusOrNil(); bus != nil {
er := NewEventRing(host.EvtRing(), int(host.Evtfd().Fd()), bus)
host.SetEvtSubscriber(er)
log.Printf("[plugin] 事件环已创建（Bus → EvtRing → eventfd）")
}
@ -183,8 +183,8 @@ func (r *Registry) onProcCrash(name string, err error) {
r.mu.Unlock()
// 2) 事件通知（webui/诊断插件可订阅）。
if r.evBus != nil {
r.evBus.Publish(&events.Event{
if bus := r.eventBusOrNil(); bus != nil {
bus.Publish(&events.Event{
Type: events.EventSystem,
Source: "plugin",
Payload: map[string]interface{}{

444
internal/plugin/registry.go
View File

@ -218,18 +218,66 @@ func NewRegistry() *Registry {
}
}
func (r *Registry) SetIOManager(iom *agentIO.IOManager) { r.iom = iom }
func (r *Registry) SetEventBus(evBus *events.Bus) { r.evBus = evBus }
func (r *Registry) SetMemory(memDB *memory.GraphDB) { r.memDB = memDB }
func (r *Registry) SetTextMemory(tm *text.Memory) { r.textMem = tm }
func (r *Registry) SetDocStore(ds *doc.Store) { r.docStore = ds }
func (r *Registry) SetKnowledge(ks *knowledge.Store) { r.ks = ks }
func (r *Registry) SetProviderManager(mgr *agentAPI.ProviderManager) { r.mgr = mgr }
func (r *Registry) SetConfigRegistry(cfgReg *internalConfig.ConfigRegistry) { r.cfgReg = cfgReg }
func (r *Registry) SetPluginDir(dir string) { r.plgDir = dir }
func (r *Registry) SetDataDir(dir string) { r.dataDir = dir }
func (r *Registry) SetLuaVM(vm *luaVM.VM) { r.lua = vm }
func (r *Registry) SetBaseAPIKey(key string) { r.baseKey = key }
func (r *Registry) SetIOManager(iom *agentIO.IOManager) {
r.mu.Lock()
defer r.mu.Unlock()
r.iom = iom
}
func (r *Registry) SetEventBus(evBus *events.Bus) {
r.mu.Lock()
defer r.mu.Unlock()
r.evBus = evBus
}
func (r *Registry) SetMemory(memDB *memory.GraphDB) {
r.mu.Lock()
defer r.mu.Unlock()
r.memDB = memDB
}
func (r *Registry) SetTextMemory(tm *text.Memory) {
r.mu.Lock()
defer r.mu.Unlock()
r.textMem = tm
}
func (r *Registry) SetDocStore(ds *doc.Store) {
r.mu.Lock()
defer r.mu.Unlock()
r.docStore = ds
}
func (r *Registry) SetKnowledge(ks *knowledge.Store) {
r.mu.Lock()
defer r.mu.Unlock()
r.ks = ks
}
func (r *Registry) SetProviderManager(mgr *agentAPI.ProviderManager) {
r.mu.Lock()
defer r.mu.Unlock()
r.mgr = mgr
}
func (r *Registry) SetConfigRegistry(cfgReg *internalConfig.ConfigRegistry) {
r.mu.Lock()
defer r.mu.Unlock()
r.cfgReg = cfgReg
}
func (r *Registry) SetPluginDir(dir string) {
r.mu.Lock()
defer r.mu.Unlock()
r.plgDir = dir
}
func (r *Registry) SetDataDir(dir string) {
r.mu.Lock()
defer r.mu.Unlock()
r.dataDir = dir
}
func (r *Registry) SetLuaVM(vm *luaVM.VM) {
r.mu.Lock()
defer r.mu.Unlock()
r.lua = vm
}
func (r *Registry) SetBaseAPIKey(key string) {
r.mu.Lock()
defer r.mu.Unlock()
r.baseKey = key
}
// kernelControlAllowed 决定某个内置插件能否拿到内核重启/关机通道。
//
@ -250,38 +298,95 @@ var kernelControlAllowed = map[string]bool{
// ★ 为何用 setter 而不是构造参数：internal/plugin 不能 import cmd/homed
//
// （那是反向依赖，且 cmd 包不可被 import）。用接口 + setter 打破方向。
func (r *Registry) SetKernelControl(k sdk.KernelControlAPI) { r.kernelCtl = k }
func (r *Registry) SetKernelControl(k sdk.KernelControlAPI) {
r.mu.Lock()
defer r.mu.Unlock()
r.kernelCtl = k
}
// kernelControlFor 按插件名决定是否授予内核控制通道。
//
// 未在白名单里的插件拿到 nil —— SDK 侧 KernelControl() 返回 nil，
// 调用方（插件自己）必须判 nil 并如实告知用户"当前环境不支持"。
func (r *Registry) kernelControlFor(name string) sdk.KernelControlAPI {
if r.kernelCtl == nil || !kernelControlAllowed[name] {
return nil
}
return r.kernelCtl
r.mu.RLock()
ctl := r.kernelCtl
r.mu.RUnlock()
return kernelControlForCtl(ctl, name)
}
func (r *Registry) SetToolRegistrar(fn sdk.ToolRegistrar) { r.regTool = fn }
func (r *Registry) SetStageRegistrar(fn sdk.StageRegistrar) { r.regStage = fn }
func (r *Registry) SetAPIRegistrar(fn sdk.APIRegistrar) { r.regAPI = fn }
func (r *Registry) SetToolCleaner(tc PluginToolCleaner) { r.toolCleaner = tc }
func (r *Registry) SetStatusProvider(sp sdk.StatusAPI) { r.status = sp }
func (r *Registry) SetSupervisor(sup sdk.SupervisorAPI) { r.sup = sup }
func (r *Registry) SetTracker(trk *tracker.Tracker) { r.trk = trk }
func (r *Registry) SetConfig(cfg *types.Config) { r.cfg = cfg }
func (r *Registry) SetStageHost(sh sdk.ToolSource) { r.stageHost = sh }
// kernelControlForCtl 是 kernelControlFor 的**无锁内核**：给定已快照的 ctl 决定授权。
//
// 拆出来的原因：buildSDK 要在**一次** RLock 里取走整组依赖（见 snapshotDeps），
// 不能再回头单独取锁 —— RWMutex 不可重入，持锁再 RLock 在写者排队时会自锁。
func kernelControlForCtl(ctl sdk.KernelControlAPI, name string) sdk.KernelControlAPI {
if ctl == nil || !kernelControlAllowed[name] {
return nil
}
return ctl
}
func (r *Registry) SetToolRegistrar(fn sdk.ToolRegistrar) {
r.mu.Lock()
defer r.mu.Unlock()
r.regTool = fn
}
func (r *Registry) SetStageRegistrar(fn sdk.StageRegistrar) {
r.mu.Lock()
defer r.mu.Unlock()
r.regStage = fn
}
func (r *Registry) SetAPIRegistrar(fn sdk.APIRegistrar) {
r.mu.Lock()
defer r.mu.Unlock()
r.regAPI = fn
}
func (r *Registry) SetToolCleaner(tc PluginToolCleaner) {
r.mu.Lock()
defer r.mu.Unlock()
r.toolCleaner = tc
}
func (r *Registry) SetStatusProvider(sp sdk.StatusAPI) {
r.mu.Lock()
defer r.mu.Unlock()
r.status = sp
}
func (r *Registry) SetSupervisor(sup sdk.SupervisorAPI) {
r.mu.Lock()
defer r.mu.Unlock()
r.sup = sup
}
func (r *Registry) SetTracker(trk *tracker.Tracker) {
r.mu.Lock()
defer r.mu.Unlock()
r.trk = trk
}
func (r *Registry) SetConfig(cfg *types.Config) {
r.mu.Lock()
defer r.mu.Unlock()
r.cfg = cfg
}
func (r *Registry) SetStageHost(sh sdk.ToolSource) {
r.mu.Lock()
defer r.mu.Unlock()
r.stageHost = sh
}
// SetDeviceAuthQuery 注入"设备是否已授权"的查询（D4）。
//
// 由 bootstrap 在 agent 构造完成后调用。注入后，ToolAPI.CanUse 才会
// 对未授权设备返回 false —— 此前 ToolAPI 路径**完全不过授权闸**。
func (r *Registry) SetDeviceAuthQuery(fn func(deviceID string) bool) {
r.mu.Lock()
r.deviceAuth = fn
r.mu.Unlock()
sdk.SetDeviceAuthQuery(fn)
}
func (r *Registry) SetIndexer(idx *memory.Indexer) { r.idx = idx }
func (r *Registry) SetIndexer(idx *memory.Indexer) {
r.mu.Lock()
defer r.mu.Unlock()
r.idx = idx
}
// SetLoadAllowlist 限制 Load 仅装载指定插件名（failback 受限启动用）。
// 空/未设置 = 装载全部。违反白名单的插件（含已注册工厂）一律跳过。
@ -334,45 +439,168 @@ func (d *channelDevice) Execute(tool string, args map[string]interface{}) (inter
}
func (d *channelDevice) ChannelDef() agentIO.ChannelDef { return d.chDef }
func (r *Registry) buildSDK(name string) *sdk.PluginSDK {
// 插件自己的安装目录：rawfile 资源与可写工作目录都挂在它下面。
pluginDir := filepath.Join(r.plgDir, name)
// sdkDeps 是 buildSDK 所需依赖的**一致性快照**（KERNEL-REGISTRY-SETLOCK）。
//
// 为什么要有这层：这批字段由 Set* 在启动图里注入，读侧若逐个裸读
// （原来的写法），每一处都是独立的竞争点；而且逐个"读一次锁一次"会让
// 同一次 buildSDK 里拿到**互相不一致**的组合（比如 lua 已是新的、cfg 还是旧的）。
// 一次 RLock 取齐既消灭竞争，也保证同一次 SDK 装配看到的是同一个状态。
type sdkDeps struct {
plgDir string
dataDir string
cfgReg *internalConfig.ConfigRegistry
iom *agentIO.IOManager
evBus *events.Bus
memDB *memory.GraphDB
textMem *text.Memory
docStore *doc.Store
ks *knowledge.Store
mgr *agentAPI.ProviderManager
lua *luaVM.VM
baseKey string
regTool sdk.ToolRegistrar
regStage sdk.StageRegistrar
regAPI sdk.APIRegistrar
status sdk.StatusAPI
sup sdk.SupervisorAPI
trk *tracker.Tracker
cfg *types.Config
stageHost sdk.ToolSource
idx *memory.Indexer
kernelCtl sdk.KernelControlAPI
}
sett := sdk.NewSettings(name, r.cfgReg)
// snapshotDeps 在一次 RLock 内取走 buildSDK 需要的全部依赖。
func (r *Registry) snapshotDeps() sdkDeps {
r.mu.RLock()
defer r.mu.RUnlock()
return sdkDeps{
plgDir: r.plgDir,
dataDir: r.dataDir,
cfgReg: r.cfgReg,
iom: r.iom,
evBus: r.evBus,
memDB: r.memDB,
textMem: r.textMem,
docStore: r.docStore,
ks: r.ks,
mgr: r.mgr,
lua: r.lua,
baseKey: r.baseKey,
regTool: r.regTool,
regStage: r.regStage,
regAPI: r.regAPI,
status: r.status,
sup: r.sup,
trk: r.trk,
cfg: r.cfg,
stageHost: r.stageHost,
idx: r.idx,
kernelCtl: r.kernelCtl,
}
}
// ioManagerOrNil 返回当前 IOManager 快照（无则 nil）。
//
// 给 buildSDK 里的**闭包**用：闭包在插件 Start 期间才被调用，那时不能持锁
// （会跨用户代码），所以每次都取一次新快照 —— 语义与原来的 r.iom 裸读一致，
// 只是变成有序读。
func (r *Registry) ioManagerOrNil() *agentIO.IOManager {
r.mu.RLock()
iom := r.iom
r.mu.RUnlock()
return iom
}
// pluginDirOrEmpty / cfgRegistryOrNil / toolCleanerOrNil / stageHostOrNil
// 是其余读点的同款访问器。集中在这里，避免各处自己拼锁走样。
func (r *Registry) pluginDirOrEmpty() string {
r.mu.RLock()
dir := r.plgDir
r.mu.RUnlock()
return dir
}
func (r *Registry) cfgRegistryOrNil() *internalConfig.ConfigRegistry {
r.mu.RLock()
cfgReg := r.cfgReg
r.mu.RUnlock()
return cfgReg
}
func (r *Registry) toolCleanerOrNil() PluginToolCleaner {
r.mu.RLock()
tc := r.toolCleaner
r.mu.RUnlock()
return tc
}
func (r *Registry) eventBusOrNil() *events.Bus {
r.mu.RLock()
bus := r.evBus
r.mu.RUnlock()
return bus
}
func (r *Registry) stageHostOrNil() sdk.ToolSource {
r.mu.RLock()
sh := r.stageHost
r.mu.RUnlock()
return sh
}
func (r *Registry) buildSDK(name string) *sdk.PluginSDK {
// ★ 依赖快照（2026-10-24，KERNEL-REGISTRY-SETLOCK）。
//
// 为什么必须**一次取齐**而不是用 r.xxx 逐个裸读：
// 这批字段由 Set* 在启动图第 4 层注入，而 buildSDK 在插件加载时同理并发执行；
// 每个 r.xxx 各自裸读 = 每个都是一处数据竞争（实测 4% 概率报 7 条）。
//
// 为什么不在整个 buildSDK 外面套一把 RLock：下面的闭包（regOutput 等）
// 只在**插件 Start 期间**才被调用，而 Start 本身可能在持锁区外跑很久；
// 持着 registry 读锁跨用户代码要冒死锁风险（插件回调进内核注册面）。
// 所以取到快照就立刻放锁，闭包里再用访问器晚绑定。
deps := r.snapshotDeps()
// 插件自己的安装目录：rawfile 资源与可写工作目录都挂在它下面。
pluginDir := filepath.Join(deps.plgDir, name)
sett := sdk.NewSettings(name, deps.cfgReg)
if sd, ok := sett.(interface{ SetDataDir(string) }); ok {
// 插件专属数据目录：<data>/plugin_data/<name>，保证存在
dir := filepath.Join(r.dataDir, "plugin_data", name)
dir := filepath.Join(deps.dataDir, "plugin_data", name)
os.MkdirAll(dir, 0755)
sd.SetDataDir(dir)
}
regTool := r.regTool
regTool := deps.regTool
if regTool == nil {
regTool = func(toolName string, def sdk.ToolDef, handler sdk.ToolHandler) error {
return nil
}
}
regStage := r.regStage
regStage := deps.regStage
if regStage == nil {
regStage = func(stage sdk.Stage, handler sdk.StageHandler) {}
}
// stage handler 注册时带上归属插件名，使卸载/崩溃时能成组摘除。
// StageHost 实现了 RegisterStageFor；其他实现（测试替身）退回无归属注册。
if h, ok := r.stageRegistrarFor(); ok {
if h, ok := stageRegistrarForSource(deps.stageHost); ok {
regStage = func(stage sdk.Stage, handler sdk.StageHandler) {
h(name, stage, handler)
}
}
regAPI := r.regAPI
regAPI := deps.regAPI
if regAPI == nil {
regAPI = func(name string) error { return nil }
}
regOutput := func(chName string, caps int, desc string, def sdk.ChannelDef, handler sdk.ToolHandler) error {
if r.iom == nil {
iom := r.ioManagerOrNil()
if iom == nil {
return nil
}
if err := r.iom.RegisterDevice(&channelDevice{
if err := iom.RegisterDevice(&channelDevice{
name: chName,
caps: agentIO.OutputCapability(caps),
desc: desc,
@ -392,43 +620,45 @@ func (r *Registry) buildSDK(name string) *sdk.PluginSDK {
// 按此刻的状态判会对它误报（实测：把 qq 报成"只声明了输出通道"）。
// 真正该问的问题是"插件 Start 结束后，这个出站通道有没有对应的入站声明" ——
// 那在 load 完成后统一判（见 warnOutputOnlyChannels）。
if _, ok := r.iom.LookupInputChannel(chName); !ok {
_ = r.iom.RegisterInputChannelFrom(name, chName, agentIO.ChannelDef(def))
if _, ok := iom.LookupInputChannel(chName); !ok {
_ = iom.RegisterInputChannelFrom(name, chName, agentIO.ChannelDef(def))
}
r.noteChannel(name, chName, true)
return nil
}
regOutputUnreg := func(chName string) error {
if r.iom == nil {
iom := r.ioManagerOrNil()
if iom == nil {
return nil
}
r.iom.UnregisterDevice(chName)
iom.UnregisterDevice(chName)
r.forgetChannel(name, chName)
return nil
}
regInput := func(chName string, def sdk.ChannelDef) error {
if r.iom == nil {
iom := r.ioManagerOrNil()
if iom == nil {
return nil
}
// inputch 是最基本的输入路由单位：登记**归属插件**，便于父 agent 看清
// "哪个插件的哪个 inputch 划给了谁"（一个插件可注册多个 inputch）。
_ = r.iom.RegisterInputChannelFrom(name, chName, agentIO.ChannelDef(def))
_ = iom.RegisterInputChannelFrom(name, chName, agentIO.ChannelDef(def))
r.noteChannel(name, chName, false)
return nil
}
return sdk.New(name, sdk.SDKConfig{
IOManager: r.iom,
EventBus: r.evBus,
IOManager: deps.iom,
EventBus: deps.evBus,
// 块自带负载（媒体字节在块的 Data），不再需要独立的媒体存储句柄。
// 插件名仅用于日志溯源（哪个插件写的块）。
Memory: sdk.NewGraphMemoryWithPlugin(name, r.memDB),
TextMemory: sdk.NewTextMemoryWithPlugin(name, r.textMem),
DocMemory: sdk.NewDocMemoryWithPlugin(name, r.docStore),
Knowledge: sdk.NewKnowledge(r.ks),
LLM: sdk.NewLLM(r.mgr, r.cfgReg, r.lua, r.baseKey),
Memory: sdk.NewGraphMemoryWithPlugin(name, deps.memDB),
TextMemory: sdk.NewTextMemoryWithPlugin(name, deps.textMem),
DocMemory: sdk.NewDocMemoryWithPlugin(name, deps.docStore),
Knowledge: sdk.NewKnowledge(deps.ks),
LLM: sdk.NewLLM(deps.mgr, deps.cfgReg, deps.lua, deps.baseKey),
Settings: sett,
// ★ 变量库：桥接到进程级 security.Global()（**不另建存储**）。
// 内核未启用安全模块时它 Available()==false（不是 nil）。
@ -444,21 +674,21 @@ func (r *Registry) buildSDK(name string) *sdk.PluginSDK {
RegOutputUnreg: regOutputUnreg,
RegInput: regInput,
PluginMgr: r,
KernelControl: r.kernelControlFor(name),
KernelControl: kernelControlForCtl(deps.kernelCtl, name),
Status: r.status,
Supervisor: r.sup,
Adapter: sdk.NewAdapter(r.lua),
Tracker: r.trk,
Config: sdk.NewConfig(r.cfg),
Tool: sdk.NewTool(r.stageHost, r.iom),
Indexer: sdk.NewIndexer(r.idx),
Status: deps.status,
Supervisor: deps.sup,
Adapter: sdk.NewAdapter(deps.lua),
Tracker: deps.trk,
Config: sdk.NewConfig(deps.cfg),
Tool: sdk.NewTool(deps.stageHost, deps.iom),
Indexer: sdk.NewIndexer(deps.idx),
})
}
func (r *Registry) Load(dir string) error {
if dir == "" {
dir = r.plgDir
dir = r.pluginDirOrEmpty()
}
os.MkdirAll(dir, 0755)
@ -549,10 +779,11 @@ func (r *Registry) Load(dir string) error {
}
func (r *Registry) isDisabled(name string) bool {
if r.cfgReg == nil {
cfgReg := r.cfgRegistryOrNil()
if cfgReg == nil {
return false
}
return r.cfgReg.IsPluginDisabled(name)
return cfgReg.IsPluginDisabled(name)
}
// pluginEntryHash 计算插件入口文件的 SHA256，用于增量重载对比。
@ -670,10 +901,18 @@ func (r *Registry) runStopHandlers(name string) {
// 以 sdk.ToolSource 存在 r.stageHost 上。能取到 RegisterStageFor 时就直接用它，
// 否则退回无归属注册（测试替身、旧集成方）。
func (r *Registry) stageRegistrarFor() (func(plugin string, stage sdk.Stage, handler sdk.StageHandler), bool) {
if r.stageHost == nil {
r.mu.RLock()
sh := r.stageHost
r.mu.RUnlock()
return stageRegistrarForSource(sh)
}
// stageRegistrarForSource 是 stageRegistrarFor 的无锁内核（同样为了快照后复用）。
func stageRegistrarForSource(sh sdk.ToolSource) (func(plugin string, stage sdk.Stage, handler sdk.StageHandler), bool) {
if sh == nil {
return nil, false
}
if h, ok := r.stageHost.(interface {
if h, ok := sh.(interface {
RegisterStageFor(plugin string, stage sdk.Stage, handler sdk.StageHandler)
}); ok {
return h.RegisterStageFor, true
@ -756,16 +995,20 @@ func (r *Registry) releasePluginChannels(plugin string) []string {
set := r.pluginChannels[plugin]
delete(r.pluginChannels, plugin)
r.channelsMu.Unlock()
if set == nil || r.iom == nil {
if set == nil {
return nil
}
iom := r.ioManagerOrNil()
if iom == nil {
return nil
}
var released []string
for ch := range set.outputs {
r.iom.UnregisterDevice(ch)
iom.UnregisterDevice(ch)
released = append(released, ch)
}
for ch := range set.inputs {
r.iom.UnregisterInputChannel(ch)
iom.UnregisterInputChannel(ch)
if !set.outputs[ch] {
released = append(released, ch)
}
@ -780,9 +1023,9 @@ func (r *Registry) releasePluginChannels(plugin string) []string {
// StopAndUnload）只调 UnregisterPluginTools，漏了 stage 与通道两项，
// 子进程崩溃路径更是三项都没做。
func (r *Registry) detachPlugin(name string) {
if r.toolCleaner != nil {
r.toolCleaner.UnregisterPluginTools(name)
if sc, ok := r.toolCleaner.(PluginStageCleaner); ok {
if tc := r.toolCleanerOrNil(); tc != nil {
tc.UnregisterPluginTools(name)
if sc, ok := tc.(PluginStageCleaner); ok {
if n := sc.UnregisterPluginStages(name); n > 0 {
log.Printf("[plugin] %s: 摘除 %d 个 stage handler", name, n)
}
@ -873,7 +1116,7 @@ func (r *Registry) StopAll() {
func (r *Registry) Reload(dir string) (string, error) {
if dir == "" {
dir = r.plgDir
dir = r.pluginDirOrEmpty()
}
// 增量重载：扫描插件目录，对比入口文件 hash，仅 Stop+重载有变更的插件。
// 未变更插件保持运行，避免 plgreload 触发全量 StopAll+Load 导致所有插件重复加载
@ -949,7 +1192,7 @@ func (r *Registry) LoadOneDir(plgDir, name string) error {
// 所有插件（含声明了依赖的）立刻加载，调度器就无事可做了。
func (r *Registry) LoadSkipping(dir string, skip map[string]bool) error {
if dir == "" {
dir = r.plgDir
dir = r.pluginDirOrEmpty()
}
os.MkdirAll(dir, 0755)
@ -1033,7 +1276,7 @@ func (r *Registry) LoadSkipping(dir string, skip map[string]bool) error {
}
func (r *Registry) ReloadOne(name string) error {
plgDir := filepath.Join(r.plgDir, name)
plgDir := filepath.Join(r.pluginDirOrEmpty(), name)
r.mu.Lock()
var removed sdk.Plugin
@ -1115,13 +1358,13 @@ func (r *Registry) IsDisabled(name string) bool {
}
func (r *Registry) Enable(name string) error {
if r.cfgReg != nil {
r.cfgReg.RemoveDisabledPlugin(name)
if cfgReg := r.cfgRegistryOrNil(); cfgReg != nil {
cfgReg.RemoveDisabledPlugin(name)
}
r.mu.Lock()
delete(r.knownDisabled, name)
r.mu.Unlock()
plgDir := filepath.Join(r.plgDir, name)
plgDir := filepath.Join(r.pluginDirOrEmpty(), name)
if r.loadOne(plgDir, name) {
log.Printf("[plugin] enabled: %s", name)
return nil
@ -1151,8 +1394,8 @@ func (r *Registry) Disable(name string) error {
r.detachPlugin(name)
if r.cfgReg != nil {
r.cfgReg.AddDisabledPlugin(name, "system")
if cfgReg := r.cfgRegistryOrNil(); cfgReg != nil {
cfgReg.AddDisabledPlugin(name, "system")
}
log.Printf("[plugin] disabled: %s", name)
return nil
@ -1197,8 +1440,8 @@ func (r *Registry) pluginInstalled(name string) bool {
if _, ok := globalFactories.Load(name); ok {
return true
}
if r.plgDir != "" {
if fi, err := os.Stat(filepath.Join(r.plgDir, name)); err == nil && fi.IsDir() {
if dir := r.pluginDirOrEmpty(); dir != "" {
if fi, err := os.Stat(filepath.Join(dir, name)); err == nil && fi.IsDir() {
return true
}
}
@ -1208,10 +1451,11 @@ func (r *Registry) pluginInstalled(name string) bool {
func (r *Registry) ListLoadedPlugins() []string { return r.List() }
func (r *Registry) ListDisabledPlugins() []sdk.DisabledPluginInfo {
if r.cfgReg == nil {
cfgReg := r.cfgRegistryOrNil()
if cfgReg == nil {
return nil
}
list, err := r.cfgReg.ListDisabledPlugins()
list, err := cfgReg.ListDisabledPlugins()
if err != nil {
return nil
}
@ -1238,9 +1482,9 @@ func (r *Registry) DisablePlugin(name, by string) error {
return fmt.Errorf("plugin %s not installed", name)
}
// Check not disabling self if running
if r.cfgReg != nil {
if cfgReg := r.cfgRegistryOrNil(); cfgReg != nil {
// If already disabled, no-op
if r.cfgReg.IsPluginDisabled(name) {
if cfgReg.IsPluginDisabled(name) {
return fmt.Errorf("plugin %s already disabled", name)
}
}
@ -1266,8 +1510,8 @@ func (r *Registry) DisablePlugin(name, by string) error {
r.detachPlugin(name)
if r.cfgReg != nil {
r.cfgReg.AddDisabledPlugin(name, by)
if cfgReg := r.cfgRegistryOrNil(); cfgReg != nil {
cfgReg.AddDisabledPlugin(name, by)
}
log.Printf("[plugin] disabled: %s (by %s)", name, by)
return nil
@ -1386,7 +1630,7 @@ func (r *Registry) StartPlugin(name string) error {
return nil
}
plgDir := filepath.Join(r.plgDir, name)
plgDir := filepath.Join(r.pluginDirOrEmpty(), name)
if r.loadOne(plgDir, name) {
if wasStopped {
log.Printf("[plugin] runtime started: %s（已清运行态关闭）", name)
@ -1477,16 +1721,16 @@ func (r *Registry) RemovePlugin(name string) error {
r.mu.Unlock()
r.detachPlugin(name)
if r.cfgReg != nil {
r.cfgReg.RemoveDisabledPlugin(name)
r.cfgReg.RemovePlugin(name)
if cfgReg := r.cfgRegistryOrNil(); cfgReg != nil {
cfgReg.RemoveDisabledPlugin(name)
cfgReg.RemovePlugin(name)
}
r.closeDynamic(removed)
log.Printf("[plugin] removed: %s", name)
return nil
}
func (r *Registry) ReloadPlugins() (string, error) { return r.Reload(r.plgDir) }
func (r *Registry) ReloadPlugins() (string, error) { return r.Reload(r.pluginDirOrEmpty()) }
// PluginRuntime 返回单个插件的运行期状态。
//
@ -1508,8 +1752,8 @@ func (r *Registry) PluginRuntime(name string) (sdk.PluginRuntimeInfo, bool) {
installed := loaded || hasFactory
var dirExists bool
if r.plgDir != "" {
if st, err := os.Stat(filepath.Join(r.plgDir, name)); err == nil && st.IsDir() {
if dir := r.pluginDirOrEmpty(); dir != "" {
if st, err := os.Stat(filepath.Join(dir, name)); err == nil && st.IsDir() {
dirExists = true
installed = true
}
@ -1531,7 +1775,7 @@ func (r *Registry) PluginRuntime(name string) (sdk.PluginRuntimeInfo, bool) {
// auto_start 以清单为准（plugin.json 显式 false 才是按需）。
// 目录不存在（纯内置）时保持缺省 true —— 内置插件本就随内核加载。
if dirExists {
if m, err := ReadManifest(filepath.Join(r.plgDir, name)); err == nil {
if m, err := ReadManifest(filepath.Join(r.pluginDirOrEmpty(), name)); err == nil {
info.AutoStart = m.ShouldAutoStart()
}
}
@ -1540,7 +1784,7 @@ func (r *Registry) PluginRuntime(name string) (sdk.PluginRuntimeInfo, bool) {
case hasFactory:
info.Channel = "builtin"
case dirExists:
info.Channel = detectEntryKind(filepath.Join(r.plgDir, name)).String()
info.Channel = detectEntryKind(filepath.Join(r.pluginDirOrEmpty(), name)).String()
}
// 子进程插件报真实 PID 与存活；其余形态与 Loaded 同值（无独立进程）。
@ -1555,8 +1799,8 @@ func (r *Registry) PluginRuntime(name string) (sdk.PluginRuntimeInfo, bool) {
info.Alive = loaded
}
if r.stageHost != nil {
for _, def := range r.stageHost.GetToolDefs() {
if sh := r.stageHostOrNil(); sh != nil {
for _, def := range sh.GetToolDefs() {
if def.Plugin == name {
info.Tools = append(info.Tools, def.Name)
}
@ -1601,8 +1845,8 @@ func (r *Registry) ListKnown() []string {
}
r.mu.RUnlock()
if r.plgDir != "" {
entries, _ := os.ReadDir(r.plgDir)
if dir := r.pluginDirOrEmpty(); dir != "" {
entries, _ := os.ReadDir(dir)
for _, e := range entries {
if e.IsDir() {
known[e.Name()] = true
@ -1639,7 +1883,7 @@ func (r *Registry) PluginMetas() map[string]sdk.PluginMeta {
}
func (r *Registry) PluginDir() string {
return r.plgDir
return r.pluginDirOrEmpty()
}
func (r *Registry) tryDynamic(plgDir, name string, config map[string]interface{}) (sdk.Plugin, error) {

260
internal/plugin/registry_setlock_test.go Normal file
View File

@ -0,0 +1,260 @@
package plugin
// KERNEL-REGISTRY-SETLOCK（2026-10-24）：Registry Set* 注入路径的**写-写竞争**判据。
//
// ## 为什么必须有这条判据
//
// Registry 上那批 Set*（SetLuaVM / SetBaseAPIKey / SetStageHost / SetIndexer …）
// 长期是单行无锁直写：`func (r *Registry) SetLuaVM(vm *luaVM.VM) { r.lua = vm }`。
// 结构体里明明有 r.mu 这个 RWMutex，但这批 setter 一次都没取过锁。
//
// 平时看不出来，因为注入通常发生在启动早期。内核模块化之后就不成立了：
// **两个 goroutine 会同时注入同一批字段** ——
//
// internal/kernelmod/agent/mod.go wirePluginSDK → SetLuaVM/SetBaseAPIKey/SetStageHost/…
// internal/kernelmod/pluginhost/mod.go injectBaseDeps → SetLuaVM/SetBaseAPIKey/SetStageHost/…
//
// 两者同属启动图第 4 层、并行执行，于是 r.lua / r.baseKey / r.stageHost / r.idx
// 被两个 goroutine 同时写。Go 内存模型下这是 data race；interface 字段尤其危险
// （type 与 data 两字不是同一轮写入 ⇒ 读到撕裂值，调用时直接崩）。
//
// ## 两条判据，一快一准
//
// 1. TestSettersConcurrentWriteRace —— **运行时**判据：并发 Set + 并发读，
// -race 下未加锁必然报 DATA RACE。这是正面证据（真跑真竞争）。
// 2. TestRegistrySettersTakeWriteLock —— **源码级**判据：逐个 setter 检查
// 函数体内必须出现 r.mu.Lock()。race 是概率性的，这条把"有人把锁删了"
// 这类回退 100% 挡在门口（变异实验里它是确定性命中的那一条）。
//
// ★ 两条都必须"活的"：把任一 setter 的锁去掉（换回直写），两条都要报红。
// ★ 本文件只依赖 Registry 的公开 setter 与 r.mu 本身，不引用任何内部访问器 ——
// 因此在**未修复**与**已修复**两版 registry.go 上都能编译，
// 才可能拿它当"修前红 / 修后绿"的**同一条**判据。
import (
"go/ast"
"go/parser"
"go/token"
"os"
"runtime"
"sync"
"testing"
agentIO "github.com/JianFeeeee/HologramAgent/internal/agent/io"
"github.com/JianFeeeee/HologramAgent/internal/lua"
"github.com/JianFeeeee/HologramAgent/internal/memory"
sdk "github.com/JianFeeeee/HologramAgent/internal/sdk"
)
// setlockRounds 是每个注入 goroutine 的轮数。
//
// 取值依据：单轮 6 次写 × 2 个写 goroutine，未加锁代码上 -race 稳定命中
// （实测连续多次运行均报红，见交付证据）；再往上只是拖长测试。
const setlockRounds = 300
// TestSettersConcurrentWriteRace 钉住 setter 之间的写-写竞争。
func TestSettersConcurrentWriteRace(t *testing.T) {
r := NewRegistry()
// 两组 goroutine 用**不同的值**写同一批字段：
// 写相同值容易被优化掉，不同值才能让 race detector 看到
// "读到组合了两次写入的状态"。
vmA, vmB := &lua.VM{}, &lua.VM{}
shA := &raceToolSource{tag: "A"}
shB := &raceToolSource{tag: "B"}
idxA := &memory.Indexer{}
idxB := &memory.Indexer{}
// IOManager 只作为被写入/读回的句柄，判据不调其方法。
iomA, iomB := &agentIO.IOManager{}, &agentIO.IOManager{}
// ★ 目录只取一次：放在循环里 t.TempDir() 会造出上万个临时目录。
const dirA, dirB = "dir-A", "dir-B"
inject := func(vm *lua.VM, key, dir string, sh sdk.ToolSource, idx *memory.Indexer, iom *agentIO.IOManager) {
for i := 0; i < setlockRounds; i++ {
// 与 wirePluginSDK / injectBaseDeps 交叠的核心项
r.SetLuaVM(vm)
r.SetBaseAPIKey(key)
r.SetStageHost(sh)
r.SetIndexer(idx)
// 同一批装配里相邻的另外两项
r.SetIOManager(iom)
r.SetPluginDir(dir)
}
}
var writers sync.WaitGroup
writers.Add(2)
go func() { defer writers.Done(); inject(vmA, "key-A", dirA, shA, idxA, iomA) }()
go func() { defer writers.Done(); inject(vmB, "key-B", dirB, shB, idxB, iomB) }()
// 读侧并行：真实内核里 buildSDK 读这些字段与注入是交织的。
// 读侧持 RLock（本次修复不动读侧）—— 同时验证"没有把并发读改成串行"。
stop := make(chan struct{})
var readers sync.WaitGroup
readers.Add(1)
go func() {
defer readers.Done()
for {
select {
case <-stop:
return
default:
vm, key, sh, idx := r.injectedSnapshot()
// 必须真的用掉读到的值，否则编译器可能把读优化掉，
// race detector 就看不到这次读。
if vm == nil && key != "" && sh != nil && idx == nil {
t.Error("setlock: 读到不可能的注入组合（读侧失去意义）")
return
}
}
}
}()
writers.Wait()
close(stop)
readers.Wait()
}
// injectedSnapshot 以 RLock 一次性读齐 4 个注入字段（模拟 buildSDK 读侧）。
func (r *Registry) injectedSnapshot() (*lua.VM, string, sdk.ToolSource, *memory.Indexer) {
r.mu.RLock()
defer r.mu.RUnlock()
return r.lua, r.baseKey, r.stageHost, r.idx
}
// raceToolSource 是 sdk.ToolSource 的最小实现（判据里只需要个非 nil 值）。
type raceToolSource struct{ tag string }
func (s *raceToolSource) GetToolDefs() []sdk.ToolDef { return nil }
func (s *raceToolSource) ToolDef(string) *sdk.ToolDef { return nil }
func (s *raceToolSource) ExecuteTool(string, map[string]interface{}) (interface{}, error) {
return s.tag, nil
}
// registrySetterNames 是需要持写锁的 setter 清单（与 registry.go 对应）。
// 新增 Set* 时必须同步登记 —— 源码判据按这份清单咬。
var registrySetterNames = []string{
"SetIOManager", "SetEventBus", "SetMemory", "SetTextMemory",
"SetDocStore", "SetKnowledge", "SetProviderManager", "SetConfigRegistry",
"SetPluginDir", "SetDataDir", "SetLuaVM", "SetBaseAPIKey",
"SetKernelControl", "SetToolRegistrar", "SetStageRegistrar", "SetAPIRegistrar",
"SetToolCleaner", "SetStatusProvider", "SetSupervisor", "SetTracker",
"SetConfig", "SetStageHost", "SetDeviceAuthQuery", "SetIndexer",
}
// TestRegistrySettersTakeWriteLock 是源码级判据：每个 Registry.Set* 函数体里
// 必须出现 r.mu.Lock()。
//
// 为什么需要它（与运行时判据并存）：
//
// race detector 依赖"这一次运行恰好交错到了"，是概率性的。
// 而"某人把锁删了"这类回退必须 100% 被挡在 CI 门口 —— 源码判据确定性命中。
func TestRegistrySettersTakeWriteLock(t *testing.T) {
_, thisFile, _, _ := runtime.Caller(0)
src, err := os.ReadFile(thisFile[:len(thisFile)-len("registry_setlock_test.go")] + "registry.go")
if err != nil {
t.Fatalf("读 registry.go 失败: %v", err)
}
fset := token.NewFileSet()
f, err := parser.ParseFile(fset, "registry.go", src, 0)
if err != nil {
t.Fatalf("解析 registry.go 失败: %v", err)
}
locked := map[string]bool{}
for _, decl := range f.Decls {
fd, ok := decl.(*ast.FuncDecl)
if !ok || fd.Recv == nil || fd.Name == nil || fd.Body == nil {
continue
}
name := fd.Name.Name
if len(name) < 4 || name[:3] != "Set" {
continue
}
// 函数体内是否出现 r.mu.Lock()。
//
// ★ 形状注意：`r.mu.Lock()` 的 CallExpr.Fun 是
// SelectorExpr{ Sel: "Lock", X: SelectorExpr{ Sel: "mu", X: Ident{"r"} } }
// 不是 X 直接为 Ident —— 按 x.Name == "r" 判会漏掉**全部**加锁的 setter，
// 判据变成"恒报红"，等于没判（本判据第一版就踩了这个坑）。
found := false
ast.Inspect(fd.Body, func(n ast.Node) bool {
call, ok := n.(*ast.CallExpr)
if !ok {
return true
}
sel, ok := call.Fun.(*ast.SelectorExpr)
if !ok || sel.Sel == nil || sel.Sel.Name != "Lock" {
return true
}
mu, ok := sel.X.(*ast.SelectorExpr)
if !ok || mu.Sel == nil || mu.Sel.Name != "mu" {
return true
}
if id, ok := mu.X.(*ast.Ident); ok && id.Name == "r" {
found = true
}
return true
})
if found {
locked[name] = true
}
}
for _, want := range registrySetterNames {
if !locked[want] {
t.Errorf("Registry.%s 未取写锁（r.mu.Lock）—— "+
"这会重新引入启动图第 4 层并行注入的 DATA RACE（KERNEL-REGISTRY-SETLOCK）。"+
"修法：函数体开头 r.mu.Lock(); defer r.mu.Unlock()", want)
}
}
}
// TestBuildSDKConcurrentWithSetters 钉住**真实读路径**与 Set* 的读写竞争。
//
// 为什么需要它（前两条判据覆盖不到的地方）：
//
// TestSettersConcurrentWriteRace 的读侧用的是本文件自己写的 injectedSnapshot
// ——它**自带 RLock**，所以只证明了"setter 之间不再互撞"，
// 完全测不到 registry.go 里 buildSDK / snapshotDeps 那条真实读路径。
// 把 snapshotDeps 的 RLock 删掉，前两条判据仍会全绿（实测过）——
// 那就是判据缺口：read 侧的回退没人管。
//
// 本判据走 buildSDK（插件加载时真实调用的函数），setter 并发写同一批字段：
// 读侧若失去 RLock，-race 必报，且报告栈会直接指向 registry.go 的读点。
func TestBuildSDKConcurrentWithSetters(t *testing.T) {
r := NewRegistry()
r.SetPluginDir(t.TempDir())
r.SetDataDir(t.TempDir())
const rounds = 200
var wg sync.WaitGroup
// 写侧：模拟 agent 与 pluginhost 两个模块并行注入。
wg.Add(2)
go func() {
defer wg.Done()
for i := 0; i < rounds; i++ {
r.SetLuaVM(nil)
r.SetConfig(nil)
}
}()
go func() {
defer wg.Done()
for i := 0; i < rounds; i++ {
r.SetStageHost(nil)
r.SetIndexer(nil)
}
}()
// 读侧：走真实路径 buildSDK（内部 snapshotDeps 取依赖快照）。
wg.Add(1)
go func() {
defer wg.Done()
for i := 0; i < rounds; i++ {
_ = r.buildSDK("setlock-probe")
}
}()
wg.Wait()
}