From 98d67559d1b455def7524a0a4bf0b44857ba88d8 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sun, 13 Sep 2026 06:10:44 +0800 Subject: [PATCH] =?UTF-8?q?fix(scheduler):=20=E8=A1=A5=E9=BD=90=20M3=20?= =?UTF-8?q?=E4=B8=8E=E8=AE=BE=E8=AE=A1=E7=A8=BF=E7=9A=84=E4=B8=A4=E5=A4=84?= =?UTF-8?q?=E8=AF=AD=E4=B9=89=E5=81=8F=E7=A6=BB=EF=BC=88=E5=8F=91=E7=8E=B0?= =?UTF-8?q?=E5=8D=B3=E4=BF=AE=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 两处都不是风格差异，而是真的偏离设计语义（其一为回归）， 已按“先写判据确认失败、再修”的方式处理，判据保留为回归测试。 1. 空闲时到达的中断永远不会被处理（设计 §5.1 ③ 未落地） 调度器空闲时只阻塞在 select{InputChan, selfInputCh, ctx.Done}， 而 pendingInterrupts 不是 channel——interceptLoop 把中断入队后 没有任何东西唤醒调度器，中断要等“下一条输入”才被看到。 修复：调度器加 wake channel，enqueueInterrupt 非阻塞 signalWake， 空闲分支增加 wake 分支。 2. 临界区内只“不让位”却仍被“取消”（设计 §4.3/§5.2） requestPreempt 不判临界区，interceptLoop 照常 cancelLLM， 于是正在流式的记忆整理被中断，stepLLM 以 error 提前结束—— 整理任务被砍掉一半，而设计要求的是“请求排队等它结束”。 修复：scheduler 增加原子 critical 标志（帧仍只由调度器读写）， runInputTask 在 prepare 后设置、结束（含挂起）时清除； requestPreempt 在临界区内不 arm、不取消，中断只入队。 - 新增 scheduler_regression_test.go 2 项（先失败后通过） - 验收：agent 全量 + -race；全仓 vet 通过 --- internal/agent/core/scheduler.go | 34 +++++- .../agent/core/scheduler_regression_test.go | 101 ++++++++++++++++++ internal/agent/core/task.go | 7 ++ 3 files changed, 139 insertions(+), 3 deletions(-) create mode 100644 internal/agent/core/scheduler_regression_test.go diff --git a/internal/agent/core/scheduler.go b/internal/agent/core/scheduler.go index 890e9ca..53a227b 100644 --- a/internal/agent/core/scheduler.go +++ b/internal/agent/core/scheduler.go @@ -21,6 +21,7 @@ import ( "log" "runtime/debug" "sync" + "sync/atomic" "time" agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" @@ -197,6 +198,12 @@ type scheduler struct { // interruptLoop 只写这两个字段与 pendingInterrupts；帧永远只由调度器读写。 preemptArmed bool preemptLevel Level + // critical 报告运行任务是否在不可抢占临界区（如记忆整理）。 + // 由于 interceptLoop 要读它，必须是原子的：帧仍只由调度器读写。 + critical atomic.Bool + // wake 用于把空闲的调度器叫醒：pendingInterrupts 不是 channel， + // 没有这个信号时“空闲时到达的中断”会一直等下一次输入（设计 §5.1 ③）。 + wake chan struct{} // maxSuspendDepth：suspendPool 深度上限（设计文档 §6.3，默认 4）。 maxSuspendDepth int } @@ -221,9 +228,23 @@ func newScheduler(maxQueue int) *scheduler { if maxQueue <= 0 { maxQueue = 256 } - return &scheduler{maxQueue: maxQueue, maxSuspendDepth: 4} + return &scheduler{maxQueue: maxQueue, maxSuspendDepth: 4, wake: make(chan struct{}, 1)} } +// signalWake 非阻塞地唤醒调度器。 +func (s *scheduler) signalWake() { + select { + case s.wake <- struct{}{}: + default: + } +} + +// setCritical 由调度器 goroutine 在任务进入/离开临界区时设置。 +func (s *scheduler) setCritical(v bool) { s.critical.Store(v) } + +// inCritical 报告运行任务是否在不可抢占临界区。 +func (s *scheduler) inCritical() bool { return s.critical.Load() } + // hasRoom 报告就绪队列是否还能接收任务。泵入侧据此节流： // 队列满则停止从 channel 取，让背压落回 channel 本身。 func (s *scheduler) hasRoom() bool { @@ -329,6 +350,7 @@ func (s *scheduler) enqueueInterrupt(t *Task) { s.stats.Rejected++ } s.pendingInterrupts = append(s.pendingInterrupts, t) + s.signalWake() } // requestPreempt 登记一次中断请求。 @@ -339,11 +361,15 @@ func (s *scheduler) enqueueInterrupt(t *Task) { // 就正常结束，中断也不会丢（它会被 nextRef 按优先级选出）。 // // 判据用**有效**优先级（饥饿防护），并受抢占冷却约束。 +// +// 临界区（如记忆整理）内不 arm、不取消：中断只入队，等临界区结束后的安全点处理， +// 这是设计 §4.3 的硬要求——那个位置的“不抢占”不能只是不让位，还必须不取消。 func (s *scheduler) requestPreempt(evt *agentIO.InputEvent, level Level) bool { s.mu.Lock() running := s.running + critical := s.critical.Load() canPreempt := false - if running != nil && level > effectiveLevel(running) { + if !critical && running != nil && level > effectiveLevel(running) { if running.LastPreemptAt.IsZero() || time.Since(running.LastPreemptAt) >= preemptCooldown { canPreempt = true s.preemptArmed = true @@ -549,12 +575,14 @@ func (a *Agent) schedulerLoop() { t, f, kind := a.sched.nextRef() if kind == nextNone { - // 无待办：阻塞等新输入或退出。 + // 无待办：阻塞等新输入、新中断（wake）或退出。 select { case evt := <-a.io.InputChan(): a.sched.enqueue(newInputTask(evt)) case msg := <-a.selfInputCh: a.sched.enqueue(newSelfTask(msg)) + case <-a.sched.wake: + // 中断已入 pendingInterrupts，回到循环顶部重新挑选。 case <-a.ctx.Done(): return } diff --git a/internal/agent/core/scheduler_regression_test.go b/internal/agent/core/scheduler_regression_test.go new file mode 100644 index 0000000..0481661 --- /dev/null +++ b/internal/agent/core/scheduler_regression_test.go @@ -0,0 +1,101 @@ +package core + +// 回归判据：M3 实现与设计稿 §5.1/§4.3 的两处偏离。 +// +// 这两条是**先写判据、确认失败、再修**的（修完保留为回归测试）。 + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api" + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" +) + +// cancelAwareProvider 第一次调用进入后阻塞，直到 ctx 取消；记录是否被取消。 +type cancelAwareProvider struct { + once sync.Once + entered chan struct{} + canceled atomic.Bool +} + +func newCancelAwareProvider() *cancelAwareProvider { + return &cancelAwareProvider{entered: make(chan struct{})} +} + +func (p *cancelAwareProvider) Name() string { return "cancel-aware" } +func (p *cancelAwareProvider) Chat(ctx context.Context, req *agentAPI.CompletionRequest) (*agentAPI.CompletionResponse, error) { + p.once.Do(func() { close(p.entered) }) + <-ctx.Done() + p.canceled.Store(true) + return nil, ctx.Err() +} +func (p *cancelAwareProvider) ChatStream(ctx context.Context, req *agentAPI.CompletionRequest) (<-chan agentAPI.StreamChunk, error) { + return nil, context.Canceled +} +func (p *cancelAwareProvider) MaxContextTokens() int { return 8192 } + +// 设计 §5.1 ③：interruptLoop 定级/决策后必须**唤醒调度器**。 +// 现状：中断只被放进 pendingInterrupts，而调度器空闲时阻塞在 select（只看 +// InputChan/selfInputCh/ctx.Done）——没有任何东西会把它叫醒。 +func TestGap_IdleInterruptIsProcessed(t *testing.T) { + sp := &countingProvider{} + a := New(AgentConfig{ + ID: "idle-intr", + Provider: sp, + ProviderManager: agentAPI.NewProviderManager(), + IO: agentIO.NewIOManager(), + StageHost: NewStageHost(), + }) + a.Start() + defer a.Stop() + + // 完全空闲时投递一条中断（模拟定时器通知/插件提醒）。 + a.io.InjectInterruptText("qq", "cli", "空闲时的通知") + + deadline := time.Now().Add(3 * time.Second) + for sp.n.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(20 * time.Millisecond) + } + if sp.n.Load() == 0 { + snap := a.DumpScheduler() + t.Fatalf("空闲时到达的中断未被处理：pendingInterrupts=%d（调度器未被唤醒）", + len(snap.PendingInterrupts)) + } +} + +// 设计 §4.3/§5.2：`_consolidation_` 是整任务临界区——抢占请求必须排队等它结束， +// 而不是取消它。现状：requestPreempt 不判临界区，interceptLoop 照常 cancelLLM， +// 于是正在流式的记忆整理被中断 → stepLLM 直接以 error 结束（整理丢一半）。 +func TestGap_ConsolidationMustNotBeCancelled(t *testing.T) { + sp := newCancelAwareProvider() + a := New(AgentConfig{ + ID: "consol", + Provider: sp, + ProviderManager: agentAPI.NewProviderManager(), + IO: agentIO.NewIOManager(), + StageHost: NewStageHost(), + }) + a.Start() + defer a.Stop() + + // 走自循环通道发起一次记忆整理。 + a.selfInputCh <- selfInputMsg{text: "合并实体", channel: channelConsolidation} + + select { + case <-sp.entered: + case <-time.After(3 * time.Second): + t.Fatal("记忆整理未进入 LLM 调用") + } + + // L4 中断到达。 + a.io.InjectInterruptText("cli", "cli", "L4 打断") + + time.Sleep(500 * time.Millisecond) + if sp.canceled.Load() { + t.Fatal("记忆整理是临界区，其 LLM 不该被取消（设计 §4.3/§5.2）") + } +} diff --git a/internal/agent/core/task.go b/internal/agent/core/task.go index 79149a5..6bc2616 100644 --- a/internal/agent/core/task.go +++ b/internal/agent/core/task.go @@ -191,6 +191,10 @@ func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response stri // // M3a 还没有抢占，因此 outcomeSuspended 只会由 M3b 的抢占检查产生。 func (a *Agent) runInputTask(evt *agentIO.InputEvent, seed []agentAPI.Message) (*TaskFrame, stepOutcome) { + // 临界区标记由调度器 goroutine 维护，任务结束（含挂起）即清。 + // interceptLoop 读它来决定“能不能取消”，因此必须是原子的。 + defer a.sched.setCritical(false) + f, term := a.prepareInputTask(evt) switch term { case terminalSkipped, terminalStageShortCircuit, terminalConsolidation: @@ -234,6 +238,9 @@ func (a *Agent) prepareInputTask(evt *agentIO.InputEvent) (*TaskFrame, taskTermi if a.currentOutputChannel == "" { a.currentOutputChannel = evt.Source } + // 进入本任务的临界区属性（记忆整理整任务不可抢占）。 + // 必须在 processConsolidation 之前设置——它就在下面同步执行。 + a.sched.setCritical(a.inCriticalSection()) if evt.OutputChannel == channelConsolidation { a.processConsolidation(evt, in.text)