diff --git a/internal/plugin/proc/corehandler.go b/internal/plugin/proc/corehandler.go index 87597ed..20a25f3 100644 --- a/internal/plugin/proc/corehandler.go +++ b/internal/plugin/proc/corehandler.go @@ -34,7 +34,7 @@ type coreHandler struct { // invokeTool/invokeCleaner/invokeStageFn/invokeOutput 反向调用插件(内核 → 插件)。 // 由 Plugin 注入,注册回调时用它们构造 handler。 invokeTool func(name string, args map[string]interface{}) (interface{}, error) - invokeCleaner func(scope, name, text string) (string, error) + invokeCleaner func(scope, name string, textRef SharedRef) (SharedRef, error) invokeStageFn func(ctx context.Context, stage string, seq uint64) error invokeOutput func(channel string, args map[string]interface{}) (interface{}, error) @@ -600,11 +600,16 @@ func (h *coreHandler) cleanerProxy(scope, name string, enabled bool) (func(strin return nil, fmt.Errorf("%s %s 声明 Cleaner,但清洗回调通道未就绪", scope, name) } return func(text string) string { - cleaned, err := h.invokeCleaner(scope, name, text) + ref, err := h.host.unified.arenaWrite([]byte(text)) if err != nil { return text } - return cleaned + resultRef, err := h.invokeCleaner(scope, name, ref) + if err != nil { + return text + } + result := h.host.unified.arenaRead(resultRef) + return string(result) }, nil } diff --git a/internal/plugin/proc/plugin.go b/internal/plugin/proc/plugin.go index aa3ff40..372e5db 100644 --- a/internal/plugin/proc/plugin.go +++ b/internal/plugin/proc/plugin.go @@ -222,23 +222,23 @@ func (p *Plugin) invokeTool(name string, args map[string]interface{}) (interface } // invokeCleaner 在插件进程内执行工具或通道注册时提供的 Cleaner 函数。 -func (p *Plugin) invokeCleaner(scope, name, text string) (string, error) { +func (p *Plugin) invokeCleaner(scope, name string, textRef SharedRef) (SharedRef, error) { if p.proc == nil { - return "", ErrProcessExited + return SharedRef{}, ErrProcessExited } raw, err := p.proc.Call(MethodCleanerInvoke, CleanerInvokeParams{ - Scope: scope, - Name: name, - Text: text, + Scope: scope, + Name: name, + TextRef: textRef, }) if err != nil { - return "", err + return SharedRef{}, 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 SharedRef{}, fmt.Errorf("proc: %s %s Cleaner %s 应答解析失败: %w", p.name, scope, name, err) } - return res.Text, nil + return res.TextRef, nil } func (p *Plugin) invokeStage(ctx context.Context, stage string, seq uint64) error { diff --git a/internal/plugin/proc/protocol.go b/internal/plugin/proc/protocol.go index 8b51be6..e79941d 100644 --- a/internal/plugin/proc/protocol.go +++ b/internal/plugin/proc/protocol.go @@ -213,14 +213,15 @@ type ToolInvokeResult struct { // CleanerInvokeParams / CleanerInvokeResult:跨进程计算层清洗。 // Scope 取 tool / input / output,Name 是工具名或通道名。 +// TextRef 指向共享内存 arena 中的实际文本数据。 type CleanerInvokeParams struct { - Scope string `json:"scope"` - Name string `json:"name"` - Text string `json:"text"` + Scope string `json:"scope"` + Name string `json:"name"` + TextRef SharedRef `json:"text_ref"` } type CleanerInvokeResult struct { - Text string `json:"text"` + TextRef SharedRef `json:"text_ref"` } const ( diff --git a/internal/plugin/proc/testdata/stageplugin.go b/internal/plugin/proc/testdata/stageplugin.go index b496049..145d6cd 100644 --- a/internal/plugin/proc/testdata/stageplugin.go +++ b/internal/plugin/proc/testdata/stageplugin.go @@ -40,6 +40,9 @@ var ( shm []byte shmSize int + region []byte // 完整统一区域 mmap + arenaOff uint32 + arenaUsed uint32 ) func send(v interface{}) { @@ -233,6 +236,8 @@ func main() { // 统一区域:前 64B 是 SuperBlock,StageContext 段在其后 ctxOff := binary.LittleEndian.Uint32(m[20:]) ctxSize := binary.LittleEndian.Uint32(m[24:]) + region = m + arenaOff = binary.LittleEndian.Uint32(m[36:]) shm = m[ctxOff : ctxOff+ctxSize] } send(response{ID: req.ID, Result: map[string]interface{}{ @@ -282,9 +287,12 @@ func main() { case "cleaner.invoke": var p struct { - Scope string `json:"scope"` - Name string `json:"name"` - Text string `json:"text"` + Scope string `json:"scope"` + Name string `json:"name"` + TextRef struct { + Offset uint32 `json:"offset"` + Length uint32 `json:"length"` + } `json:"text_ref"` } json.Unmarshal(req.Params, &p) valid := (p.Scope == "tool" && p.Name == "demo_upper") || @@ -294,8 +302,22 @@ func main() { send(response{ID: req.ID, Error: "未注册 Cleaner"}) continue } + // 从 arena 读文本 + input := string(region[p.TextRef.Offset : p.TextRef.Offset+p.TextRef.Length]) + output := p.Scope + "-cleaned:" + input + // 写回 arena + off := arenaUsed + if off == 0 { + off = 1 + } + end := off + uint32(len(output)) + copy(region[arenaOff+end-uint32(len(output)):arenaOff+end], output) + arenaUsed = end send(response{ID: req.ID, Result: map[string]interface{}{ - "text": p.Scope + "-cleaned:" + p.Text, + "text_ref": map[string]interface{}{ + "offset": arenaOff + off, + "length": uint32(len(output)), + }, }}) case "stage.invoke":