diff --git a/internal/kernelmod/agent/mod.go b/internal/kernelmod/agent/mod.go index 31e3a54a..b2a8984b 100644 --- a/internal/kernelmod/agent/mod.go +++ b/internal/kernelmod/agent/mod.go @@ -24,7 +24,6 @@ import ( "github.com/JianFeeeee/HologramAgent/internal/kernelboot" "github.com/JianFeeeee/HologramAgent/internal/kernelmod" "github.com/JianFeeeee/HologramAgent/internal/memory" - "github.com/JianFeeeee/HologramAgent/internal/supervisor" "github.com/JianFeeeee/HologramAgent/pkg/types" agentPkg "github.com/JianFeeeee/HologramAgent/internal/agent" @@ -246,15 +245,30 @@ func buildAgent(k *kernelboot.KernelState) *agentCore.Agent { return agent } -// wirePluginSDK 把内核依赖注入插件注册表(原 cmd/homed 的整块平移)。 +// wirePluginSDK 注入**只有 agent 才知道**的那两项(KERNEL-REGISTRY-SETLOCK 去重后)。 +// +// ## 为什么这里不再注入那 7 项底层依赖(2026-10-24 去重) +// +// 原先本函数与 pluginhost.injectBaseDeps 各自把 LuaVM/BaseAPIKey/Supervisor/ +// Tracker/Config/StageHost/Indexer 写一遍,而两个模块同处启动图第 4 层、 +// **并行执行** ⇒ 同一批字段被两个 goroutine 无锁写(实测 4% 概率报 DATA RACE)。 +// +// 判掉哪一侧:留 pluginhost,砍这边。理由是**必要性不对称**: +// +// · pluginhost 的注入是**必需**的 —— 它自己紧接着就 LoadSkipping 拉插件, +// 而它刻意 `Requires()` 不含 agent(核心收益:插件不必等 agent 那 35 秒)。 +// 砍掉它 ⇒ 第 4 层加载的插件拿到全 nil 依赖。 +// · agent 这边注入的这批值是**同一批 ks.* 字段的同一份值**,写第二次 +// 没有新增信息;砍掉它 ⇒ 语义不变(pluginhost 侧的 nil 守卫更严, +// Supervisor 只因 FacSupervisor 已在 Requires 里而恒非 nil)。 +// +// ## 注意:去重**不能**替代加锁 +// +// 去重后本函数仍有 SetStatusProvider / SetDeviceAuthQuery 两项要写,而 +// pluginhost 同时可能正在加载插件(buildSDK 读 status)⇒ 写读依然并发。 +// 所以 registry 侧 Set*/读侧的锁是**真正的修复**,去重只是消除了一处 +// 无意义的重复写。两件事都做,缺一不可。 func wirePluginSDK(k *kernelboot.KernelState) { - k.PluginRegistry.SetLuaVM(k.LuaVM) - k.PluginRegistry.SetBaseAPIKey(k.BaseAPIKey) - k.PluginRegistry.SetSupervisor(supervisor.NewSDKAdapter(k.Supervisor)) - k.PluginRegistry.SetTracker(k.Tracker) - k.PluginRegistry.SetConfig(k.Cfg) - k.PluginRegistry.SetStageHost(k.StageHost) - k.PluginRegistry.SetIndexer(k.Memory.Indexer) k.PluginRegistry.SetStatusProvider(k.Agent) } diff --git a/internal/plugin/dynamic_proc.go b/internal/plugin/dynamic_proc.go index 96f66969..986a02f0 100644 --- a/internal/plugin/dynamic_proc.go +++ b/internal/plugin/dynamic_proc.go @@ -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{}{ diff --git a/internal/plugin/registry.go b/internal/plugin/registry.go index 8b1d3d83..3b612800 100644 --- a/internal/plugin/registry.go +++ b/internal/plugin/registry.go @@ -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 { // 插件专属数据目录:/plugin_data/,保证存在 - 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) { diff --git a/internal/plugin/registry_setlock_test.go b/internal/plugin/registry_setlock_test.go new file mode 100644 index 00000000..38c3e83e --- /dev/null +++ b/internal/plugin/registry_setlock_test.go @@ -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() +}