重构: 插件自注册 + .so 动态加载 + 中断打断机制

- 所有内置插件 init() 自注册 (plugin.RegisterFactory), 移除 main.go 硬编码 - 新增 .so 动态加载器 (internal/plugin/dynamic.go), 插件可编译为 plugin.so - 新增 plugin.json 元数据 (internal/plugin/manifest.go) - 新增 interceptLoop 独立 goroutine: (a) cancelLLM() 取消进行中的 HTTP 请求 (b) interceptCh → drainInterrupt() 注入 [打断消息] 到 LLM 上下文 (c) InjectInput 空闲时触发新处理循环 - 新增 internal/plugins/all.go 空白导入触发所有内置插件 init() - internal/sdk/ 作为 PluginSDK 正式 Go API - internal/api/ → internal/plugins/webui/ 迁移 - 删除旧 cmd/cli/, 使用 cmd/waiter/ 替代 - 更新 PLAN.md / ARCHITECTURE.md / README.md 文档
2026-09-21 17:38:10 +00:00 · 2026-07-03 16:53:34 +08:00
parent 4442c9cea4
commit 2d314b3e9c
37 changed files with 3376 additions and 1576 deletions
--- a/internal/agent/core/agent.go
+++ b/internal/agent/core/agent.go
@ -18,7 +18,7 @@ import (
 	"gitcode.com/JianFeeeee/HomeAgent/internal/memory/social"
 	"gitcode.com/JianFeeeee/HomeAgent/internal/memory/text"
 	"gitcode.com/JianFeeeee/HomeAgent/internal/plugin"
-	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/plugin/sdk"
+	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
 	"gitcode.com/JianFeeeee/HomeAgent/internal/skill"
 	"gitcode.com/JianFeeeee/HomeAgent/internal/tracker"
 	"gitcode.com/JianFeeeee/HomeAgent/pkg/types"
@ -82,6 +82,13 @@ type Agent struct {
 	childMu      sync.Mutex
 	childNextID  int64
 	childResults map[string]string
+
+	// 高优先级打断通道：interceptLoop 注入，process() 在工具循环轮次间非阻塞读取
+	interceptCh chan string
+
+	// 进行中的 LLM 请求取消函数，interceptLoop 可调用以在请求中打断
+	cancelLLM context.CancelFunc
+	llmMu     sync.Mutex
 }

 type AgentConfig struct {
@ -148,11 +155,13 @@ func New(cfg AgentConfig) *Agent {
 		eventBus:         cfg.EventBus,
 		selfInputCh:      make(chan string, 64),
 		childResults:     make(map[string]string),
+		interceptCh:      make(chan string, 64),
 	}
 }

 func (a *Agent) Start() {
 	go a.eventLoop()
+	go a.interceptLoop()
 	go a.distillLoop()
 	log.Printf("[agent] %s started, waiting for IO interrupts", a.id)
 }
@ -191,6 +200,45 @@ func (a *Agent) eventLoop() {
 	}
 }

