mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-09-21 17:38:10 +00:00
feat(proc): Plugin 加载器 + 51 method handler + RunStage 接共享段（Part 2 完成 / Part 4 闭环）
corehandler.go —— cabi/loader.go 51 个 case 体的整块平移（§3.2）:
- 参数从「s1/s2/s3 + i1/i2 五个固定槽」改为结构化 JSON，语义不变
- CoreSDK 接口刻意只含外部插件应得能力：无 Selftest/Supervisor/Tracker/
Status/Adapter/Config/Tool/Indexer/OutputChan/Publish
→ 权限梯度从「C ABI 表达能力的意外产物」变成「显式声明并强制的策略」（§3.8）
- 事件订阅(case 23/24)与 SetToolBlocks 明确返回未实现，不再像 C ABI 那样静默成功
（静默成功后收不到事件比报错更难排查）
- ToolDef.Cleaner / ChannelDef.Cleaner 是函数，跨进程置 nil（§3.5 回调型资源）
host.go —— 共享段所有权中心:
- ❗ 全部子进程插件共享**同一块 memfd**。若每插件一段，
「内核 ctx → 段 → 插件改 → 回读 ctx」在多插件下退化成副本模型，
最后回读者覆盖前者，§8.4 的 35.8~36.8% lost update 原样复现
- stageMu 串行化整次 stage 对段的独占（内核可能并发触发 RunStage）
- 首个进入者写入段，最后离开者回读 + Compact（此时无插件持锁，满足 §3.3 前提）
stage.go —— RunStage 接线（风险 3.4 落点）:
- 并发扇出保留（§0.2 第 1 条：并发扇出是原始设计，不是缺陷）
- 插件失败时 ForceReleaseLock，避免后续插件死锁（实验 9，无需 robust mutex）
plugin.go —— registry 可加载的插件实体:
- Start: spawn（fd 3 传共享段）→ 握手 → plugin.init → plugin.start
- Close: **真 kill + wait**，对比 cabi 的 Close 只做 dlclose 而后者是 no-op（§1.1）
- invokeOutput **同步等真实结果**，失败上报 error —— §9.4 根治
验证（34 项测试全绿，含 -race，全部用真实子进程）:
- 单插件 stage 读改写经共享段回到内核 StageContext
- sanitizer(改写) + weather(只读) 并发：清洗结果不被覆盖（现网场景）
- **5 个独立进程并发 append 同一 FinalText：5 个标记全部保留，零丢失零撕裂**
（实验 8 在真实 RPC + 真实 RunStage 下的复刻）
- 工具注册可调用 / 输出通道真实失败上报 / start 期间反向调用
- 未知 method 与未实现能力被拒绝 / stage 外加锁被拒绝
接口冻结: git diff third_party/homeagent-sdk/sdk/ 为空
This commit is contained in:
dev
697
internal/plugin/proc/corehandler.go
Normal file
697
internal/plugin/proc/corehandler.go
Normal file
@ -0,0 +1,697 @@
|
||||
package proc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
|
||||
)
|
||||
|
||||
// coreHandler 把插件发来的 RPC 调用路由到内核 PluginSDK。
|
||||
//
|
||||
// 这是 cabi/loader.go 那 51 个 case 体的**整块平移**（§3.2）：
|
||||
// 参数解析、调用、错误处理逻辑不变，只把「整数 method id + 三个 C 字符串槽」
|
||||
// 换成「method 名 + 结构化 JSON 参数」。语义不变，回归风险最小。
|
||||
//
|
||||
// 平移带来的直接改善：
|
||||
// - 参数不再挤进 s1/s2/s3 + i1/i2 五个固定槽（C ABI 的形状约束）
|
||||
// - 不需要 CORE_FREE_STRING：跨进程各自 GC
|
||||
// - 错误可携带结构化信息，不只是一个字符串
|
||||
type coreHandler struct {
|
||||
// sdk 是内核为该插件构建的 PluginSDK（与内置插件同一类型）。
|
||||
sdk CoreSDK
|
||||
// name 是插件名，用于工具归属推断与日志。
|
||||
name string
|
||||
|
||||
// host 持有被全部插件共享的 StageContext 段与锁仲裁（§3.3/§3.7）。
|
||||
// ❗ 必须是"全部插件共享一个 Host"——每插件一段会退化成副本模型。
|
||||
host *Host
|
||||
|
||||
// locks 是 host.locks 的引用，供 stage.lock/unlock 路由。
|
||||
locks *lockRegistry
|
||||
|
||||
// invokeTool/invokeStageFn/invokeOutput 反向调用插件（内核 → 插件）。
|
||||
// 由 Plugin 注入，注册回调时用它们构造 handler。
|
||||
invokeTool func(name string, args map[string]interface{}) (interface{}, error)
|
||||
invokeStageFn func(ctx context.Context, stage string, seq uint64) error
|
||||
invokeOutput func(channel string, args map[string]interface{}) (interface{}, error)
|
||||
}
|
||||
|
||||
// invokeStageWithCtx 反向调用插件执行 stage。
|
||||
func (h *coreHandler) invokeStageWithCtx(ctx context.Context, stage string, seq uint64) error {
|
||||
if h.invokeStageFn == nil {
|
||||
return fmt.Errorf("插件 %s: stage 调用通道未就绪", h.name)
|
||||
}
|
||||
return h.invokeStageFn(ctx, stage, seq)
|
||||
}
|
||||
|
||||
// CoreSDK 是 coreHandler 依赖的内核能力面。
|
||||
//
|
||||
// 定义为接口而非直接依赖 internal/sdk.PluginSDK，原因：
|
||||
// 1. 避免 internal/plugin/proc → internal/sdk 的强耦合（后者已依赖 internal/plugin 的类型）
|
||||
// 2. 单测可注入假实现，无需构造完整内核
|
||||
//
|
||||
// 方法集**刻意只包含外部插件应得的能力**——`Selftest`/`Supervisor`/`Tracker`/
|
||||
// `Status`/`Adapter`/`Config`/`Tool`/`Indexer`/`OutputChan`/`Publish` 不在此列。
|
||||
// 这正是把权限梯度从「C ABI 表达能力的意外产物」变成「显式声明并强制的策略」（§3.8）。
|
||||
type CoreSDK interface {
|
||||
PluginName() string
|
||||
|
||||
Settings() pubsdk.SettingsAPI
|
||||
Memory() pubsdk.MemoryAPI
|
||||
TextMemory() pubsdk.TextMemoryAPI
|
||||
DocMemory() pubsdk.DocMemoryAPI
|
||||
Knowledge() pubsdk.KnowledgeAPI
|
||||
LLM() pubsdk.LLMAPI
|
||||
Social() pubsdk.SocialAPI
|
||||
PluginMgr() pubsdk.PluginMgrAPI
|
||||
|
||||
RegisterTool(name string, def pubsdk.ToolDef, handler pubsdk.ToolHandler) error
|
||||
RegisterStage(stage pubsdk.Stage, handler pubsdk.StageHandler, scope ...pubsdk.StageScope)
|
||||
RegisterPluginAPI(name string) error
|
||||
RegisterOutputChannel(name string, caps int, desc string, def pubsdk.ChannelDef, handler pubsdk.ToolHandler) error
|
||||
RegisterInputChannel(name string, def pubsdk.ChannelDef) error
|
||||
|
||||
InjectText(source, channel, text string)
|
||||
InjectInterruptText(source, channel, text string)
|
||||
InjectTextNoMemory(source, channel, text string)
|
||||
InjectInputSync(source, channel, text string) string
|
||||
|
||||
SetAutoRestart(enabled bool)
|
||||
}
|
||||
|
||||
// Handle 分派一次插件 → 内核的调用。
|
||||
func (h *coreHandler) Handle(method string, params json.RawMessage) (interface{}, error) {
|
||||
switch method {
|
||||
|
||||
// ---- 注册面（原 case 1/2/3/4/46）----
|
||||
case MethodToolRegister:
|
||||
return h.toolRegister(params)
|
||||
case MethodStageRegister:
|
||||
return h.stageRegister(params)
|
||||
case MethodOutputRegister:
|
||||
return h.outputRegister(params)
|
||||
case MethodAPIRegister:
|
||||
var p struct {
|
||||
Name string `json:"name"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return nil, h.sdk.RegisterPluginAPI(p.Name)
|
||||
case MethodInputRegister:
|
||||
var p struct {
|
||||
Name string `json:"name"`
|
||||
Def pubsdk.ChannelDef `json:"def"`
|
||||
}
|
||||
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})
|
||||
|
||||
// ---- IO 注入（原 case 5/6/7/47）----
|
||||
case MethodIOInjectText:
|
||||
var p injectParams
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
h.sdk.InjectText(p.Source, p.Channel, p.Text)
|
||||
return nil, nil
|
||||
case MethodIOInjectInterrupt:
|
||||
var p injectParams
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
h.sdk.InjectInterruptText(p.Source, p.Channel, p.Text)
|
||||
return nil, nil
|
||||
case MethodIOInjectTextNoMem:
|
||||
var p injectParams
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
h.sdk.InjectTextNoMemory(p.Source, p.Channel, p.Text)
|
||||
return nil, nil
|
||||
case MethodIOInjectSync:
|
||||
var p injectParams
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"reply": h.sdk.InjectInputSync(p.Source, p.Channel, p.Text)}, nil
|
||||
|
||||
// ---- 生命周期（原 case 8）----
|
||||
case MethodLifecycleAutoRestart:
|
||||
var p struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
h.sdk.SetAutoRestart(p.Enabled)
|
||||
return nil, nil
|
||||
|
||||
// ---- 图记忆（原 case 9/10/11/12/13）----
|
||||
case MethodMemoryRecall:
|
||||
mem := h.sdk.Memory()
|
||||
if mem == nil {
|
||||
return nil, errUnavailable("memory")
|
||||
}
|
||||
var p struct {
|
||||
Query []string `json:"query"`
|
||||
Depth int `json:"depth"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
entities, relations, err := mem.Recall(p.Query, p.Depth)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if entities == nil {
|
||||
entities = []pubsdk.Entity{}
|
||||
}
|
||||
if relations == nil {
|
||||
relations = []pubsdk.Relation{}
|
||||
}
|
||||
return map[string]interface{}{"entities": entities, "relations": relations}, nil
|
||||
|
||||
case MethodMemoryCommit:
|
||||
mem := h.sdk.Memory()
|
||||
if mem == nil {
|
||||
return nil, errUnavailable("memory")
|
||||
}
|
||||
var p struct {
|
||||
Triples []pubsdk.Triple `json:"triples"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return nil, mem.Commit(p.Triples)
|
||||
|
||||
case MethodMemoryIntrospect:
|
||||
mem := h.sdk.Memory()
|
||||
if mem == nil {
|
||||
return nil, errUnavailable("memory")
|
||||
}
|
||||
return mem.Introspect()
|
||||
|
||||
case MethodMemoryMerge:
|
||||
mem := h.sdk.Memory()
|
||||
if mem == nil {
|
||||
return nil, errUnavailable("memory")
|
||||
}
|
||||
var p struct {
|
||||
Source string `json:"source"`
|
||||
Target string `json:"target"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
n, err := mem.MergeEntities(p.Source, p.Target)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"merged": n}, nil
|
||||
|
||||
case MethodMemoryPurge:
|
||||
mem := h.sdk.Memory()
|
||||
if mem == nil {
|
||||
return nil, errUnavailable("memory")
|
||||
}
|
||||
var p struct {
|
||||
Criteria map[string]string `json:"criteria"`
|
||||
Mode string `json:"mode"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if p.Mode == "" {
|
||||
p.Mode = "soft"
|
||||
}
|
||||
n, err := mem.Purge(p.Criteria, p.Mode)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"purged": n}, nil
|
||||
|
||||
// ---- 文档记忆（原 case 14/32/33/34）----
|
||||
case MethodDocQuery:
|
||||
dm := h.sdk.DocMemory()
|
||||
if dm == nil {
|
||||
return nil, errUnavailable("doc memory")
|
||||
}
|
||||
var p struct {
|
||||
Text string `json:"text"`
|
||||
TopK int `json:"top_k"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
docs := dm.Query(p.Text, p.TopK)
|
||||
if docs == nil {
|
||||
docs = []*pubsdk.Doc{}
|
||||
}
|
||||
return map[string]interface{}{"docs": docs}, nil
|
||||
|
||||
case MethodDocInsert:
|
||||
dm := h.sdk.DocMemory()
|
||||
if dm == nil {
|
||||
return nil, errUnavailable("doc memory")
|
||||
}
|
||||
var p struct {
|
||||
Doc *pubsdk.Doc `json:"doc"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if p.Doc == nil {
|
||||
return nil, fmt.Errorf("doc.insert: 缺少 doc 字段")
|
||||
}
|
||||
return nil, dm.Insert(p.Doc)
|
||||
|
||||
case MethodDocRemove:
|
||||
dm := h.sdk.DocMemory()
|
||||
if dm == nil {
|
||||
return nil, errUnavailable("doc memory")
|
||||
}
|
||||
var p struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
dm.Remove(p.ID)
|
||||
return nil, nil
|
||||
|
||||
case MethodDocStats:
|
||||
dm := h.sdk.DocMemory()
|
||||
if dm == nil {
|
||||
return nil, errUnavailable("doc memory")
|
||||
}
|
||||
return dm.Stats(), nil
|
||||
|
||||
// ---- 知识库（原 case 15/35/36）----
|
||||
case MethodKnowledgeSearch:
|
||||
kn := h.sdk.Knowledge()
|
||||
if kn == nil {
|
||||
return nil, errUnavailable("knowledge")
|
||||
}
|
||||
var p struct {
|
||||
Query string `json:"query"`
|
||||
TopK int `json:"top_k"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results, err := kn.Search(p.Query, p.TopK)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if results == nil {
|
||||
results = []*pubsdk.Knowledge{}
|
||||
}
|
||||
return map[string]interface{}{"results": results}, nil
|
||||
|
||||
case MethodKnowledgeAdd:
|
||||
kn := h.sdk.Knowledge()
|
||||
if kn == nil {
|
||||
return nil, errUnavailable("knowledge")
|
||||
}
|
||||
var p struct {
|
||||
Name string `json:"name"`
|
||||
Content string `json:"content"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return nil, kn.Add(p.Name, p.Content)
|
||||
|
||||
case MethodKnowledgeList:
|
||||
kn := h.sdk.Knowledge()
|
||||
if kn == nil {
|
||||
return nil, errUnavailable("knowledge")
|
||||
}
|
||||
names, err := kn.List()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if names == nil {
|
||||
names = []string{}
|
||||
}
|
||||
return map[string]interface{}{"names": names}, nil
|
||||
|
||||
// ---- 文本记忆（原 case 41）----
|
||||
case MethodTextMemoryAppend:
|
||||
tm := h.sdk.TextMemory()
|
||||
if tm == nil {
|
||||
return nil, errUnavailable("text memory")
|
||||
}
|
||||
var p struct {
|
||||
Event pubsdk.TextEvent `json:"event"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return nil, tm.Append(p.Event)
|
||||
|
||||
// ---- 设置（原 case 16/17/18/26~31/42~45/51）----
|
||||
case MethodSettingsGet, MethodSettingsSet, MethodSettingsRegisterDef,
|
||||
MethodSettingsGetCore, MethodSettingsSetCore, MethodSettingsListCore,
|
||||
MethodSettingsGetPlugin, MethodSettingsSetPlugin, MethodSettingsListPlugin,
|
||||
MethodSettingsList, MethodSettingsDefs, MethodSettingsDump,
|
||||
MethodSettingsPlugins, MethodSettingsDataDir:
|
||||
return h.settings(method, params)
|
||||
|
||||
// ---- LLM 源（原 case 19/20/37）----
|
||||
case MethodLLMListSources:
|
||||
llm := h.sdk.LLM()
|
||||
if llm == nil {
|
||||
return nil, errUnavailable("llm")
|
||||
}
|
||||
sources := llm.ListSources()
|
||||
if sources == nil {
|
||||
sources = []string{}
|
||||
}
|
||||
return map[string]interface{}{"sources": sources}, nil
|
||||
case MethodLLMSetSource:
|
||||
llm := h.sdk.LLM()
|
||||
if llm == nil {
|
||||
return nil, errUnavailable("llm")
|
||||
}
|
||||
var p struct {
|
||||
Name string `json:"name"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return nil, llm.SetSource(p.Name)
|
||||
case MethodLLMCurrentSource:
|
||||
llm := h.sdk.LLM()
|
||||
if llm == nil {
|
||||
return nil, errUnavailable("llm")
|
||||
}
|
||||
return map[string]interface{}{"source": llm.CurrentSource()}, nil
|
||||
|
||||
// ---- 社交图（只读，原 case 21/22/38/39/40）----
|
||||
case MethodSocialGetPerson, MethodSocialGetNetwork, MethodSocialGetTrait,
|
||||
MethodSocialGetRelation, MethodSocialListPersons:
|
||||
return h.social(method, params)
|
||||
|
||||
// ---- 插件管理（原 case 48/49/50）----
|
||||
case MethodPluginReloadOne:
|
||||
pm := h.sdk.PluginMgr()
|
||||
if pm == nil {
|
||||
return nil, errUnavailable("plugin manager")
|
||||
}
|
||||
var p struct {
|
||||
Name string `json:"name"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return nil, pm.ReloadOne(p.Name)
|
||||
case MethodPluginListLoaded:
|
||||
pm := h.sdk.PluginMgr()
|
||||
if pm == nil {
|
||||
return nil, errUnavailable("plugin manager")
|
||||
}
|
||||
list := pm.ListLoadedPlugins()
|
||||
if list == nil {
|
||||
list = []string{}
|
||||
}
|
||||
return map[string]interface{}{"plugins": list}, nil
|
||||
case MethodPluginIsDisabled:
|
||||
pm := h.sdk.PluginMgr()
|
||||
if pm == nil {
|
||||
return nil, errUnavailable("plugin manager")
|
||||
}
|
||||
var p struct {
|
||||
Name string `json:"name"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"disabled": pm.IsPluginDisabled(p.Name)}, nil
|
||||
|
||||
// ---- 共享段锁仲裁（新增，§3.7）----
|
||||
case MethodStageLock:
|
||||
if h.locks == nil {
|
||||
return nil, fmt.Errorf("stage.lock: 锁仲裁未就绪")
|
||||
}
|
||||
return nil, h.locks.acquire(h.name)
|
||||
case MethodStageUnlock:
|
||||
if h.locks == nil {
|
||||
return nil, fmt.Errorf("stage.unlock: 锁仲裁未就绪")
|
||||
}
|
||||
return nil, h.locks.release(h.name)
|
||||
|
||||
// ---- 事件订阅（原 case 23/24，今日空实现）----
|
||||
case MethodEventsSubscribe, MethodEventsUnsubscribe:
|
||||
// Part 5 通知面（事件环 + eventfd）落地后接线。
|
||||
// 今日 C ABI 侧是空实现（"给不了"而非"不给"，§1.3）；
|
||||
// 明确返回未实现，比静默成功后收不到事件更容易排查。
|
||||
return nil, fmt.Errorf("%s: 事件订阅待 Part 5 通知面落地（事件环 + eventfd）", method)
|
||||
|
||||
// ---- 多模态注入（C ABI 侧空实现）----
|
||||
case MethodIOSetToolBlocks:
|
||||
// Part 4 扩展：二进制落 arena、Slice 描述符回传（§3.8）。
|
||||
return nil, fmt.Errorf("%s: 多模态注入待共享段二进制通道落地", method)
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("未知 method: %s", method)
|
||||
}
|
||||
|
||||
type injectParams struct {
|
||||
Source string `json:"source"`
|
||||
Channel string `json:"channel"`
|
||||
Text string `json:"text"`
|
||||
}
|
||||
|
||||
func unmarshal(params json.RawMessage, out interface{}) error {
|
||||
if len(params) == 0 {
|
||||
return nil
|
||||
}
|
||||
if err := json.Unmarshal(params, out); err != nil {
|
||||
return fmt.Errorf("参数解析失败: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func errUnavailable(what string) error {
|
||||
return fmt.Errorf("%s 能力在当前内核实例中不可用", what)
|
||||
}
|
||||
|
||||
// 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"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if p.Name == "" {
|
||||
return nil, fmt.Errorf("tool.register: 缺少 name")
|
||||
}
|
||||
p.Def.Plugin = h.name
|
||||
// ToolDef.Cleaner 是函数，跨进程无法传递（§3.5）——与 C ABI 路径行为一致。
|
||||
p.Def.Cleaner = nil
|
||||
|
||||
name := p.Name
|
||||
return nil, h.sdk.RegisterTool(name, p.Def, func(args map[string]interface{}) (interface{}, error) {
|
||||
return h.invokeTool(name, args)
|
||||
})
|
||||
}
|
||||
|
||||
// stageRegister 注册阶段处理器（原 case 2）。
|
||||
//
|
||||
// **与 C ABI 路径的本质差异**：这里不做「快照 → 副本 → 写回」。
|
||||
// StageContext 的数据在共享段，插件直接在同一份状态上读改写，
|
||||
// 由锁仲裁串行化——消除了副本模型的 lost update（§8.4 实测 35.8~36.8%）。
|
||||
func (h *coreHandler) stageRegister(params json.RawMessage) (interface{}, error) {
|
||||
var p struct {
|
||||
Stage string `json:"stage"`
|
||||
Scope string `json:"scope"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if p.Stage == "" {
|
||||
return nil, fmt.Errorf("stage.register: 缺少 stage")
|
||||
}
|
||||
|
||||
scope := pubsdk.StageScopeGlobal
|
||||
if p.Scope == "own_tools" {
|
||||
scope = pubsdk.StageScopeOwnTools
|
||||
}
|
||||
stage := p.Stage
|
||||
h.sdk.RegisterStage(pubsdk.Stage(stage), func(sc *pubsdk.StageContext) error {
|
||||
return h.runStage(stage, sc)
|
||||
}, scope)
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// outputRegister 注册输出通道（原 case 3）。
|
||||
//
|
||||
// **与 C ABI 路径的本质差异**：可同步等真实结果。
|
||||
// C ABI 下因 cgo 不可嵌套，只能异步 fire-and-forget，导致 output_send
|
||||
// 永远返回成功（§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"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if p.Name == "" {
|
||||
return nil, fmt.Errorf("output.register: 缺少 name")
|
||||
}
|
||||
channel := p.Name
|
||||
return nil, h.sdk.RegisterOutputChannel(channel, p.Caps, p.Desc,
|
||||
pubsdk.ChannelDef{NoMemory: p.Def.NoMemory},
|
||||
func(args map[string]interface{}) (interface{}, error) {
|
||||
return h.invokeOutput(channel, args)
|
||||
})
|
||||
}
|
||||
|
||||
func (h *coreHandler) settings(method string, params json.RawMessage) (interface{}, error) {
|
||||
sett := h.sdk.Settings()
|
||||
if sett == nil {
|
||||
return nil, errUnavailable("settings")
|
||||
}
|
||||
var p struct {
|
||||
Key string `json:"key"`
|
||||
Value interface{} `json:"value"`
|
||||
Prefix string `json:"prefix"`
|
||||
Plugin string `json:"plugin"`
|
||||
Def pubsdk.ConfigDef `json:"def"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
switch method {
|
||||
case MethodSettingsGet:
|
||||
v, err := sett.Get(p.Key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"value": v}, nil
|
||||
case MethodSettingsSet:
|
||||
return nil, sett.Set(p.Key, p.Value)
|
||||
case MethodSettingsRegisterDef:
|
||||
sett.RegisterDef(p.Def)
|
||||
return nil, nil
|
||||
case MethodSettingsGetCore:
|
||||
v, err := sett.GetCore(p.Key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"value": v}, nil
|
||||
case MethodSettingsSetCore:
|
||||
return nil, sett.SetCore(p.Key, p.Value)
|
||||
case MethodSettingsListCore:
|
||||
keys, err := sett.ListCore(p.Prefix)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"keys": orEmpty(keys)}, nil
|
||||
case MethodSettingsGetPlugin:
|
||||
v, err := sett.GetPlugin(p.Plugin, p.Key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"value": v}, nil
|
||||
case MethodSettingsSetPlugin:
|
||||
return nil, sett.SetPlugin(p.Plugin, p.Key, p.Value)
|
||||
case MethodSettingsListPlugin:
|
||||
keys, err := sett.ListPlugin(p.Plugin, p.Prefix)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"keys": orEmpty(keys)}, nil
|
||||
case MethodSettingsList:
|
||||
keys, err := sett.List(p.Prefix)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"keys": orEmpty(keys)}, nil
|
||||
case MethodSettingsDefs:
|
||||
defs := sett.Defs(p.Prefix)
|
||||
if defs == nil {
|
||||
defs = []*pubsdk.ConfigDef{}
|
||||
}
|
||||
return map[string]interface{}{"defs": defs}, nil
|
||||
case MethodSettingsDump:
|
||||
return sett.Dump(), nil
|
||||
case MethodSettingsPlugins:
|
||||
return map[string]interface{}{"plugins": orEmpty(sett.Plugins())}, nil
|
||||
case MethodSettingsDataDir:
|
||||
return map[string]interface{}{"dir": sett.DataDir()}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("未知 settings method: %s", method)
|
||||
}
|
||||
|
||||
func (h *coreHandler) social(method string, params json.RawMessage) (interface{}, error) {
|
||||
social := h.sdk.Social()
|
||||
if social == nil {
|
||||
return nil, errUnavailable("social")
|
||||
}
|
||||
var p struct {
|
||||
Name string `json:"name"`
|
||||
Trait string `json:"trait"`
|
||||
Depth int `json:"depth"`
|
||||
}
|
||||
if err := unmarshal(params, &p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
switch method {
|
||||
case MethodSocialGetPerson:
|
||||
profile, err := social.GetPerson(p.Name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"person": profile}, nil
|
||||
case MethodSocialGetNetwork:
|
||||
profiles, err := social.GetNetwork(p.Name, p.Depth)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if profiles == nil {
|
||||
profiles = []*pubsdk.PersonProfile{}
|
||||
}
|
||||
return map[string]interface{}{"network": profiles}, nil
|
||||
case MethodSocialGetTrait:
|
||||
v, ok := social.GetTrait(p.Name, p.Trait)
|
||||
return map[string]interface{}{"value": v, "found": ok}, nil
|
||||
case MethodSocialGetRelation:
|
||||
rels, err := social.GetRelations(p.Name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if rels == nil {
|
||||
rels = []pubsdk.SocialRelation{}
|
||||
}
|
||||
return map[string]interface{}{"relations": rels}, nil
|
||||
case MethodSocialListPersons:
|
||||
names, err := social.ListPersons()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]interface{}{"persons": orEmpty(names)}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("未知 social method: %s", method)
|
||||
}
|
||||
|
||||
func orEmpty(s []string) []string {
|
||||
if s == nil {
|
||||
return []string{}
|
||||
}
|
||||
return s
|
||||
}
|
||||
215
internal/plugin/proc/host.go
Normal file
215
internal/plugin/proc/host.go
Normal file
@ -0,0 +1,215 @@
|
||||
package proc
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"sync"
|
||||
|
||||
pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
|
||||
"golang.org/x/sys/unix"
|
||||
)
|
||||
|
||||
// Host 持有**被全部子进程插件共享的一块 StageContext 段**，是共享内存数据面的
|
||||
// 所有权中心（§3.3/§3.4）。
|
||||
//
|
||||
// ❗ 为什么必须共享一块段（这是一个容易走错的关键点）：
|
||||
// 若每个插件各持一块段，则「内核 ctx → 段 → 插件改 → 回读 ctx」在多插件下退化成
|
||||
// 副本模型——两个插件各写各的段、各自回读，最后回读者覆盖前者，
|
||||
// lost update 原样复现（§8.4 实测 35.8~36.8%）。
|
||||
// 实验 8 的做法是 5 个 worker 进程 mmap **同一个 memfd**，本实现与之一致。
|
||||
//
|
||||
// 生命周期：Host 由 registry 创建一次，随内核存活；每个插件 spawn 时经
|
||||
// ExtraFiles 拿到同一 memfd（fd 3），mmap 后即看到同一份物理页。
|
||||
type Host struct {
|
||||
memfd *os.File
|
||||
data []byte
|
||||
seg *Segment
|
||||
shmSize int
|
||||
|
||||
// locks 被全部插件的 coreHandler 共享——同阶段并发扇出的插件在此排队，
|
||||
// 语义等价于内置插件共享 *StageContext 的 sync.RWMutex（§0.2 第 1 条）。
|
||||
locks *lockRegistry
|
||||
|
||||
// stageMu 串行化「整次 stage 执行」对共享段的独占。
|
||||
//
|
||||
// 必要性：内核可能在不同路径并发触发 RunStage（如 emitResponse 的
|
||||
// before_output 与主循环的其他阶段）。段只有一份，两次 stage 交叠会互相污染。
|
||||
// 由首个进入的插件加锁、最后离开的插件解锁；RunStage 的 wg.Wait() 保证
|
||||
// 每个 handler 的 defer 必然执行，故 inflight 必然归零，不会死锁。
|
||||
stageMu sync.Mutex
|
||||
|
||||
coordMu sync.Mutex
|
||||
coord *stageCoordinator
|
||||
}
|
||||
|
||||
// NewHost 创建共享段（memfd + mmap + 布局初始化）。
|
||||
//
|
||||
// 用 memfd 而非 /dev/shm 文件：无需文件名、不残留（进程退出即回收）、
|
||||
// 可经 ExtraFiles 传给子进程。实验 2 已验证父子 mmap 到不同虚拟地址时
|
||||
// 相对偏移仍正确解引用。
|
||||
func NewHost() (*Host, error) {
|
||||
fd, err := unix.MemfdCreate("hastagectx", unix.MFD_CLOEXEC)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("proc: 创建共享段 memfd: %w", err)
|
||||
}
|
||||
if err := unix.Ftruncate(fd, int64(shmDefaultSize)); err != nil {
|
||||
unix.Close(fd)
|
||||
return nil, fmt.Errorf("proc: 共享段 ftruncate: %w", err)
|
||||
}
|
||||
data, err := unix.Mmap(fd, 0, shmDefaultSize,
|
||||
unix.PROT_READ|unix.PROT_WRITE, unix.MAP_SHARED)
|
||||
if err != nil {
|
||||
unix.Close(fd)
|
||||
return nil, fmt.Errorf("proc: 共享段 mmap: %w", err)
|
||||
}
|
||||
seg, err := NewSegment(data)
|
||||
if err != nil {
|
||||
unix.Munmap(data)
|
||||
unix.Close(fd)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &Host{
|
||||
memfd: os.NewFile(uintptr(fd), "hastagectx"),
|
||||
data: data,
|
||||
seg: seg,
|
||||
shmSize: shmDefaultSize,
|
||||
locks: &lockRegistry{},
|
||||
}, nil
|
||||
}
|
||||
|
||||
// shmDefaultSize 是共享 StageContext 段的大小。
|
||||
//
|
||||
// 取 256KB：StageContext 全字段 JSON 化后典型 < 4KB（工具结果中位 93B，§2.5），
|
||||
// append-only 中间垃圾由 stage 结束时 Compact 回收，256KB 给足余量。
|
||||
// 全部插件共享一块，总开销恒定，不随插件数增长。
|
||||
const shmDefaultSize = 256 * 1024
|
||||
|
||||
// Close 释放共享段。
|
||||
func (h *Host) Close() error {
|
||||
if h.data != nil {
|
||||
unix.Munmap(h.data)
|
||||
h.data = nil
|
||||
}
|
||||
if h.memfd != nil {
|
||||
err := h.memfd.Close()
|
||||
h.memfd = nil
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// beginStage 由插件 handler 进入时调用。
|
||||
//
|
||||
// 首个进入者：获取 stageMu（独占共享段）→ 把内核 StageContext 写入段。
|
||||
// 后续进入者：仅递增 inflight。
|
||||
func (h *Host) beginStage(sc *pubsdk.StageContext) (*stageCoordinator, error) {
|
||||
h.coordMu.Lock()
|
||||
first := h.coord == nil
|
||||
if first {
|
||||
// 独占共享段直到本次 stage 全部插件离开
|
||||
h.coordMu.Unlock()
|
||||
h.stageMu.Lock()
|
||||
h.coordMu.Lock()
|
||||
// 双检：等锁期间可能已有其他插件建好协调器（它们会先拿到 stageMu）
|
||||
if h.coord != nil {
|
||||
first = false
|
||||
h.stageMu.Unlock()
|
||||
} else {
|
||||
h.coord = newStageCoordinator(h.seg)
|
||||
}
|
||||
}
|
||||
coord := h.coord
|
||||
h.coordMu.Unlock()
|
||||
|
||||
if err := coord.enter(sc, first); err != nil {
|
||||
if first {
|
||||
h.coordMu.Lock()
|
||||
h.coord = nil
|
||||
h.coordMu.Unlock()
|
||||
h.stageMu.Unlock()
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
h.locks.bind(coord.lock)
|
||||
return coord, nil
|
||||
}
|
||||
|
||||
// endStage 由插件 handler 返回时调用。
|
||||
// 最后离开者：把共享段结果读回内核 StageContext → 压实 arena → 释放 stageMu。
|
||||
func (h *Host) endStage(coord *stageCoordinator) error {
|
||||
last, err := coord.leave()
|
||||
if !last {
|
||||
return err
|
||||
}
|
||||
h.coordMu.Lock()
|
||||
h.coord = nil
|
||||
h.coordMu.Unlock()
|
||||
h.stageMu.Unlock()
|
||||
return err
|
||||
}
|
||||
|
||||
// ForceReleaseLock 在插件进程崩溃时释放其可能持有的 stage 锁（实验 9 自愈机制）。
|
||||
func (h *Host) ForceReleaseLock(plugin string) bool {
|
||||
return h.locks.forceRelease(plugin)
|
||||
}
|
||||
|
||||
// Segment 暴露共享段（供诊断与测试）。
|
||||
func (h *Host) Segment() *Segment { return h.seg }
|
||||
|
||||
// stageCoordinator 跟踪一次 stage 执行中参与插件的进出。
|
||||
type stageCoordinator struct {
|
||||
seg *Segment
|
||||
lock *stageLock
|
||||
|
||||
mu sync.Mutex
|
||||
inflight int
|
||||
written bool
|
||||
ctxRef *pubsdk.StageContext
|
||||
}
|
||||
|
||||
func newStageCoordinator(seg *Segment) *stageCoordinator {
|
||||
return &stageCoordinator{seg: seg, lock: newStageLock()}
|
||||
}
|
||||
|
||||
// enter 登记一个插件进入本次 stage；first 为真时把内核状态写入共享段。
|
||||
func (c *stageCoordinator) enter(sc *pubsdk.StageContext, first bool) error {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.inflight++
|
||||
if !first || c.written {
|
||||
return nil
|
||||
}
|
||||
c.ctxRef = sc
|
||||
if err := c.seg.WriteAll(sc); err != nil {
|
||||
c.inflight--
|
||||
return fmt.Errorf("写入共享段: %w", err)
|
||||
}
|
||||
c.written = true
|
||||
return nil
|
||||
}
|
||||
|
||||
// leave 登记一个插件离开；返回是否为最后一个离开者。
|
||||
//
|
||||
// 最后离开者负责把共享段结果读回内核 StageContext，并压实 arena
|
||||
// （此时无插件持锁，满足 §3.3 的压实前提）。
|
||||
func (c *stageCoordinator) leave() (last bool, err error) {
|
||||
c.mu.Lock()
|
||||
c.inflight--
|
||||
last = c.inflight == 0
|
||||
sc := c.ctxRef
|
||||
written := c.written
|
||||
c.mu.Unlock()
|
||||
|
||||
if !last || !written || sc == nil {
|
||||
return last, nil
|
||||
}
|
||||
if rErr := c.seg.ReadInto(sc); rErr != nil {
|
||||
return last, fmt.Errorf("回读共享段: %w", rErr)
|
||||
}
|
||||
if reclaimed := c.seg.Compact(); reclaimed > 0 {
|
||||
log.Printf("[proc] stage 结束，arena 压实回收 %d 字节", reclaimed)
|
||||
}
|
||||
return last, nil
|
||||
}
|
||||
229
internal/plugin/proc/plugin.go
Normal file
229
internal/plugin/proc/plugin.go
Normal file
@ -0,0 +1,229 @@
|
||||
package proc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"sync"
|
||||
|
||||
pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
|
||||
)
|
||||
|
||||
// Plugin 是 registry 可加载的子进程插件，与内置插件同构的启停接口。
|
||||
//
|
||||
// 生命周期：
|
||||
//
|
||||
// New() 创建（尚未 spawn）
|
||||
// Start(core) spawn 子进程 → 握手（传共享段 fd）→ plugin.init → plugin.start
|
||||
// （plugin.start 期间插件反向注册工具/阶段/通道）
|
||||
// Stop() plugin.stop → 宽限期 → 必要时 Kill
|
||||
// Close() 强制结束（registry 卸载/重载路径）
|
||||
//
|
||||
// **共享段不属于 Plugin**：它属于 Host，被全部子进程插件共享。
|
||||
// 若每插件一段，「内核 ctx → 段 → 插件改 → 回读 ctx」在多插件下会退化成
|
||||
// 副本模型，lost update 原样复现（§8.4）。
|
||||
type Plugin struct {
|
||||
name string
|
||||
bin string
|
||||
dir string
|
||||
config map[string]interface{}
|
||||
|
||||
host *Host
|
||||
proc *Process
|
||||
handler *coreHandler
|
||||
|
||||
// env 追加到子进程环境变量（测试用；生产由 registry 按需设置）。
|
||||
env []string
|
||||
|
||||
// onCrash 由 registry 注入，把进程退出喂给 plugin_health.recordCrash（§2.3）。
|
||||
onCrash func(name string, err error)
|
||||
|
||||
stopOnce sync.Once
|
||||
}
|
||||
|
||||
// New 创建子进程插件（不启动进程）。
|
||||
//
|
||||
// host 必须是全部子进程插件共用的实例（由 registry 创建一次）。
|
||||
func New(name, bin, dir string, config map[string]interface{}, host *Host, onCrash func(string, error)) *Plugin {
|
||||
return &Plugin{
|
||||
name: name,
|
||||
bin: bin,
|
||||
dir: dir,
|
||||
config: config,
|
||||
host: host,
|
||||
onCrash: onCrash,
|
||||
}
|
||||
}
|
||||
|
||||
// Name 实现 sdk.Plugin。
|
||||
func (p *Plugin) Name() string { return p.name }
|
||||
|
||||
// Start 启动子进程并完成注册。
|
||||
//
|
||||
// core 是内核为该插件构建的能力面（internal/sdk.PluginSDK 天然满足 CoreSDK）。
|
||||
func (p *Plugin) Start(core CoreSDK) error {
|
||||
if p.host == nil {
|
||||
return fmt.Errorf("proc: %s 缺少共享段 Host", p.name)
|
||||
}
|
||||
|
||||
p.handler = &coreHandler{
|
||||
sdk: core,
|
||||
name: p.name,
|
||||
host: p.host,
|
||||
locks: p.host.locks,
|
||||
}
|
||||
// 反向调用闭包：注册回调时捕获，运行期经 RPC 打到插件进程。
|
||||
p.handler.invokeTool = p.invokeTool
|
||||
p.handler.invokeStageFn = p.invokeStage
|
||||
p.handler.invokeOutput = p.invokeOutput
|
||||
|
||||
proc, err := Spawn(p.name, p.bin, Options{
|
||||
Dir: p.dir,
|
||||
Env: p.env,
|
||||
// 子进程 fd 3 = 共享段 memfd（全部插件同一个，故看到同一份物理页）
|
||||
ExtraFiles: []*os.File{p.host.memfd},
|
||||
ShmSize: p.host.shmSize,
|
||||
Handler: p.handler.Handle,
|
||||
OnExit: p.handleExit,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
p.proc = proc
|
||||
|
||||
// plugin.init：构造插件实例
|
||||
if _, err := proc.Call(MethodPluginInit, PluginInitParams{
|
||||
Name: p.name,
|
||||
Config: p.config,
|
||||
}); err != nil {
|
||||
proc.Kill()
|
||||
return fmt.Errorf("proc: %s plugin.init 失败: %w", p.name, err)
|
||||
}
|
||||
|
||||
// plugin.start：插件在此期间反向注册工具/阶段/通道
|
||||
if _, err := proc.Call(MethodPluginStart, nil); err != nil {
|
||||
proc.Kill()
|
||||
return fmt.Errorf("proc: %s plugin.start 失败: %w", p.name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop 优雅停止（实现 sdk.Plugin）。
|
||||
func (p *Plugin) Stop() error {
|
||||
var err error
|
||||
p.stopOnce.Do(func() {
|
||||
if p.proc != nil {
|
||||
err = p.proc.Stop()
|
||||
}
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
// Close 强制结束子进程。
|
||||
//
|
||||
// **这里是真 kill + wait**——对比 cabi 路径的 Close 只做 dlclose，
|
||||
// 而 dlclose 对 Go c-shared 是 no-op（§1.1，热重载静默失效的根因）。
|
||||
func (p *Plugin) Close() error {
|
||||
var err error
|
||||
p.stopOnce.Do(func() {
|
||||
if p.proc != nil {
|
||||
err = p.proc.Kill()
|
||||
}
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
// handleExit 在子进程退出时把信号喂给 plugin_health（§2.3 逻辑复用），
|
||||
// 并释放该插件可能持有的 stage 锁。
|
||||
//
|
||||
// 后者是"锁仲裁回内核"的自愈价值：持锁者死亡不会导致全局死锁，
|
||||
// 无需 robust pthread_mutex（实验 9）。
|
||||
func (p *Plugin) handleExit(name string, err error) {
|
||||
if p.host != nil && p.host.ForceReleaseLock(name) {
|
||||
log.Printf("[proc] %s 退出，内核已释放其持有的 stage 锁", name)
|
||||
}
|
||||
if err != nil && p.onCrash != nil {
|
||||
p.onCrash(name, err)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- 内核 → 插件的反向调用 ----
|
||||
|
||||
func (p *Plugin) invokeTool(name string, args map[string]interface{}) (interface{}, error) {
|
||||
if p.proc == nil {
|
||||
return nil, ErrProcessExited
|
||||
}
|
||||
raw, err := p.proc.Call(MethodToolInvoke, ToolInvokeParams{Name: name, Args: args})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var res ToolInvokeResult
|
||||
if err := json.Unmarshal(raw, &res); err != nil {
|
||||
return nil, fmt.Errorf("proc: %s 工具 %s 应答解析失败: %w", p.name, name, err)
|
||||
}
|
||||
return res.Result, nil
|
||||
}
|
||||
|
||||
func (p *Plugin) invokeStage(ctx context.Context, stage string, seq uint64) error {
|
||||
if p.proc == nil {
|
||||
return ErrProcessExited
|
||||
}
|
||||
raw, err := p.proc.CallContext(ctx, MethodStageInvoke, StageInvokeParams{
|
||||
Stage: stage,
|
||||
Seq: seq,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var res StageInvokeResult
|
||||
if len(raw) > 0 {
|
||||
if err := json.Unmarshal(raw, &res); err != nil {
|
||||
return fmt.Errorf("proc: %s stage %s 应答解析失败: %w", p.name, stage, err)
|
||||
}
|
||||
}
|
||||
if res.DirtyFields > 0 {
|
||||
log.Printf("[proc] %s stage %s 改写了 %d 个字段", p.name, stage, res.DirtyFields)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// invokeOutput 经插件输出通道发送，**同步等待真实结果**。
|
||||
//
|
||||
// 这是 §9.4 的根治：C ABI 下 cgo 不可嵌套，只能异步 fire-and-forget，
|
||||
// 导致 output_send 永远返回 {status:queued} + err=nil，模型永远以为发送成功
|
||||
// （现网 7 天内 2 次消息实际发不出）。进程模型下 RPC 天然可等应答。
|
||||
func (p *Plugin) invokeOutput(channel string, args map[string]interface{}) (interface{}, error) {
|
||||
if p.proc == nil {
|
||||
return nil, ErrProcessExited
|
||||
}
|
||||
raw, err := p.proc.Call(MethodOutputInvoke, OutputInvokeParams{
|
||||
Channel: channel,
|
||||
Args: args,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err // 真实失败上报，模型可感知并重试
|
||||
}
|
||||
if len(raw) == 0 {
|
||||
return map[string]interface{}{"status": "sent"}, nil
|
||||
}
|
||||
var res map[string]interface{}
|
||||
if err := json.Unmarshal(raw, &res); err != nil {
|
||||
return map[string]interface{}{"status": "sent"}, nil
|
||||
}
|
||||
if _, ok := res["status"]; !ok {
|
||||
res["status"] = "sent"
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// 编译期确认 Plugin 具备 registry 需要的启停形状。
|
||||
var _ interface {
|
||||
Name() string
|
||||
Stop() error
|
||||
Close() error
|
||||
} = (*Plugin)(nil)
|
||||
|
||||
// 引用一下公开 SDK，确保本文件的类型假设与它同版本。
|
||||
var _ = pubsdk.StageScopeGlobal
|
||||
393
internal/plugin/proc/plugin_test.go
Normal file
393
internal/plugin/proc/plugin_test.go
Normal file
@ -0,0 +1,393 @@
|
||||
package proc
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
|
||||
)
|
||||
|
||||
// 端到端验证：内核 RunStage 并发扇出 → 真实子进程插件经共享内存读改写 → 结果回读。
|
||||
//
|
||||
// 这是**整个迁移最关键的一环闭环验证**（§4.4 风险 3.4）：
|
||||
// 机制在 shm_test.go 已被单元验证，这里验证它在真进程 + 真 RPC 下同样成立。
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
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{}{},
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fakeCoreSDK) PluginName() string { return "fake" }
|
||||
func (f *fakeCoreSDK) Settings() pubsdk.SettingsAPI { return nil }
|
||||
func (f *fakeCoreSDK) Memory() pubsdk.MemoryAPI { return nil }
|
||||
func (f *fakeCoreSDK) TextMemory() pubsdk.TextMemoryAPI { return nil }
|
||||
func (f *fakeCoreSDK) DocMemory() pubsdk.DocMemoryAPI { return nil }
|
||||
func (f *fakeCoreSDK) Knowledge() pubsdk.KnowledgeAPI { return nil }
|
||||
func (f *fakeCoreSDK) LLM() pubsdk.LLMAPI { return nil }
|
||||
func (f *fakeCoreSDK) Social() pubsdk.SocialAPI { return nil }
|
||||
func (f *fakeCoreSDK) PluginMgr() pubsdk.PluginMgrAPI { return nil }
|
||||
func (f *fakeCoreSDK) RegisterPluginAPI(name string) error { return nil }
|
||||
func (f *fakeCoreSDK) InjectText(s, c, t string) {}
|
||||
func (f *fakeCoreSDK) InjectInterruptText(s, c, t string) {}
|
||||
func (f *fakeCoreSDK) InjectTextNoMemory(s, c, t string) {}
|
||||
func (f *fakeCoreSDK) InjectInputSync(s, c, t string) string { return "" }
|
||||
func (f *fakeCoreSDK) SetAutoRestart(enabled bool) { f.autoStart = enabled }
|
||||
|
||||
func (f *fakeCoreSDK) RegisterTool(name string, def pubsdk.ToolDef, h pubsdk.ToolHandler) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.tools[name] = h
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeCoreSDK) RegisterStage(stage pubsdk.Stage, h pubsdk.StageHandler, scope ...pubsdk.StageScope) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.stages[stage] = append(f.stages[stage], h)
|
||||
}
|
||||
|
||||
func (f *fakeCoreSDK) RegisterOutputChannel(name string, caps int, desc string, def pubsdk.ChannelDef, h pubsdk.ToolHandler) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.outputs[name] = h
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeCoreSDK) RegisterInputChannel(name string, def pubsdk.ChannelDef) error { return nil }
|
||||
|
||||
func (f *fakeCoreSDK) stageHandlers(stage pubsdk.Stage) []pubsdk.StageHandler {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
out := make([]pubsdk.StageHandler, len(f.stages[stage]))
|
||||
copy(out, f.stages[stage])
|
||||
return out
|
||||
}
|
||||
|
||||
// runStageLikeKernel 复刻 internal/agent/core.StageHost.RunStage 的并发扇出语义
|
||||
// （stages.go:124 的 go func + wg.Wait），验证外部插件在同样的并发模型下正确工作。
|
||||
func runStageLikeKernel(handlers []pubsdk.StageHandler, sc *pubsdk.StageContext) []error {
|
||||
var wg sync.WaitGroup
|
||||
errCh := make(chan error, len(handlers))
|
||||
for _, h := range handlers {
|
||||
wg.Add(1)
|
||||
go func(fn pubsdk.StageHandler) {
|
||||
defer wg.Done()
|
||||
if err := fn(sc); err != nil {
|
||||
errCh <- err
|
||||
}
|
||||
}(h)
|
||||
}
|
||||
wg.Wait()
|
||||
close(errCh)
|
||||
var errs []error
|
||||
for err := range errCh {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
return errs
|
||||
}
|
||||
|
||||
// 单插件 stage 读改写：验证共享段 + RPC + 锁的完整链路。
|
||||
func TestPlugin_StageReadModifyWriteOverSharedMemory(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("sanitizer", bin, t.TempDir(), nil, host, nil)
|
||||
if err := p.Start(core); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
defer p.Close()
|
||||
|
||||
handlers := core.stageHandlers(pubsdk.StageAfterToolcall)
|
||||
if len(handlers) != 1 {
|
||||
t.Fatalf("插件应注册 1 个 after_toolcall handler，实际 %d", len(handlers))
|
||||
}
|
||||
|
||||
dirty := "结果：\x1b[31m脏数据\x1b[0m"
|
||||
clean := "结果：脏数据"
|
||||
sc := &pubsdk.StageContext{
|
||||
Phase: pubsdk.StageAfterToolcall,
|
||||
ToolResults: []pubsdk.ToolResult{{CallID: "c1", Name: "x_tool", Result: dirty}},
|
||||
}
|
||||
|
||||
if errs := runStageLikeKernel(handlers, sc); len(errs) > 0 {
|
||||
t.Fatalf("stage 执行失败: %v", errs)
|
||||
}
|
||||
|
||||
got, _ := sc.ToolResults[0].Result.(string)
|
||||
if got != clean {
|
||||
t.Fatalf("插件的清洗结果未回到内核 StageContext：期望 %q，实际 %q", clean, got)
|
||||
}
|
||||
}
|
||||
|
||||
// **核心断言**：改写型插件 + 只读插件并发时，清洗结果不被覆盖。
|
||||
// 复刻现网 sanitizer + weather 场景（§8.6 实测 C ABI 下 1.6~4.3% 被覆盖）。
|
||||
func TestPlugin_ConcurrentWriterAndReaderNoLostUpdate(t *testing.T) {
|
||||
bin := buildTestPlugin(t, "stageplugin.go")
|
||||
|
||||
// ❗ 两个插件进程**共享同一个 Host**（同一 memfd）——这是消除 lost update 的前提。
|
||||
// 若各持一段，「内核 ctx → 段 → 插件改 → 回读 ctx」会退化成副本模型，
|
||||
// 最后回读者覆盖前者，§8.4 的 35.8~36.8% 丢失原样复现。
|
||||
host, err := NewHost()
|
||||
if err != nil {
|
||||
t.Fatalf("NewHost: %v", err)
|
||||
}
|
||||
defer host.Close()
|
||||
|
||||
writerCore := newFakeCore()
|
||||
writer := New("sanitizer", bin, t.TempDir(), nil, host, nil)
|
||||
if err := writer.Start(writerCore); err != nil {
|
||||
t.Fatalf("writer Start: %v", err)
|
||||
}
|
||||
defer writer.Close()
|
||||
|
||||
readerBin := buildTestPlugin(t, "readonlyplugin.go")
|
||||
readerCore := newFakeCore()
|
||||
reader := New("weather", readerBin, t.TempDir(), nil, host, nil)
|
||||
if err := reader.Start(readerCore); err != nil {
|
||||
t.Fatalf("reader Start: %v", err)
|
||||
}
|
||||
defer reader.Close()
|
||||
|
||||
handlers := append(
|
||||
writerCore.stageHandlers(pubsdk.StageAfterToolcall),
|
||||
readerCore.stageHandlers(pubsdk.StageAfterToolcall)...,
|
||||
)
|
||||
if len(handlers) != 2 {
|
||||
t.Fatalf("应有 2 个 handler，实际 %d", len(handlers))
|
||||
}
|
||||
|
||||
dirty := "天气：晴 \x1b[31m28°C\x1b[0m"
|
||||
clean := "天气：晴 28°C"
|
||||
sc := &pubsdk.StageContext{
|
||||
Phase: pubsdk.StageAfterToolcall,
|
||||
ToolResults: []pubsdk.ToolResult{{CallID: "c1", Name: "weather_query", Result: dirty}},
|
||||
}
|
||||
|
||||
if errs := runStageLikeKernel(handlers, sc); len(errs) > 0 {
|
||||
t.Fatalf("stage 执行失败: %v", errs)
|
||||
}
|
||||
|
||||
got, _ := sc.ToolResults[0].Result.(string)
|
||||
if got != clean {
|
||||
t.Fatalf("只读插件覆盖了改写插件的清洗结果：期望 %q，实际 %q", clean, got)
|
||||
}
|
||||
}
|
||||
|
||||
// 插件注册的工具可被内核调用，并把结果带回。
|
||||
func TestPlugin_RegisteredToolInvokable(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("demo", bin, t.TempDir(), nil, host, nil)
|
||||
if err := p.Start(core); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
defer p.Close()
|
||||
|
||||
core.mu.Lock()
|
||||
h, ok := core.tools["demo_upper"]
|
||||
core.mu.Unlock()
|
||||
if !ok {
|
||||
t.Fatal("插件应注册 demo_upper 工具")
|
||||
}
|
||||
|
||||
res, err := h(map[string]interface{}{"text": "abc"})
|
||||
if err != nil {
|
||||
t.Fatalf("调用工具: %v", err)
|
||||
}
|
||||
if res != "ABC" {
|
||||
t.Fatalf("工具结果应为 ABC，实际 %v", res)
|
||||
}
|
||||
}
|
||||
|
||||
// 输出通道**同步等真实结果**：失败必须上报（§9.4 根治）。
|
||||
func TestPlugin_OutputChannelReportsRealFailure(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("demo", bin, t.TempDir(), nil, host, nil)
|
||||
if err := p.Start(core); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
defer p.Close()
|
||||
|
||||
core.mu.Lock()
|
||||
h, ok := core.outputs["demo_ch"]
|
||||
core.mu.Unlock()
|
||||
if !ok {
|
||||
t.Fatal("插件应注册 demo_ch 输出通道")
|
||||
}
|
||||
|
||||
// 成功路径
|
||||
res, err := h(map[string]interface{}{"payload": "hi", "type": "text"})
|
||||
if err != nil {
|
||||
t.Fatalf("发送应成功: %v", err)
|
||||
}
|
||||
m, _ := res.(map[string]interface{})
|
||||
if m["status"] != "sent" {
|
||||
t.Errorf("成功应返回 status=sent，实际 %v", m)
|
||||
}
|
||||
|
||||
// 失败路径：插件返回错误 → 调用方必须收到 error（而非假成功）
|
||||
_, err = h(map[string]interface{}{"payload": "fail", "type": "text"})
|
||||
if err == nil {
|
||||
t.Fatal("发送失败时必须上报 error（C ABI 路径此处永远假成功）")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "缺少 user_id") {
|
||||
t.Errorf("应透传插件的失败原因，实际: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 插件在 plugin.start 期间反向调用内核（settings/autoRestart 等）。
|
||||
func TestPlugin_ReverseCallsDuringStart(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("demo", bin, t.TempDir(), nil, host, nil)
|
||||
if err := p.Start(core); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
defer p.Close()
|
||||
|
||||
if !core.autoStart {
|
||||
t.Error("插件调用 lifecycle.autoRestart 后内核状态应更新")
|
||||
}
|
||||
}
|
||||
|
||||
// 权限梯度显式化（§3.8）：CoreSDK 不提供内核内部机制，
|
||||
// 插件请求这些能力时必须被拒绝而非静默忽略。
|
||||
func TestCoreHandler_RejectsUnknownAndUnimplementedMethods(t *testing.T) {
|
||||
h := &coreHandler{sdk: newFakeCore(), name: "x", locks: &lockRegistry{}}
|
||||
|
||||
// 未知 method
|
||||
if _, err := h.Handle("supervisor.restart", nil); err == nil {
|
||||
t.Error("内核内部机制不应可达（应报未知 method）")
|
||||
}
|
||||
|
||||
// 事件订阅：今日 C ABI 是空实现（静默成功），这里必须明确报未实现
|
||||
if _, err := h.Handle(MethodEventsSubscribe, json.RawMessage(`{}`)); err == nil {
|
||||
t.Error("事件订阅未落地时应明确报错，而非静默成功后收不到事件")
|
||||
}
|
||||
|
||||
// 多模态注入同理
|
||||
if _, err := h.Handle(MethodIOSetToolBlocks, json.RawMessage(`{}`)); err == nil {
|
||||
t.Error("多模态注入未落地时应明确报错")
|
||||
}
|
||||
}
|
||||
|
||||
// stage 锁在无进行中 stage 时申请应被拒绝（防止插件在 stage 外乱加锁）。
|
||||
func TestCoreHandler_StageLockOutsideStageRejected(t *testing.T) {
|
||||
h := &coreHandler{sdk: newFakeCore(), name: "x", locks: &lockRegistry{}}
|
||||
if _, err := h.Handle(MethodStageLock, nil); err == nil {
|
||||
t.Error("stage 外加锁应被拒绝")
|
||||
}
|
||||
if !strings.Contains(fmt.Sprint(mustErr(h.Handle(MethodStageUnlock, nil))), "无进行中的 stage") {
|
||||
t.Error("stage 外解锁的错误信息应说明原因")
|
||||
}
|
||||
}
|
||||
|
||||
func mustErr(_ interface{}, err error) error { return err }
|
||||
|
||||
// **跨进程 lost update 终极验证**：5 个独立插件进程并发读-改-写同一个
|
||||
// FinalText，全部标记必须保留。
|
||||
//
|
||||
// 这是实验 8（5 进程 × 300 轮零丢失）在真实 RPC + 真实 RunStage 并发扇出
|
||||
// 下的复刻。对照今日 C ABI 副本模型实测 35.8~36.8% 丢失（§8.4）。
|
||||
func TestPlugin_FiveProcessesConcurrentAppendNoLostUpdate(t *testing.T) {
|
||||
bin := buildTestPlugin(t, "appendplugin.go")
|
||||
|
||||
// 关键：全部插件共享同一个 Host（同一 memfd）
|
||||
host, err := NewHost()
|
||||
if err != nil {
|
||||
t.Fatalf("NewHost: %v", err)
|
||||
}
|
||||
defer host.Close()
|
||||
|
||||
tags := []string{"A", "B", "C", "D", "E"}
|
||||
var handlers []pubsdk.StageHandler
|
||||
for _, tag := range tags {
|
||||
core := newFakeCore()
|
||||
p := New("append-"+tag, bin, t.TempDir(), nil, host, nil)
|
||||
p.env = []string{"PLUGIN_TAG=" + tag}
|
||||
if err := p.Start(core); err != nil {
|
||||
t.Fatalf("插件 %s Start: %v", tag, err)
|
||||
}
|
||||
defer p.Close()
|
||||
handlers = append(handlers, core.stageHandlers(pubsdk.StageAfterToolcall)...)
|
||||
}
|
||||
if len(handlers) != len(tags) {
|
||||
t.Fatalf("应有 %d 个 handler，实际 %d", len(tags), len(handlers))
|
||||
}
|
||||
|
||||
sc := &pubsdk.StageContext{
|
||||
Phase: pubsdk.StageAfterToolcall,
|
||||
FinalText: "",
|
||||
}
|
||||
|
||||
if errs := runStageLikeKernel(handlers, sc); len(errs) > 0 {
|
||||
t.Fatalf("并发 stage 执行失败: %v", errs)
|
||||
}
|
||||
|
||||
// 断言：各标记出现次数之和 == 最终长度 == 插件数 ⇒ 无丢失、无撕裂
|
||||
total := 0
|
||||
counts := map[string]int{}
|
||||
for _, tag := range tags {
|
||||
c := strings.Count(sc.FinalText, tag)
|
||||
counts[tag] = c
|
||||
total += c
|
||||
}
|
||||
if total != len(sc.FinalText) {
|
||||
t.Fatalf("出现撕裂：各标记计数之和 %d != 最终长度 %d（final=%q counts=%v）",
|
||||
total, len(sc.FinalText), sc.FinalText, counts)
|
||||
}
|
||||
if total != len(tags) {
|
||||
t.Fatalf("出现 lost update：期望 %d 个插件的写入全部保留，实际 %d（final=%q counts=%v）",
|
||||
len(tags), total, sc.FinalText, counts)
|
||||
}
|
||||
for tag, c := range counts {
|
||||
if c != 1 {
|
||||
t.Errorf("插件 %s 的写入丢失：期望 1 次，实际 %d 次", tag, c)
|
||||
}
|
||||
}
|
||||
}
|
||||
111
internal/plugin/proc/stage.go
Normal file
111
internal/plugin/proc/stage.go
Normal file
@ -0,0 +1,111 @@
|
||||
package proc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
|
||||
)
|
||||
|
||||
// stage 执行：把内核的 RunStage 并发扇出接到共享段（§3.4，风险 3.4 的落点）。
|
||||
//
|
||||
// 执行链路：
|
||||
//
|
||||
// 内核 RunStage（并发 go func，原始设计不变）
|
||||
// └─ 外部插件 handler = coreHandler.runStage()
|
||||
// ├─ Host.beginStage：首个到达者独占共享段并写入 StageContext
|
||||
// ├─ stage.invoke RPC → 插件进程
|
||||
// │ └─ 插件侧：stage.lock → 读共享段 → handler → 只写脏字段 → stage.unlock
|
||||
// └─ Host.endStage：最后离开者把共享段读回内核 StageContext + 压实 arena
|
||||
//
|
||||
// 关键性质：
|
||||
// - **并发扇出保留**（§0.2 第 1 条：并发扇出是原始设计，不是缺陷）
|
||||
// - **无副本**：全部插件 mmap 同一 memfd，在同一份状态上读改写，锁仲裁串行化临界区
|
||||
// - **只读插件零写入**：脏字段集为空 → 不可能覆盖他人改写
|
||||
//
|
||||
// 对照今日 C ABI：每个插件拿到独立 JSON 副本，回传时无条件覆盖 10 个字段，
|
||||
// 实测 35.8~36.8% lost update（§8.4），现网量级百分之几脏数据进 LLM（§8.6）。
|
||||
|
||||
// lockRegistry 持有当前进行中 stage 的锁，供插件的 stage.lock/unlock 路由。
|
||||
type lockRegistry struct {
|
||||
mu sync.Mutex
|
||||
lock *stageLock
|
||||
}
|
||||
|
||||
func (r *lockRegistry) bind(l *stageLock) {
|
||||
r.mu.Lock()
|
||||
r.lock = l
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
func (r *lockRegistry) current() *stageLock {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return r.lock
|
||||
}
|
||||
|
||||
func (r *lockRegistry) acquire(plugin string) error {
|
||||
l := r.current()
|
||||
if l == nil {
|
||||
return fmt.Errorf("stage.lock: 当前无进行中的 stage（插件 %s 在 stage 外加锁？）", plugin)
|
||||
}
|
||||
return l.Acquire(plugin)
|
||||
}
|
||||
|
||||
func (r *lockRegistry) release(plugin string) error {
|
||||
l := r.current()
|
||||
if l == nil {
|
||||
return fmt.Errorf("stage.unlock: 当前无进行中的 stage（插件 %s）", plugin)
|
||||
}
|
||||
return l.Release(plugin)
|
||||
}
|
||||
|
||||
// forceRelease 在插件进程崩溃时释放其可能持有的锁（实验 9 的自愈机制）。
|
||||
func (r *lockRegistry) forceRelease(plugin string) bool {
|
||||
l := r.current()
|
||||
if l == nil {
|
||||
return false
|
||||
}
|
||||
return l.ForceRelease(plugin)
|
||||
}
|
||||
|
||||
// stageInvokeTimeout 是单个插件执行 stage 的上限。
|
||||
//
|
||||
// 取 30s：与内核工具超时（60s，toolcall.go）留出差距，
|
||||
// 使 stage 超时能被识别为 stage 问题而非工具问题。
|
||||
// 超时后调用方返回错误，卡住的插件进程可由上层 Kill 回收——
|
||||
// **对比 cgo 路径超时后 OS 线程永久泄漏（现网 26 次，§9.3）**。
|
||||
const stageInvokeTimeout = 30 * time.Second
|
||||
|
||||
// runStage 是注册到内核 StageHost 的 handler（每个外部插件一个）。
|
||||
func (h *coreHandler) runStage(stage string, sc *pubsdk.StageContext) error {
|
||||
if h.host == nil {
|
||||
return fmt.Errorf("插件 %s: stage %s 共享段未就绪", h.name, stage)
|
||||
}
|
||||
|
||||
coord, err := h.host.beginStage(sc)
|
||||
if err != nil {
|
||||
return fmt.Errorf("插件 %s stage %s: %w", h.name, stage, err)
|
||||
}
|
||||
defer func() {
|
||||
if endErr := h.host.endStage(coord); endErr != nil {
|
||||
log.Printf("[proc] %s stage %s 收尾失败: %v", h.name, stage, endErr)
|
||||
}
|
||||
}()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), stageInvokeTimeout)
|
||||
defer cancel()
|
||||
|
||||
if err := h.invokeStageWithCtx(ctx, stage, coord.seg.Seq()); err != nil {
|
||||
// 插件可能在持锁时失败（崩溃/超时）——强制释放，避免后续插件死锁。
|
||||
// 这正是"锁仲裁回内核"的自愈价值（实验 9）：无需 robust mutex。
|
||||
if h.host.ForceReleaseLock(h.name) {
|
||||
log.Printf("[proc] %s stage %s 失败后强制释放其持有的 stage 锁", h.name, stage)
|
||||
}
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
232
internal/plugin/proc/testdata/appendplugin.go
vendored
Normal file
232
internal/plugin/proc/testdata/appendplugin.go
vendored
Normal file
@ -0,0 +1,232 @@
|
||||
//go:build ignore
|
||||
|
||||
// appendplugin 在 stage 中把自己的标记追加到 FinalText（读-改-写）。
|
||||
//
|
||||
// 用于跨进程 lost update 验证：多个此类插件并发处理同一 stage，
|
||||
// 若全部标记都保留 ⇒ 无丢失；若少了 ⇒ 出现 lost update。
|
||||
//
|
||||
// 这是实验 8（5 进程 × 300 轮零丢失）在真实 RPC + 真实内核 RunStage
|
||||
// 下的复刻——机制单测已过，这里验证集成后同样成立。
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
type request struct {
|
||||
ID uint64 `json:"id,omitempty"`
|
||||
Method string `json:"method"`
|
||||
Params json.RawMessage `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
type response struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result interface{} `json:"result,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
var (
|
||||
out = bufio.NewWriter(os.Stdout)
|
||||
writeMu sync.Mutex
|
||||
|
||||
nextID uint64
|
||||
pendMu sync.Mutex
|
||||
pending = map[uint64]chan json.RawMessage{}
|
||||
|
||||
shm []byte
|
||||
tag string
|
||||
)
|
||||
|
||||
func send(v interface{}) {
|
||||
b, _ := json.Marshal(v)
|
||||
writeMu.Lock()
|
||||
out.Write(b)
|
||||
out.WriteByte('\n')
|
||||
out.Flush()
|
||||
writeMu.Unlock()
|
||||
}
|
||||
|
||||
func callKernel(method string, params interface{}) (json.RawMessage, bool) {
|
||||
pendMu.Lock()
|
||||
nextID++
|
||||
id := nextID
|
||||
ch := make(chan json.RawMessage, 1)
|
||||
pending[id] = ch
|
||||
pendMu.Unlock()
|
||||
|
||||
var raw json.RawMessage
|
||||
if params != nil {
|
||||
b, _ := json.Marshal(params)
|
||||
raw = b
|
||||
}
|
||||
send(request{ID: id, Method: method, Params: raw})
|
||||
r, ok := <-ch
|
||||
return r, ok
|
||||
}
|
||||
|
||||
const (
|
||||
headerSize = 64
|
||||
stageFieldCount = 18
|
||||
sliceSize = 8
|
||||
flagCount = 8
|
||||
|
||||
offArenaBase = 8
|
||||
offArenaCap = 12
|
||||
offArenaUsed = 16
|
||||
offCtxBase = 20
|
||||
offSeq = 24
|
||||
|
||||
fFinalText = 5 // 与 proc/shmcodec.go 的 stageField 枚举顺序一致
|
||||
)
|
||||
|
||||
func desc(field int) (uint32, uint32) {
|
||||
cb := binary.LittleEndian.Uint32(shm[offCtxBase:])
|
||||
o := cb + uint32(field*sliceSize)
|
||||
return binary.LittleEndian.Uint32(shm[o:]), binary.LittleEndian.Uint32(shm[o+4:])
|
||||
}
|
||||
|
||||
func setDesc(field int, off, ln uint32) {
|
||||
cb := binary.LittleEndian.Uint32(shm[offCtxBase:])
|
||||
o := cb + uint32(field*sliceSize)
|
||||
binary.LittleEndian.PutUint32(shm[o:], off)
|
||||
binary.LittleEndian.PutUint32(shm[o+4:], ln)
|
||||
}
|
||||
|
||||
func readFinalText() string {
|
||||
off, ln := desc(fFinalText)
|
||||
if off == 0 && ln == 0 {
|
||||
return ""
|
||||
}
|
||||
if ln == 0 {
|
||||
return ""
|
||||
}
|
||||
base := binary.LittleEndian.Uint32(shm[offArenaBase:])
|
||||
return string(shm[base+off : base+off+ln])
|
||||
}
|
||||
|
||||
func writeFinalText(s string) error {
|
||||
used := binary.LittleEndian.Uint32(shm[offArenaUsed:])
|
||||
if used == 0 {
|
||||
used = 1
|
||||
}
|
||||
cap_ := binary.LittleEndian.Uint32(shm[offArenaCap:])
|
||||
if used+uint32(len(s)) > cap_ {
|
||||
return fmt.Errorf("arena 空间不足")
|
||||
}
|
||||
base := binary.LittleEndian.Uint32(shm[offArenaBase:])
|
||||
copy(shm[base+used:], []byte(s))
|
||||
binary.LittleEndian.PutUint32(shm[offArenaUsed:], used+uint32(len(s)))
|
||||
setDesc(fFinalText, used, uint32(len(s)))
|
||||
// 世代号自增
|
||||
v := binary.LittleEndian.Uint64(shm[offSeq:])
|
||||
binary.LittleEndian.PutUint64(shm[offSeq:], v+1)
|
||||
return nil
|
||||
}
|
||||
|
||||
func main() {
|
||||
tag = os.Getenv("PLUGIN_TAG")
|
||||
if tag == "" {
|
||||
tag = "?"
|
||||
}
|
||||
|
||||
in := bufio.NewScanner(bufio.NewReader(os.Stdin))
|
||||
in.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||
|
||||
for in.Scan() {
|
||||
line := make([]byte, len(in.Bytes()))
|
||||
copy(line, in.Bytes())
|
||||
|
||||
var probe struct {
|
||||
ID uint64 `json:"id"`
|
||||
Method string `json:"method"`
|
||||
}
|
||||
if json.Unmarshal(line, &probe) != nil {
|
||||
continue
|
||||
}
|
||||
if probe.Method == "" {
|
||||
var resp struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result json.RawMessage `json:"result"`
|
||||
}
|
||||
json.Unmarshal(line, &resp)
|
||||
pendMu.Lock()
|
||||
ch, ok := pending[resp.ID]
|
||||
delete(pending, resp.ID)
|
||||
pendMu.Unlock()
|
||||
if ok {
|
||||
ch <- resp.Result
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
var req request
|
||||
json.Unmarshal(line, &req)
|
||||
|
||||
switch req.Method {
|
||||
case "handshake":
|
||||
var hp struct {
|
||||
ShmSize int `json:"shm_size"`
|
||||
}
|
||||
json.Unmarshal(req.Params, &hp)
|
||||
if hp.ShmSize > 0 {
|
||||
m, err := syscall.Mmap(3, 0, hp.ShmSize,
|
||||
syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_SHARED)
|
||||
if err != nil {
|
||||
send(response{ID: req.ID, Error: fmt.Sprintf("mmap: %v", err)})
|
||||
continue
|
||||
}
|
||||
shm = m
|
||||
}
|
||||
send(response{ID: req.ID, Result: map[string]interface{}{
|
||||
"protocol": 1, "sdk_version": "test",
|
||||
"plugin_name": "append-" + tag, "pid": os.Getpid(),
|
||||
}})
|
||||
|
||||
case "plugin.init":
|
||||
send(response{ID: req.ID})
|
||||
|
||||
case "plugin.start":
|
||||
go func(id uint64) {
|
||||
callKernel("stage.register", map[string]interface{}{
|
||||
"stage": "after_toolcall",
|
||||
"scope": "global",
|
||||
})
|
||||
send(response{ID: id})
|
||||
}(req.ID)
|
||||
|
||||
case "plugin.stop":
|
||||
send(response{ID: req.ID})
|
||||
out.Flush()
|
||||
os.Exit(0)
|
||||
|
||||
case "stage.invoke":
|
||||
go func(id uint64) {
|
||||
if shm == nil {
|
||||
send(response{ID: id, Error: "共享段未挂载"})
|
||||
return
|
||||
}
|
||||
// 拿锁 → 读 → 追加自己的标记 → 写回 → 放锁
|
||||
callKernel("stage.lock", nil)
|
||||
cur := readFinalText()
|
||||
err := writeFinalText(cur + tag)
|
||||
callKernel("stage.unlock", nil)
|
||||
if err != nil {
|
||||
send(response{ID: id, Error: err.Error()})
|
||||
return
|
||||
}
|
||||
send(response{ID: id, Result: map[string]interface{}{"dirty_fields": 1}})
|
||||
}(req.ID)
|
||||
|
||||
default:
|
||||
if req.ID != 0 {
|
||||
send(response{ID: req.ID})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
178
internal/plugin/proc/testdata/readonlyplugin.go
vendored
Normal file
178
internal/plugin/proc/testdata/readonlyplugin.go
vendored
Normal file
@ -0,0 +1,178 @@
|
||||
//go:build ignore
|
||||
|
||||
// readonlyplugin 是只读 stage 插件（模拟 weather 的 AfterToolcall）：
|
||||
// 读取共享段但不写回任何字段。
|
||||
//
|
||||
// **这是 lost update 修复的关键验证对象**：C ABI 副本模型下，
|
||||
// 它会把自己收到的旧快照无条件回传，覆盖 sanitizer 的清洗结果
|
||||
// （§8.6 实测现网 1.6~4.3% 被覆盖）。共享内存模型下它零写入。
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
type request struct {
|
||||
ID uint64 `json:"id,omitempty"`
|
||||
Method string `json:"method"`
|
||||
Params json.RawMessage `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
type response struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result interface{} `json:"result,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
var (
|
||||
out = bufio.NewWriter(os.Stdout)
|
||||
writeMu sync.Mutex
|
||||
|
||||
nextID uint64
|
||||
pendMu sync.Mutex
|
||||
pending = map[uint64]chan json.RawMessage{}
|
||||
|
||||
shm []byte
|
||||
)
|
||||
|
||||
func send(v interface{}) {
|
||||
b, _ := json.Marshal(v)
|
||||
writeMu.Lock()
|
||||
out.Write(b)
|
||||
out.WriteByte('\n')
|
||||
out.Flush()
|
||||
writeMu.Unlock()
|
||||
}
|
||||
|
||||
func callKernel(method string, params interface{}) json.RawMessage {
|
||||
pendMu.Lock()
|
||||
nextID++
|
||||
id := nextID
|
||||
ch := make(chan json.RawMessage, 1)
|
||||
pending[id] = ch
|
||||
pendMu.Unlock()
|
||||
|
||||
var raw json.RawMessage
|
||||
if params != nil {
|
||||
b, _ := json.Marshal(params)
|
||||
raw = b
|
||||
}
|
||||
send(request{ID: id, Method: method, Params: raw})
|
||||
return <-ch
|
||||
}
|
||||
|
||||
const (
|
||||
offArenaBase = 8
|
||||
offCtxBase = 20
|
||||
sliceSize = 8
|
||||
fToolResults = 10
|
||||
)
|
||||
|
||||
func readToolResults() []byte {
|
||||
base := binary.LittleEndian.Uint32(shm[offArenaBase:])
|
||||
cb := binary.LittleEndian.Uint32(shm[offCtxBase:])
|
||||
o := cb + uint32(fToolResults*sliceSize)
|
||||
off := binary.LittleEndian.Uint32(shm[o:])
|
||||
ln := binary.LittleEndian.Uint32(shm[o+4:])
|
||||
if off == 0 && ln == 0 {
|
||||
return nil
|
||||
}
|
||||
return shm[base+off : base+off+ln]
|
||||
}
|
||||
|
||||
func main() {
|
||||
in := bufio.NewScanner(bufio.NewReader(os.Stdin))
|
||||
in.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||
|
||||
for in.Scan() {
|
||||
line := make([]byte, len(in.Bytes()))
|
||||
copy(line, in.Bytes())
|
||||
|
||||
var probe struct {
|
||||
ID uint64 `json:"id"`
|
||||
Method string `json:"method"`
|
||||
}
|
||||
if json.Unmarshal(line, &probe) != nil {
|
||||
continue
|
||||
}
|
||||
if probe.Method == "" {
|
||||
var resp struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result json.RawMessage `json:"result"`
|
||||
}
|
||||
json.Unmarshal(line, &resp)
|
||||
pendMu.Lock()
|
||||
ch, ok := pending[resp.ID]
|
||||
delete(pending, resp.ID)
|
||||
pendMu.Unlock()
|
||||
if ok {
|
||||
ch <- resp.Result
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
var req request
|
||||
json.Unmarshal(line, &req)
|
||||
|
||||
switch req.Method {
|
||||
case "handshake":
|
||||
var hp struct {
|
||||
ShmSize int `json:"shm_size"`
|
||||
}
|
||||
json.Unmarshal(req.Params, &hp)
|
||||
if hp.ShmSize > 0 {
|
||||
m, err := syscall.Mmap(3, 0, hp.ShmSize,
|
||||
syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_SHARED)
|
||||
if err != nil {
|
||||
send(response{ID: req.ID, Error: fmt.Sprintf("mmap: %v", err)})
|
||||
continue
|
||||
}
|
||||
shm = m
|
||||
}
|
||||
send(response{ID: req.ID, Result: map[string]interface{}{
|
||||
"protocol": 1, "sdk_version": "test", "plugin_name": "readonly", "pid": os.Getpid(),
|
||||
}})
|
||||
|
||||
case "plugin.init":
|
||||
send(response{ID: req.ID})
|
||||
|
||||
case "plugin.start":
|
||||
go func(id uint64) {
|
||||
callKernel("stage.register", map[string]interface{}{
|
||||
"stage": "after_toolcall",
|
||||
"scope": "global",
|
||||
})
|
||||
send(response{ID: id})
|
||||
}(req.ID)
|
||||
|
||||
case "plugin.stop":
|
||||
send(response{ID: req.ID})
|
||||
out.Flush()
|
||||
os.Exit(0)
|
||||
|
||||
case "stage.invoke":
|
||||
go func(id uint64) {
|
||||
if shm == nil {
|
||||
send(response{ID: id, Error: "共享段未挂载"})
|
||||
return
|
||||
}
|
||||
callKernel("stage.lock", nil)
|
||||
// 只读：读了但一个字节都不写回
|
||||
_ = readToolResults()
|
||||
callKernel("stage.unlock", nil)
|
||||
send(response{ID: id, Result: map[string]interface{}{"dirty_fields": 0}})
|
||||
}(req.ID)
|
||||
|
||||
default:
|
||||
if req.ID != 0 {
|
||||
send(response{ID: req.ID})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
311
internal/plugin/proc/testdata/stageplugin.go
vendored
Normal file
311
internal/plugin/proc/testdata/stageplugin.go
vendored
Normal file
@ -0,0 +1,311 @@
|
||||
//go:build ignore
|
||||
|
||||
// stageplugin 是完整形态的测试插件：注册工具/阶段/输出通道，
|
||||
// stage 处理经共享内存读改写（模拟 sanitizer 的清洗行为）。
|
||||
//
|
||||
// 它手写 RPC 与共享段访问，不依赖公开 SDK——因为 SDK 侧的 proc 支持
|
||||
// 属于 Part 3（plugindev 工具链）的内容。这里只验证内核侧机制。
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
type request struct {
|
||||
ID uint64 `json:"id,omitempty"`
|
||||
Method string `json:"method"`
|
||||
Params json.RawMessage `json:"params,omitempty"`
|
||||
}
|
||||
|
||||
type response struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result interface{} `json:"result,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
var (
|
||||
out = bufio.NewWriter(os.Stdout)
|
||||
writeMu sync.Mutex
|
||||
|
||||
nextID uint64
|
||||
pendMu sync.Mutex
|
||||
pending = map[uint64]chan json.RawMessage{}
|
||||
|
||||
shm []byte
|
||||
shmSize int
|
||||
)
|
||||
|
||||
func send(v interface{}) {
|
||||
b, _ := json.Marshal(v)
|
||||
writeMu.Lock()
|
||||
out.Write(b)
|
||||
out.WriteByte('\n')
|
||||
out.Flush()
|
||||
writeMu.Unlock()
|
||||
}
|
||||
|
||||
func callKernel(method string, params interface{}) json.RawMessage {
|
||||
pendMu.Lock()
|
||||
nextID++
|
||||
id := nextID
|
||||
ch := make(chan json.RawMessage, 1)
|
||||
pending[id] = ch
|
||||
pendMu.Unlock()
|
||||
|
||||
var raw json.RawMessage
|
||||
if params != nil {
|
||||
b, _ := json.Marshal(params)
|
||||
raw = b
|
||||
}
|
||||
send(request{ID: id, Method: method, Params: raw})
|
||||
return <-ch
|
||||
}
|
||||
|
||||
// ---- 共享段访问（与内核 proc 包的布局一致）----
|
||||
|
||||
const (
|
||||
headerSize = 64
|
||||
// 与 proc 包保持一致：18 个字段 × 8 字节 + 8 字节标志位
|
||||
stageFieldCount = 18
|
||||
sliceSize = 8
|
||||
flagCount = 8
|
||||
ctxSize = stageFieldCount*sliceSize + flagCount
|
||||
|
||||
offArenaBase = 8
|
||||
offArenaCap = 12
|
||||
offArenaUsed = 16
|
||||
offCtxBase = 20
|
||||
offSeq = 24
|
||||
|
||||
// 字段索引（与 proc/shmcodec.go 的 stageField 枚举顺序一致）
|
||||
fToolResults = 10
|
||||
)
|
||||
|
||||
func arenaBase() uint32 { return binary.LittleEndian.Uint32(shm[offArenaBase:]) }
|
||||
func arenaCap() uint32 { return binary.LittleEndian.Uint32(shm[offArenaCap:]) }
|
||||
func ctxBase() uint32 { return binary.LittleEndian.Uint32(shm[offCtxBase:]) }
|
||||
|
||||
func descOffset(field int) uint32 { return ctxBase() + uint32(field*sliceSize) }
|
||||
|
||||
func getDesc(field int) (off, ln uint32) {
|
||||
o := descOffset(field)
|
||||
return binary.LittleEndian.Uint32(shm[o:]), binary.LittleEndian.Uint32(shm[o+4:])
|
||||
}
|
||||
|
||||
func setDesc(field int, off, ln uint32) {
|
||||
o := descOffset(field)
|
||||
binary.LittleEndian.PutUint32(shm[o:], off)
|
||||
binary.LittleEndian.PutUint32(shm[o+4:], ln)
|
||||
}
|
||||
|
||||
func readField(field int) []byte {
|
||||
off, ln := getDesc(field)
|
||||
if off == 0 && ln == 0 {
|
||||
return nil
|
||||
}
|
||||
if ln == 0 {
|
||||
return []byte{}
|
||||
}
|
||||
base := arenaBase()
|
||||
return shm[base+off : base+off+ln]
|
||||
}
|
||||
|
||||
func writeField(field int, data []byte) error {
|
||||
used := binary.LittleEndian.Uint32(shm[offArenaUsed:])
|
||||
if used == 0 {
|
||||
used = 1
|
||||
}
|
||||
end := used + uint32(len(data))
|
||||
if end > arenaCap() {
|
||||
return fmt.Errorf("arena 空间不足")
|
||||
}
|
||||
base := arenaBase()
|
||||
copy(shm[base+used:], data)
|
||||
binary.LittleEndian.PutUint32(shm[offArenaUsed:], end)
|
||||
setDesc(field, used, uint32(len(data)))
|
||||
return nil
|
||||
}
|
||||
|
||||
func bumpSeq() {
|
||||
v := binary.LittleEndian.Uint64(shm[offSeq:])
|
||||
binary.LittleEndian.PutUint64(shm[offSeq:], v+1)
|
||||
}
|
||||
|
||||
type toolResult struct {
|
||||
CallID string `json:"call_id"`
|
||||
Name string `json:"name"`
|
||||
Plugin string `json:"plugin,omitempty"`
|
||||
Success bool `json:"success"`
|
||||
Result interface{} `json:"result"`
|
||||
}
|
||||
|
||||
// handleStage 模拟 sanitizer：拿锁 → 读 ToolResults → 剥 ANSI → 只写脏字段 → 放锁
|
||||
func handleStage() (int, error) {
|
||||
callKernel("stage.lock", nil)
|
||||
defer callKernel("stage.unlock", nil)
|
||||
|
||||
raw := readField(fToolResults)
|
||||
if len(raw) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
var results []toolResult
|
||||
if err := json.Unmarshal(raw, &results); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
before := string(raw)
|
||||
for i := range results {
|
||||
if s, ok := results[i].Result.(string); ok {
|
||||
results[i].Result = strings.NewReplacer("\x1b[31m", "", "\x1b[0m", "").Replace(s)
|
||||
}
|
||||
}
|
||||
after, _ := json.Marshal(results)
|
||||
|
||||
// 只有真变了才写回 —— 这是消除 lost update 的核心
|
||||
if string(after) == before {
|
||||
return 0, nil
|
||||
}
|
||||
if err := writeField(fToolResults, after); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
bumpSeq()
|
||||
return 1, nil
|
||||
}
|
||||
|
||||
func main() {
|
||||
in := bufio.NewScanner(bufio.NewReader(os.Stdin))
|
||||
in.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||
|
||||
for in.Scan() {
|
||||
line := make([]byte, len(in.Bytes()))
|
||||
copy(line, in.Bytes())
|
||||
|
||||
var probe struct {
|
||||
ID uint64 `json:"id"`
|
||||
Method string `json:"method"`
|
||||
}
|
||||
if json.Unmarshal(line, &probe) != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
// 内核对我们反向调用的应答
|
||||
if probe.Method == "" {
|
||||
var resp struct {
|
||||
ID uint64 `json:"id"`
|
||||
Result json.RawMessage `json:"result"`
|
||||
}
|
||||
json.Unmarshal(line, &resp)
|
||||
pendMu.Lock()
|
||||
ch, ok := pending[resp.ID]
|
||||
delete(pending, resp.ID)
|
||||
pendMu.Unlock()
|
||||
if ok {
|
||||
ch <- resp.Result
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
var req request
|
||||
json.Unmarshal(line, &req)
|
||||
|
||||
switch req.Method {
|
||||
case "handshake":
|
||||
var hp struct {
|
||||
ShmSize int `json:"shm_size"`
|
||||
}
|
||||
json.Unmarshal(req.Params, &hp)
|
||||
shmSize = hp.ShmSize
|
||||
if shmSize > 0 {
|
||||
// fd 3 = 内核传入的共享段 memfd
|
||||
m, err := syscall.Mmap(3, 0, shmSize,
|
||||
syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_SHARED)
|
||||
if err != nil {
|
||||
send(response{ID: req.ID, Error: fmt.Sprintf("mmap 共享段失败: %v", err)})
|
||||
continue
|
||||
}
|
||||
shm = m
|
||||
}
|
||||
send(response{ID: req.ID, Result: map[string]interface{}{
|
||||
"protocol": 1, "sdk_version": "test", "plugin_name": "stage", "pid": os.Getpid(),
|
||||
}})
|
||||
|
||||
case "plugin.init":
|
||||
send(response{ID: req.ID})
|
||||
|
||||
case "plugin.start":
|
||||
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": "转大写"},
|
||||
})
|
||||
callKernel("stage.register", map[string]interface{}{
|
||||
"stage": "after_toolcall",
|
||||
"scope": "global",
|
||||
})
|
||||
callKernel("output.register", map[string]interface{}{
|
||||
"name": "demo_ch",
|
||||
"caps": 1,
|
||||
"desc": "测试通道",
|
||||
})
|
||||
send(response{ID: id})
|
||||
}(req.ID)
|
||||
|
||||
case "plugin.stop":
|
||||
send(response{ID: req.ID})
|
||||
out.Flush()
|
||||
os.Exit(0)
|
||||
|
||||
case "tool.invoke":
|
||||
var p struct {
|
||||
Name string `json:"name"`
|
||||
Args map[string]interface{} `json:"args"`
|
||||
}
|
||||
json.Unmarshal(req.Params, &p)
|
||||
text, _ := p.Args["text"].(string)
|
||||
send(response{ID: req.ID, Result: map[string]interface{}{
|
||||
"result": strings.ToUpper(text),
|
||||
}})
|
||||
|
||||
case "stage.invoke":
|
||||
go func(id uint64) {
|
||||
if shm == nil {
|
||||
send(response{ID: id, Error: "共享段未挂载"})
|
||||
return
|
||||
}
|
||||
n, err := handleStage()
|
||||
if err != nil {
|
||||
send(response{ID: id, Error: err.Error()})
|
||||
return
|
||||
}
|
||||
send(response{ID: id, Result: map[string]interface{}{"dirty_fields": n}})
|
||||
}(req.ID)
|
||||
|
||||
case "output.invoke":
|
||||
var p struct {
|
||||
Channel string `json:"channel"`
|
||||
Args map[string]interface{} `json:"args"`
|
||||
}
|
||||
json.Unmarshal(req.Params, &p)
|
||||
payload, _ := p.Args["payload"].(string)
|
||||
if payload == "fail" {
|
||||
// 模拟现网 qq 插件的真实失败：meta 缺 user_id
|
||||
send(response{ID: req.ID, Error: "meta 中缺少 user_id 字段"})
|
||||
continue
|
||||
}
|
||||
send(response{ID: req.ID, Result: map[string]interface{}{"status": "sent"}})
|
||||
|
||||
default:
|
||||
if req.ID != 0 {
|
||||
send(response{ID: req.ID})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user