feat(scheduler): M5 饥饿防护（抢占计数提升有效级 + 抢占冷却）

设计依据 docs/zh/input-scheduler-design.md §9、§11.5（G1/G2）。

- Task 增加 PreemptCount / LastPreemptAt；effectiveLevel(t) =
  min(L4, Level + min(PreemptCount, 2))：被抢占越多越“值钱”，
  逐步追上抢占它的流，但封顶 L4 因而抢不过真正的紧急输入
- 选择函数 taskBefore 改用有效级；requestPreempt 与 preemptGrantedFor
  同样以有效级比较
- 抢占冷却 preemptCooldown=2s：刚被抢占的任务期内不再被抢占，
  避免同一任务被反复打断到永不完结
- 新增 scheduler_starvation_test.go 4 项：提升与封顶、冷却期内不得再抢占、
  提升后同级不得抢占而更高可、选择函数确实用有效级
- 验收：agent 全量 + -race；全仓 build/vet 通过

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)

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 {

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")
 	}
 

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("同有效级时应先到先服务")
+	}
+}

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) {