From 895948b24e3e41fa500b353195bd7b053607a8f6 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Fri, 18 Sep 2026 11:39:04 +0800 Subject: [PATCH] =?UTF-8?q?fix(stop):=20=E9=85=8D=E9=A2=9D=E9=A1=BB?= =?UTF-8?q?=E8=AE=A1=E5=85=A5=E5=81=9C=E5=9C=A8=E8=BE=93=E5=85=A5=20channe?= =?UTF-8?q?l=20=E7=9A=84=E5=BE=85=E5=A4=84=E7=90=86=E6=B6=88=E6=81=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实测：停止后排队消息仍逐条跑完。原因是 armStop 只数 sched.queue， 而用户按下停止时调度器正忙于当前任务，其余消息大多还没被 pumpInbox 搬进队列、仍停在 inputCh ⇒ queued=0、配额归零。 - IOManager.PendingInputs()：暴露 channel 中待处理条数。 - armStop(pending int)：queued = len(s.queue) + pending。 - 补 TestStop_ArmCountsPendingChannelInputs 锁死该口径。 --- internal/agent/core/eventloop.go | 9 ++++- internal/agent/core/scheduler.go | 15 ++++----- internal/agent/core/stop_semantics_test.go | 39 +++++++++++++++++++--- internal/agent/io/channel.go | 13 ++++++++ 4 files changed, 63 insertions(+), 13 deletions(-) diff --git a/internal/agent/core/eventloop.go b/internal/agent/core/eventloop.go index e85ce4b..4aa8924 100644 --- a/internal/agent/core/eventloop.go +++ b/internal/agent/core/eventloop.go @@ -53,7 +53,14 @@ func (a *Agent) interceptLoop() { // 这个 `||` 在 queued=0 时会短路到 takeStop()，把标记先消费掉， // 于是 stepLLM 永远看不到它 → 取消后照样重跑一轮。 // 实测：停止被正确记录（`stop requested ... queued=0`）但生成仍跑到自然结束。 - n := a.sched.armStop() + // pending 必须算上**停在输入 channel 里**的那一段：停止时 + // 调度器多在半路忙当前任务，其余消息还没被 pumpInbox 搬进队列， + // 只数 sched.queue 会得到 0，配额随之失效（实测过）。 + pending := 0 + if a.io != nil { + pending = a.io.PendingInputs() + } + n := a.sched.armStop(pending) log.Printf("[agent] stop requested by %s/%s (queued=%d will be short-circuited at pre-action)", evt.Source, evt.OutputChannel, n) a.cancelCurrentLLM() diff --git a/internal/agent/core/scheduler.go b/internal/agent/core/scheduler.go index fe34122..de8382a 100644 --- a/internal/agent/core/scheduler.go +++ b/internal/agent/core/scheduler.go @@ -590,21 +590,20 @@ func (s *scheduler) setImmediateLocked(t *Task) bool { // 停止之后新到的输入不受影响（否则停止会变成一个永远生效的“黑洞”）。 // // 多次按停止取**较大值**而不是累加：两个客户端同时按下时配额不应翻倍。 -func (s *scheduler) armStop() int { +func (s *scheduler) armStop(pending int) int { s.mu.Lock() defer s.mu.Unlock() - if n := s.pendingQueueLenLocked(); n > s.cancelBudget { - s.cancelBudget = n + // pending 是**还没被 pumpInbox 搬进队列**的那一段（停在输入 channel 里）。 + // 用户按下停止时，调度器通常正忙于当前任务，其它消息基本都停在 channel； + // 只数 s.queue 会得到 0（实测），配额随之失效。 + queued := len(s.queue) + pending + if queued > s.cancelBudget { + s.cancelBudget = queued } s.stopArmed = true return s.cancelBudget } -// pendingQueueLenLocked 统计**尚未执行**的排队输入数量（不含中断队列/挂起）。 -func (s *scheduler) pendingQueueLenLocked() int { - return len(s.queue) -} - // takeStop 消费「当前任务应被立即结束而不是重试」这一次标记。 // // 取消 LLM 后 stepLLM 会看到 context.Canceled 并 outcomeContinue 重跑； diff --git a/internal/agent/core/stop_semantics_test.go b/internal/agent/core/stop_semantics_test.go index 4b791de..c0838b4 100644 --- a/internal/agent/core/stop_semantics_test.go +++ b/internal/agent/core/stop_semantics_test.go @@ -105,7 +105,7 @@ func TestStop_TerminatesCurrentTaskWithoutRetry(t *testing.T) { a.io.InjectInterrupt(stopEvt.Source, stopEvt.OutputChannel, stopEvt.Payload) // interceptLoop 才是 armStop 的调用点；这里模拟它已消费该指令（只 arm， // 不 takeStop——takeStop 必须由 stepLLM 去消费，那正是被测行为）。 - a.sched.armStop() + a.sched.armStop(0) a.cancelCurrentLLM() select { @@ -135,7 +135,7 @@ func TestStop_CancelBudgetIsSnapshot(t *testing.T) { // 停止那一刻队列里有 2 条。 a.sched.enqueue(newInputTask(mustTextEvent("qq", "一"))) a.sched.enqueue(newInputTask(mustTextEvent("qq", "二"))) - n := a.sched.armStop() + n := a.sched.armStop(0) if n != 2 { t.Fatalf("停止时排队深度=%d，期望 2", n) } @@ -170,10 +170,10 @@ func TestStop_ArmTwiceTakesMaxNotSum(t *testing.T) { a := newPreemptAgent(t, newPreemptProvider()) a.sched.enqueue(newInputTask(mustTextEvent("qq", "一"))) a.sched.enqueue(newInputTask(mustTextEvent("qq", "二"))) - if n := a.sched.armStop(); n != 2 { + if n := a.sched.armStop(0); n != 2 { t.Fatalf("首次 armStop=%d，期望 2", n) } - if n := a.sched.armStop(); n != 2 { + if n := a.sched.armStop(0); n != 2 { t.Fatalf("重复 armStop=%d，期望仍为 2（取 max 不累加）", n) } if !a.sched.consumeCancel() || !a.sched.consumeCancel() { @@ -189,6 +189,37 @@ func mustTextEvent(source, text string) *agentIO.InputEvent { return evt } +// 停止时必须把**停在输入 channel 里**的待处理消息也计入配额。 +// +// 这是实测踩到的坑：用户按下停止时调度器正忙于当前任务，其它消息大多还没被 +// pumpInbox 搬进就绪队列，仍然停在 inputCh。只数 sched.queue 会得到 queued=0， +// 配额归零，停止后排队消息照旧逐条跑完。 +func TestStop_ArmCountsPendingChannelInputs(t *testing.T) { + a := newPreemptAgent(t, newPreemptProvider()) + + // 两条消息只进 channel，不入队（模拟 pumpInbox 尚未搬运）。 + for _, txt := range []string{"一", "二"} { + evt, _ := textEvent("webui", txt) + a.io.InjectInput("webui", "text", map[string]interface{}{"content": txt}) + _ = evt + } + pending := a.io.PendingInputs() + if pending != 2 { + t.Fatalf("channel 待处理=%d，期望 2", pending) + } + + n := a.sched.armStop(pending) + if n != 2 { + t.Fatalf("armStop(2)=%d，期望 2（配额须包含 channel 中的待处理）", n) + } + if !a.sched.consumeCancel() || !a.sched.consumeCancel() { + t.Fatal("两条配额都应可消费") + } + if a.sched.consumeCancel() { + t.Fatal("不应有多余配额") + } +} + // 停止必须经**真实的 interceptLoop** 被消费，而不是测试自己 arm。 // // 这条用例专门镇一个已经发生过的自伤：曾经在 interceptLoop 里写成 diff --git a/internal/agent/io/channel.go b/internal/agent/io/channel.go index e132f47..14e76c4 100644 --- a/internal/agent/io/channel.go +++ b/internal/agent/io/channel.go @@ -536,6 +536,19 @@ func (m *IOManager) EmitTextTo(target, outputChannel, text string) { func (m *IOManager) InputChan() <-chan *InputEvent { return m.inputCh } func (m *IOManager) OutputChan() <-chan *OutputEvent { return m.outputCh } +// PendingInputs 返回**尚未被调度器取走**的输入条数（排队队列之外的那一段）。 +// +// 为何需要单独一个口径：输入先进 inputCh，再由 pumpInbox 搬进调度器就绪队列。 +// “停止”要统计的是“用户按下停止时还没被处理的消息”，而它们大多还躺在 +// inputCh 里（调度器正忙于处理当前任务）。只数 sched.queue 会得到 0， +// 于是短路配额失效——实测就是这个现象：停止后排队消息照旧逐条跑完。 +func (m *IOManager) PendingInputs() int { + if m == nil { + return 0 + } + return len(m.inputCh) +} + // RegisterInputChannel 注册一个 inputch（不带插件归属，兼容旧调用）。 // // inputch 是**最基本的输入路由单位**；一个插件可以注册多个。