diff --git a/internal/agent/core/scheduler.go b/internal/agent/core/scheduler.go index e5d768c..4f7c037 100644 --- a/internal/agent/core/scheduler.go +++ b/internal/agent/core/scheduler.go @@ -89,6 +89,35 @@ type Task struct { // SeedMsgs 是抢占式中断任务的只读前缀(D1=A):由被打断的任务在挂起时 // 附上,使中断任务看得见「进行到哪一步」,但其产出不合并回原任务。 SeedMsgs []agentAPI.Message + + // PreemptCount 是本任务被抢占的次数,用于饥饿防护: + // effectiveLevel = min(L4, Level + min(PreemptCount, 2))。 + PreemptCount int + // LastPreemptAt 是上次被抢占的时刻,用于抢占冷却。 + LastPreemptAt time.Time +} + +// preemptPromotionCap 是抢占计数能带来的最大提升档数。 +const preemptPromotionCap = 2 + +// preemptCooldown 是“刚被抢占过”的冷却期:期内不再被抢占, +// 避免高优先级流把同一任务反复打断到永不完结。 +const preemptCooldown = 2 * time.Second + +// effectiveLevel 返回任务的**有效**优先级(设计文档 §9 饥饿防护)。 +// +// 被抢占越多的任务越“值钱”,从而逐步追上抢占它的流;封顶 L4, +// 因此它永远不会反过来抢占真正的紧急输入。 +func effectiveLevel(t *Task) Level { + p := t.PreemptCount + if p > preemptPromotionCap { + p = preemptPromotionCap + } + l := t.Level + Level(p) + if l > LevelCritical { + l = LevelCritical + } + return l } // SchedulerStats 是调度器的累计计数(可观测性,设计文档 §11 O2)。 @@ -266,28 +295,33 @@ func (s *scheduler) enqueueInterrupt(t *Task) { // // 无论能否抢占,中断请求都进 pendingInterrupts——这样即使运行任务在抢占生效前 // 就正常结束,中断也不会丢(它会被 nextRef 按优先级选出)。 +// +// 判据用**有效**优先级(饥饿防护),并受抢占冷却约束。 func (s *scheduler) requestPreempt(evt *agentIO.InputEvent, level Level) bool { s.mu.Lock() running := s.running + canPreempt := false + if running != nil && level > effectiveLevel(running) { + if running.LastPreemptAt.IsZero() || time.Since(running.LastPreemptAt) >= preemptCooldown { + canPreempt = true + s.preemptArmed = true + s.preemptLevel = level + } + } s.mu.Unlock() s.enqueueInterrupt(newInterruptTask(evt, level)) - - if running == nil || level <= running.Level { - return false - } - s.mu.Lock() - s.preemptArmed = true - s.preemptLevel = level - s.mu.Unlock() - return true + return canPreempt } -// preemptGrantedFor 报告级别为 level 的运行任务是否应在当前安全点让位。 -func (s *scheduler) preemptGrantedFor(level Level) bool { +// preemptGrantedFor 报告运行任务是否应在当前安全点让位。 +func (s *scheduler) preemptGrantedFor() bool { s.mu.Lock() defer s.mu.Unlock() - return s.preemptArmed && s.preemptLevel > level + if !s.preemptArmed || s.running == nil { + return false + } + return s.preemptLevel > effectiveLevel(s.running) } func (s *scheduler) clearPreempt() { @@ -308,6 +342,9 @@ func (s *scheduler) suspend(t *Task, f *TaskFrame) { s.stats.Rejected++ } s.suspendPool = append(s.suspendPool, &suspendedTask{Task: t, Frame: f}) + // 饥饿防护:抢占计数 +1(提升有效级)并记录冷却起点。 + t.PreemptCount++ + t.LastPreemptAt = time.Now() if s.running == t { s.running = nil } @@ -414,10 +451,11 @@ func pickTaskIndex(q []*Task) int { return best } -// taskBefore 报告 x 是否应先于 y 执行。 +// taskBefore 报告 x 是否应先于 y 执行(按**有效**优先级)。 func taskBefore(x, y *Task) bool { - if x.Level != y.Level { - return x.Level > y.Level + lx, ly := effectiveLevel(x), effectiveLevel(y) + if lx != ly { + return lx > ly } if !x.EnqueuedAt.Equal(y.EnqueuedAt) { return x.EnqueuedAt.Before(y.EnqueuedAt) diff --git a/internal/agent/core/scheduler_critical_test.go b/internal/agent/core/scheduler_critical_test.go index bbdfb62..f4a82fd 100644 --- a/internal/agent/core/scheduler_critical_test.go +++ b/internal/agent/core/scheduler_critical_test.go @@ -61,7 +61,7 @@ func TestPreempt_DeferredDuringToolExec(t *testing.T) { t.Fatal("L4 应 arm 让位信号") } // 关键断言:信号已 arm,但任务仍在工具里 —— 绝不能挂起。 - if !a.sched.preemptGrantedFor(LevelBackground) { + if !a.sched.preemptGrantedFor() { t.Fatal("让位信号应已 arm") } if a.DumpScheduler().Running == nil { diff --git a/internal/agent/core/scheduler_preempt_test.go b/internal/agent/core/scheduler_preempt_test.go index 0a514b0..1f4e9ef 100644 --- a/internal/agent/core/scheduler_preempt_test.go +++ b/internal/agent/core/scheduler_preempt_test.go @@ -210,7 +210,7 @@ func TestPreempt_LowerOrEqualDoesNotPreempt(t *testing.T) { if a.sched.requestPreempt(e2, LevelBackground) { t.Fatal("更低级不得抢占") } - if a.sched.preemptGrantedFor(LevelInteractive) { + if a.sched.preemptGrantedFor() { t.Fatal("未 arm 让位信号,preemptGrantedFor 应为 false") } diff --git a/internal/agent/core/scheduler_starvation_test.go b/internal/agent/core/scheduler_starvation_test.go new file mode 100644 index 0000000..bc1f563 --- /dev/null +++ b/internal/agent/core/scheduler_starvation_test.go @@ -0,0 +1,118 @@ +package core + +// M5 验收测试:饥饿防护(抢占计数提升有效级 + 抢占冷却)。 +// +// 设计依据 docs/zh/input-scheduler-design.md §9、§11.5(G1/G2)。 +// +// 为什么需要:固定四级 + 「严格大于才抢占」下,一条 L4 流可以反复打断同一个 +// L1 任务,使它永不完结。提升被抢占者的**有效**优先级,让它在竞争排队时 +// 逐步追上;封顶 L4,因此它永远抢不过真正的紧急输入(紧急输入本身不被抢占)。 + +import ( + "testing" + "time" +) + +func TestStarvation_EffectiveLevelPromotion(t *testing.T) { + base := &Task{Level: LevelBackground} + if got := effectiveLevel(base); got != LevelBackground { + t.Fatalf("未抢占时有效级=%v,期望 L1", got) + } + + base.PreemptCount = 1 + if got := effectiveLevel(base); got != LevelMessage { + t.Fatalf("被抢占 1 次后有效级=%v,期望 L2", got) + } + base.PreemptCount = 2 + if got := effectiveLevel(base); got != LevelInteractive { + t.Fatalf("被抢占 2 次后有效级=%v,期望 L3", got) + } + base.PreemptCount = 99 + if got := effectiveLevel(base); got != LevelInteractive { + t.Fatalf("提升应封顶在 +2 档,实际 %v", got) + } + + // 封顶 L4:L3 任务被多次抢占也不会超过紧急级。 + high := &Task{Level: LevelInteractive, PreemptCount: 99} + if got := effectiveLevel(high); got != LevelCritical { + t.Fatalf("L3 提升后应封顶为 L4,实际 %v", got) + } +} + +func TestStarvation_CooldownBlocksImmediateRepreempt(t *testing.T) { + a := newPreemptAgent(t, newPreemptProvider()) + + low := &Task{ID: 1, Level: LevelBackground, EnqueuedAt: time.Now()} + a.sched.enqueue(low) + a.sched.nextRef() // running = low + + e1, _ := textEvent("qq", "第一次打断") + if !a.sched.requestPreempt(e1, LevelMessage) { + t.Fatal("L2 应能抢占 L1(首次)") + } + a.sched.suspend(low, a.newTaskFrame("x", a.stageCtxFromInput("x", "", ""))) + if low.PreemptCount != 1 { + t.Fatalf("PreemptCount=%d,期望 1", low.PreemptCount) + } + if low.LastPreemptAt.IsZero() { + t.Fatal("挂起必须记录 LastPreemptAt(冷却起点)") + } + + // 冷却期内:即使 L4 也不得再抢占。 + a.sched.mu.Lock() + a.sched.running = low + a.sched.mu.Unlock() + + e2, _ := textEvent("cli", "冷却期内的紧急打断") + if a.sched.requestPreempt(e2, LevelCritical) { + t.Fatal("抢占冷却期内不得再抢占") + } + if a.sched.preemptGrantedFor() { + t.Fatal("冷却期内不得 arm 让位信号") + } +} + +func TestStarvation_PromotionBlocksSameLevelPreempt(t *testing.T) { + a := newPreemptAgent(t, newPreemptProvider()) + + low := &Task{ID: 1, Level: LevelBackground, EnqueuedAt: time.Now()} + a.sched.enqueue(low) + a.sched.nextRef() + // 模拟「已被抢占过一次」:有效级 = L2。 + low.PreemptCount = 1 + low.LastPreemptAt = time.Now().Add(-time.Hour) // 冷却已过 + + e1, _ := textEvent("qq", "同级打断") + if a.sched.requestPreempt(e1, LevelMessage) { + t.Fatal("有效级 L2 时,L2 中断不得抢占(严格大于才抢占)") + } + + e2, _ := textEvent("cli", "更高级打断") + if !a.sched.requestPreempt(e2, LevelInteractive) { + t.Fatal("L3 应能抢占有效级 L2") + } + if !a.sched.preemptGrantedFor() { + t.Fatal("L3 > 有效 L2,应已 arm") + } +} + +// 选择函数必须用有效级:被抢占过的任务在排队时应当优先于同级/更低的任务。 +func TestStarvation_SelectionUsesEffectiveLevel(t *testing.T) { + base := time.Now() + promoted := &Task{ID: 1, Level: LevelBackground, PreemptCount: 2, EnqueuedAt: base} // 有效 L3 + normal := &Task{ID: 2, Level: LevelMessage, EnqueuedAt: base.Add(time.Second)} // L2 + + if !taskBefore(promoted, normal) { + t.Fatal("被抢占 2 次的 L1(有效 L3)应先于 L2 执行") + } + if taskBefore(normal, promoted) { + t.Fatal("选择函数不得只看基础级") + } + + // 提升不改变调度器自身的排序稳定性:同为有效级时按入队时刻。 + a := &Task{ID: 3, Level: LevelBackground, PreemptCount: 1, EnqueuedAt: base.Add(2 * time.Second)} // 有效 L2 + b := &Task{ID: 4, Level: LevelMessage, EnqueuedAt: base.Add(time.Second)} // L2,更早 + if !taskBefore(b, a) { + t.Fatal("同有效级时应先到先服务") + } +} diff --git a/internal/agent/core/task.go b/internal/agent/core/task.go index 7019f57..8c448a0 100644 --- a/internal/agent/core/task.go +++ b/internal/agent/core/task.go @@ -121,13 +121,12 @@ type TaskFrame struct { // 这些字段让帧覆盖 prepare → step… → finish 全生命周期:挂起发生在 run 段的 // 安全点,恢复后由 finish 段统一提交(context.Append + emitResponse + // emitMemoryCandidate),因此挂起不会重复提交。 - Evt *agentIO.InputEvent - CleanInput string - IsInterrupt bool - StartedAt time.Time - Terminal taskTerminal - Level Level - PreemptCount int + Evt *agentIO.InputEvent + CleanInput string + IsInterrupt bool + StartedAt time.Time + Terminal taskTerminal + Level Level // SeedMsgs 非空时,stepPrepare 不重建 system prompt / 记忆上下文, // 而是以它为前缀继续(D1=A:抢占式中断任务继承被打断任务的**只读前缀**)。 @@ -148,7 +147,7 @@ func (a *Agent) runTaskSteps(f *TaskFrame) stepOutcome { for i := 0; i < maxSteps; i++ { // 安全点:只在 step 之间检查让位。临界区(StepToolExec)不在此列, // 因为让位信号由 interruptLoop 置位、而本循环是唯一读帧者。 - if !a.inCriticalSection() && a.sched.preemptGrantedFor(f.Level) && a.sched.canSuspend() { + if !a.inCriticalSection() && a.sched.preemptGrantedFor() && a.sched.canSuspend() { return outcomeSuspended } switch a.step(f) {