feat(scheduler): M4 临界区显式化 + 清掉被 pendingInterrupts 取代的 interceptCh

设计依据 docs/zh/input-scheduler-design.md §4.3、§11.1（P5/P6）。

- 临界区语义显式化：让位检查**只在 step 之间**做，执行中的 step
  （工具 RPC / ONNX / CAS 落盘）天然不可抢占；_consolidation_ 整任务
  经 inCriticalSection() 判为不可抢占（它直接改图库）
- 删除 interceptCh 与 drainInterrupts：M3b 起中断一律走 pendingInterrupts，
  旧的「同行注入 + 三处 drain + 批次放弃」已无写入者，属死代码
- 新增 scheduler_critical_test.go 3 项：
  P5/P6 工具执行中 arm 了让位信号也不得挂起、必须等工具返回后的安全点；
  _consolidation_ 判为临界区；无抢占时同批工具必须全部执行（新语义回归）
- 验收：agent 全量 + -race；全仓 build/vet 通过

internal/agent/core/agent.go

@@ -110,9 +110,6 @@ type Agent struct {
 	// childSeq 给完成的任务排个序，用于有界淘汰。
 	childSeq int64
 
-	// 高优先级打断通道：interceptLoop 注入，process() 在工具循环轮次间非阻塞读取
-	interceptCh chan *agentIO.InputEvent
-
 	// 输入调度器：就绪队列、任务抽象与快照（见 scheduler.go）。
 	// M2 起取代 eventLoop 的隐式 channel 排队。
 	sched *scheduler
@@ -312,7 +309,6 @@ func New(cfg AgentConfig) *Agent {
 		eventBus:          cfg.EventBus,
 		selfInputCh:       make(chan selfInputMsg, 64),
 		childTasks:        make(map[string]*childTaskState),
-		interceptCh:       make(chan *agentIO.InputEvent, 64),
 		sched:             newScheduler(256),
 		priorityLookup:    cfg.PriorityLookup,
 		pluginHealth:      newPluginHealthTracker(),

internal/agent/core/eventloop.go

@@ -321,33 +321,6 @@ func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) {
 	a.runStage(sdk.StageAfterOutput, stageCtx)
 }
 
-func (a *Agent) drainInterrupts() []string {
-	var out []string
-	for {
-		select {
-		case evt := <-a.interceptCh:
-			if evt == nil {
-				continue
-			}
-			text, _ := evt.Payload["content"].(string)
-			if text == "" {
-				continue
-			}
-			source := evt.Source
-			if source == "" {
-				source = "unknown"
-			}
-			channel := evt.OutputChannel
-			if channel == "" {
-				channel = source
-			}
-			out = append(out, fmt.Sprintf("[打断消息][来源:%s][输出通道:%s] %s", source, channel, text))
-		default:
-			return out
-		}
-	}
-}
-
 // pruneOnInput 按声明的上下文策略裁剪上下文，返回归档的事件数。
 //
 // 默认**不裁剪**：ContextPolicy 必须在注入点（payload 的 context_policy）

internal/agent/core/scheduler_critical_test.go

@@ -0,0 +1,151 @@
+package core
+
+// M4 验收测试：临界区语义显式化 + 抢占延迟到安全点 + 批次不再被中断放弃。
+//
+// 设计依据 docs/zh/input-scheduler-design.md §4.3（临界区）、§11.1（P5/P6）。
+//
+// 关键结构事实：让位检查**只在 step 之间**进行，因此任何正在执行的 step
+// （工具 RPC、ONNX、CAS 落盘）天然不可抢占——中断只能等它返回。
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
+	agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
+	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
+)
+
+// P5/P6：工具执行期间到达的高优先级中断不得立即抢占；必须等工具返回后的安全点。
+func TestPreempt_DeferredDuringToolExec(t *testing.T) {
+	sh := NewStageHost()
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	var once sync.Once
+	sh.RegisterTool("t_slow", sdk.ToolDef{Name: "t_slow", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) {
+		once.Do(func() { close(entered) })
+		<-release
+		return "slow-done", nil
+	})
+
+	sp := &scriptProvider{script: []*agentAPI.CompletionResponse{
+		{Content: "", ToolCalls: []agentAPI.ToolCall{tc("c1", "t_slow")}}, // 低优先级任务调用慢工具
+		{Content: "low-done"},  // 恢复后收尾
+		{Content: "intr-done"}, // 中断任务
+	}}
+	a := New(AgentConfig{
+		ID:              "crit",
+		Provider:        sp,
+		ProviderManager: agentAPI.NewProviderManager(),
+		IO:              agentIO.NewIOManager(),
+		StageHost:       sh,
+	})
+
+	if _, _ = enqueueTask(t, a, LevelBackground, "qq", "低优先级任务"); true {
+	}
+	lt, _, _ := a.sched.nextRef()
+	done := make(chan struct{})
+	go func() { a.executeNewTask(lt); close(done) }()
+
+	select {
+	case <-entered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("慢工具未被调用")
+	}
+
+	// 工具执行中注入 L4 中断。
+	intrEvt, _ := textEvent("cli", "紧急打断")
+	intrEvt.Payload["interrupt"] = true
+	if !a.sched.requestPreempt(intrEvt, LevelCritical) {
+		t.Fatal("L4 应 arm 让位信号")
+	}
+	// 关键断言：信号已 arm，但任务仍在工具里 —— 绝不能挂起。
+	if !a.sched.preemptGrantedFor(LevelBackground) {
+		t.Fatal("让位信号应已 arm")
+	}
+	if a.DumpScheduler().Running == nil {
+		t.Fatal("工具执行中不得挂起（StepToolExec 是临界区）")
+	}
+	if len(a.DumpScheduler().SuspendPool) != 0 {
+		t.Fatal("工具执行中 suspendPool 应为空")
+	}
+
+	// 放行工具 → 工具返回后的安全点才挂起。
+	close(release)
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("工具返回后未挂起")
+	}
+
+	snap := a.DumpScheduler()
+	if len(snap.SuspendPool) != 1 {
+		t.Fatalf("工具返回后 suspendPool=%d，期望 1", len(snap.SuspendPool))
+	}
+	if snap.SuspendPool[0].Frame.Step != StepToolAfter {
+		t.Fatalf("应在工具执行后的安全点挂起（StepToolAfter），实际 %v", snap.SuspendPool[0].Frame.Step)
+	}
+	if len(snap.PendingInterrupts) != 1 {
+		t.Fatalf("中断请求不得丢失，pendingInterrupts=%d", len(snap.PendingInterrupts))
+	}
+}
+
+// _consolidation_ 整任务视为不可抢占（它直接改图库）。
+func TestCriticalSection_ConsolidationMarked(t *testing.T) {
+	a := newPreemptAgent(t, newPreemptProvider())
+
+	a.currentOutputChannel = "cli"
+	if a.inCriticalSection() {
+		t.Fatal("普通通道不应被判为临界区")
+	}
+	a.currentOutputChannel = channelConsolidation
+	if !a.inCriticalSection() {
+		t.Fatal("记忆整理必须是不可抢占临界区")
+	}
+}
+
+// 新语义：没有抢占时，同批的多个工具必须全部执行——不再有「中断放弃剩余批」。
+func TestBatch_NotAbandonedWithoutPreemption(t *testing.T) {
+	sh := NewStageHost()
+	var mu sync.Mutex
+	var ran []string
+	reg := func(name string) {
+		sh.RegisterTool(name, sdk.ToolDef{Name: name, Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) {
+			mu.Lock()
+			ran = append(ran, name)
+			mu.Unlock()
+			return name + "-out", nil
+		})
+	}
+	reg("t_a")
+	reg("t_b")
+
+	sp := &scriptProvider{script: []*agentAPI.CompletionResponse{
+		{Content: "", ToolCalls: []agentAPI.ToolCall{tc("c1", "t_a"), tc("c2", "t_b")}},
+		{Content: "全部完成"},
+	}}
+	a := New(AgentConfig{
+		ID:              "batch",
+		Provider:        sp,
+		ProviderManager: agentAPI.NewProviderManager(),
+		IO:              agentIO.NewIOManager(),
+		StageHost:       sh,
+	})
+
+	if _, _ = enqueueTask(t, a, LevelBackground, "cli", "跑两个工具"); true {
+	}
+	tt, _, _ := a.sched.nextRef()
+	a.executeNewTask(tt)
+
+	if len(ran) != 2 || ran[0] != "t_a" || ran[1] != "t_b" {
+		t.Fatalf("同批工具应全部按序执行，实际 %v", ran)
+	}
+	snap := a.DumpScheduler()
+	if len(snap.SuspendPool) != 0 || len(snap.PendingInterrupts) != 0 {
+		t.Fatalf("无抢占时不应有挂起或待处理中断：%+v", snap)
+	}
+	if snap.Stats.Executed != 1 {
+		t.Fatalf("Executed=%d，期望 1", snap.Stats.Executed)
+	}
+}

