mirror of
https://gitcode.com/JianFeeeee/HomeAgent.git
synced 2026-10-03 15:53:56 +00:00
fix(scheduler): inputch 划给子后输入只流向子 —— 补上「进内核之前」的输入路由
用户指出的语义(设计稿 §4.1 早已写明): **inputch 是可分配资源**,「路由发生在**进内核之前**」—— 划给某个 agent 后, 该通道的输入**只流向那个 agent**;outputch 不同,授权是**非独占**的, 父依旧可以通过它发送内容。 而代码里 inputch 划拨只做了**登记**,没有做**路由**: - 插件注入输入的 io 是**根 agent 的**(`cmd/homed` 里 `pluginReg.SetIOManager(iom)`); - 唯一消费输入的是「该 io 自己的调度器」(`scheduler.go` 读 `a.io.InputChan()`); - `ChannelRegistry.Assign` 只把 Owner 写进登记表,**没有任何转发动作**。 ⇒ 现场表现(用户线上联调):子挂 `inputch=[timer]`,**timer 的输入却打在父身上** (日志 `[agent] interrupt from timer/timer`),子侧 `轮次=0` 永远不动。 登记表里的 Owner 于是沦为标签。 改法(按 §4.1 把路由放回"进内核之前"): - `IOManager` 增加 `InputRouter`(`SetInputRouter`),并把**五处直接入队**收口到 `deliverInput`:`InjectInput` / `InjectInputSync` / `InjectInputTo` / `InjectInputSyncTo` / `InjectInterrupt`(排队与中断两条路都过路由)。 - 内核注入路由器 `Agent.routeInputByOwner`:查 inputch 的 Owner —— 归自己/未分配 ⇒ 本内核处理;归自己的某个驻留子 ⇒ `DeliverRouted` 交给它(**不再进父的队列**); 归一个不存在的 agent ⇒ **不吞输入**,父兜底 + 留痕(吞掉输入比多处理一条更糟)。 - `DeliverRouted` 是"已路由"的投递口,不再二次路由(避免成环)。 - 同步输入的 `ResponseCh` 随事件一起走 ⇒ 回答由持有者写回同一回程(§4.3)。 判据(新增 6 条): - io 层:被接管时排队/中断都**不入本内核队列**(且中断确实经过路由)/ 放行与未设 路由器时与历史行为一致 / `DeliverRouted` 不再触发路由 - 内核层:划给子的 inputch 输入进**子**(子 Executed>0)且**父 Enqueued 不变** / 归属到不存在的 agent 时父兜底(不吞)/ 未分配的 inputch 仍归父
This commit is contained in:
@ -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) {
|
||||
|
||||
74
internal/agent/io/inputroute_test.go
Normal file
74
internal/agent/io/inputroute_test.go
Normal file
@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user