From 610e9d0bbb5d8595238342b53a3df0d4cf5dc243 Mon Sep 17 00:00:00 2001 From: dev Date: Wed, 2 Sep 2026 10:41:18 +0800 Subject: [PATCH] =?UTF-8?q?feat(plugin):=20entry=20=E5=8F=8C=E9=80=9A?= =?UTF-8?q?=E9=81=93=E5=88=86=E6=B4=BE=20+=20=E5=85=B1=E4=BA=AB=E5=86=85?= =?UTF-8?q?=E5=AD=98=20stage=20=E6=95=B0=E6=8D=AE=E9=9D=A2=EF=BC=88Part=20?= =?UTF-8?q?1=20+=20Part=204=20=E6=A0=B8=E5=BF=83=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part 1 加载分派骨架（迁移可逐插件推进、随时回退的前提）: - dynamic.go: 新增 binEntry/skillEntry 常量 + entryKind 枚举 + classifyEntry/detectEntryKind manifest entry 优先级最高（改回 plugin.so 即回退 cabi）；无 manifest 时按目录探测，.bin 优先 - registry.go: tryDynamic 按 entry 分派 proc/cabi 双通道； entry 声明 .bin 但二进制缺失时报明确错误，不静默回退（否则'已迁移插件跑回旧通道'极难排查） - registry.go: pluginEntryHash 候选顺序与 detectEntryKind 对齐（.bin 优先）， 否则增量重载会用错文件算 hash - dynamic_proc_{unix,windows}.go: tryLoadProc 桩位（权限/类型校验已实现，进程管理属 Part 2） Part 4 共享内存数据面（迁移评估 §3.3/§3.4/§3.7，最关键一环）: - proc/shm.go: 段布局（Header + ShmStageCtx 描述符数组 + append-only arena） 相对偏移设计——各进程 mmap 到不同虚拟地址仍能正确解引用 arena 用尽显式报错而非静默截断（§4.4 风险登记）；Compact() 回收 append-only 垃圾 - proc/shmcodec.go: StageContext 16 字段跨进程编解码 字段级描述符消除 lost update：只改 FinalText 的插件不触碰 ToolResults 描述符 WriteDirty 只写脏字段——只读插件零写入，不可能覆盖他人改写 Snapshot 存序列化字符串（切片共享底层数组的坑，C ABI 侧修 11.3 时已踩过） Extra 4 键提升为具名字段；Response 用标志位表达 nil vs 空串 - proc/lock.go: 锁仲裁回归内核（§3.7 已裁定，零 cgo） ForceRelease 实现实验 9 的崩溃自愈——排除 robust pthread_mutex 必要性 重复加锁显式拒绝（否则死锁 30s）；等待超时有补偿 goroutine 防锁泄漏 验证: - proc 包 16 项测试全绿（含 -race）：全字段往返/只读零写回/原地改切片识别/ 现网 sanitizer+weather 场景/5插件×40轮并发零丢失/arena 耗尽报错/压实不破坏字段/ 锁互斥·串扰拒绝·崩溃自愈·临界区串行化 - entry 分派 9 项测试全绿；go build ./... exit 0；接口冻结 git diff sdk/ 为空 --- internal/plugin/dynamic.go | 80 +++- internal/plugin/dynamic_proc_unix.go | 43 ++ internal/plugin/dynamic_proc_windows.go | 25 ++ internal/plugin/entry_dispatch_test.go | 160 ++++++++ internal/plugin/manifest.go | 2 +- internal/plugin/proc/lock.go | 131 ++++++ internal/plugin/proc/lock_test.go | 184 +++++++++ internal/plugin/proc/shm.go | 307 ++++++++++++++ internal/plugin/proc/shm_test.go | 441 ++++++++++++++++++++ internal/plugin/proc/shmcodec.go | 514 ++++++++++++++++++++++++ internal/plugin/proc/unsafe.go | 15 + internal/plugin/registry.go | 25 +- 12 files changed, 1919 insertions(+), 8 deletions(-) create mode 100644 internal/plugin/dynamic_proc_unix.go create mode 100644 internal/plugin/dynamic_proc_windows.go create mode 100644 internal/plugin/entry_dispatch_test.go create mode 100644 internal/plugin/proc/lock.go create mode 100644 internal/plugin/proc/lock_test.go create mode 100644 internal/plugin/proc/shm.go create mode 100644 internal/plugin/proc/shm_test.go create mode 100644 internal/plugin/proc/shmcodec.go create mode 100644 internal/plugin/proc/unsafe.go diff --git a/internal/plugin/dynamic.go b/internal/plugin/dynamic.go index 41beba7..33aecc0 100644 --- a/internal/plugin/dynamic.go +++ b/internal/plugin/dynamic.go @@ -7,12 +7,84 @@ import ( ) const ( - soEntry = "plugin.so" - dllEntry = "plugin.dll" - luaEntry = "main.lua" - metaEntry = "plugin.json" + soEntry = "plugin.so" + dllEntry = "plugin.dll" + binEntry = "plugin.bin" // 子进程插件（纯 Go 二进制，stdio JSON-RPC） + luaEntry = "main.lua" + skillEntry = "SKILL.md" + metaEntry = "plugin.json" ) +// entryKind 描述插件入口归属的加载通道。 +// 外部插件多进程化期间 .so/.dll（cabi）与 .bin（proc）**双通道共存**， +// 按 plugin.json 的 entry 字段分派，使迁移可逐插件推进、随时回退。 +type entryKind int + +const ( + entryUnknown entryKind = iota + entryCABI // plugin.so / plugin.dll / plugin.dylib —— C ABI 动态库 + entryProc // plugin.bin —— 子进程 + stdio JSON-RPC + entryLua // main.lua + entrySkill // SKILL.md +) + +func (k entryKind) String() string { + switch k { + case entryCABI: + return "cabi" + case entryProc: + return "proc" + case entryLua: + return "lua" + case entrySkill: + return "skill" + } + return "unknown" +} + +// classifyEntry 把 manifest 的 entry 字段映射到加载通道。 +// entry 为空时返回 entryUnknown，由调用方回退到目录探测（兼容无 manifest 的旧插件）。 +func classifyEntry(entry string) entryKind { + switch entry { + case soEntry, dllEntry, "plugin.dylib": + return entryCABI + case binEntry: + return entryProc + case luaEntry: + return entryLua + case skillEntry: + return entrySkill + } + return entryUnknown +} + +// detectEntryKind 先读 manifest 的 entry，读不到则按目录内存在的入口文件推断。 +// 推断顺序：.bin 优先于 .so——迁移期间同一插件目录可能两个产物共存（升级未清理）， +// 此时应走新通道；manifest 显式声明优先级最高。 +func detectEntryKind(plgDir string) entryKind { + if mft := readManifest(plgDir); mft != nil { + if k := classifyEntry(mft.Entry); k != entryUnknown { + return k + } + } + for _, probe := range []struct { + file string + kind entryKind + }{ + {binEntry, entryProc}, + {soEntry, entryCABI}, + {"plugin.dylib", entryCABI}, + {dllEntry, entryCABI}, + {luaEntry, entryLua}, + {skillEntry, entrySkill}, + } { + if st, err := os.Stat(filepath.Join(plgDir, probe.file)); err == nil && !st.IsDir() { + return probe.kind + } + } + return entryUnknown +} + func readManifest(dir string) *PluginManifest { data, err := os.ReadFile(filepath.Join(dir, metaEntry)) if err != nil { diff --git a/internal/plugin/dynamic_proc_unix.go b/internal/plugin/dynamic_proc_unix.go new file mode 100644 index 0000000..4787c3e --- /dev/null +++ b/internal/plugin/dynamic_proc_unix.go @@ -0,0 +1,43 @@ +//go:build linux || darwin + +package plugin + +import ( + "fmt" + "os" + "path/filepath" + + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +// tryLoadProc 加载子进程插件（plugin.bin）——外部插件多进程化的加载入口。 +// +// 设计依据：docs/zh/架构迁移评估.md §3（stdio JSON-RPC 控制面 + shm 数据面 + eventfd 通知面） +// 实施计划：docs/zh/plugin-migration-plan.md Part 2 +// +// 当前状态：**分派桩位**。共享内存数据面与锁仲裁已在 internal/plugin/proc/ 落地 +// 并通过 16 项测试（含 -race），进程管理与 RPC 编解码为 Part 2 内容。 +// +// 返回 nil,nil 表示目录中没有 plugin.bin（交由后续探测通道）。 +// 找到二进制但通道未就绪时返回明确错误——不静默回退到 cabi， +// 否则"已迁移插件跑回旧通道"极难排查。 +func tryLoadProc(dir, name string, config map[string]interface{}) (sdk.Plugin, error) { + binPath := filepath.Join(dir, binEntry) + st, err := os.Stat(binPath) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, fmt.Errorf("proc plugin %s: 检查 %s: %w", name, binEntry, err) + } + if st.IsDir() { + return nil, fmt.Errorf("proc plugin %s: %s 是目录，不是可执行文件", name, binEntry) + } + if st.Mode()&0o111 == 0 { + return nil, fmt.Errorf("proc plugin %s: %s 缺少可执行权限（chmod +x）", name, binEntry) + } + + return nil, fmt.Errorf("proc plugin %s: 子进程通道尚未实现（Part 2）——"+ + "共享内存数据面已就绪（internal/plugin/proc），"+ + "如需运行请把 plugin.json 的 entry 改回 %s 走 C ABI 通道", name, soEntry) +} diff --git a/internal/plugin/dynamic_proc_windows.go b/internal/plugin/dynamic_proc_windows.go new file mode 100644 index 0000000..0942ae1 --- /dev/null +++ b/internal/plugin/dynamic_proc_windows.go @@ -0,0 +1,25 @@ +//go:build windows + +package plugin + +import ( + "fmt" + "os" + "path/filepath" + + sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +// tryLoadProc 的 Windows 桩：子进程通道本身是跨平台的（stdio JSON-RPC 无平台差异）， +// 但共享内存数据面当前基于 POSIX mmap，Windows 需改用 CreateFileMapping。 +// +// 迁移评估 §9.2 已记录：Windows DLL 路径当前能力严重退化（只下发 3 字段、无写回）， +// 迁移到子进程后三套 ABI 收敛为单一 RPC 实现，Windows 反而受益——但需要测试机验证。 +func tryLoadProc(dir, name string, config map[string]interface{}) (sdk.Plugin, error) { + for _, candidate := range []string{binEntry, "plugin.exe"} { + if st, err := os.Stat(filepath.Join(dir, candidate)); err == nil && !st.IsDir() { + return nil, fmt.Errorf("proc plugin %s: Windows 子进程通道尚未实现（Part 2 + §9.2）", name) + } + } + return nil, nil +} diff --git a/internal/plugin/entry_dispatch_test.go b/internal/plugin/entry_dispatch_test.go new file mode 100644 index 0000000..ced5aec --- /dev/null +++ b/internal/plugin/entry_dispatch_test.go @@ -0,0 +1,160 @@ +package plugin + +import ( + "os" + "path/filepath" + "testing" +) + +// entry 分派骨架（docs/zh/plugin-migration-plan.md Part 1）： +// 外部插件多进程化期间 .so/.dll（cabi）与 .bin（proc）双通道共存， +// 按 plugin.json 的 entry 字段分派，使迁移可逐插件推进、随时回退。 + +func TestClassifyEntry(t *testing.T) { + cases := []struct { + entry string + want entryKind + }{ + {"plugin.so", entryCABI}, + {"plugin.dll", entryCABI}, + {"plugin.dylib", entryCABI}, + {"plugin.bin", entryProc}, + {"main.lua", entryLua}, + {"SKILL.md", entrySkill}, + {"", entryUnknown}, + {"plugin.wasm", entryUnknown}, + } + for _, c := range cases { + if got := classifyEntry(c.entry); got != c.want { + t.Errorf("classifyEntry(%q) = %v, want %v", c.entry, got, c.want) + } + } +} + +// manifest 显式声明的 entry 优先级最高。 +func TestDetectEntryKind_ManifestWins(t *testing.T) { + dir := t.TempDir() + // 目录里放 .so，但 manifest 声明 .bin → 应走 proc + mustWrite(t, filepath.Join(dir, "plugin.so"), "fake so") + mustWrite(t, filepath.Join(dir, "plugin.bin"), "fake bin") + mustWrite(t, filepath.Join(dir, metaEntry), `{"name":"x","entry":"plugin.bin"}`) + + if got := detectEntryKind(dir); got != entryProc { + t.Fatalf("manifest 声明 plugin.bin 应走 proc，实际 %v", got) + } +} + +// manifest 声明 .so 时即便存在 .bin 也走 cabi —— 这是回退路径的保证。 +func TestDetectEntryKind_ManifestCanForceRollback(t *testing.T) { + dir := t.TempDir() + mustWrite(t, filepath.Join(dir, "plugin.so"), "fake so") + mustWrite(t, filepath.Join(dir, "plugin.bin"), "fake bin") + mustWrite(t, filepath.Join(dir, metaEntry), `{"name":"x","entry":"plugin.so"}`) + + if got := detectEntryKind(dir); got != entryCABI { + t.Fatalf("manifest 声明 plugin.so 应回退到 cabi，实际 %v", got) + } +} + +// 无 manifest（或 entry 为空）时按目录探测，.bin 优先于 .so： +// 迁移期间同目录可能两种产物共存（升级未清理），此时应走新通道。 +func TestDetectEntryKind_ProbeOrderPrefersBin(t *testing.T) { + dir := t.TempDir() + mustWrite(t, filepath.Join(dir, "plugin.so"), "fake so") + mustWrite(t, filepath.Join(dir, "plugin.bin"), "fake bin") + + if got := detectEntryKind(dir); got != entryProc { + t.Fatalf("无 manifest 时应优先 plugin.bin，实际 %v", got) + } +} + +func TestDetectEntryKind_ProbeFallbacks(t *testing.T) { + t.Run("only so", func(t *testing.T) { + dir := t.TempDir() + mustWrite(t, filepath.Join(dir, "plugin.so"), "x") + if got := detectEntryKind(dir); got != entryCABI { + t.Fatalf("got %v", got) + } + }) + t.Run("only lua", func(t *testing.T) { + dir := t.TempDir() + mustWrite(t, filepath.Join(dir, "main.lua"), "x") + if got := detectEntryKind(dir); got != entryLua { + t.Fatalf("got %v", got) + } + }) + t.Run("only skill", func(t *testing.T) { + dir := t.TempDir() + mustWrite(t, filepath.Join(dir, "SKILL.md"), "x") + if got := detectEntryKind(dir); got != entrySkill { + t.Fatalf("got %v", got) + } + }) + t.Run("empty dir", func(t *testing.T) { + if got := detectEntryKind(t.TempDir()); got != entryUnknown { + t.Fatalf("空目录应为 unknown，实际 %v", got) + } + }) +} + +// entry 声明 plugin.bin 但二进制缺失时必须报明确错误， +// 不得静默回退到 cabi —— 否则"已迁移插件跑回旧通道"极难排查。 +func TestTryLoadProc_MissingBinaryReturnsNil(t *testing.T) { + dir := t.TempDir() + plg, err := tryLoadProc(dir, "demo", nil) + if plg != nil || err != nil { + t.Fatalf("无 plugin.bin 应返回 nil,nil（交由后续探测），实际 plg=%v err=%v", plg, err) + } +} + +func TestTryLoadProc_NonExecutableRejected(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, binEntry) + mustWrite(t, path, "not executable") + if err := os.Chmod(path, 0o644); err != nil { + t.Fatalf("chmod: %v", err) + } + + _, err := tryLoadProc(dir, "demo", nil) + if err == nil { + t.Fatal("缺少可执行权限应报错") + } +} + +// pluginEntryHash 的候选顺序须与 detectEntryKind 一致（plugin.bin 优先）， +// 否则增量重载会用错文件算 hash，导致"换了 .bin 但内核以为没变"。 +func TestPluginEntryHash_PrefersBin(t *testing.T) { + dir := t.TempDir() + mustWrite(t, filepath.Join(dir, "plugin.so"), "so content") + mustWrite(t, filepath.Join(dir, "plugin.bin"), "bin content") + + h1 := pluginEntryHash(dir) + if h1 == "" { + t.Fatal("应算出 hash") + } + + // 改 .so 不应影响 hash（因为以 .bin 为准） + mustWrite(t, filepath.Join(dir, "plugin.so"), "so content CHANGED") + if h2 := pluginEntryHash(dir); h2 != h1 { + t.Error("plugin.bin 存在时 hash 不应受 plugin.so 变化影响") + } + + // 改 .bin 必须改变 hash + mustWrite(t, filepath.Join(dir, "plugin.bin"), "bin content CHANGED") + if h3 := pluginEntryHash(dir); h3 == h1 { + t.Error("plugin.bin 变化必须反映到 hash（否则增量重载失效）") + } +} + +func TestPluginEntryHash_EmptyForFactoryOnlyPlugin(t *testing.T) { + if h := pluginEntryHash(t.TempDir()); h != "" { + t.Errorf("无入口文件应返回空串（内置纯工厂插件），实际 %q", h) + } +} + +func mustWrite(t *testing.T, path, content string) { + t.Helper() + if err := os.WriteFile(path, []byte(content), 0o755); err != nil { + t.Fatalf("写 %s: %v", path, err) + } +} diff --git a/internal/plugin/manifest.go b/internal/plugin/manifest.go index 095ae68..dfb844c 100644 --- a/internal/plugin/manifest.go +++ b/internal/plugin/manifest.go @@ -19,7 +19,7 @@ type PluginManifest struct { License string `json:"license,omitempty"` Homepage string `json:"homepage,omitempty"` Repository string `json:"repository,omitempty"` - Entry string `json:"entry"` // "plugin.so" | "plugin.dll" | "main.lua" | "SKILL.md" + Entry string `json:"entry"` // "plugin.bin"(子进程) | "plugin.so" | "plugin.dll" | "main.lua" | "SKILL.md" Platforms []string `json:"platforms,omitempty"` // 声明的支持平台: ["linux","darwin","windows"] MinVersion string `json:"min_version,omitempty"` Tags []string `json:"tags,omitempty"` diff --git a/internal/plugin/proc/lock.go b/internal/plugin/proc/lock.go new file mode 100644 index 0000000..a07d244 --- /dev/null +++ b/internal/plugin/proc/lock.go @@ -0,0 +1,131 @@ +package proc + +import ( + "fmt" + "sync" + "time" +) + +// 跨进程锁：锁仲裁回归内核（§3.7 已裁定，实验 3 + 实验 9 支撑）。 +// +// 为什么不用 robust pthread_mutex： +// - PTHREAD_PROCESS_SHARED + ROBUST 属性 Go 标准库无等价物，引入它意味着 +// **为了一把锁保留 cgo**——与"C 整体退场"的目标冲突。 +// - 锁仲裁回内核后，持锁进程崩溃由 cmd.Wait()/stdio EOF 检测，内核代为释放； +// 实验 9 实测无死锁、**无需 EOWNERDEAD 处理**。 +// - 成本：一次 RPC 往返 19.4 µs（实验 3，20000 次测得）。stage handler 的加锁 +// 频率很低（每次 stage 一两次，不是每字段一次），微秒级往返可忽略。 +// +// 于是整个新架构可做到**完全无 cgo**。 + +// lockWaitTimeout 是插件申请 stage 锁的最长等待时间。 +// +// 取 30s：stage handler 自身受 60s 工具超时约束（toolcall.go），锁等待 +// 必须显著短于它，否则超时错误会指向错误的原因。超时返回错误而非 +// 静默继续——**持锁失败下改写共享段会破坏并发正确性**。 +const lockWaitTimeout = 30 * time.Second + +// stageLock 是内核侧为单个 stage 执行持有的互斥体。 +// +// 一次 RunStage 对应一个 stageLock 实例：同阶段并发扇出的所有插件 +// （含跨进程的）在此排队，语义等价于今日内置插件共享 +// *StageContext 的 sync.RWMutex——这正是"保留并发扇出原始设计" +// （§0.2 第 1 条：并发扇出是原始设计，不是缺陷）。 +type stageLock struct { + mu sync.Mutex + + // ownerMu 保护 owner/held，使 ForceRelease 能安全介入 + ownerMu sync.Mutex + owner string // 当前持锁的插件名，空表示未持有 + held bool +} + +func newStageLock() *stageLock { return &stageLock{} } + +// Acquire 为 plugin 申请写锁，带超时。 +// +// 同一插件重复 Acquire 会死锁（stage handler 不应嵌套加锁）， +// 故显式拒绝并返回错误——比让插件挂死 30s 更容易排查。 +func (l *stageLock) Acquire(plugin string) error { + l.ownerMu.Lock() + if l.held && l.owner == plugin { + l.ownerMu.Unlock() + return fmt.Errorf("proc: 插件 %s 重复申请 stage 锁（handler 内不应嵌套加锁）", plugin) + } + l.ownerMu.Unlock() + + acquired := make(chan struct{}) + go func() { + l.mu.Lock() + close(acquired) + }() + + select { + case <-acquired: + l.ownerMu.Lock() + l.owner = plugin + l.held = true + l.ownerMu.Unlock() + return nil + case <-time.After(lockWaitTimeout): + // 等待超时：上面的 goroutine 可能随后拿到锁，必须让它能释放， + // 否则锁永久泄漏。用一个补偿 goroutine 等它拿到后立刻放掉。 + go func() { + <-acquired + l.ownerMu.Lock() + stillFree := !l.held + l.ownerMu.Unlock() + if stillFree { + l.mu.Unlock() + } + }() + return fmt.Errorf("proc: 插件 %s 申请 stage 锁超时（%s）", plugin, lockWaitTimeout) + } +} + +// Release 释放写锁。非持锁者调用返回错误（防止串扰）。 +func (l *stageLock) Release(plugin string) error { + l.ownerMu.Lock() + if !l.held { + l.ownerMu.Unlock() + return fmt.Errorf("proc: 插件 %s 释放未持有的 stage 锁", plugin) + } + if l.owner != plugin { + owner := l.owner + l.ownerMu.Unlock() + return fmt.Errorf("proc: 插件 %s 试图释放 %s 持有的 stage 锁", plugin, owner) + } + l.owner = "" + l.held = false + l.ownerMu.Unlock() + l.mu.Unlock() + return nil +} + +// ForceRelease 在插件进程崩溃/退出时由内核代为释放其持有的锁（实验 9 的自愈机制）。 +// +// 返回是否实际释放了锁。**这是"无需 robust mutex"的核心**： +// 锁的所有权在内核进程，插件死亡由 cmd.Wait()/stdio EOF 检测到， +// 内核直接解锁，不存在"持锁者死亡导致全局死锁"。 +func (l *stageLock) ForceRelease(plugin string) bool { + l.ownerMu.Lock() + if !l.held || l.owner != plugin { + l.ownerMu.Unlock() + return false + } + l.owner = "" + l.held = false + l.ownerMu.Unlock() + l.mu.Unlock() + return true +} + +// Owner 返回当前持锁插件名（诊断用）。 +func (l *stageLock) Owner() string { + l.ownerMu.Lock() + defer l.ownerMu.Unlock() + if !l.held { + return "" + } + return l.owner +} diff --git a/internal/plugin/proc/lock_test.go b/internal/plugin/proc/lock_test.go new file mode 100644 index 0000000..d77dd07 --- /dev/null +++ b/internal/plugin/proc/lock_test.go @@ -0,0 +1,184 @@ +package proc + +import ( + "strings" + "sync" + "testing" + "time" +) + +// 锁仲裁回归内核（§3.7 已裁定）的行为验证，含实验 9 的崩溃自愈机制。 + +func TestStageLock_MutualExclusion(t *testing.T) { + l := newStageLock() + + if err := l.Acquire("A"); err != nil { + t.Fatalf("A 应能获得锁: %v", err) + } + if l.Owner() != "A" { + t.Errorf("Owner 应为 A，实际 %q", l.Owner()) + } + + // B 在 A 持锁期间不得进入 + entered := make(chan struct{}) + go func() { + _ = l.Acquire("B") + close(entered) + }() + select { + case <-entered: + t.Fatal("A 持锁期间 B 不应获得锁") + case <-time.After(50 * time.Millisecond): + } + + if err := l.Release("A"); err != nil { + t.Fatalf("A 释放失败: %v", err) + } + select { + case <-entered: + case <-time.After(2 * time.Second): + t.Fatal("A 释放后 B 应获得锁") + } + if l.Owner() != "B" { + t.Errorf("Owner 应为 B，实际 %q", l.Owner()) + } + _ = l.Release("B") +} + +// 非持锁者不得释放他人的锁（防止串扰导致并发正确性被破坏）。 +func TestStageLock_ReleaseByNonOwnerRejected(t *testing.T) { + l := newStageLock() + if err := l.Acquire("A"); err != nil { + t.Fatalf("Acquire: %v", err) + } + defer l.Release("A") + + err := l.Release("B") + if err == nil { + t.Fatal("非持锁者释放应被拒绝") + } + if !strings.Contains(err.Error(), "试图释放") { + t.Errorf("错误信息应说明串扰，实际: %v", err) + } + if l.Owner() != "A" { + t.Errorf("A 应仍持锁，实际 owner=%q", l.Owner()) + } +} + +func TestStageLock_ReleaseWithoutHoldRejected(t *testing.T) { + l := newStageLock() + if err := l.Release("A"); err == nil { + t.Fatal("未持锁时释放应报错") + } +} + +// handler 内嵌套加锁会死锁，应显式拒绝而不是让插件挂死到超时。 +func TestStageLock_ReentrantAcquireRejected(t *testing.T) { + l := newStageLock() + if err := l.Acquire("A"); err != nil { + t.Fatalf("Acquire: %v", err) + } + defer l.Release("A") + + err := l.Acquire("A") + if err == nil { + t.Fatal("同一插件重复加锁应被拒绝（否则死锁 30s）") + } + if !strings.Contains(err.Error(), "重复申请") { + t.Errorf("错误信息应说明重复加锁，实际: %v", err) + } +} + +// 实验 9 的核心：持锁进程崩溃后内核代为释放，后续插件不死锁。 +// 这条彻底排除了 robust pthread_mutex 的必要性 —— 整个架构零 cgo。 +func TestStageLock_ForceReleaseOnPluginCrash(t *testing.T) { + l := newStageLock() + + // 插件 X 拿锁后"崩溃"（不调用 Release） + if err := l.Acquire("X"); err != nil { + t.Fatalf("X Acquire: %v", err) + } + if !l.ForceRelease("X") { + t.Fatal("内核应能强制释放崩溃插件持有的锁") + } + if l.Owner() != "" { + t.Errorf("强制释放后应无持有者，实际 %q", l.Owner()) + } + + // 插件 Y 随后必须能正常拿到锁（无死锁） + done := make(chan error, 1) + go func() { done <- l.Acquire("Y") }() + select { + case err := <-done: + if err != nil { + t.Fatalf("Y 应能获得锁: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("X 崩溃后 Y 无法获得锁 —— 出现死锁") + } + if err := l.Release("Y"); err != nil { + t.Fatalf("Y 释放失败: %v", err) + } +} + +// ForceRelease 对非持有者/未持锁应为 no-op，不能误放他人的锁。 +func TestStageLock_ForceReleaseIsTargeted(t *testing.T) { + l := newStageLock() + if err := l.Acquire("A"); err != nil { + t.Fatalf("Acquire: %v", err) + } + defer l.Release("A") + + if l.ForceRelease("B") { + t.Error("强制释放不该动 A 持有的锁") + } + if l.Owner() != "A" { + t.Errorf("A 应仍持锁，实际 %q", l.Owner()) + } +} + +// 高并发下锁的串行化保证：临界区不重叠。 +func TestStageLock_SerializesCriticalSection(t *testing.T) { + l := newStageLock() + var ( + mu sync.Mutex + inside int + maxSeen int + ) + const workers = 8 + const iters = 50 + + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + name := string(rune('A' + id)) + for j := 0; j < iters; j++ { + if err := l.Acquire(name); err != nil { + t.Errorf("Acquire: %v", err) + return + } + mu.Lock() + inside++ + if inside > maxSeen { + maxSeen = inside + } + mu.Unlock() + + mu.Lock() + inside-- + mu.Unlock() + if err := l.Release(name); err != nil { + t.Errorf("Release: %v", err) + return + } + } + }(i) + } + wg.Wait() + + if maxSeen > 1 { + t.Fatalf("临界区出现并发：同时 %d 个持有者", maxSeen) + } +} diff --git a/internal/plugin/proc/shm.go b/internal/plugin/proc/shm.go new file mode 100644 index 0000000..1cf9ef7 --- /dev/null +++ b/internal/plugin/proc/shm.go @@ -0,0 +1,307 @@ +// Package proc 实现外部插件的子进程加载通道（plugin.bin）。 +// +// 设计依据：docs/zh/架构迁移评估.md 第三章 +// +// homed ──spawn──> plugin（纯 Go 二进制，无 cgo） +// ├── stdio JSON-RPC 控制面：51 个 method id 平移为 method 名（§3.2） +// ├── shm + 偏移 数据面：StageContext 并发改写、二进制零拷贝（§3.3） +// └── eventfd 通知面：事件环 post-and-forget（§3.6） +// +// 本文件负责数据面的共享段布局与 arena 分配器。 +package proc + +import ( + "encoding/binary" + "fmt" + "sync/atomic" +) + +// 共享段魔数与版本，用于挂载时校验对端布局一致。 +const ( + shmMagic uint32 = 0x48415348 // "HASH" — HomeAgent SHared + shmVersion uint32 = 1 +) + +// 段布局（所有偏移均相对**段起始**，arena 内偏移相对 arenaBase）： +// +// [0, headerSize) Header：魔数/版本/arena 游标 +// [headerSize, ctxEnd) ShmStageCtx：每字段一个 Slice{off,len} 描述符 +// [arenaBase, arenaBase+arenaCap) arena：append-only 变长数据区 +// +// **相对偏移是关键**（§3.3）——各进程 mmap 到不同虚拟地址仍能正确解引用。 +const ( + headerSize = 64 + + // Header 内字段偏移 + offMagic = 0 // uint32 + offVersion = 4 // uint32 + offArenaBase = 8 // uint32 + offArenaCap = 12 // uint32 + offArenaUsed = 16 // uint32（原子 bump 游标） + offCtxBase = 20 // uint32 + offSeq = 24 // uint64：每次成功写回自增，供乐观读校验 +) + +// Slice 是 arena 内变长数据的描述符，off 相对 arenaBase。 +// 长度为 0 表示空值；off==0 && len==0 表示"字段未设置"。 +type Slice struct { + Off uint32 + Len uint32 +} + +const sliceSize = 8 + +// IsUnset 报告该描述符是否表示"字段从未被写入"。 +// 注意与"写入了空字符串"区分：后者 Off 非 0、Len 为 0。 +func (s Slice) IsUnset() bool { return s.Off == 0 && s.Len == 0 } + +// stageField 枚举 StageContext 的 16 个字段在共享段中的槽位。 +// +// **字段级粒度是消除 lost update 的机制**：每个字段独立一个 Slice 描述符， +// 只改 FinalText 的插件完全不触碰 ToolResults 的描述符，因此不存在 +// "只读插件把自己收到的旧快照写回、覆盖他人改写"的问题（对比今日副本模型 +// 实测 35.8~36.8% 丢失率，见 §8.4）。 +// +// 字段内部的编码方式（原始字符串 vs JSON）不影响这一性质： +// ToolCall.Arguments 是 map[string]interface{}、ToolResult.Result 是 interface{}， +// 无法拆成定长结构，故以 JSON 存入 arena——工具结果中位数仅 93B（§2.5）， +// 序列化开销占 LLM 往返的 0.0001%，不构成瓶颈。 +type stageField int + +const ( + fRawMessage stageField = iota + fUserID + fGroupID + fLLMText + fReasoningContent + fFinalText + fResponse // 配合 fResponseSet 表达 *string 的 nil 语义 + fPhase + fContextMsgs // JSON + fToolCalls // JSON + fToolResults // JSON + fMemory // JSON + fTokenUsage // JSON + fErrors // JSON + fExtraMediaBlocks // JSON —— Extra 的 4 个键提升为具名字段（§3.3 已核实使用点） + fExtraMediaType + fExtraInputSource + fExtraOutputChannel + + stageFieldCount +) + +// 标志位区（紧跟描述符数组）：表达 bool 与指针的 nil 语义。 +const ( + flagNoMemory = 0 + flagResponseSet = 1 + flagCount = 8 // 预留到 8 字节，便于对齐与后续扩展 +) + +// ctxSize 是 ShmStageCtx 区域的总字节数。 +const ctxSize = int(stageFieldCount)*sliceSize + flagCount + +// Segment 是一块已 mmap 的共享段，内核与插件进程各持一个实例 +// （底层同一物理页，虚拟地址可不同）。 +type Segment struct { + data []byte // 完整 mmap 区域 +} + +// NewSegment 在给定的 mmap 区域上初始化段布局（内核侧调用一次）。 +func NewSegment(data []byte) (*Segment, error) { + if len(data) < headerSize+ctxSize+1 { + return nil, fmt.Errorf("proc: 共享段过小（%d 字节，至少需要 %d）", + len(data), headerSize+ctxSize+1) + } + s := &Segment{data: data} + + arenaBase := uint32(headerSize + ctxSize) + arenaCap := uint32(len(data)) - arenaBase + + binary.LittleEndian.PutUint32(data[offMagic:], shmMagic) + binary.LittleEndian.PutUint32(data[offVersion:], shmVersion) + binary.LittleEndian.PutUint32(data[offArenaBase:], arenaBase) + binary.LittleEndian.PutUint32(data[offArenaCap:], arenaCap) + binary.LittleEndian.PutUint32(data[offArenaUsed:], 0) + binary.LittleEndian.PutUint32(data[offCtxBase:], headerSize) + binary.LittleEndian.PutUint64(data[offSeq:], 0) + + // 描述符与标志位清零（IsUnset 语义依赖此） + for i := headerSize; i < headerSize+ctxSize; i++ { + data[i] = 0 + } + return s, nil +} + +// AttachSegment 挂载一块已由 NewSegment 初始化的区域（插件进程侧调用）。 +// 校验魔数与版本，避免版本不一致时静默错读。 +func AttachSegment(data []byte) (*Segment, error) { + if len(data) < headerSize+ctxSize { + return nil, fmt.Errorf("proc: 共享段过小（%d 字节）", len(data)) + } + if got := binary.LittleEndian.Uint32(data[offMagic:]); got != shmMagic { + return nil, fmt.Errorf("proc: 共享段魔数不匹配（0x%x，期望 0x%x）", got, shmMagic) + } + if got := binary.LittleEndian.Uint32(data[offVersion:]); got != shmVersion { + return nil, fmt.Errorf("proc: 共享段版本不匹配（%d，本内核 %d）——插件需用配套 plugindev 重编", + got, shmVersion) + } + return &Segment{data: data}, nil +} + +func (s *Segment) arenaBase() uint32 { return binary.LittleEndian.Uint32(s.data[offArenaBase:]) } +func (s *Segment) arenaCap() uint32 { return binary.LittleEndian.Uint32(s.data[offArenaCap:]) } +func (s *Segment) ctxBase() uint32 { return binary.LittleEndian.Uint32(s.data[offCtxBase:]) } + +// Seq 返回当前世代号。每次 WriteBack 成功后自增，供乐观读校验（§3.3）。 +func (s *Segment) Seq() uint64 { + return atomic.LoadUint64((*uint64)(ptrU64(s.data[offSeq:]))) +} + +func (s *Segment) bumpSeq() { atomic.AddUint64((*uint64)(ptrU64(s.data[offSeq:])), 1) } + +// ArenaUsed 返回 arena 已用字节数（诊断/压实判断用）。 +func (s *Segment) ArenaUsed() uint32 { + return atomic.LoadUint32((*uint32)(ptrU32(s.data[offArenaUsed:]))) +} + +// ArenaCap 返回 arena 容量。 +func (s *Segment) ArenaCap() uint32 { return s.arenaCap() } + +// alloc 在 arena 上分配 n 字节并返回相对 arenaBase 的偏移。 +// +// **append-only（§3.3）**：插件把 FinalText 从 10 字节改成 10KB 时分配新区域、 +// 更新描述符，旧区域留作垃圾；arena 用尽由内核在 stage 结束后（此时无插件持锁） +// 整体压实。代价是单次 stage 内写入总量有上限——**上限必须显式报错而非静默截断** +// （§4.4 风险登记）。 +// +// 调用方须持有 stage 写锁（锁仲裁见 lock.go），故这里用非原子的读-改-写即可； +// 仍用原子操作是为了让未持锁的诊断读取（ArenaUsed）不产生数据竞争。 +func (s *Segment) alloc(n int) (uint32, error) { + if n < 0 { + return 0, fmt.Errorf("proc: 非法分配长度 %d", n) + } + // 偏移 0 保留给"字段未设置"语义，故 arena 从 1 开始分配。 + used := s.ArenaUsed() + if used == 0 { + used = 1 + } + end := uint64(used) + uint64(n) + if end > uint64(s.arenaCap()) { + return 0, fmt.Errorf("proc: arena 空间不足——需要 %d 字节，剩余 %d 字节（容量 %d，已用 %d）；"+ + "单次 stage 写入总量超限，请减少写入或等待内核压实", + n, int64(s.arenaCap())-int64(used), s.arenaCap(), used) + } + atomic.StoreUint32((*uint32)(ptrU32(s.data[offArenaUsed:])), uint32(end)) + return used, nil +} + +// write 把 b 写入 arena 并返回描述符。空切片返回 {Off:1, Len:0} +// （非 IsUnset —— 表达"写入了空值"，与"未设置"区分）。 +func (s *Segment) write(b []byte) (Slice, error) { + if len(b) == 0 { + return Slice{Off: 1, Len: 0}, nil + } + off, err := s.alloc(len(b)) + if err != nil { + return Slice{}, err + } + base := s.arenaBase() + copy(s.data[base+off:base+off+uint32(len(b))], b) + return Slice{Off: off, Len: uint32(len(b))}, nil +} + +// read 按描述符取出 arena 中的字节（返回的是段内切片视图，调用方须在持锁期间使用）。 +func (s *Segment) read(sl Slice) ([]byte, error) { + if sl.IsUnset() || sl.Len == 0 { + return nil, nil + } + base := s.arenaBase() + if uint64(sl.Off)+uint64(sl.Len) > uint64(s.arenaCap()) { + return nil, fmt.Errorf("proc: 描述符越界（off=%d len=%d cap=%d）", sl.Off, sl.Len, s.arenaCap()) + } + return s.data[base+sl.Off : base+sl.Off+sl.Len], nil +} + +// descOffset 返回字段 f 的描述符在段内的绝对偏移。 +func (s *Segment) descOffset(f stageField) uint32 { + return s.ctxBase() + uint32(int(f)*sliceSize) +} + +func (s *Segment) getDesc(f stageField) Slice { + o := s.descOffset(f) + return Slice{ + Off: binary.LittleEndian.Uint32(s.data[o:]), + Len: binary.LittleEndian.Uint32(s.data[o+4:]), + } +} + +func (s *Segment) setDesc(f stageField, sl Slice) { + o := s.descOffset(f) + binary.LittleEndian.PutUint32(s.data[o:], sl.Off) + binary.LittleEndian.PutUint32(s.data[o+4:], sl.Len) +} + +func (s *Segment) flagsOffset() uint32 { + return s.ctxBase() + uint32(int(stageFieldCount)*sliceSize) +} + +func (s *Segment) getFlag(bit int) bool { + return s.data[s.flagsOffset()+uint32(bit)] != 0 +} + +func (s *Segment) setFlag(bit int, v bool) { + b := byte(0) + if v { + b = 1 + } + s.data[s.flagsOffset()+uint32(bit)] = b +} + +// Compact 回收 arena 垃圾：把仍被描述符引用的数据紧凑重排到段头部。 +// +// 必须在**无插件持锁**时调用（§3.3：由内核在 stage 结束后执行）。 +// 返回回收的字节数。 +func (s *Segment) Compact() uint32 { + before := s.ArenaUsed() + + // 收集现存描述符指向的数据，按字段顺序重新写入。 + type kept struct { + f stageField + data []byte + } + var live []kept + for f := stageField(0); f < stageFieldCount; f++ { + sl := s.getDesc(f) + if sl.IsUnset() { + continue + } + b, err := s.read(sl) + if err != nil { + // 描述符损坏：丢弃该字段而非让压实失败（诊断由上层日志承担） + s.setDesc(f, Slice{}) + continue + } + cp := make([]byte, len(b)) + copy(cp, b) + live = append(live, kept{f: f, data: cp}) + } + + // 重置游标后按序回填 + atomic.StoreUint32((*uint32)(ptrU32(s.data[offArenaUsed:])), 0) + for _, k := range live { + sl, err := s.write(k.data) + if err != nil { + // 压实后仍放不下：理论上不可能（总量未增），保守清空该字段 + s.setDesc(k.f, Slice{}) + continue + } + s.setDesc(k.f, sl) + } + + after := s.ArenaUsed() + if before > after { + return before - after + } + return 0 +} diff --git a/internal/plugin/proc/shm_test.go b/internal/plugin/proc/shm_test.go new file mode 100644 index 0000000..270b1aa --- /dev/null +++ b/internal/plugin/proc/shm_test.go @@ -0,0 +1,441 @@ +package proc + +import ( + "fmt" + "strings" + "sync" + "testing" + + pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk" +) + +// 本文件验证共享内存 stage 并发的正确性——**整个迁移最关键的一环**（§4.4 风险 3.4）。 +// +// 对照基线（今日 C ABI 副本模型）： +// - 内置插件（共享 *StageContext + RWMutex）：0% 丢失 +// - 外部插件（快照-副本-写回）：35.8~36.8% 丢失（实验 12），现网量级百分之几脏数据 +// +// 目标：共享内存 + 锁仲裁下，跨进程并发改写收敛到 0% 丢失。 + +// newTestSegment 造一块内存段模拟 mmap 区域（单测无需真 mmap： +// 编解码与 arena 逻辑与底层是 mmap 还是普通内存无关）。 +func newTestSegment(t *testing.T, size int) *Segment { + t.Helper() + buf := make([]byte, size) + seg, err := NewSegment(buf) + if err != nil { + t.Fatalf("NewSegment: %v", err) + } + return seg +} + +func TestSegment_AttachValidatesMagicAndVersion(t *testing.T) { + buf := make([]byte, 8192) + if _, err := NewSegment(buf); err != nil { + t.Fatalf("NewSegment: %v", err) + } + if _, err := AttachSegment(buf); err != nil { + t.Fatalf("AttachSegment 应成功: %v", err) + } + + // 魔数损坏 + bad := make([]byte, len(buf)) + copy(bad, buf) + bad[0] ^= 0xFF + if _, err := AttachSegment(bad); err == nil { + t.Error("魔数不匹配应报错（避免版本不一致时静默错读）") + } + + // 版本不匹配 + badVer := make([]byte, len(buf)) + copy(badVer, buf) + badVer[4] = 99 + if _, err := AttachSegment(badVer); err == nil { + t.Error("版本不匹配应报错") + } +} + +// 全部 16 个字段可跨进程往返——今日经 C ABI 只有 10 个字段可见（§8.3）。 +func TestSegment_RoundTripAllFields(t *testing.T) { + seg := newTestSegment(t, 16384) + + resp := "短路响应" + src := &pubsdk.StageContext{ + RawMessage: "原始输入", + UserID: "u1", + GroupID: "g1", + LLMText: "模型输出", + ReasoningContent: "思考过程", // C ABI 下外部插件看不到 + FinalText: "最终文本", + Response: &resp, + Phase: pubsdk.StageAfterToolcall, + NoMemory: true, + ContextMsgs: []map[string]interface{}{{"role": "user", "content": "hi"}}, // C ABI 看不到 + ToolCalls: []pubsdk.ToolCall{{ID: "t1", Name: "weather_query", Plugin: "weather"}}, + ToolResults: []pubsdk.ToolResult{{CallID: "t1", Name: "weather_query", Success: true, Result: "晴"}}, + Memory: []pubsdk.MemItem{{Role: "user", Content: "记忆", Score: 0.9}}, // C ABI 看不到 + TokenUsage: map[string]int{"prompt": 100, "completion": 50}, // C ABI 看不到 + Errors: []string{"err1"}, // C ABI 看不到 + Extra: map[string]interface{}{ + ExtraKeyMediaType: "image", + ExtraKeyInputSource: "qq", + ExtraKeyOutputChannel: "qq", + }, + } + + if err := seg.WriteAll(src); err != nil { + t.Fatalf("WriteAll: %v", err) + } + + var dst pubsdk.StageContext + if err := seg.ReadInto(&dst); err != nil { + t.Fatalf("ReadInto: %v", err) + } + + if dst.RawMessage != src.RawMessage || dst.UserID != src.UserID || dst.GroupID != src.GroupID { + t.Errorf("标量字段不一致: raw=%q uid=%q gid=%q", dst.RawMessage, dst.UserID, dst.GroupID) + } + if dst.ReasoningContent != "思考过程" { + t.Errorf("ReasoningContent 应可见（C ABI 下不可见）: %q", dst.ReasoningContent) + } + if len(dst.ContextMsgs) != 1 { + t.Errorf("ContextMsgs 应可见: %v", dst.ContextMsgs) + } + if len(dst.Memory) != 1 || dst.Memory[0].Score != 0.9 { + t.Errorf("Memory 应可见: %v", dst.Memory) + } + if dst.TokenUsage["prompt"] != 100 { + t.Errorf("TokenUsage 应可见: %v", dst.TokenUsage) + } + if len(dst.Errors) != 1 { + t.Errorf("Errors 应可见: %v", dst.Errors) + } + if dst.Response == nil || *dst.Response != resp { + t.Errorf("Response 应往返: %v", dst.Response) + } + if !dst.NoMemory { + t.Error("NoMemory 标志应往返") + } + if len(dst.ToolResults) != 1 || dst.ToolResults[0].Result != "晴" { + t.Errorf("ToolResults 应往返: %v", dst.ToolResults) + } + if dst.Extra[ExtraKeyMediaType] != "image" { + t.Errorf("Extra 提升字段应往返: %v", dst.Extra) + } +} + +// nil Response 与空字符串 Response 必须可区分（短路语义依赖此）。 +func TestSegment_ResponseNilVsEmpty(t *testing.T) { + seg := newTestSegment(t, 8192) + + if err := seg.WriteAll(&pubsdk.StageContext{RawMessage: "x"}); err != nil { + t.Fatalf("WriteAll: %v", err) + } + var d1 pubsdk.StageContext + if err := seg.ReadInto(&d1); err != nil { + t.Fatalf("ReadInto: %v", err) + } + if d1.Response != nil { + t.Errorf("未设置的 Response 应为 nil，实际 %q", *d1.Response) + } + + empty := "" + seg2 := newTestSegment(t, 8192) + if err := seg2.WriteAll(&pubsdk.StageContext{Response: &empty}); err != nil { + t.Fatalf("WriteAll: %v", err) + } + var d2 pubsdk.StageContext + if err := seg2.ReadInto(&d2); err != nil { + t.Fatalf("ReadInto: %v", err) + } + if d2.Response == nil { + t.Error("显式设为空串的 Response 不应读成 nil（短路语义会丢）") + } else if *d2.Response != "" { + t.Errorf("Response 应为空串，实际 %q", *d2.Response) + } +} + +// 只读插件的 WriteDirty 必须零写入——**这是消除 lost update 的核心断言**。 +func TestSegment_WriteDirty_ReadOnlyPluginWritesNothing(t *testing.T) { + seg := newTestSegment(t, 16384) + base := &pubsdk.StageContext{ + RawMessage: "查天气", + ToolResults: []pubsdk.ToolResult{{CallID: "c1", Name: "weather_query", Result: "已清洗"}}, + } + if err := seg.WriteAll(base); err != nil { + t.Fatalf("WriteAll: %v", err) + } + + // 插件侧：读入 → 只读 → 写回 + var local pubsdk.StageContext + if err := seg.ReadInto(&local); err != nil { + t.Fatalf("ReadInto: %v", err) + } + snap := TakeSnapshot(&local) + _ = local.ToolResults[0].Result // 只读，不改 + + n, err := seg.WriteDirty(&local, snap) + if err != nil { + t.Fatalf("WriteDirty: %v", err) + } + if n != 0 { + t.Fatalf("只读插件应零写回，实际写回 %d 个字段（会覆盖他人改写）", n) + } +} + +// 原地改切片元素必须被识别为脏 —— C ABI 侧修 11.3 时踩过的坑。 +func TestSegment_WriteDirty_InPlaceSliceMutationDetected(t *testing.T) { + seg := newTestSegment(t, 16384) + if err := seg.WriteAll(&pubsdk.StageContext{ + ToolResults: []pubsdk.ToolResult{{CallID: "c1", Result: "带\x1b[31mANSI\x1b[0m"}}, + }); err != nil { + t.Fatalf("WriteAll: %v", err) + } + + var local pubsdk.StageContext + if err := seg.ReadInto(&local); err != nil { + t.Fatalf("ReadInto: %v", err) + } + snap := TakeSnapshot(&local) + local.ToolResults[0].Result = "带ANSI" // 原地改元素（sanitizer 的实际行为） + + n, err := seg.WriteDirty(&local, snap) + if err != nil { + t.Fatalf("WriteDirty: %v", err) + } + if n != 1 { + t.Fatalf("原地改切片元素应被识别为 1 个脏字段，实际 %d", n) + } + + var after pubsdk.StageContext + if err := seg.ReadInto(&after); err != nil { + t.Fatalf("ReadInto: %v", err) + } + if after.ToolResults[0].Result != "带ANSI" { + t.Errorf("清洗结果未写回: %v", after.ToolResults[0].Result) + } +} + +// 复刻现网场景（实验 13）：sanitizer 改写 + weather 只读并发，清洗结果不得被覆盖。 +// 这是 C ABI 副本模型下量级百分之几脏数据的直接来源。 +func TestSegment_ProductionScenario_SanitizerNotOverwrittenByWeather(t *testing.T) { + seg := newTestSegment(t, 16384) + dirty := "天气：晴 \x1b[31m28°C\x1b[0m" + clean := "天气：晴 28°C" + + if err := seg.WriteAll(&pubsdk.StageContext{ + RawMessage: "查天气", + Phase: pubsdk.StageAfterToolcall, + ToolResults: []pubsdk.ToolResult{{CallID: "c1", Name: "weather_query", Result: dirty}}, + }); err != nil { + t.Fatalf("WriteAll: %v", err) + } + + lock := newStageLock() + + // sanitizer：拿锁 → 读 → 清洗 → 写脏字段 → 放锁 + runSanitizer := func() error { + if err := lock.Acquire("sanitizer"); err != nil { + return err + } + defer lock.Release("sanitizer") + var local pubsdk.StageContext + if err := seg.ReadInto(&local); err != nil { + return err + } + snap := TakeSnapshot(&local) + if len(local.ToolResults) > 0 { + s, _ := local.ToolResults[0].Result.(string) + local.ToolResults[0].Result = strings.NewReplacer("\x1b[31m", "", "\x1b[0m", "").Replace(s) + } + _, err := seg.WriteDirty(&local, snap) + return err + } + + // weather：拿锁 → 读 → 只读 → 零写回 → 放锁 + runWeather := func() error { + if err := lock.Acquire("weather"); err != nil { + return err + } + defer lock.Release("weather") + var local pubsdk.StageContext + if err := seg.ReadInto(&local); err != nil { + return err + } + snap := TakeSnapshot(&local) + if len(local.ToolResults) > 0 { + _ = local.ToolResults[0].Result // 只读 + } + n, err := seg.WriteDirty(&local, snap) + if err != nil { + return err + } + if n != 0 { + return fmt.Errorf("weather 只读却写回 %d 个字段", n) + } + return nil + } + + // 并发扇出（保留原始设计），weather 后完成是最坏情形 + var wg sync.WaitGroup + errs := make(chan error, 2) + wg.Add(2) + go func() { defer wg.Done(); errs <- runSanitizer() }() + go func() { defer wg.Done(); errs <- runWeather() }() + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatalf("插件执行失败: %v", err) + } + } + + var final pubsdk.StageContext + if err := seg.ReadInto(&final); err != nil { + t.Fatalf("ReadInto: %v", err) + } + got, _ := final.ToolResults[0].Result.(string) + if got != clean { + t.Fatalf("清洗结果被覆盖：期望 %q，实际 %q", clean, got) + } +} + +// 多插件高并发累加同一字段：总写入次数必须等于最终长度（零丢失零撕裂）。 +// 对应实验 8（5 进程 × 300 轮），这里在单进程内用 goroutine 模拟并发扇出， +// 验证共享段 + 锁仲裁 + 脏字段写回三者组合的正确性。 +func TestSegment_ConcurrentAppend_NoLostUpdate(t *testing.T) { + // arena 需容纳 append-only 的中间垃圾：每轮写入长度递增， + // 5 插件 × 40 轮 → 最长 200 字符，累计约 200*201/2 = 20100 字节，留足余量。 + seg := newTestSegment(t, 128*1024) + if err := seg.WriteAll(&pubsdk.StageContext{FinalText: ""}); err != nil { + t.Fatalf("WriteAll: %v", err) + } + + lock := newStageLock() + tags := []string{"A", "B", "C", "D", "E"} + const iters = 40 + + var wg sync.WaitGroup + errCh := make(chan error, len(tags)*iters) + + for _, tag := range tags { + wg.Add(1) + go func(tag string) { + defer wg.Done() + for i := 0; i < iters; i++ { + if err := lock.Acquire(tag); err != nil { + errCh <- err + return + } + var local pubsdk.StageContext + if err := seg.ReadInto(&local); err != nil { + lock.Release(tag) + errCh <- err + return + } + snap := TakeSnapshot(&local) + local.FinalText += tag // 读-改-写 + if _, err := seg.WriteDirty(&local, snap); err != nil { + lock.Release(tag) + errCh <- err + return + } + if err := lock.Release(tag); err != nil { + errCh <- err + return + } + } + }(tag) + } + wg.Wait() + close(errCh) + for err := range errCh { + if err != nil { + t.Fatalf("并发写入失败: %v", err) + } + } + + var final pubsdk.StageContext + if err := seg.ReadInto(&final); err != nil { + t.Fatalf("ReadInto: %v", err) + } + + // 关键断言：各标记出现次数之和 == 最终长度 ⇒ 无丢失、无撕裂 + total := 0 + counts := map[string]int{} + for _, tag := range tags { + c := strings.Count(final.FinalText, tag) + counts[tag] = c + total += c + } + if total != len(final.FinalText) { + t.Fatalf("出现撕裂：各标记计数之和 %d != 最终长度 %d（counts=%v）", + total, len(final.FinalText), counts) + } + want := len(tags) * iters + if total != want { + t.Fatalf("出现 lost update：期望 %d 次写入全部保留，实际 %d（counts=%v）", + want, total, counts) + } + for tag, c := range counts { + if c != iters { + t.Errorf("插件 %s 的写入丢失：期望 %d 次，实际 %d 次", tag, iters, c) + } + } +} + +// arena 用尽必须显式报错，不得静默截断（§4.4 风险登记）。 +func TestSegment_ArenaExhaustionReturnsError(t *testing.T) { + seg := newTestSegment(t, headerSize+ctxSize+256) // 极小 arena + big := strings.Repeat("x", 1024) + err := seg.WriteAll(&pubsdk.StageContext{FinalText: big}) + if err == nil { + t.Fatal("arena 不足应报错，而非静默截断") + } + if !strings.Contains(err.Error(), "arena 空间不足") { + t.Errorf("错误信息应说明 arena 不足，实际: %v", err) + } +} + +// 压实回收 append-only 垃圾，且不破坏现存字段。 +func TestSegment_CompactReclaimsGarbage(t *testing.T) { + seg := newTestSegment(t, 32*1024) + if err := seg.WriteAll(&pubsdk.StageContext{FinalText: "初始"}); err != nil { + t.Fatalf("WriteAll: %v", err) + } + + // 反复改写同一字段，制造 append-only 垃圾 + for i := 0; i < 50; i++ { + var local pubsdk.StageContext + if err := seg.ReadInto(&local); err != nil { + t.Fatalf("ReadInto: %v", err) + } + snap := TakeSnapshot(&local) + local.FinalText = fmt.Sprintf("第%d次改写内容", i) + if _, err := seg.WriteDirty(&local, snap); err != nil { + t.Fatalf("WriteDirty: %v", err) + } + } + + usedBefore := seg.ArenaUsed() + var beforeCtx pubsdk.StageContext + if err := seg.ReadInto(&beforeCtx); err != nil { + t.Fatalf("ReadInto: %v", err) + } + + reclaimed := seg.Compact() + if reclaimed == 0 { + t.Error("应回收到垃圾空间") + } + if seg.ArenaUsed() >= usedBefore { + t.Errorf("压实后已用空间应下降：%d → %d", usedBefore, seg.ArenaUsed()) + } + + var afterCtx pubsdk.StageContext + if err := seg.ReadInto(&afterCtx); err != nil { + t.Fatalf("压实后 ReadInto: %v", err) + } + if afterCtx.FinalText != beforeCtx.FinalText { + t.Errorf("压实破坏了字段内容：%q → %q", beforeCtx.FinalText, afterCtx.FinalText) + } +} diff --git a/internal/plugin/proc/shmcodec.go b/internal/plugin/proc/shmcodec.go new file mode 100644 index 0000000..c758eae --- /dev/null +++ b/internal/plugin/proc/shmcodec.go @@ -0,0 +1,514 @@ +package proc + +import ( + "encoding/json" + "fmt" + + pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk" +) + +// StageContext 的跨进程编解码（§3.3 数据面 / §3.4 SDK 封装全部复杂度）。 +// +// 设计要点： +// +// 1. **插件作者永远不接触 Slice{off,len}**。插件进程内保留原生 +// *pubsdk.StageContext，handler 照常读写字段；stage 入口从共享段 +// 反序列化成本地对象，handler 返回时把**脏字段**写回共享段。 +// +// 2. **字段级描述符消除 lost update**。只改 FinalText 的插件不触碰 +// ToolResults 的描述符，故不存在"只读插件把旧快照写回覆盖他人改写" +// （今日副本模型实测 35.8~36.8% 丢失，§8.4）。 +// +// 3. **全部 16 个字段可见**。今日经 C ABI 只下发 10 个字段，外部插件永远 +// 看不到 ContextMsgs/ReasoningContent/TokenUsage/Memory/Extra/Errors（§8.3）； +// 共享内存下全部可见可改——接口形式不变，能力变强。 +// +// 4. **Extra 的 4 个键提升为具名字段**（§3.3 已核实全部使用点仅这 4 个）： +// media_blocks / media_type / input_source / output_channel。 +// 它们都是内核写、插件读，无并发改写需求；真正需要多插件并发改的 +// （LLMText/FinalText/ToolCalls/Errors）全是强类型字段。 + +// extra 中被提升为具名字段的键。 +const ( + ExtraKeyMediaBlocks = "media_blocks" + ExtraKeyMediaType = "media_type" + ExtraKeyInputSource = "input_source" + ExtraKeyOutputChannel = "output_channel" +) + +// WriteAll 把整个 StageContext 写入共享段（内核侧在 stage 开始前调用一次）。 +// 调用方须持有写锁。 +func (s *Segment) WriteAll(sc *pubsdk.StageContext) error { + sc.RLock() + snap := captureLocal(sc) + sc.RUnlock() + return s.writeLocal(snap) +} + +// localCtx 是 StageContext 的值快照，用于在不持有 sc 锁的情况下做编解码。 +type localCtx struct { + RawMessage string + UserID string + GroupID string + LLMText string + ReasoningContent string + FinalText string + Response *string + Phase string + NoMemory bool + ContextMsgs []map[string]interface{} + ToolCalls []pubsdk.ToolCall + ToolResults []pubsdk.ToolResult + Memory []pubsdk.MemItem + TokenUsage map[string]int + Errors []string + ExtraMediaBlocks interface{} + ExtraMediaType interface{} + ExtraInputSource interface{} + ExtraOutputChan interface{} +} + +func captureLocal(sc *pubsdk.StageContext) *localCtx { + l := &localCtx{ + RawMessage: sc.RawMessage, + UserID: sc.UserID, + GroupID: sc.GroupID, + LLMText: sc.LLMText, + ReasoningContent: sc.ReasoningContent, + FinalText: sc.FinalText, + Response: sc.Response, + Phase: string(sc.Phase), + NoMemory: sc.NoMemory, + ContextMsgs: sc.ContextMsgs, + ToolCalls: sc.ToolCalls, + ToolResults: sc.ToolResults, + Memory: sc.Memory, + TokenUsage: sc.TokenUsage, + Errors: sc.Errors, + } + if sc.Extra != nil { + l.ExtraMediaBlocks = sc.Extra[ExtraKeyMediaBlocks] + l.ExtraMediaType = sc.Extra[ExtraKeyMediaType] + l.ExtraInputSource = sc.Extra[ExtraKeyInputSource] + l.ExtraOutputChan = sc.Extra[ExtraKeyOutputChannel] + } + return l +} + +func (s *Segment) writeLocal(l *localCtx) error { + putStr := func(f stageField, v string) error { + sl, err := s.write([]byte(v)) + if err != nil { + return fmt.Errorf("写入字段 %s: %w", f, err) + } + s.setDesc(f, sl) + return nil + } + putJSON := func(f stageField, v interface{}) error { + if v == nil { + s.setDesc(f, Slice{}) + return nil + } + b, err := json.Marshal(v) + if err != nil { + return fmt.Errorf("序列化字段 %s: %w", f, err) + } + sl, err := s.write(b) + if err != nil { + return fmt.Errorf("写入字段 %s: %w", f, err) + } + s.setDesc(f, sl) + return nil + } + + for _, step := range []struct { + f stageField + v string + }{ + {fRawMessage, l.RawMessage}, + {fUserID, l.UserID}, + {fGroupID, l.GroupID}, + {fLLMText, l.LLMText}, + {fReasoningContent, l.ReasoningContent}, + {fFinalText, l.FinalText}, + {fPhase, l.Phase}, + } { + if err := putStr(step.f, step.v); err != nil { + return err + } + } + + // Response 是 *string：用标志位表达 nil，避免 "" 与 nil 混淆 + if l.Response != nil { + if err := putStr(fResponse, *l.Response); err != nil { + return err + } + s.setFlag(flagResponseSet, true) + } else { + s.setDesc(fResponse, Slice{}) + s.setFlag(flagResponseSet, false) + } + s.setFlag(flagNoMemory, l.NoMemory) + + // 切片/映射字段：nil 与空切片都写成"未设置"，避免插件收到 [] 后误以为 + // 内核显式清空过（与今日 writable 的 len>0 才下发语义一致）。 + jsonFields := []struct { + f stageField + v interface{} + }{ + {fContextMsgs, sliceOrNil(len(l.ContextMsgs), l.ContextMsgs)}, + {fToolCalls, sliceOrNil(len(l.ToolCalls), l.ToolCalls)}, + {fToolResults, sliceOrNil(len(l.ToolResults), l.ToolResults)}, + {fMemory, sliceOrNil(len(l.Memory), l.Memory)}, + {fTokenUsage, sliceOrNil(len(l.TokenUsage), l.TokenUsage)}, + {fErrors, sliceOrNil(len(l.Errors), l.Errors)}, + {fExtraMediaBlocks, l.ExtraMediaBlocks}, + {fExtraMediaType, l.ExtraMediaType}, + {fExtraInputSource, l.ExtraInputSource}, + {fExtraOutputChannel, l.ExtraOutputChan}, + } + for _, step := range jsonFields { + if err := putJSON(step.f, step.v); err != nil { + return err + } + } + s.bumpSeq() + return nil +} + +// sliceOrNil 让长度为 0 的容器写成 nil（未设置），非零则原样返回。 +func sliceOrNil(n int, v interface{}) interface{} { + if n == 0 { + return nil + } + return v +} + +// ReadInto 从共享段读出全部字段填充到 sc（插件进程侧 stage 入口调用）。 +// 调用方须持有读锁或写锁。 +func (s *Segment) ReadInto(sc *pubsdk.StageContext) error { + getStr := func(f stageField) (string, error) { + b, err := s.read(s.getDesc(f)) + if err != nil { + return "", fmt.Errorf("读取字段 %s: %w", f, err) + } + return string(b), nil + } + getJSON := func(f stageField, out interface{}) error { + b, err := s.read(s.getDesc(f)) + if err != nil { + return fmt.Errorf("读取字段 %s: %w", f, err) + } + if len(b) == 0 { + return nil + } + if err := json.Unmarshal(b, out); err != nil { + return fmt.Errorf("反序列化字段 %s: %w", f, err) + } + return nil + } + + raw, err := getStr(fRawMessage) + if err != nil { + return err + } + uid, err := getStr(fUserID) + if err != nil { + return err + } + gid, err := getStr(fGroupID) + if err != nil { + return err + } + llm, err := getStr(fLLMText) + if err != nil { + return err + } + reason, err := getStr(fReasoningContent) + if err != nil { + return err + } + final, err := getStr(fFinalText) + if err != nil { + return err + } + phase, err := getStr(fPhase) + if err != nil { + return err + } + + var ctxMsgs []map[string]interface{} + var toolCalls []pubsdk.ToolCall + var toolResults []pubsdk.ToolResult + var mem []pubsdk.MemItem + var usage map[string]int + var errs []string + if err := getJSON(fContextMsgs, &ctxMsgs); err != nil { + return err + } + if err := getJSON(fToolCalls, &toolCalls); err != nil { + return err + } + if err := getJSON(fToolResults, &toolResults); err != nil { + return err + } + if err := getJSON(fMemory, &mem); err != nil { + return err + } + if err := getJSON(fTokenUsage, &usage); err != nil { + return err + } + if err := getJSON(fErrors, &errs); err != nil { + return err + } + + extra := map[string]interface{}{} + for _, pair := range []struct { + f stageField + key string + }{ + {fExtraMediaBlocks, ExtraKeyMediaBlocks}, + {fExtraMediaType, ExtraKeyMediaType}, + {fExtraInputSource, ExtraKeyInputSource}, + {fExtraOutputChannel, ExtraKeyOutputChannel}, + } { + var v interface{} + if err := getJSON(pair.f, &v); err != nil { + return err + } + if v != nil { + extra[pair.key] = v + } + } + + var respPtr *string + if s.getFlag(flagResponseSet) { + r, err := getStr(fResponse) + if err != nil { + return err + } + respPtr = &r + } + + sc.Lock() + defer sc.Unlock() + sc.RawMessage = raw + sc.UserID = uid + sc.GroupID = gid + sc.LLMText = llm + sc.ReasoningContent = reason + sc.FinalText = final + sc.Phase = pubsdk.Stage(phase) + sc.NoMemory = s.getFlag(flagNoMemory) + sc.Response = respPtr + sc.ContextMsgs = ctxMsgs + sc.ToolCalls = toolCalls + sc.ToolResults = toolResults + sc.Memory = mem + sc.TokenUsage = usage + sc.Errors = errs + if len(extra) > 0 { + sc.Extra = extra + } else { + sc.Extra = nil + } + return nil +} + +// WriteDirty 只把与 base 快照不同的字段写回共享段（插件进程侧 handler 返回后调用）。 +// +// **这是消除 lost update 的关键**：只读插件的 dirty 集为空 → 零写入 → +// 不可能覆盖其他插件的改写。对比今日副本模型无条件回传 10 个字段的行为 +// （§8.4 实测 35.8~36.8% 丢失，现网量级百分之几的脏数据进 LLM）。 +// +// 返回实际写回的字段数，便于诊断与测试断言。 +func (s *Segment) WriteDirty(sc *pubsdk.StageContext, base *Snapshot) (int, error) { + sc.RLock() + cur := captureLocal(sc) + sc.RUnlock() + + now := newSnapshotFromLocal(cur) + changed := 0 + + putStr := func(f stageField, v string) error { + sl, err := s.write([]byte(v)) + if err != nil { + return fmt.Errorf("写回字段 %s: %w", f, err) + } + s.setDesc(f, sl) + changed++ + return nil + } + putRaw := func(f stageField, raw string) error { + if raw == "" { + s.setDesc(f, Slice{}) + changed++ + return nil + } + sl, err := s.write([]byte(raw)) + if err != nil { + return fmt.Errorf("写回字段 %s: %w", f, err) + } + s.setDesc(f, sl) + changed++ + return nil + } + + for _, step := range []struct { + f stageField + v string + }{ + {fRawMessage, cur.RawMessage}, + {fUserID, cur.UserID}, + {fGroupID, cur.GroupID}, + {fLLMText, cur.LLMText}, + {fReasoningContent, cur.ReasoningContent}, + {fFinalText, cur.FinalText}, + {fPhase, cur.Phase}, + } { + if base.strs[step.f] != now.strs[step.f] { + if err := putStr(step.f, step.v); err != nil { + return changed, err + } + } + } + + // JSON 字段：比较序列化结果 + for f := range now.jsons { + if base.jsons[f] != now.jsons[f] { + if err := putRaw(f, now.jsons[f]); err != nil { + return changed, err + } + } + } + + // Response 的 nil 语义变化也算脏 + if base.responseSet != now.responseSet || base.response != now.response { + if now.responseSet { + if err := putStr(fResponse, now.response); err != nil { + return changed, err + } + s.setFlag(flagResponseSet, true) + } else { + // 插件把 Response 置回 nil：短路语义不应被撑销，故不清空内核已设的值。 + // 与 C ABI 路径 applyStageResult 的行为保持一致。 + s.setFlag(flagResponseSet, s.getFlag(flagResponseSet)) + } + } + if base.noMemory != now.noMemory { + s.setFlag(flagNoMemory, now.noMemory) + changed++ + } + + if changed > 0 { + s.bumpSeq() + } + return changed, nil +} + +// Snapshot 是 handler 运行前的字段快照，用于计算脏字段。 +// +// ❗ 必须存**序列化后的字符串**而非 Go 值：StageContext 的切片字段与 +// 调用方共享底层数组，handler 原地改元素（sc.ToolResults[0].Result = x） +// 时直接持有的 Go 值快照会跟着变，脏字段计算失效——这个坑在 C ABI 侧 +// 修 11.3 时已经踩过一次（见 SDK 仓 stagediff_test.go 的注释）。 +type Snapshot struct { + strs map[stageField]string + jsons map[stageField]string + response string + responseSet bool + noMemory bool +} + +// Snapshot 抓取当前 StageContext 的快照（插件进程侧 handler 前调用）。 +func TakeSnapshot(sc *pubsdk.StageContext) *Snapshot { + sc.RLock() + l := captureLocal(sc) + sc.RUnlock() + return newSnapshotFromLocal(l) +} + +func newSnapshotFromLocal(l *localCtx) *Snapshot { + sn := &Snapshot{ + strs: map[stageField]string{}, + jsons: map[stageField]string{}, + } + sn.strs[fRawMessage] = l.RawMessage + sn.strs[fUserID] = l.UserID + sn.strs[fGroupID] = l.GroupID + sn.strs[fLLMText] = l.LLMText + sn.strs[fReasoningContent] = l.ReasoningContent + sn.strs[fFinalText] = l.FinalText + sn.strs[fPhase] = l.Phase + + marshal := func(v interface{}) string { + if v == nil { + return "" + } + b, err := json.Marshal(v) + if err != nil { + return "" + } + return string(b) + } + sn.jsons[fContextMsgs] = marshal(sliceOrNil(len(l.ContextMsgs), l.ContextMsgs)) + sn.jsons[fToolCalls] = marshal(sliceOrNil(len(l.ToolCalls), l.ToolCalls)) + sn.jsons[fToolResults] = marshal(sliceOrNil(len(l.ToolResults), l.ToolResults)) + sn.jsons[fMemory] = marshal(sliceOrNil(len(l.Memory), l.Memory)) + sn.jsons[fTokenUsage] = marshal(sliceOrNil(len(l.TokenUsage), l.TokenUsage)) + sn.jsons[fErrors] = marshal(sliceOrNil(len(l.Errors), l.Errors)) + sn.jsons[fExtraMediaBlocks] = marshal(l.ExtraMediaBlocks) + sn.jsons[fExtraMediaType] = marshal(l.ExtraMediaType) + sn.jsons[fExtraInputSource] = marshal(l.ExtraInputSource) + sn.jsons[fExtraOutputChannel] = marshal(l.ExtraOutputChan) + + if l.Response != nil { + sn.response = *l.Response + sn.responseSet = true + } + sn.noMemory = l.NoMemory + return sn +} + +// String 让字段枚举在错误信息里可读。 +func (f stageField) String() string { + switch f { + case fRawMessage: + return "raw_message" + case fUserID: + return "user_id" + case fGroupID: + return "group_id" + case fLLMText: + return "llm_text" + case fReasoningContent: + return "reasoning_content" + case fFinalText: + return "final_text" + case fResponse: + return "response" + case fPhase: + return "phase" + case fContextMsgs: + return "context_msgs" + case fToolCalls: + return "tool_calls" + case fToolResults: + return "tool_results" + case fMemory: + return "memory" + case fTokenUsage: + return "token_usage" + case fErrors: + return "errors" + case fExtraMediaBlocks: + return "extra." + ExtraKeyMediaBlocks + case fExtraMediaType: + return "extra." + ExtraKeyMediaType + case fExtraInputSource: + return "extra." + ExtraKeyInputSource + case fExtraOutputChannel: + return "extra." + ExtraKeyOutputChannel + } + return fmt.Sprintf("field(%d)", int(f)) +} diff --git a/internal/plugin/proc/unsafe.go b/internal/plugin/proc/unsafe.go new file mode 100644 index 0000000..093f2ba --- /dev/null +++ b/internal/plugin/proc/unsafe.go @@ -0,0 +1,15 @@ +package proc + +import "unsafe" + +// ptrU32 / ptrU64 把段内字节切片起始处重解释为原子操作可用的指针。 +// +// 共享段由 mmap 得到，其起始地址天然按页对齐（4096），本包所有原子字段 +// （offArenaUsed=16 对齐 4、offSeq=24 对齐 8）都落在对齐位置， +// 因此该重解释是安全的。 +// +// 这是本包唯一使用 unsafe 的地方，且**不涉及 cgo**—— +// 迁移的一个目标就是整个新架构零 cgo（§3.7 锁仲裁回归内核）。 +func ptrU32(b []byte) unsafe.Pointer { return unsafe.Pointer(&b[0]) } + +func ptrU64(b []byte) unsafe.Pointer { return unsafe.Pointer(&b[0]) } diff --git a/internal/plugin/registry.go b/internal/plugin/registry.go index 75a387d..53ce5a9 100644 --- a/internal/plugin/registry.go +++ b/internal/plugin/registry.go @@ -360,10 +360,12 @@ func (r *Registry) isDisabled(name string) bool { return r.cfgReg.IsPluginDisabled(name) } -// pluginEntryHash 计算插件入口文件（plugin.so 或 main.lua）的 SHA256，用于增量重载对比。 +// pluginEntryHash 计算插件入口文件的 SHA256，用于增量重载对比。 // 无入口文件（内置纯工厂插件）返回空字符串（始终视为已加载）。 +// plugin.bin 排在最前：与 detectEntryKind 保持一致的优先级，迁移期间同目录 +// 两种产物共存时以子进程产物为准。 func pluginEntryHash(plgDir string) string { - for _, candidate := range []string{"plugin.so", "plugin.dll", "main.lua", "SKILL.md"} { + for _, candidate := range []string{binEntry, soEntry, dllEntry, "plugin.dylib", luaEntry, skillEntry} { path := filepath.Join(plgDir, candidate) if data, err := os.ReadFile(path); err == nil && len(data) > 0 { sum := sha256.Sum256(data) @@ -763,6 +765,7 @@ func (r *Registry) DisablePlugin(name, by string) error { } func (r *Registry) EnablePlugin(name string) error { return r.Enable(name) } + // StopAndUnload 停止并从注册表移除插件，但保留其配置表（config_）。 // 供插件更新/升级流程使用：换 so/文件不动配置，重装后配置原样生效。 // 不执行 onRemove 回调（那是删除专用语义）。目录由调用方管理。 @@ -888,7 +891,23 @@ func (r *Registry) PluginDir() string { } func (r *Registry) tryDynamic(plgDir, name string, config map[string]interface{}) (sdk.Plugin, error) { - // 尝试顺序：.so (Go plugin on Linux) → .dll (Windows) → .lua (跨平台) + // 按 manifest entry 分派到对应加载通道（外部插件多进程化：.so/.dll 与 .bin 双通道共存）。 + // 这使迁移可逐插件推进、随时回退——把 entry 改回 plugin.so 即回到旧通道。 + if detectEntryKind(plgDir) == entryProc { + plg, err := tryLoadProc(plgDir, name, config) + if err != nil { + return nil, err + } + if plg != nil { + log.Printf("[plugin] %s: 经 proc 通道加载（子进程）", name) + return plg, nil + } + // entry 声明了 plugin.bin 但文件不存在/不可用 → 不隐式回退到 cabi， + // 否则"已迁移插件静默跑回旧通道"极难排查。 + return nil, fmt.Errorf("plugin %s: entry 声明 %s 但未找到可用二进制", name, binEntry) + } + + // 既有探测顺序（保持不变）：.so → .dll → .lua for _, try := range []struct { name string fn func(string, string, map[string]interface{}) (sdk.Plugin, error)