internal/agent/core/task.go

@@ -455,13 +455,6 @@ func (a *Agent) stepPrepare(f *TaskFrame) stepOutcome {
 // 取消（context.Canceled 且 agent 未退出）时**留在本 step 并 Turn++**——等价于
 // 原实现的 `continue`：重新排空中断、补占位、重新请求。抢占挂起将在 M3 从这里接管。
 func (a *Agent) stepLLM(f *TaskFrame) stepOutcome {
-	for _, interrupt := range a.drainInterrupts() {
-		f.Msgs = append(f.Msgs, agentAPI.Message{
-			Role:    "system",
-			Content: "[中断消息] " + interrupt,
-		})
-	}
-
 	// zen 兼容网关要求请求的最后一条消息必须是 user(thinking 续写模式校验),
 	// 工具轮产出的 tool/assistant 消息作结尾会被 400 拒绝,故补一条 user 占位。
 	f.Msgs = dropContinuationPlaceholders(f.Msgs)
@@ -569,22 +562,6 @@ func (a *Agent) stepToolBegin(f *TaskFrame) stepOutcome {
 	}
 	tc := f.PendingTools[f.ToolIdx]
 
-	if len(a.interceptCh) > 0 {
-		for _, interrupt := range a.drainInterrupts() {
-			f.Msgs = append(f.Msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt})
-		}
-		a.publishEvent(events.EventToolCall, map[string]interface{}{
-			"tool":    tc.Name,
-			"plugin":  a.resolveToolPlugin(tc.Name),
-			"args":    tc.Arguments,
-			"status":  "interrupted",
-			"reason":  "user interrupt before execution",
-			"channel": a.currentOutputChannel,
-		})
-		f.Step = StepTurnEnd
-		return outcomeContinue
-	}
-
 	f.ToolsUsed = append(f.ToolsUsed, tc.Name)
 	pluginName := a.resolveToolPlugin(tc.Name)
 	log.Printf("[agent] executing tool: %s (plugin=%s, id=%s)", tc.Name, pluginName, tc.ID)
@@ -744,13 +721,6 @@ func (a *Agent) stepToolAfter(f *TaskFrame) stepOutcome {
 	})
 
 	f.ToolIdx++
