mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-11-05 16:57:32 +00:00
refactor(kernel): 统一挂载点 kstate + 插件声明式系统提示词段
## 起因(生产故障)
A2A 分诊助手(驻留子)把 terminal_create 当一次性命令执行器:
terminal_create(command='cd /usr/lib/node_modules/@deepseek-ai && ls -d */ | head -60 && …')
→ read /dev/ptmx: input/output error
→ 再开 term_7、term_8、term_9 …
链路拆开四层:
① cmd 插件 auto_start=false ⇒ 工具表里没有 cmd_run
② 需要 cmd-tools skill 说明"先 plugin_start cmd"
③ 但 skill 索引没进那个 agent 的提示词
④ 因为 skillIndex 只是根 agent 的字段、由 bootstrap 手工接线
④ 才是总根。而它之所以必须手工接线:agent/core 读不到内核的共享数据。
## 环的精确形状(实测,非推测)
kernelboot 因为持有 KernelState 而依赖 19 个业务包:
kernelboot ──▶ agent/core ──▶ …
──▶ plugin
──▶ security
──▶ memory / knowledge / text / coldcontext / …
而 kernelboot.go / scheduler.go / funcmodule.go / syncbus.go / lifecycle.go
**一行业务代码都不需要** —— 依赖全在 kernelstate.go 这一个数据文件里。
后果是四个底层包各自造通道:
core → Agent 字段(每种 agent 构造点都要记得传)
plugin → 12 个 SetXxx
sdk → SDKConfig 传参
security → 包级单例 Global() / RuntimeGlobal() / ConfigCodec()
四条通道是同一个根因的四个表现。
## 本次做的事
### 1. 抽出 internal/kstate(断环)
kernelstate.go / syncbus.go / lifecycle.go 搬出 kernelboot:
internal/kstate 纯数据 + 唯一挂载点(叶子,谁都能 import)
internal/kernelboot 装配引擎,零业务包
挂载点 API:Mount / Get(nil 安全)/ MountForTest(返回 restore,并行测试隔离)。
验证:core 现在能 import kstate(此前 import cycle not allowed)。
### 2. StageHost / Agent 字段存 any
它们是 core 自己的类型(core 是环的一极,这是真环)。按用户裁定的方向:
> 是其他包在kstate挂共享内容啊
存 any,消费方断言回具体类型 —— 与本仓既有的 PluginCatalog /
PluginScheduler / UpdateCapabilities 同型。
★ 其余字段保持**具体类型**,没有一律改 any。判据是"它会不会反向依赖
agent/core":实测 plugin / security / sdk 都不依赖 core,所以它们作为
具体类型不构成环,而类型化字段让"拼错/取错类型"是编译错误(本结构体的
设计原则)。
### 3. 插件声明式系统提示词段
SDK 新增 SystemPromptAPI(见 submodule 提交)。内核侧:
kstate.State.PromptSections *promptsec.Registry ← 唯一事实来源
Agent.renderSystemPrompt() 每轮重建时读它
skillmgr 从"实现 SkillIndexProvider 让内核来问"改为"声明自己的段":
DeclarePromptSection{Section:"skillmgr.skills", Order:100, Volatile:true}
并挂到全部变更点(启动全扫后、移交事件后、load/unload/create/install/
enable/disable 后、Stop 时撤销)。
### 4. 内核自身的提示词段也走同一管理器
按用户裁定「把现在内核所有 systemprompt 的相关注入都接入到这个统一的
systemprompt 管理器上」,把 buildSystemPrompt 里硬拼的十几块
(人格卡/任务提示词/记忆说明/输出规则/工具目录…)改为声明式:
kernelSections() 当轮计算 → 与插件段按 Order 合并渲染。
顺序由 Order 常量决定,改结构变成调常量而不是挪代码块。
## 判据(均经变异验证)
TestCoreCanImportKstate 真编译探针;环重现则红(实测)
TestKstateDoesNotDependOnCore 静态版,指向具体字段
TestKstateIsLeaf kstate 不得反依赖上层
TestRTChildSeesPromptSections 子 agent 能看到插件段
TestResidentSeesPromptSections 生产实际中招的那一类;变异后红(实测)
TestRootAgentSeesPromptSections 确认没改坏根路径
TestNoPromptSectionsMount 未装配时不 panic、不凭空出现段
TestDeclareTwiceRendersOnce ★ 回应"会不会多份":声明 5 次渲染 1 份
TestSectionRemovalTakesEffect Stop 后段消失
TestRuntimeUpdateVisibleToExistingAgent 同一实例下一轮看到新内容
TestSectionOrderIsRespected Order 生效且稳定
## 验证
go build ./... PASS
go vet ./... PASS
gofmt -l internal/ cmd/ pkg/ 干净
go test ./... -count=1 63 包全过,0 FAIL
干净检出(git worktree) build + 4 包测试全绿
## 未做(见 kernelplan.md)
T3 security 三单例收敛(含 Store 复用 Codec 的惰性迁移)
T5 plugin Setter 收敛
T6 防腐判据(部分已随本次落地)
This commit is contained in:
@ -1,5 +1,7 @@
|
||||
package kernelboot
|
||||
|
||||
import "github.com/JianFeeeee/HologramAgent/internal/kstate"
|
||||
|
||||
// 把既有的内核 `init*` 函数**适配**成 Module,而不必重写它们的签名。
|
||||
//
|
||||
// ## 为什么不直接改 init* 的签名
|
||||
@ -20,7 +22,7 @@ package kernelboot
|
||||
// Name: "lua",
|
||||
// Requires: []string{"config"},
|
||||
// Provides: []string{"lua"},
|
||||
// Init: func(ks *kernelboot.KernelState) error {
|
||||
// Init: func(ks *kstate.State) error {
|
||||
// vm, closeFn := initLuaVM(ks.Cfg) // 依赖的产物:直接读字段
|
||||
// ks.LuaVM = vm // 自己的产物:写回字段
|
||||
// ks.AddCloser(closeFn) // 关闭函数统一登记
|
||||
@ -45,7 +47,7 @@ type FuncSpec struct {
|
||||
Requires []string
|
||||
Provides []string
|
||||
// Init 执行装配:从 ctx 取依赖、调既有函数、把产物 Put 回 ctx。
|
||||
Init func(ks *KernelState) error
|
||||
Init func(ks *kstate.State) error
|
||||
}
|
||||
|
||||
// funcModule 是 FuncSpec 的 Module 实现。
|
||||
@ -56,7 +58,7 @@ type funcModule struct {
|
||||
func (m *funcModule) Name() string { return m.spec.Name }
|
||||
func (m *funcModule) Provides() []string { return m.spec.Provides }
|
||||
func (m *funcModule) Requires() []string { return m.spec.Requires }
|
||||
func (m *funcModule) Init(ks *KernelState) error {
|
||||
func (m *funcModule) Init(ks *kstate.State) error {
|
||||
if m.spec.Init == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -36,6 +36,7 @@ package kernelboot
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/kstate"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
@ -60,7 +61,7 @@ type Module interface {
|
||||
// 模块从中**读**已就绪的依赖,把自己产出的公共数据**写**回它的字段。
|
||||
//
|
||||
// 返回错误表示装配失败(**必须传播,不得吞掉**)。
|
||||
Init(ks *KernelState) error
|
||||
Init(ks *kstate.State) error
|
||||
}
|
||||
|
||||
// Plan 是一次依赖解析的结果:分好层的调度计划。
|
||||
@ -278,7 +279,7 @@ type RunOptions struct {
|
||||
//
|
||||
// 为何由调用方传入而不是 Run 内部自建:产物在 Run 返回后**仍然要用**
|
||||
// (如 LuaVM、Memory 要一直活着),所以它必须比 Run 活得久。
|
||||
State *KernelState
|
||||
State *kstate.State
|
||||
|
||||
// Bus 是状态同步总线。非 nil 时,Run 会在**每个层边界**把该层模块贡献的
|
||||
// 设施标记为就绪(设计 §2.6.2)。
|
||||
@ -287,7 +288,7 @@ type RunOptions struct {
|
||||
// 若模块自己 MarkReady,顺序取决于**层内并发的完成顺序** ⇒ 不可复现;
|
||||
// 而且模块可能"忘记标记"(静默漏掉一个设施 ⇒ 插件永远等不到它就绪)。
|
||||
// 由 Run 在层边界统一写,既确定又不可能漏。
|
||||
Bus *SyncBus
|
||||
Bus *kstate.SyncBus
|
||||
}
|
||||
|
||||
// RunResult 是执行结果。
|
||||
@ -381,7 +382,7 @@ func Run(plan *Plan, opts RunOptions) (*RunResult, error) {
|
||||
|
||||
ks := opts.State
|
||||
if ks == nil {
|
||||
ks = NewKernelState()
|
||||
ks = kstate.New()
|
||||
}
|
||||
|
||||
// ★ 执行前把**全部**模块声明的设施登记为"内核注册过"。
|
||||
|
||||
@ -15,6 +15,7 @@ package kernelboot
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/kstate"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
@ -40,7 +41,7 @@ type fakeModule struct {
|
||||
func (m *fakeModule) Name() string { return m.name }
|
||||
func (m *fakeModule) Provides() []string { return m.provides }
|
||||
func (m *fakeModule) Requires() []string { return m.requires }
|
||||
func (m *fakeModule) Init(_ *KernelState) error {
|
||||
func (m *fakeModule) Init(_ *kstate.State) error {
|
||||
if m.initCount != nil {
|
||||
atomic.AddInt32(m.initCount, 1)
|
||||
}
|
||||
|
||||
@ -1,396 +0,0 @@
|
||||
package kernelboot
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"sync"
|
||||
|
||||
"github.com/JianFeeeee/HologramAgent/internal/agent/api"
|
||||
agentAPI "github.com/JianFeeeee/HologramAgent/internal/agent/api"
|
||||
agentCore "github.com/JianFeeeee/HologramAgent/internal/agent/core"
|
||||
agentIO "github.com/JianFeeeee/HologramAgent/internal/agent/io"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/config"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/events"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/knowledge"
|
||||
luapkg "github.com/JianFeeeee/HologramAgent/internal/lua"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/memory"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/memory/coldcontext"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/memory/pipeline"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/memory/social"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/memory/text"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/memory/vector"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/plugin"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/security"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/supervisor"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/tracker"
|
||||
"github.com/JianFeeeee/HologramAgent/pkg/types"
|
||||
)
|
||||
|
||||
// MemoryBundle 是"记忆子系统"的打包(memory 模块产出)。
|
||||
//
|
||||
// ★ 为何打包成一个类型而不是拆成 4 个字段:
|
||||
// 这 4 个东西**同生共死**(同一个 initMemoryStack 产出、同一个 close 回收、
|
||||
// 它们的持有者都假设彼此一致)。拆开会让"只填了 db 忘了 indexer"变成可能。
|
||||
//
|
||||
// ★ 为何放在 kernelboot 而不是 cmd/homed:
|
||||
// KernelState 要引用它的类型 ⇒ 类型必须与 KernelState 同包
|
||||
// (否则 kernelboot 得 import cmd/homed,那是不可能的)。
|
||||
type MemoryBundle struct {
|
||||
DB *memory.GraphDB
|
||||
Indexer *memory.Indexer
|
||||
Social *social.SocialStore
|
||||
Distiller *pipeline.Distiller
|
||||
}
|
||||
|
||||
// KernelState 是内核**唯一的公共数据持有者**(用户设计)。
|
||||
//
|
||||
// ## 设计意图(用户原话)
|
||||
//
|
||||
// > 内核通过一个结构体收集全部包级共享变量,其他模块通过这个结构体访问数据,
|
||||
// > 各个模块的公共数据也统一注册到内核这个公共结构体中。
|
||||
//
|
||||
// > 结构体成员初始置 nil 加读取前检查不就好了
|
||||
//
|
||||
// ## 为什么这样比"每个包各持全局变量"好
|
||||
//
|
||||
// 本仓实测的隐藏依赖(设计 §2.8.2)大多来自**散落各包的包级变量**:
|
||||
//
|
||||
// internal/nlp/parser.go: var defaultParser Parser // SetDefaultParser 写,
|
||||
// // NewExtractor 运行期读
|
||||
//
|
||||
// 这类变量的共同问题:
|
||||
//
|
||||
// · **看不见**:不读那个包的源码就不知道它存在、谁写谁读
|
||||
// · **拼错不报**:`SetDefaultParser` 忘调 ⇒ 运行期静默降级
|
||||
// (实测:图记忆实体退化成整句复合值,跨维度检索全失效)
|
||||
// · **顺序无保证**:写入时机靠"碰巧"(原 main 里蒸馏先启动、解析器后装上,
|
||||
// 靠 Interval=10min 兜着 —— 见 §2.8.2 A)
|
||||
//
|
||||
// 收到一个结构体后:
|
||||
//
|
||||
// · **一处可见**:一眼看全内核有哪些共享状态(可审计)
|
||||
// · **类型化**:字段拼错 / 取错类型 ⇒ **编译错误**
|
||||
// · **有归属**:字段属于某个模块的产物 ⇒ 顺序由依赖图保证
|
||||
//
|
||||
// ## ★ 取值方式:直接读字段 + 调用方检查 nil(用户指定的做法)
|
||||
//
|
||||
// 不用 map[string]any、不用泛型取用、不用"未声明依赖就取不到"的拦截。
|
||||
// 理由:
|
||||
//
|
||||
// · 类型化字段 ⇒ 拼错/取错类型是**编译错误**(比任何运行期检查都早)
|
||||
// · nil 检查 ⇒ 读得越直观越好(`if ks.LuaVM == nil { ... }`)
|
||||
// · 少一层间接 ⇒ 少一处可能出错的地方
|
||||
//
|
||||
// 顺序正确性由**声明**驱动:`Build()` 校验缺依赖 / 环 / 重名,
|
||||
// `Run()` 在执行每一层之前再核一次"实际已产出的设施"是否涵盖本层依赖
|
||||
// (声明对、但计划层顺序被改坏时也能拦住,见 DependencyNotReadyError)。
|
||||
// 判据断言关键顺序(见 internal/kernelmod/all/bootgraph_test.go 的
|
||||
// TestLayerTopologyIsRespected:用真实模块登记表装配整机,逐层断言层间有序)。
|
||||
//
|
||||
// ## ★ 并发
|
||||
//
|
||||
// 同层模块**并行** Init,各自只写**自己那一个字段**
|
||||
// ⇒ 字段互不相同,无数据竞争(判据用 -race 验证)。
|
||||
//
|
||||
// 内核**不**在字段上加锁:加锁会给调用方"可以乱写"的错觉,
|
||||
// 正确约束是"各模块只写自己的字段"(由代码审查 + 判据守)。
|
||||
type KernelState struct {
|
||||
// ── Group 0 产出(core 自己立起来的东西,所有模块只读) ──
|
||||
|
||||
// Cfg 是配置快照(config 模块产出)。
|
||||
//
|
||||
// ★ 它就是原先"无主"的那个 cfg:`cfg := cfgReg.ToConfig()` 之后被手工
|
||||
// 传给 11 个 init*。归到这里后,任何模块读 ks.Cfg 即可。
|
||||
// 它在依赖图上是 config 模块的产物(第 0 层,前提由 Group 0 保证)。
|
||||
Cfg *types.Config
|
||||
|
||||
// CfgReg 是配置中心(Group 0 产出,所有模块只读用)。
|
||||
CfgReg *config.ConfigRegistry
|
||||
|
||||
// DataDir 是数据目录(Group 0 产出)。
|
||||
DataDir string
|
||||
|
||||
// AgentWorkDir 是 agent 工作目录(ensureDataDirs 产出)。
|
||||
AgentWorkDir string
|
||||
|
||||
// CLISocket 是 CLI 插件的 unix socket 路径(--cli-socket 产出)。
|
||||
//
|
||||
// 放在公共结构体里的理由:它是插件读的配置,必须与插件系统同时就绪
|
||||
// (stage-host 模块用它设 cli.DefaultSocket —— 原先那行在 main 里
|
||||
// 紧邻 loadPlugins)。
|
||||
CLISocket string
|
||||
|
||||
// HTTPAddr 是 WebUI 监听地址覆盖值(--http-addr 产出)。
|
||||
HTTPAddr string
|
||||
|
||||
// ── 各模块注册的公共数据(模块只写自己那几个字段) ──
|
||||
|
||||
// Memory 是图记忆栈(memory 模块产出)。
|
||||
Memory *MemoryBundle
|
||||
|
||||
// LuaVM 是 Lua 协议适配 VM(lua 模块产出)。
|
||||
LuaVM *luapkg.VM
|
||||
|
||||
// Supervisor 是代理生命周期守护(supervisor 模块产出)。
|
||||
Supervisor *supervisor.Daemon
|
||||
|
||||
// Tracker 是变更追踪(tracker 模块产出)。
|
||||
Tracker *tracker.Tracker
|
||||
|
||||
// IO 是内核 IO 抽象层(kernel-api 模块产出)。
|
||||
IO *agentIO.IOManager
|
||||
|
||||
// EventBus 是事件总线(kernel-api 模块产出)。
|
||||
EventBus *events.Bus
|
||||
|
||||
// TextMemory 是文本记忆(text-memory 模块产出)。
|
||||
TextMemory *text.Memory
|
||||
|
||||
// Embedder 是共享词嵌入(memory 模块产出)。
|
||||
//
|
||||
// 蒸馏提取与 Agent 上下文**复用同一实例**,避免同一模型被二次加载
|
||||
// (约 200k×300 维 ≈ 数百 MB 内存)。
|
||||
Embedder *memory.StaticEmbedder
|
||||
|
||||
// LLMManager 是多源 provider 管理(llm 模块产出)。
|
||||
LLMManager *agentAPI.ProviderManager
|
||||
|
||||
// LLMProvider 是默认 provider(llm 模块产出,= LLMManager.Default())。
|
||||
LLMProvider agentAPI.Provider
|
||||
|
||||
// BaseAPIKey 是网关基础 key(llm 模块产出)。
|
||||
BaseAPIKey string
|
||||
|
||||
// DocStore 是文档记忆(doc-store 模块产出)。
|
||||
DocStore *coldcontext.Store
|
||||
|
||||
// Knowledge 是知识库(knowledge-store 模块产出)。
|
||||
Knowledge *knowledge.Store
|
||||
|
||||
// MultimodalSpace 是多模态向量空间(multimodal 模块产出)。
|
||||
MultimodalSpace vector.MultimodalEmbedder
|
||||
|
||||
// MultimodalProviderName / MultimodalErr 是"配了哪个 provider/为什么没启用"
|
||||
// (multimodal 模块产出,供 healthcheck 状态报告)。
|
||||
MultimodalProviderName string
|
||||
MultimodalErr string
|
||||
|
||||
// Update 是更新子系统的公共数据(update 模块产出)。
|
||||
//
|
||||
// 用接口而不是具体类型:kernelboot 不能 import internal/update
|
||||
// (那会形成 kernelboot → update → kernelboot 的循环)。
|
||||
// 模块在 Init 里赋 *update.State,读取方用类型断言取回。
|
||||
//
|
||||
// ★ 它在依赖图第 0 层(update 模块 Requires 为空),所以任何模块
|
||||
// 读它时都已就绪 —— 包括"启动失败回滚"的判定结果。
|
||||
Update UpdateStateProvider
|
||||
|
||||
// UpdateCapabilities 是本机**升级能力**的快照(update 模块产出)。
|
||||
//
|
||||
// ## 为什么它是一个独立字段而不是塞进 Update.State
|
||||
//
|
||||
// State 回答"我这台机器现在是什么版本、上次回滚过没有";
|
||||
// Capabilities 回答"我这台机器**能不能**自己升级"。
|
||||
// 前者随时间变,后者在进程生命周期内恒定(安装形态不变)。
|
||||
//
|
||||
// 消费者不同:
|
||||
// · WebUI / healthcheck 展示 State(版本、回滚轨迹)
|
||||
// · kernel_update 工具 / CLI 决策看 Capabilities(走哪条升级通道)
|
||||
//
|
||||
// 合并成一个结构会让“只关心能不能升级”的调用方也拿到一堆版本字段,
|
||||
// 反之亦然——那类混搭在这仓已经出过问题(healthcheck 把工具表塞进快照)。
|
||||
//
|
||||
// ★ 用 interface{} 而不是定义接口(与 PluginCatalog 同一范式)。
|
||||
//
|
||||
// 尝试过定义 UpdateCapabilitiesProvider 接口,但它要求方法名
|
||||
// Variant()/InstallKind() —— 而 Capabilities 的**字段**就叫 Variant,
|
||||
// 方法名与字段名在同一个类型上不能共存(编译错误)。
|
||||
// 改方法名(VariantName())只会让接口变丑而无收益:读取方本来
|
||||
// 就要完整结构体(WebUI 要 bin_path / reason / package_name)。
|
||||
//
|
||||
// 所以:与 ks.PluginCatalog 一致,存 any,读取方断言回
|
||||
// *update.Capabilities。代价是一次类型断言,收益是零重复定义。
|
||||
UpdateCapabilities any
|
||||
|
||||
// StageHost 是阶段管道(stage-host 模块产出)。
|
||||
StageHost *agentCore.StageHost
|
||||
|
||||
// PluginRegistry 是插件注册表(stage-host 模块产出)。
|
||||
PluginRegistry *plugin.Registry
|
||||
|
||||
// PluginCatalog 是**插件元数据缓存**(plugin-host 模块产出)。
|
||||
//
|
||||
// ## 为什么需要它(用户指出的设计)
|
||||
//
|
||||
// > 插件管理器不应该在被拉起后就去读插件列表获取元数据并写入缓存准备拉起吗?
|
||||
//
|
||||
// 对。原先元数据散在两处、且**每次都要重扫磁盘**:
|
||||
// · cmd/homed 的 collectPluginSpecs(决定启动时机)
|
||||
// · pluginmgr 的 listPlugins(人看的列表)
|
||||
// 两者各读各的 plugin.json,且调度器每轮评估都要用到。
|
||||
//
|
||||
// 现在统一由 plugin-host 模块**扫一次**、缓存下来,供:
|
||||
// · 调度器(判断时机)
|
||||
// · pluginmgr(列列表/状态)
|
||||
// · 显式启用路径(校验名字是否存在)
|
||||
//
|
||||
// 类型是 any 以避免 import 环(kernelboot 不能依赖 plugin 包之外的
|
||||
// 插件元数据类型)—— 取用处断言为 *PluginCatalog。
|
||||
PluginCatalog any
|
||||
|
||||
// SyncBus 是内核**就绪状态总线**(由 Run 创建并填入)。
|
||||
//
|
||||
// ## ★ 为什么它必须在这里(而不只是 Run 的局部变量)
|
||||
//
|
||||
// 模块的 `Init(ks)` 只能看到 KernelState —— 若 bus 只在 Run 里,
|
||||
// **模块拿不到它**,于是无法:
|
||||
// · 订阅状态变化(plugin-host 要据此拉起"等更晚设施"的插件)
|
||||
// · 标记就绪(虽然 Run 会自动做,但模块可能需要主动补标)
|
||||
//
|
||||
// 实测撞到:plugin-host 想在 Init 里挂"等 agent 就绪再拉起某些插件"的
|
||||
// 回调,却找不到 bus。把它放进 KernelState(内核公共数据,用户设计)
|
||||
// 是最自然的落点 —— 它本来就是**内核级共享状态**。
|
||||
SyncBus *SyncBus
|
||||
|
||||
// PluginScheduler 是**插件拉起调度器**(plugin-host 模块产出)。
|
||||
//
|
||||
// 用途:诊断(还剩哪些插件挂起)、以及将来"显式启用某插件"时
|
||||
// 判断它的依赖是否就绪。
|
||||
//
|
||||
// 类型是 any 以避免 kernelboot 反向依赖插件包(import 环)。
|
||||
// 取用处断言为 *Scheduler。
|
||||
PluginScheduler any
|
||||
|
||||
// PluginsReady 标记"插件调度已完成"(plugin-host 模块产出)。
|
||||
//
|
||||
// 用途:main 在装配后据此知道"插件该起的都起了",再往下走
|
||||
// (如 compactConfigDB / 启动 runtime)。
|
||||
PluginsReady bool
|
||||
|
||||
// Agent 是主 agent(agent 模块产出)。
|
||||
Agent *agentCore.Agent
|
||||
|
||||
// Security 是**安全防火墙运行时**(security 模块产出)。
|
||||
//
|
||||
// ★ 它不是"某项能力",而是横在 Core 与全部外部模块之间的边界:
|
||||
// 出向(Core → 外部)与入向(外部 → Core)都要经它审查。
|
||||
//
|
||||
// 为什么是具体类型而不是接口:internal/security 不依赖本包(无环),
|
||||
// 而类型化字段能让"取错类型/拼错字段"变成编译错误 —— 这正是本结构体
|
||||
// 存在的理由(见包注释)。未装配时为 nil;配置显式关闭时是
|
||||
// Disabled Runtime(所有审查零成本透传),不是 nil。
|
||||
Security *security.Runtime
|
||||
|
||||
// Device 是**工具/通道数据面**(device 模块产出)。
|
||||
//
|
||||
// 类型是低层接口(internal/agent/api.DeviceSurface)而不是
|
||||
// *kernelmod/device.Runtime:后者 import kernelboot,直接用具体类型会成环。
|
||||
// 接口只暴露"读面"(工具清单),执行仍经 Agent —— 那里才有授权闸与超时。
|
||||
Device api.DeviceSurface
|
||||
|
||||
// ── 运行期(装配阶段建立,后续流程用) ──
|
||||
|
||||
// Ctx / Stop 是 agent 运行期的取消上下文。
|
||||
//
|
||||
// ★ 为何在这里而不是模块内部:main 在装配之后还要把它传给
|
||||
// startAgentRuntime / startIPCServer。生命周期必须一致 ——
|
||||
// 装配时创建的 ctx 若不交出去,那些组件会用另一个 ctx,
|
||||
// 停机信号就传不进去。
|
||||
Ctx context.Context
|
||||
Stop context.CancelFunc
|
||||
|
||||
// Closers 是各模块注册的关闭函数(按**注册顺序**记录)。
|
||||
//
|
||||
// ★ 为何按注册顺序而不是按名字排:名字与依赖序**无关**。
|
||||
// 实测反例:注册序 config→memory→llm,按名字逆序得 memory,llm,config(错),
|
||||
// 正确的是 llm,memory,config。关闭时**逆序**执行(LIFO,与 defer 一致)。
|
||||
//
|
||||
// ★★ 为什么需要独立的锁(closerMu):
|
||||
// **同层的模块是并行 Init 的**,而它们都会调 AddCloser。
|
||||
// 实测(压力测试 + -race):第 1 层有 8 个模块并行,其中 5 个
|
||||
// (docstore/lua/memory/multimodal/textmem)都注册关闭函数
|
||||
// ⇒ 并发 append 到同一 slice ⇒ **数据竞争**。
|
||||
//
|
||||
// 后果不只是 race 警告:切片 append 在容量增长时可能**丢失元素**
|
||||
// (两个 goroutine 同时读到旧 len、各自写同一槽)⇒ 某个模块的
|
||||
// 关闭函数永远不被调用 ⇒ **资源泄漏**(且无任何报错)。
|
||||
//
|
||||
// 为什么不用 KernelState 上"模块只写自己字段"的约定来避免:
|
||||
// Closers 不属于任一模块 —— 它是**共享的回收站**。
|
||||
// 共享可变状态就必须有锁(这与"各写各的字段"是两条不同的保证)。
|
||||
Closers []func()
|
||||
|
||||
// closerMu 保护 Closers(见上)。
|
||||
closerMu sync.Mutex
|
||||
|
||||
// MigrationPreflight 是数据目录迁移体检(由 migration 模块填充,可 nil)。
|
||||
//
|
||||
// 用 func 字段而不是具体类型:kernelboot 不能依赖 migration 模块
|
||||
// (那会成环),而迁移能力本身是**可选**的 —— nil 时调用方跳过即可。
|
||||
MigrationPreflight func(target string) (interface{}, error)
|
||||
|
||||
// Quiescers 是可**暂停/恢复**的写者(伪关机用,见 lifecycle.go)。
|
||||
//
|
||||
// 与 Closers 的区别:Closers 是**一次性**的(关了就没了),
|
||||
// 而迁移需要"停下来 → 搬数据 → 再起来" ⇒ 必须可逆。
|
||||
Quiescers []Quiescer
|
||||
|
||||
quiesceMu sync.Mutex
|
||||
}
|
||||
|
||||
// NewKernelState 创建公共结构体(所有字段零值 = nil,由各模块填充)。
|
||||
func NewKernelState() *KernelState {
|
||||
return &KernelState{}
|
||||
}
|
||||
|
||||
// AddCloser 注册一个关闭函数(模块在 Init 里调用)。
|
||||
//
|
||||
// 关闭时按**注册顺序的逆序**执行(LIFO,与 `defer` 语义一致):
|
||||
// 先初始化的一般是底层(config → memory → llm),停机应反向拆。
|
||||
func (s *KernelState) AddCloser(fn func()) {
|
||||
if fn == nil {
|
||||
return
|
||||
}
|
||||
// ★ 必须加锁:同层模块**并行** Init 时都会调它(见 Closers 字段的注释)。
|
||||
s.closerMu.Lock()
|
||||
s.Closers = append(s.Closers, fn)
|
||||
s.closerMu.Unlock()
|
||||
}
|
||||
|
||||
// CloseAll 逆序执行全部关闭函数,返回首个错误。
|
||||
//
|
||||
// ★ 失败策略:**继续执行剩余关闭**(返回首个错误)。
|
||||
// 一个模块关不掉不该阻止其他模块关 —— 那会造成更多泄漏。
|
||||
// 关闭函数 panic 也被隔离(否则一个模块的问题会连带其他模块都关不掉)。
|
||||
func (s *KernelState) CloseAll() error {
|
||||
// ★ 先取快照再执行(不持锁执行回调):关闭函数可能很慢
|
||||
// (停进程/flush 数据库),持锁会让并发调用 AddCloser 的模块卡住;
|
||||
// 且回调可能反查 KernelState ⇒ 死锁。
|
||||
s.closerMu.Lock()
|
||||
fns := make([]func(), len(s.Closers))
|
||||
copy(fns, s.Closers)
|
||||
s.Closers = nil
|
||||
s.closerMu.Unlock()
|
||||
|
||||
for i := len(fns) - 1; i >= 0; i-- {
|
||||
fn := fns[i]
|
||||
func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
log.Printf("[kernelboot] shutdown function panicked (isolated): %v", r)
|
||||
}
|
||||
}()
|
||||
fn()
|
||||
}()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpdateStateProvider 是更新子系统暴露给内核的最小接口。
|
||||
//
|
||||
// 定义在 kernelboot 而不是 internal/update:避免 kernelboot → update 的
|
||||
// import 循环(update 的模块实现要 import kernelboot)。
|
||||
type UpdateStateProvider interface {
|
||||
// Summary 返回给 healthcheck / 状态面的版本摘要。
|
||||
Summary() map[string]interface{}
|
||||
}
|
||||
@ -14,6 +14,7 @@ package kernelboot
|
||||
// (判据的"数量"不是指标,"每条原判据都有对应"才是)。
|
||||
|
||||
import (
|
||||
"github.com/JianFeeeee/HologramAgent/internal/kstate"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@ -25,7 +26,7 @@ import (
|
||||
// 为何逆序:先初始化的一般是底层(config → memory → llm),
|
||||
// 停机应**反向拆**,否则可能拆掉别人还在用的东西。
|
||||
func TestKernelStateCloseAllIsReverseOrder(t *testing.T) {
|
||||
ks := NewKernelState()
|
||||
ks := kstate.New()
|
||||
var mu sync.Mutex
|
||||
var order []string
|
||||
|
||||
@ -48,7 +49,7 @@ func TestKernelStateCloseAllIsReverseOrder(t *testing.T) {
|
||||
|
||||
// ★★★ 判据:**CloseAll 不因单个关闭函数 panic 而中断**。
|
||||
func TestKernelStateCloseAllSurvivesPanic(t *testing.T) {
|
||||
ks := NewKernelState()
|
||||
ks := kstate.New()
|
||||
closed := false
|
||||
ks.AddCloser(func() { panic("关闭时炸了") })
|
||||
ks.AddCloser(func() { closed = true })
|
||||
@ -61,7 +62,7 @@ func TestKernelStateCloseAllSurvivesPanic(t *testing.T) {
|
||||
|
||||
// ★★★ 判据:**CloseAll 幂等**(清空后再次调用不重复执行)。
|
||||
func TestKernelStateCloseAllIsIdempotent(t *testing.T) {
|
||||
ks := NewKernelState()
|
||||
ks := kstate.New()
|
||||
var n int32
|
||||
ks.AddCloser(func() { atomic.AddInt32(&n, 1) })
|
||||
|
||||
@ -78,7 +79,7 @@ func TestKernelStateCloseAllIsIdempotent(t *testing.T) {
|
||||
// 第 1 层有 8 个模块并行,其中 5 个注册关闭函数 ⇒ 并发 append
|
||||
// ⇒ 数据竞争 + 可能丢失元素(某个模块的关闭函数永不调用 ⇒ 资源泄漏)。
|
||||
func TestKernelStateAddCloserConcurrent(t *testing.T) {
|
||||
ks := NewKernelState()
|
||||
ks := kstate.New()
|
||||
const n = 500
|
||||
|
||||
var wg sync.WaitGroup
|
||||
@ -99,7 +100,7 @@ func TestKernelStateAddCloserConcurrent(t *testing.T) {
|
||||
|
||||
// ★★★ 判据:**nil 关闭函数被忽略**(不 panic、不占位)。
|
||||
func TestKernelStateAddCloserIgnoresNil(t *testing.T) {
|
||||
ks := NewKernelState()
|
||||
ks := kstate.New()
|
||||
ks.AddCloser(nil)
|
||||
if len(ks.Closers) != 0 {
|
||||
t.Fatalf("nil 关闭函数应被忽略,实际记录 %d 个", len(ks.Closers))
|
||||
|
||||
@ -1,123 +0,0 @@
|
||||
package kernelboot
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
)
|
||||
|
||||
// 内核生命周期:**伪关机(Quiesce)** 与 **唤醒(Resume)**。
|
||||
//
|
||||
// ## 为什么需要它(用户 2026-10-17 要求)
|
||||
//
|
||||
// 数据目录迁移时**绝不能有模块在写数据**:distiller 每 10 分钟写一次
|
||||
// memory/graph.db,text memory 滚动写 memory/text,tracker 在 overlay 里写,
|
||||
// 40 个插件各写自己的 plugin_data。迁移 21G 数据要几分钟,期间任何写入
|
||||
// 都会造成"迁移到一半的数据"或迁移后新旧不一致。
|
||||
//
|
||||
// 用户原话:
|
||||
//
|
||||
// 「迁移模块要先让 core 暂停其他模块运行(伪关机)完成迁移后再唤醒」
|
||||
//
|
||||
// ## 为什么不能用现成的 CloseAll
|
||||
//
|
||||
// `CloseAll` 是**一次性**的:它执行完就把 `Closers` 清空(切片置 nil),
|
||||
// 且各模块的 Init 有副作用(建库、播种、迁移扫描),**不能简单重跑**。
|
||||
// 用它做伪关机 = 关掉就再也起不来,等于要求重启进程 —— 那就不是"伪"关机了。
|
||||
//
|
||||
// ## 本文件的做法
|
||||
//
|
||||
// 分两级,都**可逆**:
|
||||
//
|
||||
// Quiesce() → 让所有"活动写者"停下来,但**保留连接/句柄**
|
||||
// Resume() → 把停下来的重新启动
|
||||
//
|
||||
// 参与方式:模块在 Init 里通过 `AddQuiescer(name, pause, resume)` 登记。
|
||||
// 与 CloseAll 一样**逆序**唤醒(LIFO)——后启动的先恢复,保证依赖顺序。
|
||||
//
|
||||
// ## ★ 关键约束(写进类型里)
|
||||
//
|
||||
// 1. pause 必须**幂等**且**不关闭文件/DB 句柄**(关了就无法 resume)
|
||||
// 2. pause 必须**阻塞到确实不再写**(否则迁移仍有竞态)
|
||||
// 3. resume 失败必须**可见**(返回错误),不能静默"半醒"
|
||||
// 4. 只有**持有写句柄**的模块才需要登记 —— 纯读模块不必(如 config)
|
||||
type Quiescer struct {
|
||||
// Name 用于日志与诊断(哪个模块停不下来,一眼看到)。
|
||||
Name string
|
||||
// Pause 停下活动写入并**等到确实停了**才返回。
|
||||
Pause func() error
|
||||
// Resume 恢复写入。失败必须返回错误(不能假装成功)。
|
||||
Resume func() error
|
||||
}
|
||||
|
||||
// AddQuiescer 登记一个可暂停/恢复的写者(在模块 Init 里调用)。
|
||||
func (s *KernelState) AddQuiescer(name string, pause, resume func() error) {
|
||||
if pause == nil {
|
||||
return
|
||||
}
|
||||
s.quiesceMu.Lock()
|
||||
s.Quiescers = append(s.Quiescers, Quiescer{Name: name, Pause: pause, Resume: resume})
|
||||
s.quiesceMu.Unlock()
|
||||
}
|
||||
|
||||
// Quiesce 伪关机:**逆序**暂停全部已登记的写者。
|
||||
//
|
||||
// 返回成功暂停的清单 —— Resume 只唤醒这些(没有的不要"凭空唤醒")。
|
||||
// 任何一个 Pause 失败都会**停止继续暂停**并返回错误:宁可只关一半
|
||||
// 让调用方放弃迁移,也不要"看起来关了但仍有写者在跑"。
|
||||
func (s *KernelState) Quiesce() ([]string, error) {
|
||||
s.quiesceMu.Lock()
|
||||
qs := make([]Quiescer, len(s.Quiescers))
|
||||
copy(qs, s.Quiescers)
|
||||
s.quiesceMu.Unlock()
|
||||
|
||||
// 逆序暂停:后启动的先停(与启动顺序相反,先停"消费者"再停"生产者")
|
||||
var paused []string
|
||||
for i := len(qs) - 1; i >= 0; i-- {
|
||||
q := qs[i]
|
||||
if err := q.Pause(); err != nil {
|
||||
// 已暂停的按**反序**回滚(把它唤醒回去),别留半关状态
|
||||
for j := len(paused) - 1; j >= 0; j-- {
|
||||
if r := s.resumeOne(qs, paused[j]); r != nil {
|
||||
log.Printf("[quiesce] rollback wake %s failed: %v", paused[j], r)
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("暂停 %s 失败(已回滚 %d 个): %w", q.Name, len(paused), err)
|
||||
}
|
||||
paused = append(paused, q.Name)
|
||||
log.Printf("[quiesce] paused: %s", q.Name)
|
||||
}
|
||||
return paused, nil
|
||||
}
|
||||
|
||||
// Resume 唤醒:按**暂停的逆序**(即原启动顺序)恢复。
|
||||
func (s *KernelState) Resume(paused []string) error {
|
||||
s.quiesceMu.Lock()
|
||||
qs := make([]Quiescer, len(s.Quiescers))
|
||||
copy(qs, s.Quiescers)
|
||||
s.quiesceMu.Unlock()
|
||||
|
||||
var firstErr error
|
||||
for i := len(paused) - 1; i >= 0; i-- {
|
||||
if err := s.resumeOne(qs, paused[i]); err != nil {
|
||||
log.Printf("[quiesce] wake %s failed: %v", paused[i], err)
|
||||
if firstErr == nil {
|
||||
firstErr = fmt.Errorf("唤醒 %s: %w", paused[i], err)
|
||||
}
|
||||
} else {
|
||||
log.Printf("[quiesce] woken: %s", paused[i])
|
||||
}
|
||||
}
|
||||
return firstErr
|
||||
}
|
||||
|
||||
func (s *KernelState) resumeOne(qs []Quiescer, name string) error {
|
||||
for _, q := range qs {
|
||||
if q.Name == name {
|
||||
if q.Resume == nil {
|
||||
return nil
|
||||
}
|
||||
return q.Resume()
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("未登记的写者: %s", name)
|
||||
}
|
||||
@ -25,6 +25,7 @@ package kernelboot
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/kstate"
|
||||
"log"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
@ -87,7 +88,7 @@ type PluginQuiescer interface {
|
||||
// defer sess.End() // 无论成功失败都唤醒
|
||||
// // ... 外部脚本迁移 ...
|
||||
type MigrateSession struct {
|
||||
ks *KernelState
|
||||
ks *kstate.State
|
||||
plugins PluginQuiescer
|
||||
paused []string
|
||||
stopped []string // 已运行态停掉的插件(按停止顺序,唤醒时逆序)
|
||||
@ -97,7 +98,7 @@ type MigrateSession struct {
|
||||
// BeginMigration 进入伪关机:停插件(运行态)→ 逆序暂停内核写者 → 等在途归零。
|
||||
//
|
||||
// 返回的 session 必须由调用方 `defer sess.End()`。
|
||||
func (s *KernelState) BeginMigration(pq PluginQuiescer) (*MigrateSession, error) {
|
||||
func BeginMigration(s *kstate.State, pq PluginQuiescer) (*MigrateSession, error) {
|
||||
sess := &MigrateSession{ks: s, plugins: pq}
|
||||
|
||||
// ① 先停**插件**(运行态停启,可逆)。
|
||||
|
||||
@ -21,9 +21,11 @@ import (
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/JianFeeeee/HologramAgent/internal/kstate"
|
||||
)
|
||||
|
||||
func newTestKS() *KernelState { return NewKernelState() }
|
||||
func newTestKS() *kstate.State { return kstate.New() }
|
||||
|
||||
// ★★★ 判据:Quiesce 逆序暂停、Resume 逆序恢复。
|
||||
func TestQuiescePausesInReverseOrder(t *testing.T) {
|
||||
@ -121,7 +123,7 @@ func TestMigrateSessionEndIsIdempotent(t *testing.T) {
|
||||
return nil
|
||||
})
|
||||
|
||||
sess, err := ks.BeginMigration(nil)
|
||||
sess, err := BeginMigration(ks, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@ -185,7 +187,7 @@ func TestMigrationStopsAndRestoresAllPlugins(t *testing.T) {
|
||||
ks := newTestKS()
|
||||
fp := &fakePlugins{running: []string{"webui", "rss", "calendar", "commondev"}}
|
||||
|
||||
sess, err := ks.BeginMigration(fp)
|
||||
sess, err := BeginMigration(ks, fp)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@ -209,7 +211,7 @@ func TestMigrationRollsBackPluginsOnFailure(t *testing.T) {
|
||||
// 让 calendar 失败 ⇒ 已停的 commondev 必须被拉回
|
||||
fp := &fakePlugins{running: []string{"webui", "rss", "calendar", "commondev"}, failOn: "calendar"}
|
||||
|
||||
if _, err := ks.BeginMigration(fp); err == nil {
|
||||
if _, err := BeginMigration(ks, fp); err == nil {
|
||||
t.Fatal("停插件失败时 BeginMigration 必须返回错误")
|
||||
}
|
||||
got := fp.RunningPlugins()
|
||||
|
||||
@ -29,6 +29,7 @@ package kernelboot
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/kstate"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
@ -56,7 +57,7 @@ type PluginSpec struct {
|
||||
// Scheduler 按 syncbus 状态调度插件。
|
||||
type Scheduler struct {
|
||||
mu sync.Mutex
|
||||
bus SyncState
|
||||
bus kstate.SyncState
|
||||
pending []PluginSpec
|
||||
launched map[string]bool
|
||||
launch func(PluginSpec) error
|
||||
@ -67,9 +68,9 @@ type Scheduler struct {
|
||||
|
||||
// NewScheduler 创建调度器。launch 是实际拉起动作(由插件管理器提供)。
|
||||
//
|
||||
// ★ bus 参数类型是 **SyncState**(只读接口),不是 *SyncBus。
|
||||
// ★ bus 参数类型是 **kstate.SyncState**(只读接口),不是 *kstate.SyncBus。
|
||||
// 这就是设计 §2.6.3 的权限隔离:调度器**在编译期就没有**写 syncbus 的能力。
|
||||
func NewScheduler(bus SyncState, launch func(PluginSpec) error) *Scheduler {
|
||||
func NewScheduler(bus kstate.SyncState, launch func(PluginSpec) error) *Scheduler {
|
||||
s := &Scheduler{
|
||||
bus: bus,
|
||||
launch: launch,
|
||||
@ -82,8 +83,8 @@ func NewScheduler(bus SyncState, launch func(PluginSpec) error) *Scheduler {
|
||||
//
|
||||
// 若 bus 不支持订阅(只传了个只读快照实现),退化为"只在 Submit 时评估"——
|
||||
// 那种情况下调度器仍可用,只是不会自动跟进后续状态变化。
|
||||
if sub, ok := bus.(interface{ OnUpdate(func(SyncState)) }); ok {
|
||||
sub.OnUpdate(func(SyncState) { s.evaluate() })
|
||||
if sub, ok := bus.(interface{ OnUpdate(func(kstate.SyncState)) }); ok {
|
||||
sub.OnUpdate(func(kstate.SyncState) { s.evaluate() })
|
||||
}
|
||||
return s
|
||||
}
|
||||
@ -293,7 +294,7 @@ func (s *Scheduler) WaitString(p PluginSpec) string {
|
||||
if len(p.RequiresFacilities) == 0 {
|
||||
return "等内核完全初始化"
|
||||
}
|
||||
// AllReady 需要 *SyncBus;这里只做只读渲染,用 Snapshot 判断
|
||||
// AllReady 需要 *kstate.SyncBus;这里只做只读渲染,用 Snapshot 判断
|
||||
snap := s.bus.Snapshot()
|
||||
var missing []string
|
||||
for _, f := range p.RequiresFacilities {
|
||||
|
||||
@ -25,17 +25,18 @@ package kernelboot
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/kstate"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ★★★ 压力:**SyncBus 读写并发**(内核写 + N 个插件读 + 订阅回调)。
|
||||
// ★★★ 压力:**kstate.SyncBus 读写并发**(内核写 + N 个插件读 + 订阅回调)。
|
||||
//
|
||||
// 用 -race 跑时,任何未保护的状态都会报数据竞争。
|
||||
func TestStressSyncBusConcurrentReadWrite(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
const readers = 64
|
||||
const facilities = 200
|
||||
|
||||
@ -43,7 +44,7 @@ func TestStressSyncBusConcurrentReadWrite(t *testing.T) {
|
||||
// 订阅者**读状态**(这正是死锁的触发条件:若 fire 持锁调用回调,
|
||||
// 回调里 Snapshot() 会等读锁 ⇒ 死锁。压力下必现)
|
||||
for i := 0; i < 8; i++ {
|
||||
bus.OnUpdate(func(s SyncState) {
|
||||
bus.OnUpdate(func(s kstate.SyncState) {
|
||||
_ = s.Snapshot()
|
||||
_ = s.AllReady([]string{"llm"})
|
||||
atomic.AddInt64(&callbacks, 1)
|
||||
@ -109,7 +110,7 @@ func TestStressSyncBusConcurrentReadWrite(t *testing.T) {
|
||||
// · 不丢插件(全部最终被拉起)
|
||||
// · 无数据竞争
|
||||
func TestStressSchedulerConcurrentSubmit(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
bus.MarkDeclared([]string{"config", "llm", "memory"})
|
||||
// ★ 必须 MarkLayerReady:MarkDeclared 只登记"名字存在",
|
||||
// 不代表就绪。插件要求 config **就绪**才拉起。
|
||||
@ -164,7 +165,7 @@ func TestStressSchedulerConcurrentSubmit(t *testing.T) {
|
||||
|
||||
// ★★★ 压力:**状态变化 + 提交并发**(真实时序:内核在标记就绪的同时插件在提交)。
|
||||
func TestStressSchedulerConcurrentStateChange(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
bus.MarkDeclared([]string{"config", "llm", "memory", "stagehost"})
|
||||
|
||||
var launched int64
|
||||
@ -291,9 +292,9 @@ func TestStressRunRespectsMaxParallelAtScale(t *testing.T) {
|
||||
t.Logf("%d 个模块同层,上限 %d,观测峰值 %d", n, maxP, observed)
|
||||
}
|
||||
|
||||
// ★★★ 压力:**KernelState 同层并行写**(各写自己字段,无竞争)。
|
||||
// ★★★ 压力:**kstate.State 同层并行写**(各写自己字段,无竞争)。
|
||||
func TestStressKernelStateParallelWrites(t *testing.T) {
|
||||
ks := NewKernelState()
|
||||
ks := kstate.New()
|
||||
const n = 256
|
||||
|
||||
var wg sync.WaitGroup
|
||||
|
||||
@ -1,290 +0,0 @@
|
||||
package kernelboot
|
||||
|
||||
// syncbus:内核就绪状态的同步快照,驱动插件调度(设计 §2.6)。
|
||||
//
|
||||
// ## 一句话
|
||||
//
|
||||
// 内核每完成一层装配就把该层贡献的设施标记为"就绪";
|
||||
// 插件管理器**只读**这份状态,据此拉起"依赖已齐"的插件。
|
||||
//
|
||||
// ## 为什么要有它(解决什么)
|
||||
//
|
||||
// 现状(`cmd/homed/main.go:291 loadPlugins`)是**一刀切**:
|
||||
// 12 个 init* 全部做完才加载全部插件。
|
||||
// 于是"只读本地文件"的插件也要等多模态空间(ONNX 加载)+ 知识库建索引做完。
|
||||
//
|
||||
// 而内核那 12 步本身有明确先后(§2.3 算出 4 层),"阶段"是天然存在的,
|
||||
// 只是此前没有对外发布。syncbus 就是那个发布物。
|
||||
//
|
||||
// ## ★ 权限:插件管理器只读(这不是习惯,是设计)
|
||||
//
|
||||
// 若插件管理器能写 syncbus,它就能**谎报内核状态**,让插件在自己依赖的东西
|
||||
// 还不存在时启动 —— 那正是病 A(静默顺序错)。
|
||||
//
|
||||
// 所以类型层面分离:内核拿 `*SyncBus`(可写),
|
||||
// 插件管理器只拿 `SyncState` 接口(**上面没有写方法**)。
|
||||
// 想写也写不了 —— 编译期保证,而非"约定不要写"。
|
||||
// 同 `Get[T]` 泛型取用(类型错=编译不过)是一个思路。
|
||||
|
||||
import (
|
||||
"sort"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// SyncState 是 syncbus 的**只读**视图。
|
||||
//
|
||||
// ★ 刻意不含 MarkReady/Reset:插件管理器拿到的就是这个接口,
|
||||
// 因此它**在编译期就不可能**篡改内核状态。
|
||||
type SyncState interface {
|
||||
// IsReady 报告某设施是否已就绪。
|
||||
IsReady(facility string) bool
|
||||
// Snapshot 返回当前全部就绪状态的副本(调用方改不动内部)。
|
||||
Snapshot() map[string]bool
|
||||
// ReadyInOrder 按**就绪顺序**返回设施名(调试/判据用)。
|
||||
ReadyInOrder() []string
|
||||
// KernelComplete 报告内核是否已完全初始化(最后一层完成)。
|
||||
KernelComplete() bool
|
||||
// AllReady 报告给定设施是否**全部**就绪(插件调度器的核心判断)。
|
||||
//
|
||||
// ★ 空列表返回 false(不是 true):见实现处的说明 ——
|
||||
// "没有依赖"与"依赖全就绪"是不同语义,混淆会让未声明依赖的插件提前拉起。
|
||||
AllReady(facilities []string) bool
|
||||
|
||||
// Known 报告某设施名**是否被内核注册过**(不论是否已就绪)。
|
||||
//
|
||||
// ## ★ 为什么必须有它(修一个"静默永不启动"的缺口)
|
||||
//
|
||||
// 插件声明 `requires_facilities` 是**字符串**。若拼错(如写 "llm " 或
|
||||
// "llmservice"),原本的行为是:它永远等不到 → **永不拉起**,
|
||||
// 且**没有任何报错**、没有超时兜底(我第一版设计文档写了"超时兜底"
|
||||
// 但从未实现)。
|
||||
//
|
||||
// 这正是本仓反复吃亏的"静默降级"形态:系统看着正常,某插件就是不工作。
|
||||
//
|
||||
// ⇒ 有了 Known,调度器能在**提交时**立刻报错:
|
||||
// "插件 X 声明了内核从未注册过的设施 Y(可能是拼写错误)"
|
||||
// 并列出全部已知设施名供对照。
|
||||
Known(facility string) bool
|
||||
|
||||
// KnownFacilities 返回内核**注册过的全部设施名**(含尚未就绪的)。
|
||||
//
|
||||
// 用途:错误信息里列出可选值(让"拼错了"当场可修)。
|
||||
KnownFacilities() []string
|
||||
}
|
||||
|
||||
// SyncBus 是可写的同步总线(内核持有)。
|
||||
type SyncBus struct {
|
||||
mu sync.RWMutex
|
||||
// known 是内核**注册过的全部设施名**(不论是否已就绪)。
|
||||
//
|
||||
// ★ 它存在的唯一目的是让"插件拼错设施名"从**静默永不启动**
|
||||
// 变成**提交时立刻报错**(见 SyncState.Known 的说明)。
|
||||
known map[string]bool
|
||||
ready map[string]bool
|
||||
order []string
|
||||
complete bool
|
||||
onUpdate []func(SyncState)
|
||||
notifyMu sync.Mutex
|
||||
}
|
||||
|
||||
// NewSyncBus 创建空总线。
|
||||
func NewSyncBus() *SyncBus {
|
||||
return &SyncBus{ready: map[string]bool{}, known: map[string]bool{}}
|
||||
}
|
||||
|
||||
// MarkReady 标记一个设施就绪。**只有内核该调它。**
|
||||
//
|
||||
// 幂等:重复标记同一设施不报错(便于层边界回调重入)。
|
||||
// 但**不重复记录**顺序,保证 ReadyInOrder 可复现。
|
||||
func (b *SyncBus) MarkReady(facility string) {
|
||||
if facility == "" {
|
||||
return
|
||||
}
|
||||
b.mu.Lock()
|
||||
if b.ready[facility] {
|
||||
b.mu.Unlock()
|
||||
return
|
||||
}
|
||||
b.ready[facility] = true
|
||||
b.order = append(b.order, facility)
|
||||
b.mu.Unlock()
|
||||
b.fire()
|
||||
}
|
||||
|
||||
// MarkLayerReady 在**层边界**标记该层所有模块贡献的设施(设计 §2.6.2)。
|
||||
//
|
||||
// ★ 为何以"层"而不是"单个模块"为同步点:
|
||||
// 层内是**并发**执行的,若层内逐个模块标记就绪,就绪顺序会**不确定**,
|
||||
// 而判据要求可复现。层边界天然是一个确定的同步点。
|
||||
func (b *SyncBus) MarkLayerReady(provides []string) {
|
||||
// 先在锁内一次性写入,再统一通知(避免 N 次通知、N 次唤醒)
|
||||
b.mu.Lock()
|
||||
changed := false
|
||||
for _, f := range provides {
|
||||
if f == "" {
|
||||
continue
|
||||
}
|
||||
b.known[f] = true // 注册过(不论是否首次就绪)
|
||||
if b.ready[f] {
|
||||
continue
|
||||
}
|
||||
b.ready[f] = true
|
||||
b.order = append(b.order, f)
|
||||
changed = true
|
||||
}
|
||||
b.mu.Unlock()
|
||||
if changed {
|
||||
b.fire()
|
||||
}
|
||||
}
|
||||
|
||||
// MarkDeclared 在**执行前**登记一批设施名(视为"内核注册过")。
|
||||
//
|
||||
// ## ★ 为什么需要它(时序)
|
||||
//
|
||||
// 插件的 `requires_facilities` 校验要在**插件提交时**做(那是错误最该被发现
|
||||
// 的时刻,紧挨着写错的插件作者)。但插件的提交发生在内核装配**完成之后**
|
||||
// —— 那时所有层都跑完了,Known 自然齐了。
|
||||
//
|
||||
// 但仍然需要本方法,因为:
|
||||
//
|
||||
// · 装配**中途失败**时,后面的层没跑 ⇒ 那些设施名从未登记 ⇒
|
||||
// 若此时还有插件提交,会报"内核从未注册过 X"(**误报**:其实是没跑到)
|
||||
// · 判据要能在**不执行**的情况下校验"插件声明 ⊆ 内核声明"
|
||||
// (只看声明,不看执行结果)
|
||||
//
|
||||
// ⇒ 装配开始时把**全部模块的 Provides** 一次性登记为 known。
|
||||
// 这样 Known 的语义精确为"**模块声明过**这个设施",与"是否已就绪"解耦。
|
||||
func (b *SyncBus) MarkDeclared(facilities []string) {
|
||||
b.mu.Lock()
|
||||
for _, f := range facilities {
|
||||
if f != "" {
|
||||
b.known[f] = true
|
||||
}
|
||||
}
|
||||
b.mu.Unlock()
|
||||
}
|
||||
|
||||
// MarkKernelComplete 标记内核已完全初始化。
|
||||
//
|
||||
// 这是**未声明依赖的插件**的启动信号(默认路径 = 现行为,§2.6.5)。
|
||||
func (b *SyncBus) MarkKernelComplete() {
|
||||
b.mu.Lock()
|
||||
if b.complete {
|
||||
b.mu.Unlock()
|
||||
return
|
||||
}
|
||||
b.complete = true
|
||||
b.mu.Unlock()
|
||||
b.fire()
|
||||
}
|
||||
|
||||
// OnUpdate 注册状态变化回调(插件调度器用它重新检查挂起的插件)。
|
||||
//
|
||||
// 回调在**不持锁**的情况下调用 —— 否则回调里再读 syncbus 会死锁。
|
||||
// (这个坑很容易踩:回调显然要读状态,持锁调用必然死锁。)
|
||||
func (b *SyncBus) OnUpdate(fn func(SyncState)) {
|
||||
b.notifyMu.Lock()
|
||||
b.onUpdate = append(b.onUpdate, fn)
|
||||
b.notifyMu.Unlock()
|
||||
}
|
||||
|
||||
func (b *SyncBus) fire() {
|
||||
b.notifyMu.Lock()
|
||||
cbs := make([]func(SyncState), len(b.onUpdate))
|
||||
copy(cbs, b.onUpdate)
|
||||
b.notifyMu.Unlock()
|
||||
for _, fn := range cbs {
|
||||
fn(b) // b 实现了 SyncState(只读方法)
|
||||
}
|
||||
}
|
||||
|
||||
// --- SyncState 实现(只读) ---
|
||||
|
||||
func (b *SyncBus) IsReady(facility string) bool {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
return b.ready[facility]
|
||||
}
|
||||
|
||||
func (b *SyncBus) Snapshot() map[string]bool {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
out := make(map[string]bool, len(b.ready))
|
||||
for k, v := range b.ready {
|
||||
out[k] = v
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (b *SyncBus) ReadyInOrder() []string {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
out := make([]string, len(b.order))
|
||||
copy(out, b.order)
|
||||
return out
|
||||
}
|
||||
|
||||
func (b *SyncBus) KernelComplete() bool {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
return b.complete
|
||||
}
|
||||
|
||||
// Known 报告某设施名是否被内核注册过(不论是否已就绪)。
|
||||
//
|
||||
// ★ 为何"注册过但未就绪"也算 Known:那是**正常的中间状态**
|
||||
// (如 llm 声明了但还在第 1 层)。真正的错误是"内核从未注册过这个名字"。
|
||||
func (b *SyncBus) Known(facility string) bool {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
return b.known[facility]
|
||||
}
|
||||
|
||||
// KnownFacilities 返回内核注册过的全部设施名(升序,便于对照阅读)。
|
||||
func (b *SyncBus) KnownFacilities() []string {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
out := make([]string, 0, len(b.known))
|
||||
for k := range b.known {
|
||||
out = append(out, k)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// AllReady 报告给定的设施**全部**就绪(插件调度器的核心判断)。
|
||||
//
|
||||
// 空列表 = 真的全就绪 → 返回 false。
|
||||
//
|
||||
// ★ 为何空列表返回 false 而不是 true:
|
||||
// "没有依赖" 与 "依赖全就绪" 是**不同语义**,混淆会让"未声明依赖的插件"
|
||||
// 被当成"依赖已满足"而**提前拉起** —— 那就破坏了 §2.6.5 的存量兼容。
|
||||
// 未声明走的是 KernelComplete 那条路径,与这条刻意分开。
|
||||
func (b *SyncBus) AllReady(facilities []string) bool {
|
||||
if len(facilities) == 0 {
|
||||
return false
|
||||
}
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
for _, f := range facilities {
|
||||
if !b.ready[f] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// Missing 返回尚未就绪的设施名(诊断用:告诉用户"还差什么")。
|
||||
func (b *SyncBus) Missing(facilities []string) []string {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
var out []string
|
||||
for _, f := range facilities {
|
||||
if !b.ready[f] {
|
||||
out = append(out, f)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
@ -9,6 +9,7 @@ package kernelboot
|
||||
// 静默改变启动时机 = 静默引入竞态。
|
||||
|
||||
import (
|
||||
"github.com/JianFeeeee/HologramAgent/internal/kstate"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync"
|
||||
@ -24,7 +25,7 @@ import (
|
||||
// 反例(若实现错误):把"未声明"当成"依赖集为空",而空集在数学上已满足
|
||||
// ⇒ 第一层完成后就拉起 ⇒ 比现状早得多。
|
||||
func TestSchedulerUndeclaredWaitsForKernelComplete(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
var launched []string
|
||||
var mu sync.Mutex
|
||||
sch := NewScheduler(bus, func(p PluginSpec) error {
|
||||
@ -63,7 +64,7 @@ func TestSchedulerUndeclaredWaitsForKernelComplete(t *testing.T) {
|
||||
//
|
||||
// 断言"早于内核完成" —— 否则这条判据在"等内核完成"的实现下也会通过(假绿)。
|
||||
func TestSchedulerDeclaredLaunchesEarly(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
var launched []string
|
||||
var mu sync.Mutex
|
||||
sch := NewScheduler(bus, func(p PluginSpec) error {
|
||||
@ -98,7 +99,7 @@ func TestSchedulerDeclaredLaunchesEarly(t *testing.T) {
|
||||
|
||||
// ★★★ 判据:**依赖未齐则挂起**,不会误拉起。
|
||||
func TestSchedulerWaitsForMissingDependency(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
var launched []string
|
||||
sch := NewScheduler(bus, func(p PluginSpec) error {
|
||||
launched = append(launched, p.Name)
|
||||
@ -128,21 +129,21 @@ func TestSchedulerWaitsForMissingDependency(t *testing.T) {
|
||||
|
||||
// ★★★ 判据:**只读隔离**(编译期保证,不是约定)。
|
||||
//
|
||||
// 断言 Scheduler 拿到的接口**没有**写方法 —— 如果给它 *SyncBus,
|
||||
// 断言 Scheduler 拿到的接口**没有**写方法 —— 如果给它 *kstate.SyncBus,
|
||||
// 它就能谎报内核状态,让插件在依赖不存在时启动(病 A)。
|
||||
func TestSchedulerCannotWriteSyncState(t *testing.T) {
|
||||
// ★ 检查必须是**接口类型的方法集**,不是某个具体值。
|
||||
//
|
||||
// 我第一次写错了:用 `var st SyncState = NewSyncBus()` 再断言 `st.(writer)` ——
|
||||
// 那检查的是**动态类型**(*SyncBus 当然有 MarkReady),判据当场报假红。
|
||||
// 要约束的是"拿到 SyncState 的人**能不能**调写方法" ⇒ 必须查接口的方法集。
|
||||
rt := reflect.TypeOf((*SyncState)(nil)).Elem()
|
||||
// 我第一次写错了:用 `var st kstate.SyncState = kstate.NewSyncBus()` 再断言 `st.(writer)` ——
|
||||
// 那检查的是**动态类型**(*kstate.SyncBus 当然有 MarkReady),判据当场报假红。
|
||||
// 要约束的是"拿到 kstate.SyncState 的人**能不能**调写方法" ⇒ 必须查接口的方法集。
|
||||
rt := reflect.TypeOf((*kstate.SyncState)(nil)).Elem()
|
||||
forbidden := []string{"MarkReady", "MarkKernelComplete", "MarkLayerReady"}
|
||||
for i := 0; i < rt.NumMethod(); i++ {
|
||||
m := rt.Method(i).Name
|
||||
for _, bad := range forbidden {
|
||||
if m == bad {
|
||||
t.Fatalf("★ SyncState(插件管理器拿到的接口)暴露了写方法 %s —— "+
|
||||
t.Fatalf("★ kstate.SyncState(插件管理器拿到的接口)暴露了写方法 %s —— "+
|
||||
"它能谎报内核状态,让插件在依赖不存在时启动(病 A)", m)
|
||||
}
|
||||
}
|
||||
@ -150,23 +151,23 @@ func TestSchedulerCannotWriteSyncState(t *testing.T) {
|
||||
// 反向确认:读方法必须在(否则接口退化到没用了)
|
||||
for _, want := range []string{"IsReady", "Snapshot", "AllReady", "KernelComplete"} {
|
||||
if _, ok := rt.MethodByName(want); !ok {
|
||||
t.Errorf("SyncState 缺少读方法 %s", want)
|
||||
t.Errorf("kstate.SyncState 缺少读方法 %s", want)
|
||||
}
|
||||
}
|
||||
// Scheduler 持有的字段类型也必须是 SyncState(不能是 *SyncBus)
|
||||
// Scheduler 持有的字段类型也必须是 kstate.SyncState(不能是 *kstate.SyncBus)
|
||||
st := reflect.TypeOf(Scheduler{})
|
||||
f, ok := st.FieldByName("bus")
|
||||
if !ok {
|
||||
t.Fatal("Scheduler 应有 bus 字段")
|
||||
}
|
||||
if f.Type.String() != "kernelboot.SyncState" {
|
||||
t.Fatalf("★ Scheduler.bus 的类型是 %s —— 必须是只读的 SyncState", f.Type)
|
||||
if f.Type.String() != "kstate.SyncState" {
|
||||
t.Fatalf("★ Scheduler.bus 的类型是 %s —— 必须是只读的 kstate.SyncState", f.Type)
|
||||
}
|
||||
}
|
||||
|
||||
// ★★★ 判据:**不伴随启动的插件即使依赖齐了也不自动拉起**(§2.6.6)。
|
||||
func TestSchedulerAutoStartFalseIsNotLaunched(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
var launched []string
|
||||
sch := NewScheduler(bus, func(p PluginSpec) error {
|
||||
launched = append(launched, p.Name)
|
||||
@ -196,7 +197,7 @@ func TestSyncBusReadyOrderIsDeterministic(t *testing.T) {
|
||||
// 模拟"层内并发完成":同一层 8 个设施,用 goroutine 乱序 MarkReady
|
||||
var first []string
|
||||
for round := 0; round < 5; round++ {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
var wg sync.WaitGroup
|
||||
fac := []string{"f1", "f2", "f3", "f4", "f5", "f6", "f7", "f8"}
|
||||
for i, f := range fac {
|
||||
@ -225,7 +226,7 @@ func TestSyncBusReadyOrderIsDeterministic(t *testing.T) {
|
||||
// ★ 真正的可复现性断言:MarkLayerReady 一次性写入
|
||||
var order []string
|
||||
for round := 0; round < 3; round++ {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
bus.MarkLayerReady([]string{"a", "b", "c"})
|
||||
bus.MarkLayerReady([]string{"d", "e"})
|
||||
got := bus.ReadyInOrder()
|
||||
@ -250,7 +251,7 @@ func TestSyncBusReadyOrderIsDeterministic(t *testing.T) {
|
||||
//
|
||||
// 这条是补的:变异"去掉幂等检查"第一次没被咬住 —— 因为原判据
|
||||
// 只跑了一份声明、没重复标记,测不到幂等性。
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
bus.MarkLayerReady([]string{"a", "b"})
|
||||
bus.MarkLayerReady([]string{"a", "b"}) // 重复
|
||||
bus.MarkLayerReady([]string{"c"})
|
||||
@ -263,7 +264,7 @@ func TestSyncBusReadyOrderIsDeterministic(t *testing.T) {
|
||||
//
|
||||
// 这条守的是:装配中途失败 ⇒ 未声明依赖的插件**不**被拉起。
|
||||
func TestRunDoesNotMarkCompleteOnFailure(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
mods := []Module{
|
||||
&fakeModule{name: "A", provides: []string{"a"}},
|
||||
&fakeModule{name: "B", provides: []string{"b"}, initErr: errFake},
|
||||
@ -285,7 +286,7 @@ func TestRunDoesNotMarkCompleteOnFailure(t *testing.T) {
|
||||
// 为何由 Run 做:模块自己标记会漏(静默漏一个设施 ⇒ 插件永远等不到),
|
||||
// 且顺序取决于层内并发完成序。
|
||||
func TestRunPublishesFacilitiesAtLayerBoundary(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
mods := []Module{
|
||||
&fakeModule{name: "config", provides: []string{"config", "config-write"}},
|
||||
&fakeModule{name: "llm", provides: []string{"llm"}, requires: []string{"config"}},
|
||||
@ -311,7 +312,7 @@ func TestRunPublishesFacilitiesAtLayerBoundary(t *testing.T) {
|
||||
//
|
||||
// 若这条反了,未声明依赖的插件会被提前拉起(见第一条判据)。
|
||||
func TestAllReadyEmptyIsFalse(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
bus.MarkKernelComplete()
|
||||
if bus.AllReady(nil) {
|
||||
t.Fatal("★ AllReady(空) 返回 true —— \"没有依赖\"被当成\"依赖全就绪\",未声明插件会提前拉起")
|
||||
@ -328,7 +329,7 @@ func TestAllReadyEmptyIsFalse(t *testing.T) {
|
||||
|
||||
// ★★★ 判据:**拉起失败不阻塞其他插件**(一个坏插件不该拖垮全局)。
|
||||
func TestSchedulerLaunchFailureDoesNotBlockOthers(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
var okLaunched []string
|
||||
sch := NewScheduler(bus, func(p PluginSpec) error {
|
||||
if p.Name == "bad" {
|
||||
@ -357,7 +358,7 @@ func TestSchedulerLaunchFailureDoesNotBlockOthers(t *testing.T) {
|
||||
|
||||
// ★★★ 判据:**重复 Submit 不重复拉起**(幂等)。
|
||||
func TestSchedulerIsIdempotent(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
n := 0
|
||||
sch := NewScheduler(bus, func(PluginSpec) error { n++; return nil })
|
||||
bus.MarkKernelComplete()
|
||||
@ -378,11 +379,11 @@ func TestSchedulerIsIdempotent(t *testing.T) {
|
||||
// (单个设施的标记),若它持锁回调,只有"既调 MarkReady 又订阅"的场景才会卡住。
|
||||
// 补这条判据把两条路径都盖住。
|
||||
func TestSyncBusMarkReadyNotifiesWithoutDeadlock(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
got := make(chan string, 4)
|
||||
|
||||
// 订阅者**读状态**(真实回调必然会读,这正是死锁触发条件)
|
||||
bus.OnUpdate(func(s SyncState) {
|
||||
bus.OnUpdate(func(s kstate.SyncState) {
|
||||
_ = s.Snapshot() // 持锁调用的话,这里会等 RLock ⇒ 死锁
|
||||
got <- "ok"
|
||||
})
|
||||
|
||||
@ -12,13 +12,14 @@ package kernelboot
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/JianFeeeee/HologramAgent/internal/kstate"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// ★★★ 判据:**拼错的设施名在 Submit 时被拒绝**(不再静默挂起)。
|
||||
func TestSchedulerRejectsUnknownFacility(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
bus.MarkDeclared([]string{"config", "llm", "memory"}) // 内核注册过这些
|
||||
|
||||
var launched []string
|
||||
@ -57,7 +58,7 @@ func TestSchedulerRejectsUnknownFacility(t *testing.T) {
|
||||
|
||||
// ★★★ 判据:**正确的设施名不受影响**(不能把正常路径也拦了)。
|
||||
func TestSchedulerAcceptsKnownFacility(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
bus.MarkDeclared([]string{"config", "llm"})
|
||||
// ★ 必须同时 MarkLayerReady —— 真实路径里 Run 两者都会做:
|
||||
// MarkDeclared(执行前,登记名字)与 MarkLayerReady(每层完成,标记就绪)。
|
||||
@ -86,7 +87,7 @@ func TestSchedulerAcceptsKnownFacility(t *testing.T) {
|
||||
// ★ 这条很重要:llm 声明了但还在第 1 层时,Known 为 true、Ready 为 false。
|
||||
// 若把"未就绪"当"拼错",所有多依赖插件都会在提交时被误拒。
|
||||
func TestSchedulerAcceptsDeclaredButNotYetReady(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
bus.MarkDeclared([]string{"config", "llm"}) // 两者都已声明
|
||||
bus.MarkLayerReady([]string{"config"}) // 只有 config 就绪
|
||||
|
||||
@ -111,7 +112,7 @@ func TestSchedulerAcceptsDeclaredButNotYetReady(t *testing.T) {
|
||||
// 那些设施名就从未被 MarkLayerReady 登记 ⇒ 会**误报**"内核从未注册过"。
|
||||
// 所以 Run 必须**执行前**就 MarkDeclared。
|
||||
func TestRunDeclaresAllFacilitiesUpfront(t *testing.T) {
|
||||
bus := NewSyncBus()
|
||||
bus := kstate.NewSyncBus()
|
||||
mods := []Module{
|
||||
&fakeModule{name: "a", provides: []string{"fa"}},
|
||||
&fakeModule{name: "b", provides: []string{"fb"}, requires: []string{"fa"},
|
||||
|
||||
Reference in New Issue
Block a user