+// interceptLoop 独立 goroutine 监控中断通道。
+// 两种路径投递：
+//   a) 通过 cancelLLM + interceptCh 直接打断进行中的 LLM 请求
+//   b) 通过 a.io.InjectInput() → InputChan → eventLoop（代理空闲时触发新处理循环）
+func (a *Agent) interceptLoop() {
+	for {
+		select {
+		case evt := <-a.io.InputInterruptChan():
+			text, _ := evt.Payload["content"].(string)
+			if text == "" {
+				continue
+			}
+			log.Printf("[agent] interrupt from %s: %s", evt.Source, truncateStr(text, 80))
+
+			// (a) 直接取消进行中的 LLM 请求
+			a.llmMu.Lock()
+			if a.cancelLLM != nil {
+				a.cancelLLM()
+				log.Printf("[agent] LLM request cancelled by interrupt")
+			}
+			a.llmMu.Unlock()
+
+			// 注入拦截通道 — process() 在工具循环中非阻塞读取
+			select {
+			case a.interceptCh <- text:
+			default:
+			}
+
+			// (b) 投递为新输入 — 代理空闲时 eventLoop 会消费
+			a.io.InjectInput("interrupt", "text", map[string]interface{}{
+				"content": fmt.Sprintf("[interrupt] %s: %s", evt.Source, text),
+			})
+
+		case <-a.ctx.Done():
+			return
+		}
+	}
+}
+
 // handleSelfInput 处理自循环输入（内部任务，不经过 IO 层）
 func (a *Agent) handleSelfInput(task string) {
 	a.processTextInput(&agentIO.InputEvent{
@ -361,6 +409,15 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri
 	}

 	for turn := 0; turn < a.maxTurns; turn++ {
+		// === 高优先级打断：每次 LLM 调用前检查拦截通道 ===
+		if text := a.drainInterrupt(); text != "" {
+			msgs = append(msgs, agentAPI.Message{
+				Role:    "system",
+				Content: fmt.Sprintf("[打断消息] 用户发来一条紧急消息，请优先处理:\n%s", text),
+			})
+			log.Printf("[agent] interrupt injected before LLM call (turn %d)", turn)
+		}
+
 		req := &agentAPI.CompletionRequest{
 			Messages:   msgs,
 			MaxTokens:  4096,
@ -371,7 +428,19 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri
 			},
 		}

-		resp, err := a.provider.Chat(a.ctx, req)
+		// 可取消的 LLM 调用：interceptLoop 通过 cancelLLM 打断进行中的请求
+		reqCtx, reqCancel := context.WithCancel(a.ctx)
+		a.llmMu.Lock()
+		a.cancelLLM = reqCancel
+		a.llmMu.Unlock()
+
+		resp, err := a.provider.Chat(reqCtx, req)
+
+		a.llmMu.Lock()
+		a.cancelLLM = nil
+		a.llmMu.Unlock()
+		reqCancel()
+
 		if err != nil {
 			return "", toolsUsed, fmt.Errorf("provider: %w", err)
 		}
@ -1969,9 +2038,19 @@ func getFloat(m map[string]interface{}, key string) float64 {
 }

 func truncateStr(s string, max int) string {
-	runes := []rune(s)
-	if len(runes) > max {
-		return string(runes[:max]) + "..."
+	if len(s) <= max {
+		return s
+	}
+	return s[:max] + "..."
+}
+
+// drainInterrupt 非阻塞读取 interceptCh 中的一条打断消息。
+// 若有多条，只取最先到达的一条（丢弃后续）。
+func (a *Agent) drainInterrupt() string {
+	select {
+	case text := <-a.interceptCh:
+		return text
+	default:
+		return ""
 	}
-	return s
 }
--- a/internal/agent/core/stages.go
+++ b/internal/agent/core/stages.go
@ -2,73 +2,89 @@ package core

 import (
 	"fmt"
+	"sync"

-	"gitcode.com/JianFeeeee/HomeAgent/internal/plugin"
-	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/plugin/sdk"
+	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
 )

 type StageHost struct {
-	plugins  []*sdk.PluginAPI
+	mu       sync.RWMutex
 	toolDefs []sdk.ToolDef
 	tools    map[string]sdk.ToolHandler
+	stages   map[sdk.Stage][]sdk.StageHandler
 }

 func NewStageHost() *StageHost {
 	return &StageHost{
-		tools: make(map[string]sdk.ToolHandler),
+		tools:  make(map[string]sdk.ToolHandler),
+		stages: make(map[sdk.Stage][]sdk.StageHandler),
 	}
 }

-func (h *StageHost) RegisterPlugin(api *sdk.PluginAPI) {
-	h.plugins = append(h.plugins, api)
-	for name, handler := range api.Tools() {
-		h.tools[name] = handler
-		h.toolDefs = append(h.toolDefs, sdk.ToolDef{Name: name})
+func (h *StageHost) RegisterTool(name string, def sdk.ToolDef, handler sdk.ToolHandler) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	if _, exists := h.tools[name]; exists {
+		return fmt.Errorf("tool %s already registered", name)
 	}
+	h.tools[name] = handler
+	h.toolDefs = append(h.toolDefs, def)
+	return nil
 }

-// SyncFromRegistry 从插件注册表同步 SDK 插件
-func (h *StageHost) SyncFromRegistry(reg *plugin.Registry) {
-	if reg == nil {
-		return
-	}
-	for _, td := range reg.GetAllSDKToolDefs() {
-		h.toolDefs = append(h.toolDefs, td)
-	}
+func (h *StageHost) RegisterStage(stage sdk.Stage, handler sdk.StageHandler) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.stages[stage] = append(h.stages[stage], handler)
 }

 func (h *StageHost) GetToolDefs() []sdk.ToolDef {
-	return h.toolDefs
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	defs := make([]sdk.ToolDef, len(h.toolDefs))
+	copy(defs, h.toolDefs)
+	return defs
 }

 func (h *StageHost) ExecuteTool(name string, args map[string]interface{}) (interface{}, error) {
-	if handler, ok := h.tools[name]; ok {
-		return handler(args)
+	h.mu.RLock()
+	handler, ok := h.tools[name]
+	h.mu.RUnlock()
+	if !ok {
+		return nil, fmt.Errorf("tool %s not found in any plugin", name)
 	}
-	return nil, fmt.Errorf("tool %s not found in any plugin", name)
+	return handler(args)
 }

+// RunStage 并行调用同阶段所有注册的处理函数。
+// 各 handler 共享 *StageContext，通过其内置 RWMutex 安全读写：
+//   - 只读操作先调用 ctx.RLock() / defer ctx.RUnlock()
+//   - 写操作（如设置 ctx.Response）先调用 ctx.Lock() / defer ctx.Unlock()
+// 如果任意 handler 设置了 Response，后续 handler 可通过 ctx.IsResponded() 判断后提前返回。
 func (h *StageHost) RunStage(stage sdk.Stage, ctx *sdk.StageContext) {
-	for _, p := range h.plugins {
-		for _, handler := range p.StageHandlers(stage) {
-			if err := handler(ctx); err != nil {
-				return
-			}
-			if ctx.Response != nil {
-				return
-			}
-		}
+	h.mu.RLock()
+	handlers := h.stages[stage]
+	h.mu.RUnlock()
+	if len(handlers) == 0 {
+		return
 	}
+	var wg sync.WaitGroup
+	for _, handler := range handlers {
+		wg.Add(1)
+		go func(fn sdk.StageHandler) {
+			defer wg.Done()
+			fn(ctx)
+		}(handler)
+	}
+	wg.Wait()
 }

 func (h *StageHost) RunStageAll(stage sdk.Stage, ctx *sdk.StageContext) {
-	for _, p := range h.plugins {
-		for _, handler := range p.StageHandlers(stage) {
-			handler(ctx)
-		}
-	}
+	h.RunStage(stage, ctx)
 }

-func (h *StageHost) PluginCount() int {
-	return len(h.plugins)
+func (h *StageHost) ToolCount() int {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	return len(h.tools)
 }
--- a/internal/agent/core/stages_test.go
+++ b/internal/agent/core/stages_test.go
@ -1,23 +1,21 @@
 package core

 import (
+	"sync"
+	"sync/atomic"
 	"testing"

-	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/plugin/sdk"
+	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
 )

-func TestStageHostRegisterPlugin(t *testing.T) {
+func TestStageHostRegisterTool(t *testing.T) {
 	host := NewStageHost()
-	api := sdk.NewPluginAPI("test", "1.0.0", nil, nil, nil)

-	api.RegisterTool("test_tool", func(args map[string]interface{}) (interface{}, error) {
+	err := host.RegisterTool("test_tool", sdk.ToolDef{Name: "test_tool"}, func(args map[string]interface{}) (interface{}, error) {
 		return "ok", nil
 	})
-
-	host.RegisterPlugin(api)
-
-	if host.PluginCount() != 1 {
-		t.Errorf("expected 1 plugin, got %d", host.PluginCount())
+	if err != nil {
+		t.Fatalf("register: %v", err)
 	}

 	defs := host.GetToolDefs()
@ -29,16 +27,22 @@ func TestStageHostRegisterPlugin(t *testing.T) {
 	}
 }

+func TestStageHostRegisterToolDuplicate(t *testing.T) {
+	host := NewStageHost()
+	host.RegisterTool("dup", sdk.ToolDef{Name: "dup"}, nil)
+	err := host.RegisterTool("dup", sdk.ToolDef{Name: "dup"}, nil)
+	if err == nil {
+		t.Error("expected error on duplicate tool")
+	}
+}
+
 func TestStageHostExecuteTool(t *testing.T) {
 	host := NewStageHost()
-	api := sdk.NewPluginAPI("test", "1.0.0", nil, nil, nil)

-	api.RegisterTool("hello", func(args map[string]interface{}) (interface{}, error) {
+	host.RegisterTool("hello", sdk.ToolDef{Name: "hello"}, func(args map[string]interface{}) (interface{}, error) {
 		return "world", nil
 	})

-	host.RegisterPlugin(api)
-
 	result, err := host.ExecuteTool("hello", nil)
 	if err != nil {
 		t.Fatalf("execute: %v", err)
@ -55,16 +59,13 @@ func TestStageHostExecuteTool(t *testing.T) {

 func TestStageHostRunStage(t *testing.T) {
 	host := NewStageHost()
-	api := sdk.NewPluginAPI("test", "1.0.0", nil, nil, nil)

 	var called bool
-	api.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error {
+	host.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error {
 		called = true
 		return nil
 	})

-	host.RegisterPlugin(api)
-
 	ctx := &sdk.StageContext{RawMessage: "hello"}
 	host.RunStage(sdk.StageOnInput, ctx)

@ -73,56 +74,78 @@ func TestStageHostRunStage(t *testing.T) {
 	}
 }

-func TestStageHostRunStageShortCircuit(t *testing.T) {
+func TestStageHostRunStageParallel(t *testing.T) {
 	host := NewStageHost()

-	api1 := sdk.NewPluginAPI("p1", "1.0.0", nil, nil, nil)
-	api1.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error {
-		resp := "short-circuited"
-		ctx.Response = &resp
+	// Two handlers that both try to set Response under Lock.
+	// Only the first to acquire Lock actually wins; the second sees IsResponded() and skips.
+	host.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error {
+		ctx.Lock()
+		if ctx.Response == nil {
+			resp := "from-first"
+			ctx.Response = &resp
+		}
+		ctx.Unlock()
 		return nil
 	})
-
-	var api2called bool
-	api2 := sdk.NewPluginAPI("p2", "1.0.0", nil, nil, nil)
-	api2.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error {
-		api2called = true
+	host.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error {
+		ctx.Lock()
+		if ctx.Response == nil {
+			resp := "from-second"
+			ctx.Response = &resp
+		}
+		ctx.Unlock()
 		return nil
 	})

-	host.RegisterPlugin(api1)
-	host.RegisterPlugin(api2)
-
 	ctx := &sdk.StageContext{RawMessage: "hello"}
 	host.RunStage(sdk.StageOnInput, ctx)

-	if ctx.Response == nil || *ctx.Response != "short-circuited" {
-		t.Errorf("expected short-circuited, got %v", ctx.Response)
+	if ctx.Response == nil {
+		t.Fatal("expected a response to be set")
 	}
-	if api2called {
-		t.Error("api2 should not have been called after short circuit")
+	if *ctx.Response != "from-first" && *ctx.Response != "from-second" {
+		t.Errorf("expected either from-first or from-second, got %s", *ctx.Response)
+	}
+}
+
+func TestStageHostRunStageConcurrency(t *testing.T) {
+	host := NewStageHost()
+
+	var counter int32
+	n := 10
+	for i := 0; i < n; i++ {
+		host.RegisterStage(sdk.StageAfterOutput, func(ctx *sdk.StageContext) error {
+			atomic.AddInt32(&counter, 1)
+			return nil
+		})
+	}
+
+	host.RunStage(sdk.StageAfterOutput, &sdk.StageContext{})
+
+	if int(counter) != n {
+		t.Errorf("expected %d handlers called, got %d", n, counter)
 	}
 }

 func TestStageHostRunStageAll(t *testing.T) {
 	host := NewStageHost()

+	var mu sync.Mutex
 	count := 0
-	api1 := sdk.NewPluginAPI("p1", "1.0.0", nil, nil, nil)
-	api1.RegisterStage(sdk.StageAfterOutput, func(ctx *sdk.StageContext) error {
+	host.RegisterStage(sdk.StageAfterOutput, func(ctx *sdk.StageContext) error {
+		mu.Lock()
 		count++
+		mu.Unlock()
 		return nil
 	})
-
-	api2 := sdk.NewPluginAPI("p2", "1.0.0", nil, nil, nil)
-	api2.RegisterStage(sdk.StageAfterOutput, func(ctx *sdk.StageContext) error {
+	host.RegisterStage(sdk.StageAfterOutput, func(ctx *sdk.StageContext) error {
+		mu.Lock()
 		count++
+		mu.Unlock()
 		return nil
 	})

-	host.RegisterPlugin(api1)
-	host.RegisterPlugin(api2)
-
 	host.RunStageAll(sdk.StageAfterOutput, &sdk.StageContext{})

 	if count != 2 {
@ -130,42 +153,11 @@ func TestStageHostRunStageAll(t *testing.T) {
 	}
 }

-func TestStageHostMultiplePlugins(t *testing.T) {
-	host := NewStageHost()
-
-	p1 := sdk.NewPluginAPI("p1", "1.0.0", nil, nil, nil)
-	p1.RegisterTool("tool1", func(args map[string]interface{}) (interface{}, error) {
-		return "from_p1", nil
-	})
-
-	p2 := sdk.NewPluginAPI("p2", "1.0.0", nil, nil, nil)
-	p2.RegisterTool("tool2", func(args map[string]interface{}) (interface{}, error) {
-		return "from_p2", nil
-	})
-
-	host.RegisterPlugin(p1)
-	host.RegisterPlugin(p2)
-
-	if host.PluginCount() != 2 {
-		t.Errorf("expected 2 plugins, got %d", host.PluginCount())
-	}
-
-	r1, _ := host.ExecuteTool("tool1", nil)
-	if r1.(string) != "from_p1" {
-		t.Errorf("expected from_p1, got %v", r1)
-	}
-
-	r2, _ := host.ExecuteTool("tool2", nil)
-	if r2.(string) != "from_p2" {
-		t.Errorf("expected from_p2, got %v", r2)
-	}
-}
-
 func TestStageHostEmpty(t *testing.T) {
 	host := NewStageHost()

-	if host.PluginCount() != 0 {
-		t.Errorf("expected 0 plugins, got %d", host.PluginCount())
+	if host.ToolCount() != 0 {
+		t.Errorf("expected 0 tools, got %d", host.ToolCount())
 	}

 	defs := host.GetToolDefs()
@ -178,6 +170,30 @@ func TestStageHostEmpty(t *testing.T) {
 		t.Error("expected error on empty host")
 	}

-	// RunStage on empty host should not panic
 	host.RunStage(sdk.StageOnInput, &sdk.StageContext{})
 }
+
+func TestStageHostMultipleTools(t *testing.T) {
+	host := NewStageHost()
+
+	host.RegisterTool("tool1", sdk.ToolDef{Name: "tool1"}, func(args map[string]interface{}) (interface{}, error) {
+		return "from_p1", nil
+	})
+	host.RegisterTool("tool2", sdk.ToolDef{Name: "tool2"}, func(args map[string]interface{}) (interface{}, error) {
+		return "from_p2", nil
+	})
+
+	if host.ToolCount() != 2 {
+		t.Errorf("expected 2 tools, got %d", host.ToolCount())
+	}
+
+	r1, _ := host.ExecuteTool("tool1", nil)
+	if r1.(string) != "from_p1" {
+		t.Errorf("expected from_p1, got %v", r1)
+	}
+
+	r2, _ := host.ExecuteTool("tool2", nil)
+	if r2.(string) != "from_p2" {
+		t.Errorf("expected from_p2, got %v", r2)
+	}
+}