-	if len(a.interceptCh) > 0 {
-		for _, interrupt := range a.drainInterrupts() {
-			f.Msgs = append(f.Msgs, agentAPI.Message{Role: "system", Content: "[中断消息] " + interrupt})
-		}
-		f.Step = StepTurnEnd
-		return outcomeContinue
-	}
 	f.Step = StepToolBegin
 	return outcomeContinue
 }

internal/agent/core/task_test.go

@@ -152,54 +152,6 @@ func TestTaskFrame_X3_TerminatesWithinBudget(t *testing.T) {
 	}
 }
 
-// 中断（interceptCh）在工具批中途到达时：本批**剩余工具被放弃**，直接进入下一轮。
-//
-// 这是原实现的 `break` 语义（process.go 旧版工具循环尾部），必须保持。
-func TestTaskFrame_InterruptAbandonsRemainingBatch(t *testing.T) {
-	sh := NewStageHost()
-	var a *Agent
-	sh.RegisterTool("t_first", sdk.ToolDef{Name: "t_first", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) {
-		// 工具执行期间产生一次中断（模拟插件在工具里注入打断）。
-		a.interceptCh <- &agentIO.InputEvent{Source: "t", Payload: map[string]interface{}{"content": "新的用户输入"}}
-		return "first-out", nil
-	})
-	sh.RegisterTool("t_second", sdk.ToolDef{Name: "t_second", Plugin: "t"}, func(args map[string]interface{}) (interface{}, error) {
-		t.Fatal("批内第二个工具不应被执行：中断必须放弃剩余批次")
-		return nil, nil
-	})
-
-	sp := &scriptProvider{script: []*agentAPI.CompletionResponse{
-		{Content: "", ToolCalls: []agentAPI.ToolCall{tc("c1", "t_first"), tc("c2", "t_second")}},
-		{Content: "处理完中断后的答复"},
-	}}
-	a = newTaskTestAgent(t, sp, sh)
-
-	resp, toolsUsed, toolResults, err := a.process("开始", a.stageCtxFromInput("开始", "", ""))
-	if err != nil {
-		t.Fatalf("process 返回错误: %v", err)
-	}
-	if resp != "处理完中断后的答复" {
-		t.Fatalf("响应=%q", resp)
-	}
-	if len(toolsUsed) != 1 || toolsUsed[0] != "t_first" {
-		t.Fatalf("toolsUsed=%v，期望只有 t_first", toolsUsed)
-	}
-	if len(toolResults) != 1 || toolResults[0].Output != "first-out" {
-		t.Fatalf("toolResults=%+v，期望只有 t_first 的结果", toolResults)
-	}
-
-	// 第二轮请求里必须出现被打断内容（system 角色、[中断消息] 前缀）。
-	found := false
-	for _, m := range sp.reqs[1].Messages {
-		if m.Role == "system" && len(m.Content) > 0 && m.Content[0:len("[中断消息]")] == "[中断消息]" {
-			found = true
-		}
-	}
-	if !found {
-		t.Fatalf("第二轮请求缺少 [中断消息] system 消息")
-	}
-}
-
 // 状态机对未知 step 必须失败退出而不是空转。
 func TestTaskFrame_UnknownStepFails(t *testing.T) {
 	sp := &scriptProvider{}