diff --git a/internal/plugin/proc/capability_test.go b/internal/plugin/proc/capability_test.go index 7071946..2332b44 100644 --- a/internal/plugin/proc/capability_test.go +++ b/internal/plugin/proc/capability_test.go @@ -19,12 +19,13 @@ func TestCapability_AllMethodsClassified(t *testing.T) { // 与 protocol.go 的 method 常量对齐。内核→插件的 7 个调用不经 Handle， // 故不需要能力归属。 kernelToPlugin := map[string]bool{ - MethodPluginInit: true, - MethodPluginStart: true, - MethodPluginStop: true, - MethodToolInvoke: true, - MethodStageInvoke: true, - MethodOutputInvoke: true, + MethodPluginInit: true, + MethodPluginStart: true, + MethodPluginStop: true, + MethodToolInvoke: true, + MethodCleanerInvoke: true, + MethodStageInvoke: true, + MethodOutputInvoke: true, } // 插件→内核的全部 method（手工清单，与 protocol.go 对照） diff --git a/internal/plugin/proc/corehandler.go b/internal/plugin/proc/corehandler.go index b059891..87597ed 100644 --- a/internal/plugin/proc/corehandler.go +++ b/internal/plugin/proc/corehandler.go @@ -31,9 +31,10 @@ type coreHandler struct { // locks 是 host.locks 的引用，供 stage.lock/unlock 路由。 locks *lockRegistry - // invokeTool/invokeStageFn/invokeOutput 反向调用插件（内核 → 插件）。 + // invokeTool/invokeCleaner/invokeStageFn/invokeOutput 反向调用插件（内核 → 插件）。 // 由 Plugin 注入，注册回调时用它们构造 handler。 invokeTool func(name string, args map[string]interface{}) (interface{}, error) + invokeCleaner func(scope, name, text string) (string, error) invokeStageFn func(ctx context.Context, stage string, seq uint64) error invokeOutput func(channel string, args map[string]interface{}) (interface{}, error) @@ -128,15 +129,24 @@ func (h *coreHandler) Handle(method string, params json.RawMessage) (interface{} return nil, h.sdk.RegisterPluginAPI(p.Name) case MethodInputRegister: var p struct { - Name string `json:"name"` - Def pubsdk.ChannelDef `json:"def"` + Name string `json:"name"` + Def pubsdk.ChannelDef `json:"def"` + HasCleaner bool `json:"has_cleaner"` } if err := unmarshal(params, &p); err != nil { return nil, err } - // 注意 ChannelDef.Cleaner 是函数，无法跨进程传递（§3.5 回调型资源）。 - // NoMemory 可传；Cleaner 若插件需要，须在插件侧对文本预处理后再注入。 - return nil, h.sdk.RegisterInputChannel(p.Name, pubsdk.ChannelDef{NoMemory: p.Def.NoMemory}) + if p.Name == "" { + return nil, fmt.Errorf("input.register: 缺少 name") + } + cleaner, err := h.cleanerProxy(CleanerScopeInput, p.Name, p.HasCleaner) + if err != nil { + return nil, fmt.Errorf("input.register: %w", err) + } + return nil, h.sdk.RegisterInputChannel(p.Name, pubsdk.ChannelDef{ + NoMemory: p.Def.NoMemory, + Cleaner: cleaner, + }) // ---- IO 注入（原 case 5/6/7/47）---- case MethodIOInjectText: @@ -580,11 +590,30 @@ func errUnavailable(what string) error { return fmt.Errorf("%s 能力在当前内核实例中不可用", what) } +// cleanerProxy 把进程内函数式 Cleaner 恢复成内核侧透明代理。 +// RPC 失败时返回原文：清洗是计算层优化，不能因插件暂时离线而丢失内容。 +func (h *coreHandler) cleanerProxy(scope, name string, enabled bool) (func(string) string, error) { + if !enabled { + return nil, nil + } + if h.invokeCleaner == nil { + return nil, fmt.Errorf("%s %s 声明 Cleaner，但清洗回调通道未就绪", scope, name) + } + return func(text string) string { + cleaned, err := h.invokeCleaner(scope, name, text) + if err != nil { + return text + } + return cleaned + }, nil +} + // toolRegister 注册插件工具，handler 反向调用插件执行（原 case 1）。 func (h *coreHandler) toolRegister(params json.RawMessage) (interface{}, error) { var p struct { - Name string `json:"name"` - Def pubsdk.ToolDef `json:"def"` + Name string `json:"name"` + Def pubsdk.ToolDef `json:"def"` + HasCleaner bool `json:"has_cleaner"` } if err := unmarshal(params, &p); err != nil { return nil, err @@ -593,8 +622,12 @@ func (h *coreHandler) toolRegister(params json.RawMessage) (interface{}, error) return nil, fmt.Errorf("tool.register: 缺少 name") } p.Def.Plugin = h.name - // ToolDef.Cleaner 是函数，跨进程无法传递（§3.5）——与 C ABI 路径行为一致。 - p.Def.Cleaner = nil + // 函数本身不进 JSON；has_cleaner 只声明其存在，实际执行回到插件进程。 + cleaner, err := h.cleanerProxy(CleanerScopeTool, p.Name, p.HasCleaner) + if err != nil { + return nil, fmt.Errorf("tool.register: %w", err) + } + p.Def.Cleaner = cleaner name := p.Name return nil, h.sdk.RegisterTool(name, p.Def, func(args map[string]interface{}) (interface{}, error) { @@ -637,10 +670,11 @@ func (h *coreHandler) stageRegister(params json.RawMessage) (interface{}, error) // 永远返回成功（§9.4，现网 2 次消息发不出而模型以为成功）。 func (h *coreHandler) outputRegister(params json.RawMessage) (interface{}, error) { var p struct { - Name string `json:"name"` - Caps int `json:"caps"` - Desc string `json:"desc"` - Def pubsdk.ChannelDef `json:"def"` + Name string `json:"name"` + Caps int `json:"caps"` + Desc string `json:"desc"` + Def pubsdk.ChannelDef `json:"def"` + HasCleaner bool `json:"has_cleaner"` } if err := unmarshal(params, &p); err != nil { return nil, err @@ -649,8 +683,12 @@ func (h *coreHandler) outputRegister(params json.RawMessage) (interface{}, error return nil, fmt.Errorf("output.register: 缺少 name") } channel := p.Name + cleaner, err := h.cleanerProxy(CleanerScopeOutput, channel, p.HasCleaner) + if err != nil { + return nil, fmt.Errorf("output.register: %w", err) + } return nil, h.sdk.RegisterOutputChannel(channel, p.Caps, p.Desc, - pubsdk.ChannelDef{NoMemory: p.Def.NoMemory}, + pubsdk.ChannelDef{NoMemory: p.Def.NoMemory, Cleaner: cleaner}, func(args map[string]interface{}) (interface{}, error) { return h.invokeOutput(channel, args) }) diff --git a/internal/plugin/proc/plugin.go b/internal/plugin/proc/plugin.go index 70a4be9..aa3ff40 100644 --- a/internal/plugin/proc/plugin.go +++ b/internal/plugin/proc/plugin.go @@ -119,6 +119,7 @@ func (p *Plugin) Start(core CoreSDK) error { } // 反向调用闭包：注册回调时捕获，运行期经 RPC 打到插件进程。 p.handler.invokeTool = p.invokeTool + p.handler.invokeCleaner = p.invokeCleaner p.handler.invokeStageFn = p.invokeStage p.handler.invokeOutput = p.invokeOutput @@ -220,6 +221,26 @@ func (p *Plugin) invokeTool(name string, args map[string]interface{}) (interface return res.Result, nil } +// invokeCleaner 在插件进程内执行工具或通道注册时提供的 Cleaner 函数。 +func (p *Plugin) invokeCleaner(scope, name, text string) (string, error) { + if p.proc == nil { + return "", ErrProcessExited + } + raw, err := p.proc.Call(MethodCleanerInvoke, CleanerInvokeParams{ + Scope: scope, + Name: name, + Text: text, + }) + if err != nil { + return "", err + } + var res CleanerInvokeResult + if err := json.Unmarshal(raw, &res); err != nil { + return "", fmt.Errorf("proc: %s %s Cleaner %s 应答解析失败: %w", p.name, scope, name, err) + } + return res.Text, nil +} + func (p *Plugin) invokeStage(ctx context.Context, stage string, seq uint64) error { if p.proc == nil { return ErrProcessExited diff --git a/internal/plugin/proc/plugin_test.go b/internal/plugin/proc/plugin_test.go index 253b668..193b8a8 100644 --- a/internal/plugin/proc/plugin_test.go +++ b/internal/plugin/proc/plugin_test.go @@ -17,20 +17,26 @@ import ( // fakeCoreSDK 是最简 CoreSDK 实现，记录注册行为。 type fakeCoreSDK struct { - mu sync.Mutex - tools map[string]pubsdk.ToolHandler - stages map[pubsdk.Stage][]pubsdk.StageHandler - outputs map[string]pubsdk.ToolHandler - settings map[string]interface{} - autoStart bool + mu sync.Mutex + tools map[string]pubsdk.ToolHandler + toolDefs map[string]pubsdk.ToolDef + stages map[pubsdk.Stage][]pubsdk.StageHandler + outputs map[string]pubsdk.ToolHandler + outputDefs map[string]pubsdk.ChannelDef + inputDefs map[string]pubsdk.ChannelDef + settings map[string]interface{} + autoStart bool } func newFakeCore() *fakeCoreSDK { return &fakeCoreSDK{ - tools: map[string]pubsdk.ToolHandler{}, - stages: map[pubsdk.Stage][]pubsdk.StageHandler{}, - outputs: map[string]pubsdk.ToolHandler{}, - settings: map[string]interface{}{}, + tools: map[string]pubsdk.ToolHandler{}, + toolDefs: map[string]pubsdk.ToolDef{}, + stages: map[pubsdk.Stage][]pubsdk.StageHandler{}, + outputs: map[string]pubsdk.ToolHandler{}, + outputDefs: map[string]pubsdk.ChannelDef{}, + inputDefs: map[string]pubsdk.ChannelDef{}, + settings: map[string]interface{}{}, } } @@ -59,6 +65,7 @@ func (f *fakeCoreSDK) RegisterTool(name string, def pubsdk.ToolDef, h pubsdk.Too f.mu.Lock() defer f.mu.Unlock() f.tools[name] = h + f.toolDefs[name] = def return nil } @@ -72,10 +79,16 @@ func (f *fakeCoreSDK) RegisterOutputChannel(name string, caps int, desc string, f.mu.Lock() defer f.mu.Unlock() f.outputs[name] = h + f.outputDefs[name] = def return nil } -func (f *fakeCoreSDK) RegisterInputChannel(name string, def pubsdk.ChannelDef) error { return nil } +func (f *fakeCoreSDK) RegisterInputChannel(name string, def pubsdk.ChannelDef) error { + f.mu.Lock() + defer f.mu.Unlock() + f.inputDefs[name] = def + return nil +} func (f *fakeCoreSDK) stageHandlers(stage pubsdk.Stage) []pubsdk.StageHandler { f.mu.Lock() @@ -232,6 +245,36 @@ func TestPlugin_RegisteredToolInvokable(t *testing.T) { if res != "ABC" { t.Fatalf("工具结果应为 ABC，实际 %v", res) } + + core.mu.Lock() + def, ok := core.toolDefs["demo_upper"] + core.mu.Unlock() + if !ok { + t.Fatal("内核未保存 demo_upper 的 ToolDef") + } + if def.Cleaner == nil { + t.Fatal("跨进程注册后 Cleaner 不应丢失") + } + if got := def.Cleaner("raw-output"); got != "tool-cleaned:raw-output" { + t.Fatalf("跨进程工具 Cleaner 结果错误：got %q, want %q", got, "tool-cleaned:raw-output") + } + + core.mu.Lock() + inputDef, inputOK := core.inputDefs["demo_in"] + outputDef, outputOK := core.outputDefs["demo_ch"] + core.mu.Unlock() + if !inputOK || inputDef.Cleaner == nil { + t.Fatal("跨进程注册后输入通道 Cleaner 不应丢失") + } + if got := inputDef.Cleaner("raw-input"); got != "input-cleaned:raw-input" { + t.Fatalf("跨进程输入 Cleaner 结果错误：got %q", got) + } + if !outputOK || outputDef.Cleaner == nil { + t.Fatal("跨进程注册后输出通道 Cleaner 不应丢失") + } + if got := outputDef.Cleaner("raw-output"); got != "output-cleaned:raw-output" { + t.Fatalf("跨进程输出 Cleaner 结果错误：got %q", got) + } } // 输出通道**同步等真实结果**：失败必须上报（§9.4 根治）。 diff --git a/internal/plugin/proc/protocol.go b/internal/plugin/proc/protocol.go index f674863..8b51be6 100644 --- a/internal/plugin/proc/protocol.go +++ b/internal/plugin/proc/protocol.go @@ -39,7 +39,7 @@ type Response struct { Error string `json:"error,omitempty"` } -// ---- kernel → plugin（内核调用插件，对应今日 7 个 //export）---- +// ---- kernel → plugin（内核调用插件；工具执行、Cleaner、stage、输出等）---- const ( // MethodPluginInit 传插件名与配置，插件构造实例但不启动。 MethodPluginInit = "plugin.init" @@ -49,6 +49,10 @@ const ( MethodPluginStop = "plugin.stop" // MethodToolInvoke 执行插件工具。 MethodToolInvoke = "tool.invoke" + // MethodCleanerInvoke 在插件进程内执行工具或通道声明的 Cleaner。 + // Cleaner 是函数，不能随注册请求 JSON 序列化；内核保留 RPC 回调闭包， + // 需要参与向量化/蒸馏时把原文送回插件执行真正的 Cleaner。 + MethodCleanerInvoke = "cleaner.invoke" // MethodStageInvoke 执行阶段处理器。数据经共享段传递，参数只带阶段名与段世代号。 MethodStageInvoke = "stage.invoke" // MethodOutputInvoke 经插件输出通道发送。 @@ -207,6 +211,24 @@ type ToolInvokeResult struct { Result interface{} `json:"result,omitempty"` } +// CleanerInvokeParams / CleanerInvokeResult：跨进程计算层清洗。 +// Scope 取 tool / input / output，Name 是工具名或通道名。 +type CleanerInvokeParams struct { + Scope string `json:"scope"` + Name string `json:"name"` + Text string `json:"text"` +} + +type CleanerInvokeResult struct { + Text string `json:"text"` +} + +const ( + CleanerScopeTool = "tool" + CleanerScopeInput = "input" + CleanerScopeOutput = "output" +) + // OutputInvokeParams：输出通道发送（原 go_invoke_output）。 // // 与 C ABI 路径的关键差异：**可同步等待真实结果**。 diff --git a/internal/plugin/proc/shm_security.go b/internal/plugin/proc/shm_security.go new file mode 100644 index 0000000..9a9b2dc --- /dev/null +++ b/internal/plugin/proc/shm_security.go @@ -0,0 +1,81 @@ +package proc + +import ( + "log" + "os" + "strings" + "sync" +) + +// ShmSecurityMode 控制共享内存段的可见性与审计强度。 +// +// 设计前提（已确认）： +// - Unix（Linux/macOS）默认走匿名 memfd / unlink 临时文件，经 fork 继承 fd。 +// 没有文件系统名字，只有内核主动 spawn 的子进程能映射 → 安全几乎免费。 +// - Windows 没有 fd 继承语义，段是命名 FileMapping，任何拿到名字的进程都能打开。 +// 此处安全边界弱，需要额外措施。 +// +// 映射进程数 ≠ 加载插件数：它不是安全判据，只作异常告警。 +// 合法时也可能不等（一个插件映射两次、debug 工具挂载），非法时可能恰好相等。 +// 它的价值在「发现泄漏句柄或未授权挂载」，不在「授权访问」。 +type ShmSecurityMode string + +const ( + // ShmModeSafe 默认。Unix 匿名 memfd（fork 继承即唯一访问）； + // Windows 命名段用 crypto 随机 nonce 名 + 仅记录期望子 PID。 + ShmModeSafe ShmSecurityMode = "safe" + // ShmModeDebug 暴露 /proc 反查与稳定名字，便于 strace/lldb 挂载。 + ShmModeDebug ShmSecurityMode = "debug" + // ShmModeFull 允许命名段（/dev/shm 或固定名），多实例可共享。 + ShmModeFull ShmSecurityMode = "full" +) + +const envShmSecurityMode = "HOMEAGENT_SHM_SECURITY" + +var ( + shmModeOnce sync.Once + shmModeVal ShmSecurityMode +) + +// ShmSecurityMode 返回当前模式（从 HOMEAGENT_SHM_SECURITY 读，默认 safe）。 +func ShmSecurityModeOf() ShmSecurityMode { + shmModeOnce.Do(func() { + shmModeVal = ShmModeSafe + switch strings.ToLower(strings.TrimSpace(os.Getenv(envShmSecurityMode))) { + case "debug": + shmModeVal = ShmModeDebug + case "full": + shmModeVal = ShmModeFull + } + }) + return shmModeVal +} + +// ShmAudit 记录一次映射数审计的期望与实测。 +// +// expectedPlugins 是内核已 spawn 且握手的插件子进程数；actualMappings 是 +// 当前能观测到的、映射了本段的对象数（Linux 下读 /proc，Windows 计数句柄）。 +// delta = actual - expected > 0 时可能有未授权挂载或泄漏句柄。 +// +// 本结果只用于告警，不用于授权决策：真正阻止未授权访问的是 Unix 匿名 memfd +// + 仅 fork 继承，而非计数相等。 +type ShmAudit struct { + ExpectedPlugins int + ActualMappings int + Delta int +} + +// AuditShmMappings 在 Host 上做一次映射数审计。 +// +// 平台无探针时返回 ok=false（调用方应跳过告警，不要误报）。 +func (h *Host) AuditShmMappings(expectedPlugins int) (ShmAudit, bool) { + a := ShmAudit{ExpectedPlugins: expectedPlugins} + var ok bool + a.ActualMappings, ok = h.countOurShmMappings() + a.Delta = a.ActualMappings - a.ExpectedPlugins + if ok && a.Delta > 0 { + log.Printf("[proc] 共享内存映射审计：期望 %d 实测 %d（+delta=%d）—— 可能有泄漏句柄或未授权挂载", + a.ExpectedPlugins, a.ActualMappings, a.Delta) + } + return a, ok +} diff --git a/internal/plugin/proc/shm_security_linux.go b/internal/plugin/proc/shm_security_linux.go new file mode 100644 index 0000000..95cb91d --- /dev/null +++ b/internal/plugin/proc/shm_security_linux.go @@ -0,0 +1,62 @@ +//go:build linux + +package proc + +import ( + "fmt" + "os" + "strings" +) + +// countOurShmMappings 在 Linux 上审计本 Host 的共享段被多少进程映射。 +// +// 原理：memfd 是匿名对象，文件系统无名，但在每个映射它的进程的 +// /proc//maps 里会有一行形如： +// +// 7f.. 7f.. rw-s 00000000 00:00 7312 .../memfd:hastagectx (deleted) +// +// 我们遍历 Supervisor.List() 给出的全部子进程 PID，统计它们的 maps 中 +// 出现 "memfd:hastagectx" 的行数。 +// homed 自身的映射不算（它本来就该有一份）。 +// +// 局限：/proc//maps 需要同 UID/同命名空间读。读不到的 PID 直接跳过， +// 返回 ok=true 但少计——审计是告警不是判据，宁缺毋滥。 +func (h *Host) countOurShmMappings() (int, bool) { + if h.sup == nil { + return 0, false + } + procs := h.sup.List() + if len(procs) == 0 { + return 0, true + } + // memfd 的"文件名"在 shmalloc_linux.go 里是 "hastagectx"。 + const memfdTag = "memfd:hastagectx" + total := 0 + for _, p := range procs { + if !p.Alive { + continue + } + if n, err := countMemfdLinesForPID(p.PID, memfdTag); err == nil { + total += n + } + } + return total, true +} + +// countMemfdLinesForPID 读 /proc//maps，统计含 tag 的行数。 +// +// 一个进程对同一 memfd 多次 mmap 会产生多行；我们计数"行数"而非 +// "是否出现"，这样能发现一个插件进程泄漏了多余映射的情况。 +func countMemfdLinesForPID(pid int, tag string) (int, error) { + data, err := os.ReadFile(procMapsPath(pid)) + if err != nil { + return 0, fmt.Errorf("读 /proc/%d/maps: %w", pid, err) + } + return strings.Count(string(data), tag), nil +} + +// procMapsPath 组装 /proc//maps 路径。pid 是内核自己 spawn 的进程号， +// 无用户输入注入面。 +func procMapsPath(pid int) string { + return fmt.Sprintf("/proc/%d/maps", pid) +} diff --git a/internal/plugin/proc/shm_security_other.go b/internal/plugin/proc/shm_security_other.go new file mode 100644 index 0000000..65813b8 --- /dev/null +++ b/internal/plugin/proc/shm_security_other.go @@ -0,0 +1,11 @@ +//go:build !linux + +package proc + +// countOurShmMappings 在非 Linux 平台返回 ok=false（无 /proc 探针）。 +// +// macOS 无 /proc；Windows 无 fd 继承，命名对象的映射计数需走 Win32 API， +// 当前阶段不实现，留给调用方跳过告警。 +func (h *Host) countOurShmMappings() (int, bool) { + return 0, false +} diff --git a/internal/plugin/proc/shm_security_test.go b/internal/plugin/proc/shm_security_test.go new file mode 100644 index 0000000..6665e16 --- /dev/null +++ b/internal/plugin/proc/shm_security_test.go @@ -0,0 +1,63 @@ +package proc + +import ( + "testing" +) + +func TestShmSecurityMode_DefaultsToSafe(t *testing.T) { + if ShmSecurityModeOf() != ShmModeSafe { + t.Fatalf("默认应为 safe，实际 %q", ShmSecurityModeOf()) + } +} + +func TestShmAudit_ZeroPluginsNoMapping(t *testing.T) { + h, err := NewHost() + if err != nil { + t.Fatalf("NewHost: %v", err) + } + defer h.Close() + + a, ok := h.AuditShmMappings(0) + if !ok { + // 非 Linux 平台无探针，本测试断言"平台探针不误报" + t.Logf("平台无探针（ok=false），跳过") + return + } + // 无子进程时实际映射数应为 0 + if a.ActualMappings != 0 { + t.Fatalf("无子进程时应无映射，实际 %d", a.ActualMappings) + } + if a.Delta != 0 { + t.Fatalf("Delta 应为 0，实际 %d", a.Delta) + } +} + +// 子进程挂载共享段后，Linux 探针应能看到对应映射。 +func TestShmAudit_SeesChildMapping(t *testing.T) { + bin := buildTestPlugin(t, "stageplugin.go") + core := newFakeCore() + + host, err := NewHost() + if err != nil { + t.Fatalf("NewHost: %v", err) + } + defer host.Close() + + p := New("audit", bin, t.TempDir(), nil, host, nil) + if err := p.Start(core); err != nil { + t.Fatalf("Start: %v", err) + } + defer p.Close() + + a, ok := host.AuditShmMappings(1) + if !ok { + t.Logf("平台无探针（ok=false），跳过") + return + } + // 至少应看到 1 个儿童映射 + if a.ActualMappings < 1 { + t.Fatalf("子进程应映射共享段，实际映射数 %d", a.ActualMappings) + } + // 期望 1、实际 ≥1 → delta 不应为负且不应报警（delta>0 才告警） + _ = a +} diff --git a/internal/plugin/proc/shmalloc_windows.go b/internal/plugin/proc/shmalloc_windows.go index 8ed9d1e..8964cbf 100644 --- a/internal/plugin/proc/shmalloc_windows.go +++ b/internal/plugin/proc/shmalloc_windows.go @@ -3,6 +3,7 @@ package proc import ( + "crypto/rand" "fmt" "os" "sync" @@ -64,7 +65,7 @@ var ( // 返回的 *os.File 为 nil：Windows 不经 fd 传递段，插件按名字打开。 // 名字通过 procEnvForShm 注入子进程环境变量。 func allocShm(size int) (*os.File, []byte, error) { - name := fmt.Sprintf("%s_%d_%d", shmNamePrefix, os.Getpid(), shmNameSeq.Add(1)) + name := shmNameForMode() shm, data, err := createNamedMapping(name, size) if err != nil { return nil, nil, err @@ -75,6 +76,25 @@ func allocShm(size int) (*os.File, []byte, error) { return nil, data, nil } +// shmNameForMode 按安全模式生成命名段名。 +// +// safe/debug：随机 nonce 名，防猜测；唯一在 debug 下额外暴露到 stderr。 +// full：固定后缀，便于多实例按名共享。 +func shmNameForMode() string { + seq := shmNameSeq.Add(1) + switch ShmSecurityModeOf() { + case ShmModeFull: + return fmt.Sprintf("%s_%d_%d", shmNamePrefix, os.Getpid(), seq) + default: // safe / debug + var b [12]byte + if _, err := cryptorand.Read(b[:]); err != nil { + // 退化为 PID+seq（极端情况，crypto rand 几乎不会失败） + return fmt.Sprintf("%s_%d_%d", shmNamePrefix, os.Getpid(), seq) + } + return fmt.Sprintf("%s_%d_%x", shmNamePrefix, os.Getpid(), b) + } +} + // createNamedMapping 建命名段并映射为 []byte。 func createNamedMapping(name string, size int) (*namedShm, []byte, error) { namePtr, err := windows.UTF16PtrFromString(name) diff --git a/internal/plugin/proc/testdata/stageplugin.go b/internal/plugin/proc/testdata/stageplugin.go index 91baf3c..3e76861 100644 --- a/internal/plugin/proc/testdata/stageplugin.go +++ b/internal/plugin/proc/testdata/stageplugin.go @@ -243,17 +243,20 @@ func main() { go func(id uint64) { callKernel("lifecycle.autoRestart", map[string]interface{}{"enabled": true}) callKernel("tool.register", map[string]interface{}{ - "name": "demo_upper", - "def": map[string]interface{}{"name": "demo_upper", "description": "转大写"}, + "name": "demo_upper", + "def": map[string]interface{}{"name": "demo_upper", "description": "转大写"}, + "has_cleaner": true, }) callKernel("stage.register", map[string]interface{}{ "stage": "after_toolcall", "scope": "global", }) + callKernel("input.register", map[string]interface{}{ + "name": "demo_in", "def": map[string]interface{}{}, "has_cleaner": true, + }) callKernel("output.register", map[string]interface{}{ - "name": "demo_ch", - "caps": 1, - "desc": "测试通道", + "name": "demo_ch", "caps": 1, "desc": "测试通道", + "def": map[string]interface{}{}, "has_cleaner": true, }) send(response{ID: id}) }(req.ID) @@ -274,6 +277,24 @@ func main() { "result": strings.ToUpper(text), }}) + case "cleaner.invoke": + var p struct { + Scope string `json:"scope"` + Name string `json:"name"` + Text string `json:"text"` + } + json.Unmarshal(req.Params, &p) + valid := (p.Scope == "tool" && p.Name == "demo_upper") || + (p.Scope == "input" && p.Name == "demo_in") || + (p.Scope == "output" && p.Name == "demo_ch") + if !valid { + send(response{ID: req.ID, Error: "未注册 Cleaner"}) + continue + } + send(response{ID: req.ID, Result: map[string]interface{}{ + "text": p.Scope + "-cleaned:" + p.Text, + }}) + case "stage.invoke": go func(id uint64) { if shm == nil { diff --git a/plan.md b/plan.md index ca83676..b2d1a15 100644 --- a/plan.md +++ b/plan.md @@ -736,7 +736,7 @@ goroutine 里的 `pluginInvokeOutput` 才是 cgo 调用，**不构成嵌套**。 **现网触发点**：`AfterToolcall` 上有两个外部插件 | Stage | 注册者 | 风险 | -|---|---|---| +| --- | --- | --- | | `AfterToolcall` | **sanitizer**(Global,改写 ToolResults) + **weather**(own_tools,只读) | ⚠️ 真实冲突 | | `PreAction` | memo(外部) + webui(内置) | ⚡ | | `BeforeToolcall` | qq(外部) + webui(内置) + cmd(内置) | ⚡ | @@ -783,7 +783,7 @@ diff := changedFieldsOnly(before, stageContextWritable(sc)) `lua_plugin.go:726` 直接读 `sc.RawMessage` 等字段，**未持 `sc.RLock()`**： | 路径 | 快照时是否持锁 | -|---|---| +| --- | --- | | cabi（`loader.go:412`） | ✅ `sc.RLock()` | | Lua（`lua_plugin.go:726`） | ❌ 无锁 | @@ -806,7 +806,7 @@ return nil // ← 无 resultOut，无 applyStageResult ``` | 路径 | 下发字段 | 写回 | -|---|---|---| +| --- | --- | --- | | Linux cabi | 10 | ✅ | | Lua | 10 | ✅ | | **Windows DLL** | **3** | ❌ **完全没有** | @@ -850,7 +850,7 @@ return nil // ← 无 resultOut，无 applyStageResult **11 项可行性实验全部通过**（详见评估文档第七章）： | 验证项 | 结果 | -|---|---| +| --- | --- | | eventfd 走 netpoller | ✅ 200 等待者仅 +1 线程 | | 跨进程偏移解引用 | ✅ 父子 mmap 不同基址仍正确 | | 锁仲裁 RPC 成本 | ✅ 19.4 µs/次 | @@ -886,7 +886,7 @@ return nil // ← 无 resultOut，无 applyStageResult 按「影响 × 成本」排序，前 4 项不依赖迁移决策： | 序 | 项 | 规模 | 现网影响 | -|---|---|---|---| +| --- | --- | --- | --- | | 1 | **11.1** output_send 同步等结果 | M | ❗ 用户收不到消息且模型以为成功 | | 2 | **11.3** stageContextWritable diff 回传 | S | ❗ 脏数据进 LLM（量级百分之几） | | 3 | **11.6** reload 语义修正（3 项） | S | 误导模型白跑重载 | @@ -930,7 +930,7 @@ context 累积导致的内存增长。 四个决策点均已落定并执行： | # | 决策点 | 最终选择 | -|---|---|---| +| --- | --- | --- | | 1 | merge 方式 | **`--no-ff`** —— commit message 记录了「为何共享同一块 memfd」「为何 procCore 不能嵌入」等踩坑过程，压成一条就没了 | | 2 | 合回后是否删 feature 分支 | **删**（`feature/plugin-proc-migration`、`feature/memory-media` 均已删，本地 + 远端） | | 3 | release 构建是否再替换生产二进制 | **换**，且此后每个正式版都走同一流程（备份二进制 + `sqlite3 .backup` 配置库 + 记插件清单 → `install -m 0755` → restart → 健康检查） | @@ -949,6 +949,7 @@ context 累积导致的内存增长。 已发布：`v1.0.0` / `v1.0.1` / `v1.0.3` / `v1.0.4`（1.0.x 线）、`v1.1.0` / `v1.1.0-beta.1` / `v1.1.1`（1.1.x 线）， SDK 仓 `v1.0.0` / `v1.1.0`。main 的版本路牌现为 `1.2.0`（尚无 tag）。 + ``` --- @@ -1083,3 +1084,219 @@ SDK 仓 `v1.0.0` / `v1.1.0`。main 的版本路牌现为 `1.2.0`（尚无 tag） 额外收益：权限梯度从「C ABI 表达能力的意外产物」变成**显式三道闸** （类型层 `procCore` 命名字段 + manifest 能力声明 + RPC 边界明确拒绝）。 + +--- + +## 13. 子进程化后遗留改造（逐步推进） + +每个步骤独立提交、独立验证，遵循「小步 → 测试 → 提交」循环。 + +### 步骤分组 + +| 步骤 | 内容 | 依赖 | 产出 | +|---|---|---|---| +| 13.1 | 统一共享内存布局 | 无 | 单一 memfd，StageContext+EvtRing 成为区内 segment | +| 13.2 | 共享内存扩缩容 | 13.1 | ftruncate/remap，generation 感知 | +| 13.3 | ToolCall lane | 13.1 | 跨进程工具调用走 SharedRef，RPC 退化为控制信号 | +| 13.4 | Cleaner 迁移至 SharedRef | 13.1 | cleaner.invoke 参数/结果走 SharedRef | +| 13.5 | InputChannel lane | 13.1 | 输入通道消息走共享内存 | +| 13.6 | OutputChannel lane | 13.1 | 输出通道消息走共享内存 | +| 13.7 | RuntimeManager + 分组 worker | 13.1~13.6 | 一个 RuntimeManager，少量 worker，多插件共享 transport | +| 13.8 | ContextPolicy tool 上下文策略 | 无 | ToolDef.ContextPolicy = none/prune | +| 13.9 | llmsproxy 上下文溢出感知 | 无 | 溢出错误归一化 + AUTO 截断放宽 | +| 13.10 | AgentMail 三个 bug | 无 | 提示词修正 / 回信标识 / relay_key 限长 | +| 13.11 | WebUI 修复全清单 | 无 | 11 项逐步推进 | +| 13.12 | L3 原生多模态 | 13.1~13.7 | 媒体作为图节点/边，L2→L3 引用迁移 | + +### 13.1 统一共享内存布局 + +**现状**：两块独立 memfd（StageContext 256KB + EvtRing ~320KB），fd 3/4，eventfd 占 fd 5。 +**目标**：合并为单一 memfd 占 fd 3；eventfd 占 fd 4。段内偏移表定位各 segment。 + +```text +┌─────────────────────────────────────────┐ +│ Unified Shared Region（单 memfd） │ +├─────────────────────────────────────────┤ +│ [SuperBlock 64B] magic/version/cap/gen │ +│ [StageContext segment] 布局不变 │ +│ [EvtRing segment] 布局不变 │ +│ [Reserved: ToolCall] 后续步骤填充 │ +│ [Reserved: InputCh] 后续步骤填充 │ +│ [Reserved: OutputCh] 后续步骤填充 │ +│ [Dynamic Arena] 自由分配区 │ +└─────────────────────────────────────────┘ +``` + +**实施**： + +1. 定义 SuperBlock 布局（magic/version/segment 偏移表） +2. NewHost() 一次 allocShm，内含两段 +3. Segment + EvtRing 从 SuperBlock 读偏移 +4. fd 传递从 3 个降为 2 个 +5. procExtraFilesForShm 返回 2 个 fd +6. 插件侧模板解析 SuperBlock，自行定位两段 +7. 所有现有测试不变 +8. go test -race ./internal/plugin/proc/ + +**验证**： + +- [ ] SuperBlock 写入读回一致 +- [ ] StageContext 并发改写 0 lost update +- [ ] 事件环 post-and-forget 仍工作 +- [ ] TestPlugin_ConcurrentWriterAndReaderNoLostUpdate 通过 +- [ ] fd 数从 3 降到 2 +- [ ] git commit -m "feat(shm): unified shared memory region" + +### 13.2 共享内存扩缩容 + +**目标**：ftruncate 扩展 + mremap/mmap 重映射 + generation 感知。 + +**实施**： + +1. SuperBlock 加 generation + capacity 字段 +2. Host.Grow(newSize)：ftruncate → 更新 capacity/gen → 通知子进程 +3. 子进程下次 stage 读 gen → remap +4. Host.Shrink()：compact 后 free > 50% 且持续 5min → 缩容 +5. 无活跃 slot 时才允许缩容 + +**验证**： + +- [ ] Grow 后子进程正确读写 +- [ ] Shrink 不裁掉活跃 slot +- [ ] git commit -m "feat(shm): grow/shrink with generation awareness" + +### 13.3 ToolCall lane + +**目标**：工具调用参数/结果从 JSON RPC 改为 SharedRef。 + +**实施**： + +1. ToolCall Slot 状态机：FREE→WRITING→READY→READING→DONE→FREE +2. 内核写请求 → READY → eventfd 通知 +3. 插件读请求 → 执行 → 写结果 → DONE → eventfd 通知 +4. RPC tool.invoke 改为只传 {requestID, argsRef, resultRef} +5. has_cleaner 保持不变 + +**验证**： + +- [ ] TestPlugin_RegisteredToolInvokable 通过 +- [ ] Bench: ToolInvoke 延迟改善 +- [ ] git commit -m "feat(shm): toolcall lane with slot state machine" + +### 13.4 Cleaner 迁移至 SharedRef + +**目标**：cleaner.invoke 参数/结果走 SharedRef。 + +**实施**： + +1. CleanerInvokeParams 的 Text 改为 SharedRef +2. 插件从 SharedRef 读文本、写结果 +3. 内核从 SharedRef 读结果 +4. RPC 帧从 ~200B 降到 ~16B + +**验证**： + +- [ ] 三类 Cleaner 结果正确 +- [ ] git commit -m "feat(shm): cleaner invoke via shared refs" + +### 13.5 InputChannel lane + +**目标**：输入通道消息走共享内存。 + +**实施**： + +1. InputChannel Slot：source/channel/text/blocks 写入 arena +2. 插件消费后设 DONE +3. 同步注入仍走 RPC + SharedRef +4. 异步注入改写 arena + eventfd + +**验证**： + +- [ ] 真实 QQ 消息注入测试 +- [ ] git commit -m "feat(shm): input channel lane" + +### 13.6 OutputChannel lane + +**目标**：输出通道消息走共享内存。 + +**验证**： + +- [ ] QQ 输出正常 +- [ ] git commit -m "feat(shm): output channel lane" + +### 13.7 RuntimeManager + 分组 worker + +**目标**：一个 RuntimeManager + 少量 worker + 多插件共享 transport + 每插件独立 PluginContext。 + +**实施**： + +1. RuntimeManager 类型：管理 worker 池 + 调度 +2. Worker 类型：一个进程，共享 RuntimeClient +3. PluginContext 类型：独立身份，共享 transport +4. manifest 新增 worker_group 字段 +5. 默认所有 proc 插件归同一 worker（兼容迁移） +6. 高风险插件可声明独立 worker_group + +**验证**： + +- [ ] 默认分组 = 现有行为 +- [ ] git commit -m "feat(runtime): RuntimeManager with grouped workers" + +### 13.8 ContextPolicy tool 上下文策略 + +**目标**：ToolDef.ContextPolicy = none/prune。 + +**实施**： + +1. SDK ToolDef 加 ContextPolicy string 字段 +2. StageAfterToolcall 检查当前 tool 的 ContextPolicy +3. prune 时执行 RelevanceContext.Prune +4. 默认 none + +**验证**： + +- [ ] qq_get_message 加 prune 后上下文精简 +- [ ] git commit -m "feat(ctx): context policy for tool results" + +### 13.9 llmsproxy 上下文溢出感知 + +**实施**： + +1. OVERFLOW_PATTERNS 补充 "Context window is full" +2. AUTO 截断宽度 80→160 + +**验证**： + +- [ ] go test ./internal/ai/... +- [ ] git commit -m "fix(ai): context overflow pattern + truncation width" + +### 13.10 AgentMail 三个 bug + +**实施**： + +1. 提示词修正 +2. InReplyTo 字段 +3. relay_key ≤ 64 字节 + +**验证**： + +- [ ] Agent→Agent 不再套话 6 轮 +- [ ] git commit -m "fix(agentmail): prompt / in-reply-to / relay key" + +### 13.11 WebUI 修复全清单 + +11 项，每项独立提交： +SSE Last-Event-ID → 超时 → api 状态码 → renderAll 增量 → XSS 消毒 → CSS → DesignSystem → handleAgents 持久化 → handleKnowledge 吞错 → GUI 重构 + +### 13.12 L3 原生多模态 + +**实施**： + +1. L3 node_type 新增 media/media_block +2. L3 edge_type 新增 depicts/contains +3. L2→L3 迁移时保留 media 引用边 + +**验证**： + +- [ ] 图查询返回媒体节点 +- [ ] git commit -m "feat(l3): native multimodal nodes/edges"