diff --git a/internal/agent/core/offload.go b/internal/agent/core/offload.go index ffeef00..37616a1 100644 --- a/internal/agent/core/offload.go +++ b/internal/agent/core/offload.go @@ -120,12 +120,8 @@ func (a *Agent) drainInboxToQueue() { } } -// residentInputChannel 返回"父给某个子投递输入"用的 inputch 名。 -// -// 与 SendToResident 用的是同一个(sub/):子是**不配插件 inputch** 的 -// (opts.InputChs 为空),它的入站口就只有父给它的这一条,因此必须与 -// SendToResident 保持一致,否则转投的消息会落到一个父不知道的通道名上。 -func residentInputChannel(id string) string { return "sub/" + id } +// 注:转投不走 inputch 名字(那会在投递时重建事件、丢掉 ResponseCh), +// 而是直接跨 agent 推原事件 —— 见 forwardInputToResident。 // offloadPendingTasks 检查是否需要转投,需要则拉起/复用一个驻留子并搬运任务。 // @@ -212,11 +208,16 @@ func (a *Agent) offloadPendingTasks(opts OffloadOptions) int { return moved } -// forwardInputToResident 把一条输入原文投给指定驻留子的 inputch。 +// forwardInputToResident 把一条输入**原文**投给指定驻留子的队列。 // -// 走 InjectInputTo(排队输入,非中断):转投的是"待办工作",不是"打断子"。 -// 子的 io 上有 inputRouter(routeInputByOwner),但投递目标是**它自己的** inputch -// 且 Owner 就是它,因此不会被再次路由走。 +// ❗必须推**原事件**(DeliverRouted),不能重建:原事件带 ResponseCh, +// 而 cli / a2a / webui 这些**同步**调用方正阻塞等它。重建事件(如用 +// InjectInputTo)会把 ResponseCh 丢掉 ⇒ 任务被子处理完、调用方却永远收不到回执。 +// 实测:转投生效、子也正常处理(各 ~3s 日志可见),但 HTTP 请求一直挂着不返回。 +// 仓库反复警告同一件事(见 Agent.Stop 对 drainPendingInterrupts 的注释: +// “带 ResponseCh 的同步注入方会永久挂起”),这里必须走既有的跨 agent 投递原语。 +// +// 用排队语义(isInterrupt=false):转投的是"待办工作",不是"打断子"。 func (a *Agent) forwardInputToResident(residentID string, evt *agentIO.InputEvent) error { a.residentMu.Lock() rc := a.residents[residentID] @@ -225,16 +226,14 @@ func (a *Agent) forwardInputToResident(residentID string, evt *agentIO.InputEven return fmt.Errorf("驻留子 %s 不存在或不可用", residentID) } - payload := map[string]interface{}{} - for k, v := range evt.Payload { - payload[k] = v + // 带上来源线索(不重建事件,只补充 payload,保留 ResponseCh/RequestID)。 + if evt.Payload == nil { + evt.Payload = map[string]interface{}{} } - // 带上来源线索,让子知道这条不是父当前任务的续接,而是转投的独立请求。 - payload["offloaded_from"] = string(a.id) - payload["offloaded_at"] = time.Now().Format(time.RFC3339) + evt.Payload["offloaded_from"] = string(a.id) + evt.Payload["offloaded_at"] = time.Now().Format(time.RFC3339) - ch := residentInputChannel(residentID) - rc.agent.io.InjectInputToOpts(evt.Source, ch, evt.Type, payload, agentIO.InjectOptions{}) + rc.agent.io.DeliverRouted(evt, false) return nil } diff --git a/internal/agent/core/offload_test.go b/internal/agent/core/offload_test.go index 7592c40..37abc58 100644 --- a/internal/agent/core/offload_test.go +++ b/internal/agent/core/offload_test.go @@ -345,3 +345,66 @@ func TestOffloadSeesInputsStuckInChannel(t *testing.T) { t.Fatalf("应拉起 1 个驻留子,实际 %d", len(root.Residents())) } } + +// ★★ 回归:转投必须保留 ResponseCh,否则同步调用方永久挂起。 +// +// 这是我在线上真踩的第二个 bug:第一版用 InjectInputTo 重建事件 ⇒ ResponseCh +// 被丢掉 ⇒ 日志显示子**正常处理完了**(各 ~3s),但 webui 的 HTTP 请求一直挂着 +// 不返回,最终 504。仓库反复警告同一件事(Agent.Stop 的注释:"带 ResponseCh 的 +// 同步注入方(cli / clawhubadapter 均无超时)会永久挂起")。 +// +// 正确做法是走既有的跨 agent 投递原语 DeliverRouted:它推**原事件**。 +// +// 断言方式是**最强的那个**:真的等同步回执回来。 +// (不用读子的 InputChan 来断言:SpawnResident 会启动子自己的调度循环, +// +// 它会与测试抢同一个 channel —— 那样写出来的测试是 flaky 的,实测过一次挂死。) +func TestForwardKeepsResponseCh(t *testing.T) { + root, main := newRootWithoutSchedulerLoop(t) + defer main.Close() + + opts := DefaultOffloadOptions() + opts.Enabled = true + opts.BusyAfter = time.Nanosecond + opts.MinPending = 1 + opts.MaxResidents = 1 + + root.sched.enqueue(makeQueuedInput(1)) + root.sched.nextRef() + + // 一条**带同步回执通道**的输入(模拟 cli/webui 这类调用方) + respCh := make(chan *agentIO.OutputEvent, 1) + evt := &agentIO.InputEvent{ + RequestID: "sync-1", Source: "webui", Type: "text", + OutputChannel: "webui", + Payload: map[string]interface{}{"content": "sync request"}, + ResponseCh: respCh, + } + root.sched.enqueue(newInputTask(evt)) + + if n := root.offloadPendingTasks(opts); n != 1 { + t.Fatalf("应转投 1 条,实际 %d", n) + } + + // 转投的是**同一个事件对象**(所以 payload 上的标注能在这里被看到), + // 而不是重建的副本 —— 副本会丢掉 ResponseCh。 + if evt.Payload["offloaded_from"] != "parent" { + t.Errorf("应标注转投来源,实际 %v", evt.Payload["offloaded_from"]) + } + if evt.ResponseCh == nil { + t.Fatal("★ 原事件的 ResponseCh 被清掉了") + } + + // ★ 决定性断言:同步调用方真的收到回执。 + select { + case out := <-respCh: + if out == nil { + t.Fatal("收到空回执") + } + if out.RequestID != "sync-1" { + t.Errorf("回执应带原 RequestID,实际 %q", out.RequestID) + } + case <-time.After(10 * time.Second): + t.Fatal("★ 同步调用方没收到回执:转投丢了 ResponseCh(线上表现为 HTTP 挂起 504)") + } +}