diff --git a/internal/agent/core/agent.go b/internal/agent/core/agent.go index abd31ec..59c5b23 100644 --- a/internal/agent/core/agent.go +++ b/internal/agent/core/agent.go @@ -325,7 +325,7 @@ func New(cfg AgentConfig) *Agent { } } - return &Agent{ + a := &Agent{ id: cfg.ID, startTime: time.Now(), provider: cfg.Provider, @@ -376,6 +376,14 @@ func New(cfg AgentConfig) *Agent { noMergeMarkers: make(map[string]int), lastInput: make(map[string]time.Time), } + + // 输入路由:inputch 是可分配资源,划给某个 agent 后输入**只**流向那个 agent + // (设计 §4.1「路由发生在进内核之前」)。io 层不认识 agent,所以在这里把路由器 + // 注入进去:插件注入输入时先问它,被别的 agent 接管就不再进本内核队列。 + if a.io != nil { + a.io.SetInputRouter(a.routeInputByOwner) + } + return a } // SetSkillIndexProvider 注入技能索引提供者(skillmgr 插件加载后由 main 接线)。 diff --git a/internal/agent/core/inputroute.go b/internal/agent/core/inputroute.go new file mode 100644 index 0000000..05a0506 --- /dev/null +++ b/internal/agent/core/inputroute.go @@ -0,0 +1,44 @@ +package core + +import ( + "log" + + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" +) + +// routeInputByOwner 实现**输入路由**:inputch 是最基本的输入路由单位, +// 划给某个 agent 之后,该通道的输入**只流向那个 agent**,本内核看不到它 +// (docs/zh/resident-subagent-design.md §4.1:「路由发生在进内核之前」)。 +// +// 为什么必须在进内核之前做:插件注入输入的收口是**父**的 IOManager +// (`cmd/homed` 里 pluginReg 拿到的就是它),而父的内核是该 io 唯一的消费者。 +// 如果不按归属路由,登记表里的 Owner 就只是个标签 —— 现场表现正是如此: +// 子挂着 `inputch=[timer]`,timer 的输入却打在父身上,子的轮次永远是 0。 +// +// 返回 true = 本次注入已被"持有该 inputch 的 agent"接管,本内核不再处理。 +// +// 已知边界:路由只在本 agent 的**直接**驻留子里找。若孙辈的 inputch 由子划拨, +// 而插件注入打在根 io 上,根解析不到那个 owner ⇒ 兜底给根处理(有日志)。 +// 这一层要等"孙辈 + 根可见的 agent 表"再收口,此处不静默丢输入。 +func (a *Agent) routeInputByOwner(evt *agentIO.InputEvent, isInterrupt bool) bool { + if evt == nil || evt.OutputChannel == "" || a.io == nil { + return false + } + entry, ok := a.io.LookupInputChannel(evt.OutputChannel) + if !ok || entry.Owner == "" || entry.Owner == string(a.id) { + return false // 未分配 / 归自己 ⇒ 本内核处理 + } + + a.residentMu.Lock() + rc := a.residents[entry.Owner] + a.residentMu.Unlock() + if rc == nil || rc.agent == nil || rc.agent.io == nil { + // 归属到一个不存在(或已销毁、登记表尚未归还)的 agent: + // **不吞输入** —— 由本内核兜底处理并留痕。吞掉一条输入比多处理一条更糟: + // 用户会看到"消息发出去了却没人理",而日志里什么都没有。 + log.Printf("[route] inputch %s 归属 %s 无对应 agent,输入由 %s 兜底", evt.OutputChannel, entry.Owner, a.id) + return false + } + rc.agent.io.DeliverRouted(evt, isInterrupt) + return true +} diff --git a/internal/agent/core/inputroute_test.go b/internal/agent/core/inputroute_test.go new file mode 100644 index 0000000..a2c378b --- /dev/null +++ b/internal/agent/core/inputroute_test.go @@ -0,0 +1,80 @@ +package core + +import ( + "path/filepath" + "testing" + + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" +) + +// 输入路由是**独占**的:inputch 划给子之后,该通道的输入只流向子,父不再收到。 +// +// 现场缺陷(用户线上联调实录):子挂着 inputch=[timer],timer 的输入却打在父身上 +// (日志 `[agent] interrupt from timer/timer`),子的轮次永远是 0 —— 因为 +// `Assign` 只把 Owner 写进登记表,注入路径根本没有按归属路由。 +func TestResident_InputchRoutingIsExclusive(t *testing.T) { + parent, _, dir := newRootForResidents(t) + reg := parent.io.ChannelRegistry() + if err := reg.Register(agentIO.InputChannel{Name: "sub/in", Plugin: "sub"}); err != nil { + t.Fatal(err) + } + spawnTestResident(t, parent, dir, "r-route", "sub/in") + child := parent.residents["r-route"].agent + + before := parent.DumpScheduler().Stats.Enqueued + // 插件往"已划给子"的 inputch 投输入 + parent.io.InjectTextTo("plugin-sub", "sub/in", "去查一下这个") + + waitFor(t, "子处理了划给它的输入", func() bool { + if child.DumpScheduler().Stats.Executed > 0 { + return true + } + return parent.residents["r-route"].info().TableSize > 0 + }) + if got := parent.DumpScheduler().Stats.Enqueued; got != before { + t.Fatalf("划给子的 inputch,父不应再入队(before=%d after=%d)", before, got) + } +} + +// 归属到一个不存在(或已销毁)的 agent 时**不吞输入**:父兜底处理。 +// 吞掉一条输入比多处理一条更糟 —— 用户会看到"消息发出去了却没人理",日志里什么都没有。 +func TestResident_InputchRoutingFallsBackWhenOwnerMissing(t *testing.T) { + parent, _, dir := newRootForResidents(t) + reg := parent.io.ChannelRegistry() + if err := reg.Register(agentIO.InputChannel{Name: "ghost/in", Plugin: "ghost"}); err != nil { + t.Fatal(err) + } + // 故意划给一个不存在的 agent id + if err := reg.Assign("ghost/in", "no-such-agent", 0); err != nil { + t.Fatal(err) + } + _ = dir + + before := parent.DumpScheduler().Stats.Enqueued + parent.io.InjectTextTo("plugin-ghost", "ghost/in", "兜底测试") + waitFor(t, "父兜底处理了无人认领的输入", func() bool { + return parent.DumpScheduler().Stats.Enqueued > before + }) +} + +// 未划拨的 inputch(Owner 为空)仍然由父处理 —— 路由不能把默认路径也改掉。 +func TestResident_UnassignedInputchStaysWithParent(t *testing.T) { + parent, _, dir := newRootForResidents(t) + reg := parent.io.ChannelRegistry() + if err := reg.Register(agentIO.InputChannel{Name: "own/in", Plugin: "own"}); err != nil { + t.Fatal(err) + } + // 造一个子在跑,确保路由逻辑是"有子存在"的情形 + spawnTestResident(t, parent, dir, "r-other", "own/in") + _ = filepath.Join(dir, "residents") + + // 把通道退还给父(未分配) + if err := reg.Assign("own/in", "", 0); err != nil { + t.Fatal(err) + } + before := parent.DumpScheduler().Stats.Enqueued + parent.io.InjectTextTo("plugin-own", "own/in", "还是我的") + waitFor(t, "未分配的 inputch 仍由父处理", func() bool { + return parent.DumpScheduler().Stats.Enqueued > before + }) +} diff --git a/internal/agent/io/channel.go b/internal/agent/io/channel.go index 9a75461..d48417b 100644 --- a/internal/agent/io/channel.go +++ b/internal/agent/io/channel.go @@ -118,6 +118,15 @@ type IOManager struct { // 回退只解决"看得见",不解决"能不能用"。 parent *IOManager + // inputRouter 决定一条输入是否被"别的 agent"接管(返回 true = 已接管)。 + // + // 为什么放在 io:inputch 是**最基本的输入路由单位**,而**路由发生在进内核之前** + // (docs/zh/resident-subagent-design.md §4.1)。插件注入输入的收口就在这里, + // 所以路由必须在这里生效 —— inputch 划给某个 agent 后,输入**只流向那个 agent**, + // 本内核根本看不到它。io 层不认识 agent,路由器由内核注入 + // (见 core.Agent.routeInputByOwner)。 + inputRouter InputRouter + // toolBlocks:插件工具注入多模态内容块,process.go 在下一条 tool message 时消费。 // 用 interface{}[] 避免 import api.ContentBlock 导致的循环依赖。 toolBlocksMu sync.Mutex @@ -134,6 +143,47 @@ func NewIOManager() *IOManager { } } +// InputRouter 是输入路由器的签名。 +// +// evt 待投递的输入事件(OutputChannel 即它的 inputch) +// isInterrupt 该输入是中断还是排队(两者都要按归属路由) +// 返回 true = 已被别的 agent 接管,本内核不再处理 +type InputRouter func(evt *InputEvent, isInterrupt bool) bool + +// SetInputRouter 注入输入路由器(nil = 不路由,行为与以前完全一致)。 +func (m *IOManager) SetInputRouter(r InputRouter) { + m.mu.Lock() + m.inputRouter = r + m.mu.Unlock() +} + +// deliverInput 是**本内核**接收一条外部输入的收口:先按 inputch 归属路由, +// 被别的 agent 接管就不进本内核队列(划给子的 inputch,父不再收到 —— 这是「划拨」 +// 的语义,不是"父也顺便看一眼")。 +func (m *IOManager) deliverInput(evt *InputEvent, isInterrupt bool) { + m.mu.RLock() + router := m.inputRouter + m.mu.RUnlock() + if router != nil && router(evt, isInterrupt) { + return + } + m.pushLocal(evt, isInterrupt) +} + +// DeliverRouted 把**已被路由**的事件放进本内核队列(不再二次路由)。 +// 由路由器实现调用:父把输入交给持有该 inputch 的子。 +func (m *IOManager) DeliverRouted(evt *InputEvent, isInterrupt bool) { + m.pushLocal(evt, isInterrupt) +} + +func (m *IOManager) pushLocal(evt *InputEvent, isInterrupt bool) { + if isInterrupt { + m.interruptCh <- evt + return + } + m.inputCh <- evt +} + // SetParentIO 设置上级 IOManager(nil 表示无上级,行为与以前完全一致)。 // 见 parent 字段的说明:用于驻留子继承父的输出通道/设备视图。 func (m *IOManager) SetParentIO(p *IOManager) { @@ -232,50 +282,52 @@ func (m *IOManager) StopAll() { } func (m *IOManager) InjectInput(source string, eventType string, payload map[string]interface{}) { - m.inputCh <- &InputEvent{ + m.deliverInput(&InputEvent{ RequestID: m.nextRequestID(), Source: source, Type: eventType, Payload: payload, OutputChannel: source, - } + }, false) } func (m *IOManager) InjectInputSync(source string, eventType string, payload map[string]interface{}) *OutputEvent { ch := make(chan *OutputEvent, 1) - m.inputCh <- &InputEvent{ + m.deliverInput(&InputEvent{ RequestID: m.nextRequestID(), Source: source, Type: eventType, Payload: payload, ResponseCh: ch, OutputChannel: source, - } + }, false) + // 被路由走时,回答由持有该 inputch 的 agent 写进同一个 ResponseCh + //(§4.3:同步输入的回程是事前定好的)——所以这里照常等待。 return <-ch } // InjectInputTo 注入输入事件并指定输出通道 func (m *IOManager) InjectInputTo(source, outputChannel, eventType string, payload map[string]interface{}) { - m.inputCh <- &InputEvent{ + m.deliverInput(&InputEvent{ RequestID: m.nextRequestID(), Source: source, Type: eventType, Payload: payload, OutputChannel: outputChannel, - } + }, false) } // InjectInputSyncTo 注入输入事件(同步等待)并指定输出通道 func (m *IOManager) InjectInputSyncTo(source, outputChannel, eventType string, payload map[string]interface{}) *OutputEvent { ch := make(chan *OutputEvent, 1) - m.inputCh <- &InputEvent{ + m.deliverInput(&InputEvent{ RequestID: m.nextRequestID(), Source: source, Type: eventType, Payload: payload, ResponseCh: ch, OutputChannel: outputChannel, - } + }, false) return <-ch } @@ -370,13 +422,13 @@ func (m *IOManager) InjectInterrupt(source, channel string, payload map[string]i payload = map[string]interface{}{} } evtType, _ := payload["type"].(string) - m.interruptCh <- &InputEvent{ + m.deliverInput(&InputEvent{ RequestID: m.nextRequestID(), Source: source, Type: evtType, Payload: payload, OutputChannel: channel, - } + }, true) } func (m *IOManager) InjectInterruptText(source, channel, text string) { diff --git a/internal/agent/io/inputroute_test.go b/internal/agent/io/inputroute_test.go new file mode 100644 index 0000000..b8e224b --- /dev/null +++ b/internal/agent/io/inputroute_test.go @@ -0,0 +1,74 @@ +package io + +import "testing" + +// 输入路由:路由器说"已被别的 agent 接管"时,事件**不得**进本内核队列。 +// +// 语义(设计 §4.1):inputch 是可分配资源,划给某个 agent 后输入只流向它 —— +// "父也顺便看到一份"是错的。 +func TestInputRouter_TakesOverExclusively(t *testing.T) { + m := NewIOManager() + var got []*InputEvent + var sawInterrupt bool + m.SetInputRouter(func(evt *InputEvent, isInterrupt bool) bool { + got = append(got, evt) + sawInterrupt = sawInterrupt || isInterrupt + return true // 全部接管 + }) + + m.InjectInputTo("plugin-x", "sub/in", "text", map[string]interface{}{"content": "a"}) + m.InjectInterruptTextOpts("plugin-x", "sub/in", "b", InjectOptions{}) + + if len(got) < 2 { + t.Fatalf("路由器应被调用(含中断路径),实际 %d 次", len(got)) + } + if n := len(m.InputChan()); n != 0 { + t.Fatalf("被接管的排队输入不得进本内核队列,实际 %d 条", n) + } + if !sawInterrupt { + t.Fatal("中断注入也必须经过路由(否则中断会绕过 inputch 归属直投父)") + } + if got[0].OutputChannel != "sub/in" { + t.Fatalf("路由器应拿到事件的 inputch,得到 %q", got[0].OutputChannel) + } +} + +// 路由器放行(返回 false)或未设置时,行为与以前完全一致。 +func TestInputRouter_PassthroughKeepsOldBehaviour(t *testing.T) { + m := NewIOManager() + calls := 0 + m.SetInputRouter(func(evt *InputEvent, isInterrupt bool) bool { calls++; return false }) + + m.InjectInputTo("plugin-x", "sub/in", "text", map[string]interface{}{"content": "a"}) + if calls != 1 { + t.Fatalf("路由器应被调用一次,实际 %d", calls) + } + if n := len(m.InputChan()); n != 1 { + t.Fatalf("放行的输入应进本内核队列,实际 %d 条", n) + } + if _, ok := <-m.InputChan(); !ok { + t.Fatal("队列应可读") + } + + // 未设路由器:直接入队(历史行为) + m2 := NewIOManager() + m2.InjectInterruptText("plugin-x", "sub/in", "c") + if n := len(m2.InputInterruptChan()); n != 1 { + t.Fatalf("未设路由器时中断应直接入队,实际 %d 条", n) + } +} + +// DeliverRouted 是不再二次路由的投递口(路由器实现把事件交给持有者)。 +func TestDeliverRouted_SkipsSecondRouting(t *testing.T) { + m := NewIOManager() + routerCalls := 0 + m.SetInputRouter(func(evt *InputEvent, isInterrupt bool) bool { routerCalls++; return true }) + + m.DeliverRouted(&InputEvent{OutputChannel: "sub/in"}, false) + if routerCalls != 0 { + t.Fatalf("DeliverRouted 不应再触发路由(会成环),实际 %d 次", routerCalls) + } + if n := len(m.InputChan()); n != 1 { + t.Fatalf("应已入队,实际 %d 条", n) + } +}