fix(scheduler): 补齐 M3 与设计稿的两处语义偏离(发现即修)

两处都不是风格差异,而是真的偏离设计语义(其一为回归),
已按“先写判据确认失败、再修”的方式处理,判据保留为回归测试。

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 通过
This commit is contained in:
JianFeeeee
2026-09-13 06:10:44 +08:00
parent 6bd3313a22
commit 98d67559d1
3 changed files with 139 additions and 3 deletions

View File

@ -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{}
// maxSuspendDepthsuspendPool 深度上限(设计文档 §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
}