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 仍归父

(cherry picked from commit 3c263ed0b4)
This commit is contained in:
JianFeeeee
2026-09-13 15:35:54 +08:00
parent cd7aebe224
commit f876be6f63
5 changed files with 269 additions and 11 deletions

View File

@ -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 接线)。

View File

@ -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
}

View File

@ -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
})
}