feat(scheduler): M3a+M3b 任务生命周期重构 + 四级优先级抢占

设计依据 docs/zh/input-scheduler-design.md §3–§8、§14。

M3a（行为等价的所有权重构）：
- processInput 拆为 prepareInputTask / runTaskSteps / finishInputTask，
  帧覆盖 prepare→step…→finish；提交与回执只在 finish 段发生一次，
  为安全点挂起做准备（挂起不重复提交）
- process() 不再持 a.mu（挂起不能持锁），a.mu 字段随之移除
- TaskFrame 增加任务层现场（Evt/CleanInput/IsInterrupt/StartedAt/Terminal/
  Level/SeedMsgs）与 taskTerminal / outcomeSuspended
- 新增 task_lifecycle_test.go 5 项：正常恰好一次终态、去重 skipped、
  on_input 短路、错误终态、consolidation 路由

M3b（优先级与抢占）：
- interceptLoop 重写：只做「收中断 → 定级 → requestPreempt → 必要时取消
  LLM」，绝不触碰帧（不变量 I2）；三条降级路径与 interceptCh 兜底退场，
  改为统一的 pendingInterrupts
- scheduler：pendingInterrupts / suspendPool / 让位信号，nextRef 在三集合上
  按统一排序键取值；深度上限 4（canSuspend 在安全点拦下）
- 抢占判据 incoming.level > running.level；相等与更低只入队
- 安全点只在 step 之间；执行中的 step（工具 RPC/ONNX/CAS）天然不可抢占；
  _consolidation_ 整任务视为临界区
- 恢复走 resumeTask：从 frame.Step 继续，不重跑 prepare
- D1=A：suspend 把被打断任务的只读前缀交给抢占比它的中断任务（SeedMsgs）
- 新增 scheduler_preempt_test.go 5 项：抢占-挂起-恢复（含 R1/R5）、同级更低
  不抢占、深度上限、空闲中断不丢、seed 路径不污染标志位

验收：agent 全量 + -race 通过；全仓 build/vet 通过

internal/agent/core/agent.go

@@ -26,8 +26,12 @@ import (
 // ContextEvent 和 RelevanceContext 定义在 context.go
 
 // Agent — 单 agent，不区分会话/实例
+//
+// 并发现状（M3a 起）：所有任务状态只由 **schedulerLoop goroutine** 独占读写，
+// 因此不再有保护整轮执行的互斥量——挂起不能持锁（见 docs/zh/input-scheduler-design.md §8.1 I3）。
+// 仍需跨 goroutine 保护的是：childMu/llmMu/lastInputMu/noMergeMu 与各子系统自己的锁；
+// interceptLoop 只允许触碰 preemptionRequest 与 cancelLLM（经 llmMu）。
 type Agent struct {
-	mu              sync.Mutex
 	id              types.AgentID
 	provider        agentAPI.Provider
 	providerManager *agentAPI.ProviderManager
@@ -113,6 +117,9 @@ type Agent struct {
 	// M2 起取代 eventLoop 的隐式 channel 排队。
 	sched *scheduler
 
+	// 优先级策略表（可空）；见 AgentConfig.PriorityLookup。
+	priorityLookup func(source, channel string) Level
+
 	// 进行中的 LLM 请求取消函数，interceptLoop 可调用以在请求中打断
 	cancelLLM context.CancelFunc
 	llmMu     sync.Mutex
@@ -132,7 +139,7 @@ type Agent struct {
 	//
 	// 需要缓存而不是当场挂到事件上：媒体在 process() 执行期间被捕获，
 	// 而承载它的 ContextEvent 要等 process() 返回后才 Append——此刻还没有 owner_id。
-	// 与 pendingMedia 同受 a.mu 保护。
+	// 由 schedulerLoop goroutine 独占读写。
 	pendingMediaDigests []string
 
 	// 当前输入是否为工具提醒/中断（以 system 角色注入，避免被当成用户消息）
@@ -219,6 +226,10 @@ type AgentConfig struct {
 	SkillIndexProvider SkillIndexProvider
 
 	InputProcessing types.InputProcessingConfig // 非文本输入处理配置
+
+	// PriorityLookup 是内核的优先级策略表（设计文档 §3.2）。
+	// 返回 L1..L4；返回 0 或越界值表示“无策略”，由 Agent 的通道名兜底决定。
+	PriorityLookup func(source, channel string) Level
 }
 
 func New(cfg AgentConfig) *Agent {
@@ -303,6 +314,7 @@ func New(cfg AgentConfig) *Agent {
 		childTasks:        make(map[string]*childTaskState),
 		interceptCh:       make(chan *agentIO.InputEvent, 64),
 		sched:             newScheduler(256),
+		priorityLookup:    cfg.PriorityLookup,
 		pluginHealth:      newPluginHealthTracker(),
 		thinkingEnabled:   cfg.ThinkingEnabled,
 		inputCfg:          cfg.InputProcessing,

internal/agent/core/eventloop.go

@@ -50,43 +50,12 @@ func (a *Agent) interceptLoop() {
 			clone.Payload["interrupt_source"] = evt.Source
 			clone.Payload["interrupt_channel"] = evt.OutputChannel
 
-			a.llmMu.Lock()
-			hasActiveLLM := a.cancelLLM != nil
-			if hasActiveLLM {
-				a.cancelLLM()
-				log.Printf("[agent] LLM request cancelled by interrupt")
-			}
-			a.llmMu.Unlock()
-
-			if hasActiveLLM {
-				if a.currentOutputChannel == "_consolidation_" {
-					log.Printf("[agent] consolidation interrupted, re-injecting input for %s/%s", evt.Source, evt.OutputChannel)
-					a.io.InjectInputTo(evt.Source, evt.OutputChannel, "text", map[string]interface{}{
-						"content":           text,
-						"interrupt":         true,
-						"interrupt_source":  evt.Source,
-						"interrupt_channel": evt.OutputChannel,
-					})
-				} else {
-					select {
-					case a.interceptCh <- clone:
-					default:
-						log.Printf("[agent] intercept channel full, queuing input for %s", evt.Source)
-						a.io.InjectInputTo(evt.Source, evt.OutputChannel, "text", map[string]interface{}{
-							"content":           text,
-							"interrupt":         true,
-							"interrupt_source":  evt.Source,
-							"interrupt_channel": evt.OutputChannel,
-						})
-					}
-				}
-			} else {
-				a.io.InjectInputTo(evt.Source, evt.OutputChannel, "text", map[string]interface{}{
-					"content":           text,
-					"interrupt":         true,
-					"interrupt_source":  evt.Source,
-					"interrupt_channel": evt.OutputChannel,
-				})
+			// 决策交给调度器：requestPreempt 总是登记中断（进 pendingInterrupts，
+			// 因而不会丢），仅当它会真抢占时才告诉我“该取消可取消的步骤”。
+			// 本 goroutine 不碰任何帧——只写 pendingInterrupts 与让位信号。
+			level := a.taskLevel(evt.Source, evt.OutputChannel)
+			if a.sched.requestPreempt(clone, level) {
+				a.cancelCurrentLLM()
 			}
 
 		case <-a.ctx.Done():
@@ -95,6 +64,20 @@ func (a *Agent) interceptLoop() {
 	}
 }
 
+// cancelCurrentLLM 取消正在进行的 LLM 请求（若有）。
+//
+// 只有 LLM 流式步骤是可取消的；工具 RPC / ONNX / CAS 在 v1 是临界区，
+// 取消对它们无效——让位信号会等它们自然结束后的安全点（设计文档 D2）。
+func (a *Agent) cancelCurrentLLM() {
+	a.llmMu.Lock()
+	cancel := a.cancelLLM
+	a.llmMu.Unlock()
+	if cancel != nil {
+		cancel()
+		log.Printf("[agent] LLM request cancelled by preemption")
+	}
+}
+
 // channelConsolidation 标记记忆整理类自输入：无记忆路径处理，
 // 不写入对话上下文、不向任何输出通道 emit 响应。
 const channelConsolidation = "_consolidation_"
@@ -108,22 +91,27 @@ type selfInputMsg struct {
 	channel string
 }
 
-func (a *Agent) handleSelfInput(msg selfInputMsg) {
+// selfEvent 把内核自循环消息归一成输入事件。
+func selfEvent(msg selfInputMsg) *agentIO.InputEvent {
 	if msg.channel == "" {
 		msg.channel = channelConsolidation // 兼容空值：默认走整理路径
 	}
-	a.processInput(&agentIO.InputEvent{
+	return &agentIO.InputEvent{
 		Source:        "system",
 		Type:          "text",
 		Payload:       map[string]interface{}{"content": msg.text},
 		OutputChannel: msg.channel,
-	})
+	}
+}
+
+func (a *Agent) handleSelfInput(msg selfInputMsg) {
+	_, _ = a.runInputTask(selfEvent(msg), nil)
 }
 
 func (a *Agent) handleInput(evt *agentIO.InputEvent) {
 	switch evt.Type {
 	case "text", "image", "audio":
-		a.processInput(evt)
+		_, _ = a.runInputTask(evt, nil)
 
 	case "event":
 		log.Printf("[agent] event from %s: %v", evt.Source, evt.Payload)
@@ -283,155 +271,6 @@ func (a *Agent) mediaToBlocks(payload map[string]interface{}, mediaType string,
 	return blocks, alt
 }
 
-// processInput 是全部模态输入的唯一主干。
-//
-// 文本、用户上传的图/音频、插件注入的多模态块走同一条路径，因此去重、
-// no_memory、通道 Cleaner、中断语义、EventRawInput、媒体入 CAS、媒体记忆绑定
-// 对所有模态一致——不会再出现「文本路径加了功能、媒体路径没跟上」。
-func (a *Agent) processInput(evt *agentIO.InputEvent) {
-	start := time.Now()
-
-	in, ok := a.resolveInput(evt)
-	if !ok {
-		return
-	}
-
-	// 去重按文本做：webui/GUI 断线重连会重放未确认消息。
-	// 带媒体时跳过——媒体输入的 alt 文案（"[从 qq 收到了 image]"）对不同图片
-	// 是同一句，拿它去重会把连发的两张图误判成重复。
-	if len(in.blocks) == 0 && a.isDuplicateInput(evt.Source, in.text) {
-		log.Printf("[agent] dropped duplicate input from %s: %s", evt.Source, truncateStr(in.text, 60))
-		return
-	}
-
-	a.currentOutputChannel = evt.OutputChannel
-	if a.currentOutputChannel == "" {
-		a.currentOutputChannel = evt.Source
-	}
-
-	if evt.OutputChannel == "_consolidation_" {
-		a.processConsolidation(evt, in.text)
-		return
-	}
-
-	// pendingMedia 让 describe_image / transcribe_audio / ocr_image 拿到本轮媒体的
-	// 原始 data/url，也是这三个工具是否出现在工具表里的开关。仅对用户直接上传成立
-	//（payload 里才有 data/url）；插件注入的是成品 block，取不到原始数据。
-	if evt.Type == "image" || evt.Type == "audio" {
-		a.pendingMedia = evt.Payload
-		defer func() { a.pendingMedia = nil }()
-	}
-
-	// 媒体先落进 CAS。不存的后果是 ContextEvent.Input 只剩一句 alt 文本，
-	// base64 随 message 数组发给模型后就丢了。
-	if len(in.blocks) > 0 {
-		a.stageMediaDigests(a.captureBlockMedia(in.blocks, in.captureTool)...)
-	}
-
-	noMemory := false
-	if v, ok := evt.Payload["no_memory"].(bool); ok {
-		noMemory = v
-	}
-	if !noMemory && a.io != nil {
-		if chDef, ok := a.io.GetInputChannelDef(evt.Source); ok && chDef.NoMemory {
-			noMemory = true
-		}
-	}
-
-	// 工具提醒/中断（terminal_watch、timer 等）不是用户发言：
-	// 以 system 角色注入 LLM，且不写入用户对话履历。
-	isInterrupt, _ := evt.Payload["interrupt"].(bool)
-	a.mu.Lock()
-	a.interruptInput = isInterrupt
-	a.mu.Unlock()
-	if isInterrupt {
-		noMemory = true
-	}
-
-	stageCtx := a.stageCtxFromInput(in.text, evt.Source, "")
-	stageCtx.Extra["input_source"] = evt.Source
-	stageCtx.Extra["output_channel"] = evt.OutputChannel
-	if len(in.blocks) > 0 {
-		stageCtx.Extra["media_blocks"] = in.blocks
-		stageCtx.Extra["media_type"] = in.mediaType
-	}
-	if noMemory {
-		stageCtx.NoMemory = true
-	}
-	a.injectSourceContext(stageCtx, evt)
-
-	if a.runStage(sdk.StageOnInput, stageCtx) {
-		a.emitResponse(evt, *stageCtx.Response)
-		return
-	}
-
-	input := stageCtx.RawMessage
-
-	// 计算层用的清洗文本（不改原文）：通道 Cleaner 提取语义内容后用于向量化/提关键词
-	cleanInput := input
-	if a.io != nil {
-		if chDef, ok := a.io.GetInputChannelDef(evt.Source); ok && chDef.Cleaner != nil {
-			cleanInput = chDef.Cleaner(input)
-		}
-	}
-
-	// upload_* 字段一并转发：webui 的 EventRawInput 订阅方靠它们还原附件卡片。
-	// 媒体路径此前把整个 payload 塞进 content（一个 map），订阅方按 string 断言
-	// 直接失败 → 用户发的图从不出现在聊天记录里。
-	rawPayload := map[string]interface{}{"content": input, "source": evt.Source}
-	for _, k := range []string{"upload_url", "upload_type", "upload_size", "upload_name"} {
-		if v, ok := evt.Payload[k]; ok {
-			rawPayload[k] = v
-		}
-	}
-	a.publishEvent(events.EventRawInput, rawPayload)
-
-	archived := a.pruneOnInput(evt, cleanInput)
-	if archived > 0 {
-		log.Printf("[agent] pruned %d low-relevance events to document memory", archived)
-	}
-
-	if !isInterrupt {
-		a.context.Append(ContextEvent{
-			Timestamp: start,
-			Source:    evt.Source,
-			Input:     input,
-		})
-	}
-
-	response, toolsUsed, toolResults, err := a.process(input, stageCtx)
-	if err != nil {
-		log.Printf("[agent] process %s error: %v", evt.Type, err)
-		resp := fmt.Sprintf("处理错误: %v", err)
-		a.emitResponse(evt, resp)
-		a.context.Append(ContextEvent{Timestamp: time.Now(), Source: "agent", Input: input, Response: resp})
-		return
-	}
-
-	elapsed := time.Since(start)
-	log.Printf("[agent] %s from %s → response (%dms, tools=%v)", evt.Type, evt.Source, elapsed.Milliseconds(), toolsUsed)
-
-	// 本轮捕获的媒体一起挂到这条事件上：用户上传的、插件注入的，以及模型调
-	// multimodal_see_picture / see_video 时经 SetToolBlocks 注入的（后者在
-	// process() 里被捕获，纯文本输入也会有）。
-	turnEvt := ContextEvent{
-		Timestamp:   time.Now(),
-		Source:      "agent",
-		Input:       cleanInput,
-		Response:    response,
-		ToolsUsed:   toolsUsed,
-		ToolResults: toolResults,
-	}
-	a.bindEventMedia(&turnEvt, a.drainMediaDigests())
-	a.context.Append(turnEvt)
-
-	a.emitResponse(evt, response)
-
-	if !stageCtx.NoMemory {
-		a.emitMemoryCandidate(evt.Source, cleanInput, response, toolResults, toolsUsed)
-	}
-}
-
 func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) {
 	stageCtx := &sdk.StageContext{
 		FinalText: response,

internal/agent/core/mediaref.go

@@ -138,7 +138,7 @@ func (a *Agent) embedMediaOnIngest(digest, mime string, data []byte) {
 //
 // 为何要缓存而不是当场建块：媒体在 process() 执行期间被捕获，而承载它的
 // ContextEvent 要等 process() 返回后才 Append——此刻还没有 owner_id。
-// 与既有的 a.pendingMedia 同一手法（都在 a.mu 保护下）。
+// 与既有的 a.pendingMedia 同一手法（均由 schedulerLoop goroutine 独占读写）。
 func (a *Agent) stageMediaDigests(digests ...string) {
 	if len(digests) == 0 {
 		return
@@ -197,7 +197,7 @@ func mediaLabel(it *media.Item) string {
 //
 // 沿用 document.Store 的 doc_<unixnano> 手法（同一份代码库里保持一致，
 // 也避免为此引入 uuid 依赖）。纳秒精度足够：同一 Agent 的事件由
-// a.mu 串行化 Append，不存在同纳秒两条。
+// schedulerLoop 单 goroutine 串行 Append，不存在同纳秒两条。
 func newEventID() string {
 	return fmt.Sprintf("evt_%d", time.Now().UnixNano())
 }

internal/agent/core/scheduler.go

@@ -23,6 +23,7 @@ import (
 	"sync"
 	"time"
 
+	agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
 	agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
 )
 
@@ -84,6 +85,10 @@ type Task struct {
 
 	Event *agentIO.InputEvent // Kind == TaskKindInput
 	Self  selfInputMsg        // Kind == TaskKindSelf
+
+	// SeedMsgs 是抢占式中断任务的只读前缀（D1=A）：由被打断的任务在挂起时
+	// 附上，使中断任务看得见「进行到哪一步」，但其产出不合并回原任务。
+	SeedMsgs []agentAPI.Message
 }
 
 // SchedulerStats 是调度器的累计计数（可观测性，设计文档 §11 O2）。
@@ -97,9 +102,11 @@ type SchedulerStats struct {
 
 // SchedulerSnapshot 是调度器的原子快照。
 type SchedulerSnapshot struct {
-	Running *Task
-	Queue   []*Task
-	Stats   SchedulerStats
+	Running           *Task
+	Queue             []*Task
+	PendingInterrupts []*Task
+	SuspendPool       []*suspendedTask
+	Stats             SchedulerStats
 }
 
 type scheduler struct {
@@ -109,13 +116,41 @@ type scheduler struct {
 	seq      uint64
 	stats    SchedulerStats
 	maxQueue int
+
+	// pendingInterrupts：因优先级不足（或运行任务在临界区）而未立即抢占的中断请求。
+	// 与 readyQueue 分离：取出时以中断语义启动（设计文档 D3）。
+	pendingInterrupts []*Task
+	// suspendPool：被抢占后保存了现场、等待恢复的任务（**不是栈**，按优先级取）。
+	suspendPool []*suspendedTask
+	// preemptArmed/preemptLevel：运行任务的“让位信号”。
+	// interruptLoop 只写这两个字段与 pendingInterrupts；帧永远只由调度器读写。
+	preemptArmed bool
+	preemptLevel Level
+	// maxSuspendDepth：suspendPool 深度上限（设计文档 §6.3，默认 4）。
+	maxSuspendDepth int
 }
 
+// suspendedTask 是一个被抢占任务的现场。
+type suspendedTask struct {
+	Task  *Task
+	Frame *TaskFrame
+}
+
+// nextSelection 标识 nextRef 从哪个集合取出任务。
+type nextSelection int
+
+const (
+	nextNone nextSelection = iota
+	nextReady
+	nextPending
+	nextSuspended
+)
+
 func newScheduler(maxQueue int) *scheduler {
 	if maxQueue <= 0 {
 		maxQueue = 256
 	}
-	return &scheduler{maxQueue: maxQueue}
+	return &scheduler{maxQueue: maxQueue, maxSuspendDepth: 4}
 }
 
 // hasRoom 报告就绪队列是否还能接收任务。泵入侧据此节流：
@@ -145,17 +180,164 @@ func (s *scheduler) enqueue(t *Task) bool {
 }
 
 // next 取出下一个要执行的任务；队列空返回 nil。
+//
+// 保留该签名供已有测试使用；调度器自用 nextRef（需要区分是否携带现场）。
 func (s *scheduler) next() *Task {
+	t, _, _ := s.nextRef()
+	return t
+}
+
+// nextRef 从三个集合中按统一排序键取出下一个任务。
+//
+// 设计文档 §4.1：高有效级先；同级先到先服务。挂起任务保留其**原始**入队时刻，
+// 因此同级时天然倾向“先把旧任务做完”，抑制饥饿。
+func (s *scheduler) nextRef() (*Task, *TaskFrame, nextSelection) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	if len(s.queue) == 0 {
-		return nil
+
+	var bestTask *Task
+	var bestFrame *TaskFrame
+	bestKind := nextNone
+	consider := func(t *Task, k nextSelection, fr *TaskFrame) {
+		if bestTask == nil || taskBefore(t, bestTask) {
+			bestTask, bestKind, bestFrame = t, k, fr
+		}
 	}
-	i := pickTaskIndex(s.queue)
-	t := s.queue[i]
-	s.queue = append(s.queue[:i], s.queue[i+1:]...)
-	s.running = t
-	return t
+	for _, t := range s.queue {
+		consider(t, nextReady, nil)
+	}
+	for _, t := range s.pendingInterrupts {
+		consider(t, nextPending, nil)
+	}
+	for _, st := range s.suspendPool {
+		consider(st.Task, nextSuspended, st.Frame)
+	}
+	if bestTask == nil {
+		return nil, nil, nextNone
+	}
+
+	switch bestKind {
+	case nextReady:
+		s.queue = removeTask(s.queue, bestTask)
+	case nextPending:
+		s.pendingInterrupts = removeTask(s.pendingInterrupts, bestTask)
+	case nextSuspended:
+		for i, st := range s.suspendPool {
+			if st.Task == bestTask {
+				s.suspendPool = append(s.suspendPool[:i], s.suspendPool[i+1:]...)
+				break
+			}
+		}
+	}
+	s.running = bestTask
+	return bestTask, bestFrame, bestKind
+}
+
+func removeTask(list []*Task, target *Task) []*Task {
+	for i, t := range list {
+		if t == target {
+			return append(list[:i], list[i+1:]...)
+		}
+	}
+	return list
+}
+
+// enqueueInterrupt 把一个未立即抢占的中断请求放进 pendingInterrupts。
+//
+// 有界：满了丢**最老**的一条并计数（中断是提示性输入，宁可丢旧保新）。
+func (s *scheduler) enqueueInterrupt(t *Task) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.seq++
+	t.ID = s.seq
+	if t.EnqueuedAt.IsZero() {
+		t.EnqueuedAt = time.Now()
+	}
+	if len(s.pendingInterrupts) >= s.maxQueue {
+		s.pendingInterrupts = s.pendingInterrupts[1:]
+		s.stats.Rejected++
+	}
+	s.pendingInterrupts = append(s.pendingInterrupts, t)
+}
+
+// requestPreempt 登记一次中断请求。
+//
+// 返回 true 表示“应该尝试取消运行任务正在进行的可取消步骤（LLM 流式）”。
+//
+// 无论能否抢占，中断请求都进 pendingInterrupts——这样即使运行任务在抢占生效前
+// 就正常结束，中断也不会丢（它会被 nextRef 按优先级选出）。
+func (s *scheduler) requestPreempt(evt *agentIO.InputEvent, level Level) bool {
+	s.mu.Lock()
+	running := s.running
+	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
+}
+
+// preemptGrantedFor 报告级别为 level 的运行任务是否应在当前安全点让位。
+func (s *scheduler) preemptGrantedFor(level Level) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.preemptArmed && s.preemptLevel > level
+}
+
+func (s *scheduler) clearPreempt() {
+	s.mu.Lock()
+	s.preemptArmed = false
+	s.preemptLevel = 0
+	s.mu.Unlock()
+}
+
+// suspend 保存现场。
+//
+// 深度上限（设计文档 §6.3）：安全点上的 canSuspend 已提前拦下超限情况，
+// 此处仅在竞态下兜底计数——绝不丢弃帧（帧丢了会丢副作用记录）。
+func (s *scheduler) suspend(t *Task, f *TaskFrame) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if len(s.suspendPool) >= s.maxSuspendDepth {
+		s.stats.Rejected++
+	}
+	s.suspendPool = append(s.suspendPool, &suspendedTask{Task: t, Frame: f})
+	if s.running == t {
+		s.running = nil
+	}
+
+	// D1=A：把被抢占任务的只读前缀交给造成本次抢占的中断任务。
+	// 选最高优先级的待处理中断；若它已有前缀（嵌套抢占）则不覆盖。
+	if s.preemptArmed {
+		var victim *Task
+		for _, it := range s.pendingInterrupts {
+			if it.Level < s.preemptLevel {
+				continue
+			}
+			if victim == nil || taskBefore(victim, it) {
+				victim = it
+			}
+		}
+		if victim != nil && len(victim.SeedMsgs) == 0 {
+			victim.SeedMsgs = append([]agentAPI.Message(nil), f.Msgs...)
+		}
+	}
+
+	s.preemptArmed = false
+	s.preemptLevel = 0
+}
+
+// canSuspend 报告还有下潜余量（安全点用它决定是否真的让位）。
+func (s *scheduler) canSuspend() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return len(s.suspendPool) < s.maxSuspendDepth
 }
 
 // done 标记任务执行结束。
@@ -165,9 +347,59 @@ func (s *scheduler) done(t *Task) {
 	if s.running == t {
 		s.running = nil
 	}
+	// 任务正常结束：让位信号不再有意义（中断已在 pendingInterrupts 里）。
+	s.preemptArmed = false
+	s.preemptLevel = 0
 	s.stats.Executed++
 }
 
+// currentLevel 返回当前正在执行任务的级别；无 running 时为默认级。
+//
+// 用于在 prepare 段把级别写进帧（抢占比较的基准）。
+func (s *scheduler) currentLevel() Level {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.running != nil {
+		return s.running.Level
+	}
+	return DefaultLevel
+}
+
+// taskLevel 是内核的优先级策略：四级的来源（设计文档 §3.2）。
+//
+// 优先走注入的查找函数（配置表）；未命中则用通道名兜底：
+// cli/webui/http 为人机交互（L3），system/_consolidation_ 为后台（L1），
+// 其余一律默认级（L1）。显式才是特权：没有策略就不给抢占权。
+func (a *Agent) taskLevel(source, channel string) Level {
+	if a.priorityLookup != nil {
+		if l := a.priorityLookup(source, channel); l >= LevelBackground && l <= LevelCritical {
+			return l
+		}
+	}
+	switch channel {
+	case "cli", "webui", "http":
+		return LevelInteractive
+	case channelConsolidation, "system":
+		return LevelBackground
+	}
+	switch source {
+	case "cli", "webui":
+		return LevelInteractive
+	case "system":
+		return LevelBackground
+	}
+	return DefaultLevel
+}
+
+// inCriticalSection 报告运行任务是否处于不可抢占区。
+//
+// M3b 只处理「整个任务不可抢占」的情形（记忆整理）。工具执行、ONNX、
+// CAS 落盘属于**单步**临界区——它们由「只在 step 之间检查让位」天然保护，
+// 不需要在这里列（M4 会把清单显式化）。
+func (a *Agent) inCriticalSection() bool {
+	return a.currentOutputChannel == channelConsolidation
+}
+
 // pickTaskIndex 返回下一个要执行的任务下标（设计文档 §4.1 的选择函数）。
 //
 // 排序键：优先级降序 → 入队时刻升序 → ID 升序。
@@ -197,6 +429,11 @@ func newInputTask(evt *agentIO.InputEvent) *Task {
 	return &Task{Kind: TaskKindInput, Level: DefaultLevel, Event: evt, EnqueuedAt: time.Now()}
 }
 
+// newInterruptTask 把一个中断请求包装成任务。
+func newInterruptTask(evt *agentIO.InputEvent, level Level) *Task {
+	return &Task{Kind: TaskKindInput, Level: level, Event: evt, EnqueuedAt: time.Now()}
+}
+
 func newSelfTask(msg selfInputMsg) *Task {
 	return &Task{Kind: TaskKindSelf, Level: DefaultLevel, Self: msg, EnqueuedAt: time.Now()}
 }
@@ -210,6 +447,8 @@ func (a *Agent) DumpScheduler() SchedulerSnapshot {
 	defer a.sched.mu.Unlock()
 	snap := SchedulerSnapshot{Running: a.sched.running, Stats: a.sched.stats}
 	snap.Queue = append(snap.Queue, a.sched.queue...)
+	snap.PendingInterrupts = append(snap.PendingInterrupts, a.sched.pendingInterrupts...)
+	snap.SuspendPool = append(snap.SuspendPool, a.sched.suspendPool...)
 	return snap
 }
 
@@ -226,9 +465,9 @@ func (a *Agent) schedulerLoop() {
 	for {
 		a.pumpInbox()
 
-		t := a.sched.next()
-		if t == nil {
-			// 就绪队列空：阻塞等新输入或退出。
+		t, f, kind := a.sched.nextRef()
+		if kind == nextNone {
+			// 无待办：阻塞等新输入或退出。
 			select {
 			case evt := <-a.io.InputChan():
 				a.sched.enqueue(newInputTask(evt))
@@ -239,7 +478,12 @@ func (a *Agent) schedulerLoop() {
 			}
 			continue
 		}
-		a.executeTask(t)
+
+		if kind == nextSuspended {
+			a.resumeTask(t, f)
+			continue
+		}
+		a.executeNewTask(t)
 	}
 }
 
@@ -264,11 +508,19 @@ func (a *Agent) pumpInbox() {
 	}
 }
 
-// executeTask 执行一个任务，并做**任务级 panic 隔离**（不变量 I6）。
+// executeTask 执行一个任务（测试与旧调用方的入口）；见 executeNewTask。
+func (a *Agent) executeTask(t *Task) {
+	a.executeNewTask(t)
+}
+
+// executeNewTask 执行一个**新建**任务，并做任务级 panic 隔离（不变量 I6）。
 //
 // 与改造前的差异（有意）：原 eventLoop 在 panic 后重启整个循环，
 // 现在一个任务的 panic 只丢弃该任务，调度器与其它任务不受影响。
-func (a *Agent) executeTask(t *Task) {
+func (a *Agent) executeNewTask(t *Task) {
+	var f *TaskFrame
+	var out stepOutcome = outcomeDone
+
 	func() {
 		defer func() {
 			if r := recover(); r != nil {
@@ -278,10 +530,36 @@ func (a *Agent) executeTask(t *Task) {
 		}()
 		switch t.Kind {
 		case TaskKindInput:
-			a.handleInput(t.Event)
+			f, out = a.runInputTask(t.Event, t.SeedMsgs)
 		case TaskKindSelf:
-			a.handleSelfInput(t.Self)
+			f, out = a.runInputTask(selfEvent(t.Self), nil)
 		}
 	}()
+
+	if out == outcomeSuspended && f != nil {
+		a.sched.suspend(t, f)
+		return
+	}
+	a.sched.done(t)
+}
+
+// resumeTask 从保存的现场继续一个被抢占的任务。
+//
+// 关键：不重建帧、不重跑 prepare 段——否则会重复提交上下文与事件。
+func (a *Agent) resumeTask(t *Task, f *TaskFrame) {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Printf("[agent] resume task#%d panic recovered: %v\n%s",
+				t.ID, r, debug.Stack())
+			a.sched.done(t)
+		}
+	}()
+
+	out := a.runTaskSteps(f)
+	if out == outcomeSuspended {
+		a.sched.suspend(t, f)
+		return
+	}
+	a.finishInputTask(f, out)
 	a.sched.done(t)
 }

internal/agent/core/scheduler_preempt_test.go

@@ -0,0 +1,297 @@
+package core
+
+// M3b 验收测试：四级优先级 + 严格大于抢占 + 现场保存/恢复 + suspendPool。
+//
+// 设计依据 docs/zh/input-scheduler-design.md §11（P1–P4、R1、R5、D1T、Q3）。
+//
+// 测试手法：用一个「第一次调用阻塞到 ctx 取消、之后按脚本返回」的 provider，
+// 让测试可以确定性地把运行任务停在 S_LLM 上，再注入中断观察让位与恢复。
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
+	agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
+)
+
+// preemptProvider 第 1 次 Chat 阻塞直到 ctx 取消；第 2 次起返回脚本。
+type preemptProvider struct {
+	mu        sync.Mutex
+	calls     int
+	entered   chan struct{}
+	enteredOn sync.Once
+	responses []string
+}
+
+func newPreemptProvider(responses ...string) *preemptProvider {
+	return &preemptProvider{entered: make(chan struct{}), responses: responses}
+}
+
+func (p *preemptProvider) Name() string { return "preempt" }
+
+func (p *preemptProvider) Chat(ctx context.Context, req *agentAPI.CompletionRequest) (*agentAPI.CompletionResponse, error) {
+	p.mu.Lock()
+	p.calls++
+	n := p.calls
+	p.mu.Unlock()
+
+	if n == 1 {
+		p.enteredOn.Do(func() { close(p.entered) })
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	i := n - 2
+	if i < len(p.responses) {
+		return &agentAPI.CompletionResponse{Content: p.responses[i]}, nil
+	}
+	return &agentAPI.CompletionResponse{Content: "done"}, nil
+}
+
+func (p *preemptProvider) ChatStream(ctx context.Context, req *agentAPI.CompletionRequest) (<-chan agentAPI.StreamChunk, error) {
+	return nil, context.Canceled
+}
+func (p *preemptProvider) MaxContextTokens() int { return 8192 }
+
+func (p *preemptProvider) callCount() int {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.calls
+}
+
+func newPreemptAgent(t *testing.T, sp agentAPI.Provider) *Agent {
+	t.Helper()
+	return New(AgentConfig{
+		ID:              "preempt",
+		Provider:        sp,
+		ProviderManager: agentAPI.NewProviderManager(),
+		IO:              agentIO.NewIOManager(),
+		StageHost:       NewStageHost(),
+	})
+}
+
+func enqueueTask(t *testing.T, a *Agent, level Level, source, content string) (*Task, *agentIO.InputEvent) {
+	t.Helper()
+	evt, _ := textEvent(source, content)
+	task := &Task{Kind: TaskKindInput, Level: level, Event: evt, EnqueuedAt: time.Now()}
+	if !a.sched.enqueue(task) {
+		t.Fatal("入队失败")
+	}
+	return task, evt
+}
+
+// P1 + R1 + R5 + D1=A：高优先级抢占 → 挂起在 S_LLM → 中断任务带只读前缀 →
+// 恢复后从 S_LLM 重发，且原任务的 msgs 未被改动。
+func TestPreempt_HigherPreemptsAndResumes(t *testing.T) {
+	sp := newPreemptProvider("intr-done", "low-done")
+	a := newPreemptAgent(t, sp)
+
+	lowTask, _ := enqueueTask(t, a, LevelBackground, "qq", "低优先级任务")
+	lt, _, kind := a.sched.nextRef()
+	if kind != nextReady || lt != lowTask {
+		t.Fatalf("应取到低优先级任务，kind=%v", kind)
+	}
+
+	done := make(chan struct{})
+	go func() { a.executeNewTask(lt); close(done) }()
+
+	select {
+	case <-sp.entered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("provider 第 1 次调用未发生")
+	}
+
+	// 注入 L4 中断（cli）
+	intrEvt, _ := textEvent("cli", "紧急打断")
+	intrEvt.Payload["interrupt"] = true
+	if !a.sched.requestPreempt(intrEvt, LevelCritical) {
+		t.Fatal("L4 应请求抢占并返回 true（应取消 LLM）")
+	}
+	a.cancelCurrentLLM()
+
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("低优先级任务未在取消后挂起")
+	}
+
+	snap := a.DumpScheduler()
+	if snap.Running != nil {
+		t.Fatal("挂起后不应还有 running")
+	}
+	if len(snap.SuspendPool) != 1 {
+		t.Fatalf("suspendPool=%d，期望 1", len(snap.SuspendPool))
+	}
+	if len(snap.PendingInterrupts) != 1 {
+		t.Fatalf("pendingInterrupts=%d，期望 1", len(snap.PendingInterrupts))
+	}
+	sf := snap.SuspendPool[0].Frame
+	if sf.Terminal != terminalSuspended {
+		t.Fatalf("挂起任务终态=%v，期望 terminalSuspended", sf.Terminal)
+	}
+	if sf.Step != StepLLM {
+		t.Fatalf("应在 StepLLM 安全点挂起，实际 step=%v", sf.Step)
+	}
+	msgsBefore := len(sf.Msgs)
+
+	// R5：第一次 LLM 调用被丢弃，未产生新消息。
+	if sp.callCount() != 1 {
+		t.Fatalf("挂起前 LLM 调用=%d，期望 1（不完整请求被丢弃）", sp.callCount())
+	}
+
+	// Q3：三集合统一比较 → 下一轮取中断（L4 > L1）。
+	it, _, k := a.sched.nextRef()
+	if k != nextPending || it.Level != LevelCritical {
+		t.Fatalf("应取到 pending 中断，kind=%v level=%v", k, it.Level)
+	}
+	// D1=A：抢占式中断任务继承被打断任务的只读前缀。
+	if len(it.SeedMsgs) == 0 {
+		t.Fatal("抢占式中断任务必须继承只读前缀（D1=A）")
+	}
+	a.executeNewTask(it)
+	if len(a.DumpScheduler().PendingInterrupts) != 0 {
+		t.Fatal("中断任务执行后 pendingInterrupts 应清空")
+	}
+
+	// R1：恢复被抢占任务；msgs 与被抢占前逐字节一致（长度不变），并从 S_LLM 重发。
+	rt, rf, k2 := a.sched.nextRef()
+	if k2 != nextSuspended || rt != lowTask {
+		t.Fatalf("应恢复被抢占任务，kind=%v", k2)
+	}
+	if rf.Step != StepLLM {
+		t.Fatalf("恢复游标=%v，期望 StepLLM", rf.Step)
+	}
+	if len(rf.Msgs) != msgsBefore {
+		t.Fatalf("恢复后 msgs 长度=%d，期望 %d（不得被中断污染）", len(rf.Msgs), msgsBefore)
+	}
+	a.resumeTask(rt, rf)
+
+	if sp.callCount() != 3 {
+		t.Fatalf("LLM 总调用=%d，期望 3（丢弃 1 + 中断 1 + 恢复 1）", sp.callCount())
+	}
+	snap = a.DumpScheduler()
+	if len(snap.SuspendPool) != 0 || snap.Running != nil {
+		t.Fatalf("全部结束后应无挂起与运行任务：%+v", snap)
+	}
+	if snap.Stats.Executed != 2 {
+		t.Fatalf("Executed=%d，期望 2（中断任务 + 被抢占任务）", snap.Stats.Executed)
+	}
+}
+
+// P2/P3：同级与更低级都不得抢占，请求进 pendingInterrupts。
+func TestPreempt_LowerOrEqualDoesNotPreempt(t *testing.T) {
+	sp := newPreemptProvider("low-done", "intr-done")
+	a := newPreemptAgent(t, sp)
+
+	lowTask, _ := enqueueTask(t, a, LevelInteractive, "cli", "运行中的 L3")
+	if _, _, kind := a.sched.nextRef(); kind != nextReady {
+		t.Fatal("应取到运行任务")
+	}
+
+	done := make(chan struct{})
+	go func() { a.executeNewTask(lowTask); close(done) }()
+	select {
+	case <-sp.entered:
+	case <-time.After(3 * time.Second):
+		t.Fatal("provider 未被调用")
+	}
+
+	// 同级 L3
+	e1, _ := textEvent("webui", "同级打断")
+	if a.sched.requestPreempt(e1, LevelInteractive) {
+		t.Fatal("同级不得抢占")
+	}
+	// 更低级 L1
+	e2, _ := textEvent("system", "低优先级打断")
+	if a.sched.requestPreempt(e2, LevelBackground) {
+		t.Fatal("更低级不得抢占")
+	}
+	if a.sched.preemptGrantedFor(LevelInteractive) {
+		t.Fatal("未 arm 让位信号，preemptGrantedFor 应为 false")
+	}
+
+	// 运行任务没有被取消，仍在等它自己的 ctx；用取消让它收尾，便于清理。
+	a.cancelCurrentLLM()
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("运行任务未结束")
+	}
+	if len(a.DumpScheduler().PendingInterrupts) != 2 {
+		t.Fatalf("两条未抢占中断都应保留在 pendingInterrupts，实际 %d",
+			len(a.DumpScheduler().PendingInterrupts))
+	}
+}
+
+// D1T：suspendPool 满时不再下潜（canSuspend=false），让位信号也不会 arm。
+func TestPreempt_DepthCapBlocksSuspension(t *testing.T) {
+	a := newPreemptAgent(t, newPreemptProvider())
+
+	if a.sched.maxSuspendDepth != 4 {
+		t.Fatalf("默认深度上限=%d，期望 4", a.sched.maxSuspendDepth)
+	}
+	frame := func() *TaskFrame { return a.newTaskFrame("x", a.stageCtxFromInput("x", "", "")) }
+	for i := 0; i < a.sched.maxSuspendDepth; i++ {
+		a.sched.suspend(&Task{ID: uint64(i + 1), Level: LevelBackground}, frame())
+	}
+	if a.sched.canSuspend() {
+		t.Fatal("深度已达上限，canSuspend 应为 false")
+	}
+	// 超限兜底：仍保留帧（不丢副作用记录），但计数 Rejected。
+	before := a.DumpScheduler().Stats.Rejected
+	a.sched.suspend(&Task{ID: 99, Level: LevelBackground}, frame())
+	if a.DumpScheduler().Stats.Rejected != before+1 {
+		t.Fatal("超限挂起必须计数 Rejected")
+	}
+	if len(a.DumpScheduler().SuspendPool) != a.sched.maxSuspendDepth+1 {
+		t.Fatal("兜底路径必须保留帧而不是丢弃")
+	}
+}
+
+// 中断不丢：空闲时请求抢占 → 不 arm 信号，但请求进 pendingInterrupts 并被选出。
+func TestPreempt_IdleInterruptIsQueuedNotLost(t *testing.T) {
+	a := newPreemptAgent(t, &scriptProvider{script: []*agentAPI.CompletionResponse{{Content: "已处理中断"}}})
+
+	evt, respCh := textEvent("qq", "空闲时的中断")
+	if a.sched.requestPreempt(evt, LevelMessage) {
+		t.Fatal("空闲时不应请求取消 LLM（没有运行任务）")
+	}
+	if len(a.DumpScheduler().PendingInterrupts) != 1 {
+		t.Fatal("空闲时的中断必须进 pendingInterrupts")
+	}
+	task, _, kind := a.sched.nextRef()
+	if kind != nextPending || task.Level != LevelMessage {
+		t.Fatalf("应取到待处理中断，kind=%v", kind)
+	}
+	a.executeNewTask(task)
+	if len(respCh) != 1 {
+		t.Fatal("中断任务应完成并回执")
+	}
+}
+
+// 带媒体/中断标记的输入走新任务路径时不得污染下一个任务的尾部消息。
+func TestPreempt_SeedPathDoesNotLeakInterruptFlag(t *testing.T) {
+	sp := &scriptProvider{script: []*agentAPI.CompletionResponse{{Content: "ok"}}}
+	a := newPreemptAgent(t, sp)
+
+	f := a.newTaskFrame("打断文本", a.stageCtxFromInput("打断文本", "", ""))
+	f.SeedMsgs = []agentAPI.Message{{Role: "system", Content: "S"}, {Role: "user", Content: "U"}}
+	a.interruptInput = true
+	if out := a.stepPrepare(f); out != outcomeContinue {
+		t.Fatalf("seed 路径应继续，实际 %v", out)
+	}
+	if a.interruptInput {
+		t.Fatal("seed 路径必须消费 interruptInput，否则下一个任务尾部会被误改")
+	}
+	last := f.Msgs[len(f.Msgs)-1]
+	if last.Role != "user" || last.Content != "打断文本" {
+		t.Fatalf("seed 路径尾部应为中断输入本身，实际 %+v", last)
+	}
+	if f.Step != StepLLM {
+		t.Fatalf("seed 路径应直接进入 StepLLM，实际 %v", f.Step)
+	}
+}

internal/agent/core/task.go

@@ -28,6 +28,7 @@ import (
 	"time"
 
 	agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
+	agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
 	"gitcode.com/JianFeeeee/HomeAgent/internal/events"
 	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
 	pubsdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk"
@@ -61,6 +62,28 @@ const (
 	outcomeDone
 	// outcomeFailed 任务失败结束，错误在 frame.Err。
 	outcomeFailed
+	// outcomeSuspended 任务在安全点被抢占挂起，帧已保存（M3b 起使用）。
+	outcomeSuspended
+)
+
+// taskTerminal 是任务的终态种类（设计文档 §7：每个任务恰有一个终态）。
+type taskTerminal int
+
+const (
+	// terminalNone 任务尚未结束，需进入 run 段。
+	terminalNone taskTerminal = iota
+	// terminalOK 正常完成（已提交上下文并回执）。
+	terminalOK
+	// terminalError 执行出错（已提交错误响应）。
+	terminalError
+	// terminalStageShortCircuit 被 on_input 阶段短路（响应已发出）。
+	terminalStageShortCircuit
+	// terminalSkipped 未进入执行：解析失败或被去重。
+	terminalSkipped
+	// terminalConsolidation 走记忆整理专用路径，已处理完毕。
+	terminalConsolidation
+	// terminalSuspended 被抢占挂起，等待恢复（M3b 起使用）。
+	terminalSuspended
 )
 
 // TaskFrame 承载一个任务在安全点之间必须存活的所有状态。
@@ -92,37 +115,259 @@ type TaskFrame struct {
 	Step     Step
 	Response string
 	Err      error
+
+	// ---- 任务层现场（原 processInput 的局部变量）----
+	//
+	// 这些字段让帧覆盖 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
+
+	// SeedMsgs 非空时，stepPrepare 不重建 system prompt / 记忆上下文，
+	// 而是以它为前缀继续（D1=A：抢占式中断任务继承被打断任务的**只读前缀**）。
+	SeedMsgs []agentAPI.Message
 }
 
 func (a *Agent) newTaskFrame(input string, stageCtx *sdk.StageContext) *TaskFrame {
 	return &TaskFrame{Input: input, StageCtx: stageCtx, Step: StepPrepare}
 }
 
-// process 驱动状态机直到任务结束，返回与原实现完全相同的四元组。
+// runTaskSteps 驱动状态机直到任务结束或被抢占挂起。
 //
-// 保留该签名是为了让 M1 成为纯内部重构：所有调用方（processInput /
-// processConsolidation / 测试）无需改动。
-func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response string, toolsUsed []string, toolResults []ToolResultItem, err error) {
-	a.mu.Lock()
-	defer a.mu.Unlock()
-
-	if a.provider == nil {
-		return "", nil, nil, fmt.Errorf("agent: no LLM provider configured")
-	}
-
-	f := a.newTaskFrame(input, stageCtx)
+// 这是 M1 的驱动循环，M3a 从 process() 抽出来，使调用方可以拿到
+// outcomeSuspended 并把帧留给调度器保存。
+func (a *Agent) runTaskSteps(f *TaskFrame) stepOutcome {
 	// 步数上限只是防"转移缺失导致死循环"的护栏；正常任务远达不到。
 	const maxSteps = 1 << 20
 	for i := 0; i < maxSteps; i++ {
+		// 安全点：只在 step 之间检查让位。临界区（StepToolExec）不在此列，
+		// 因为让位信号由 interruptLoop 置位、而本循环是唯一读帧者。
+		if !a.inCriticalSection() && a.sched.preemptGrantedFor(f.Level) && a.sched.canSuspend() {
+			return outcomeSuspended
+		}
 		switch a.step(f) {
 		case outcomeDone:
-			return f.Response, f.ToolsUsed, f.ToolResults, nil
+			return outcomeDone
 		case outcomeFailed:
-			return "", f.ToolsUsed, f.ToolResults, f.Err
+			return outcomeFailed
+		case outcomeSuspended:
+			return outcomeSuspended
 		}
 	}
-	return "", f.ToolsUsed, f.ToolResults,
-		fmt.Errorf("agent: task step budget exhausted（状态机未收敛，疑似转移缺失）")
+	f.Err = fmt.Errorf("agent: task step budget exhausted（状态机未收敛，疑似转移缺失）")
+	return outcomeFailed
+}
+
+// process 是保留给 processConsolidation 与测试的薄壳，返回与原实现相同的四元组。
+//
+// 注意：M3a 起**不再持 a.mu**——调度器是唯一执行者，而挂起不能持锁。
+func (a *Agent) process(input string, stageCtx *sdk.StageContext) (response string, toolsUsed []string, toolResults []ToolResultItem, err error) {
+	if a.provider == nil {
+		return "", nil, nil, fmt.Errorf("agent: no LLM provider configured")
+	}
+	f := a.newTaskFrame(input, stageCtx)
+	switch a.runTaskSteps(f) {
+	case outcomeDone:
+		return f.Response, f.ToolsUsed, f.ToolResults, nil
+	case outcomeFailed:
+		return "", f.ToolsUsed, f.ToolResults, f.Err
+	default:
+		// 不该发生：process() 不参与挂起（只有 runInputTask 会）。
+		return "", f.ToolsUsed, f.ToolResults,
+			fmt.Errorf("agent: task suspended outside scheduler")
+	}
+}
+
+// runInputTask 是一个输入任务的完整生命周期：prepare → run → finish。
+//
+// 它是原 processInput 的全部职责，被拆成三段而不是一个大函数，目的只有一个：
+// 让帧可以跨安全点被挂起——挂起后由调度器保存，恢复时接着 run 段继续，
+// 而 finish 段（上下文提交与回执）只在任务真正结束时执行一次。
+//
+// M3a 还没有抢占，因此 outcomeSuspended 只会由 M3b 的抢占检查产生。
+func (a *Agent) runInputTask(evt *agentIO.InputEvent, seed []agentAPI.Message) (*TaskFrame, stepOutcome) {
+	f, term := a.prepareInputTask(evt)
+	switch term {
+	case terminalSkipped, terminalStageShortCircuit, terminalConsolidation:
+		return nil, outcomeDone
+	}
+	f.SeedMsgs = seed
+
+	out := a.runTaskSteps(f)
+	if out == outcomeSuspended {
+		f.Terminal = terminalSuspended
+		return f, outcomeSuspended
+	}
+	a.finishInputTask(f, out)
+	return f, out
+}
+
+// prepareInputTask 执行 processInput 的前半段（去重、通道解析、阶段、裁剪、
+// 输入事件落上下文）。返回终态不为 terminalNone 时调用方不得进入 run 段。
+func (a *Agent) prepareInputTask(evt *agentIO.InputEvent) (*TaskFrame, taskTerminal) {
+	start := time.Now()
+
+	in, ok := a.resolveInput(evt)
+	if !ok {
+		return nil, terminalSkipped
+	}
+
+	// 去重按文本做：webui/GUI 断线重连会重放未确认消息。
+	// 带媒体时跳过——媒体输入的 alt 文案（"[从 qq 收到了 image]"）对不同图片
+	// 是同一句，拿它去重会把连发的两张图误判成重复。
+	if len(in.blocks) == 0 && a.isDuplicateInput(evt.Source, in.text) {
+		log.Printf("[agent] dropped duplicate input from %s: %s", evt.Source, truncateStr(in.text, 60))
+		return nil, terminalSkipped
+	}
+
+	a.currentOutputChannel = evt.OutputChannel
+	if a.currentOutputChannel == "" {
+		a.currentOutputChannel = evt.Source
+	}
+
+	if evt.OutputChannel == channelConsolidation {
+		a.processConsolidation(evt, in.text)
+		return nil, terminalConsolidation
+	}
+
+	// pendingMedia 让 describe_image / transcribe_audio / ocr_image 拿到本轮媒体的
+	// 原始 data/url，也是这三个工具是否出现在工具表里的开关。仅对用户直接上传成立
+	//（payload 里才有 data/url）；插件注入的是成品 block，取不到原始数据。
+	if evt.Type == "image" || evt.Type == "audio" {
+		a.pendingMedia = evt.Payload
+	}
+
+	// 媒体先落进 CAS。不存的后果是 ContextEvent.Input 只剩一句 alt 文本，
+	// base64 随 message 数组发给模型后就丢了。
+	if len(in.blocks) > 0 {
+		a.stageMediaDigests(a.captureBlockMedia(in.blocks, in.captureTool)...)
+	}
+
+	noMemory := false
+	if v, ok := evt.Payload["no_memory"].(bool); ok {
+		noMemory = v
+	}
+	if !noMemory && a.io != nil {
+		if chDef, ok := a.io.GetInputChannelDef(evt.Source); ok && chDef.NoMemory {
+			noMemory = true
+		}
+	}
+
+	// 工具提醒/中断（terminal_watch、timer 等）不是用户发言：
+	// 以 system 角色注入 LLM，且不写入用户对话履历。
+	isInterrupt, _ := evt.Payload["interrupt"].(bool)
+	a.interruptInput = isInterrupt
+	if isInterrupt {
+		noMemory = true
+	}
+
+	stageCtx := a.stageCtxFromInput(in.text, evt.Source, "")
+	stageCtx.Extra["input_source"] = evt.Source
+	stageCtx.Extra["output_channel"] = evt.OutputChannel
+	if len(in.blocks) > 0 {
+		stageCtx.Extra["media_blocks"] = in.blocks
+		stageCtx.Extra["media_type"] = in.mediaType
+	}
+	if noMemory {
+		stageCtx.NoMemory = true
+	}
+	a.injectSourceContext(stageCtx, evt)
+
+	if a.runStage(sdk.StageOnInput, stageCtx) {
+		a.emitResponse(evt, *stageCtx.Response)
+		return nil, terminalStageShortCircuit
+	}
+
+	input := stageCtx.RawMessage
+
+	// 计算层用的清洗文本（不改原文）：通道 Cleaner 提取语义内容后用于向量化/提关键词
+	cleanInput := input
+	if a.io != nil {
+		if chDef, ok := a.io.GetInputChannelDef(evt.Source); ok && chDef.Cleaner != nil {
+			cleanInput = chDef.Cleaner(input)
+		}
+	}
+
+	// upload_* 字段一并转发：webui 的 EventRawInput 订阅方靠它们还原附件卡片。
+	rawPayload := map[string]interface{}{"content": input, "source": evt.Source}
+	for _, k := range []string{"upload_url", "upload_type", "upload_size", "upload_name"} {
+		if v, ok := evt.Payload[k]; ok {
+			rawPayload[k] = v
+		}
+	}
+	a.publishEvent(events.EventRawInput, rawPayload)
+
+	archived := a.pruneOnInput(evt, cleanInput)
+	if archived > 0 {
+		log.Printf("[agent] pruned %d low-relevance events to document memory", archived)
+	}
+
+	if !isInterrupt {
+		a.context.Append(ContextEvent{
+			Timestamp: start,
+			Source:    evt.Source,
+			Input:     input,
+		})
+	}
+
+	f := a.newTaskFrame(input, stageCtx)
+	f.Evt = evt
+	f.CleanInput = cleanInput
+	f.IsInterrupt = isInterrupt
+	f.StartedAt = start
+	f.Level = a.sched.currentLevel()
+	return f, terminalNone
+}
+
+// finishInputTask 执行 processInput 的后半段（日志、上下文提交、回执、记忆候选）。
+//
+// 只在任务真正结束时调用一次——这正是不变量 I5（每任务恰一次终态）的落点。
+func (a *Agent) finishInputTask(f *TaskFrame, out stepOutcome) {
+	evt := f.Evt
+
+	// pendingMedia 是「本轮」语义：任务结束即清（挂起时保留，见 runInputTask）。
+	if evt != nil && (evt.Type == "image" || evt.Type == "audio") {
+		a.pendingMedia = nil
+	}
+
+	if out == outcomeFailed {
+		log.Printf("[agent] process %s error: %v", evt.Type, f.Err)
+		resp := fmt.Sprintf("处理错误: %v", f.Err)
+		a.emitResponse(evt, resp)
+		a.context.Append(ContextEvent{Timestamp: time.Now(), Source: "agent", Input: f.Input, Response: resp})
+		f.Terminal = terminalError
+		return
+	}
+
+	elapsed := time.Since(f.StartedAt)
+	log.Printf("[agent] %s from %s → response (%dms, tools=%v)",
+		evt.Type, evt.Source, elapsed.Milliseconds(), f.ToolsUsed)
+
+	// 本轮捕获的媒体一起挂到这条事件上：用户上传的、插件注入的，以及模型调
+	// multimodal_see_picture / see_video 时经 SetToolBlocks 注入的。
+	turnEvt := ContextEvent{
+		Timestamp:   time.Now(),
+		Source:      "agent",
+		Input:       f.CleanInput,
+		Response:    f.Response,
+		ToolsUsed:   f.ToolsUsed,
+		ToolResults: f.ToolResults,
+	}
+	a.bindEventMedia(&turnEvt, a.drainMediaDigests())
+	a.context.Append(turnEvt)
+
+	a.emitResponse(evt, f.Response)
+
+	if !f.StageCtx.NoMemory {
+		a.emitMemoryCandidate(evt.Source, f.CleanInput, f.Response, f.ToolResults, f.ToolsUsed)
+	}
+	f.Terminal = terminalOK
 }
 
 // step 执行恰好一个 step。
@@ -148,6 +393,18 @@ func (a *Agent) step(f *TaskFrame) stepOutcome {
 
 // stepPrepare 构建本轮任务的初始帧。
 func (a *Agent) stepPrepare(f *TaskFrame) stepOutcome {
+	// 抢占式中断任务：继承被打断任务的只读前缀（D1=A），不重建上下文。
+	if len(f.SeedMsgs) > 0 {
+		f.Tools = a.buildToolDefs()
+		f.Msgs = append([]agentAPI.Message(nil), f.SeedMsgs...)
+		f.Msgs = append(f.Msgs, agentAPI.Message{Role: "user", Content: f.Input})
+		// 前缀路径不重写尾部消息（那是「同行注入」的旧形态）；
+		// 必须消费掉标志位，否则下一个普通任务的尾部会被误改。
+		a.interruptInput = false
+		f.Step = StepLLM
+		return outcomeContinue
+	}
+
 	budget := ComputeTokenBudget(a.provider, a.systemPrompt)
 
 	memContext := a.buildMemoryContext(f.Input, budget.MemoryTokens)

internal/agent/core/task_lifecycle_test.go

@@ -0,0 +1,204 @@
+package core
+
+// M3a 验收测试：任务生命周期（prepare → run → finish）与「每任务恰一次终态」。
+//
+// 设计依据 docs/zh/input-scheduler-design.md §11.3（X2/X3/X4 的 M3a 形态）：
+// 帧覆盖全生命周期后，提交（context.Append）与回执（emitResponse）只能在
+// finish 段发生一次——挂起不会重复提交。
+
+import (
+	"strings"
+	"testing"
+
+	agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
+	agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
+	"gitcode.com/JianFeeeee/HomeAgent/internal/events"
+	sdk "gitcode.com/JianFeeeee/HomeAgent/internal/sdk"
+)
+
+func newLifecycleAgent(t *testing.T, sp agentAPI.Provider, bus *events.Bus, sh *StageHost) *Agent {
+	t.Helper()
+	return New(AgentConfig{
+		ID:              "lifecycle",
+		Provider:        sp,
+		ProviderManager: agentAPI.NewProviderManager(),
+		IO:              agentIO.NewIOManager(),
+		StageHost:       sh,
+		EventBus:        bus,
+	})
+}
+
+func textEvent(source, content string) (*agentIO.InputEvent, chan *agentIO.OutputEvent) {
+	ch := make(chan *agentIO.OutputEvent, 1)
+	return &agentIO.InputEvent{
+		RequestID:     "req-1",
+		Source:        source,
+		Type:          "text",
+		Payload:       map[string]interface{}{"content": content},
+		OutputChannel: source,
+		ResponseCh:    ch,
+	}, ch
+}
+
+// X3（M3a 形态）：正常任务在 finish 段**恰好**提交一次并回执一次。
+func TestLifecycle_NormalCommitsOnceAndReplies(t *testing.T) {
+	bus := events.NewBus()
+	var outputs, rawInputs int
+	bus.Subscribe(events.EventAgentOutput, func(*events.Event) { outputs++ })
+	bus.Subscribe(events.EventRawInput, func(*events.Event) { rawInputs++ })
+
+	sp := &scriptProvider{script: []*agentAPI.CompletionResponse{{Content: "答复"}}}
+	a := newLifecycleAgent(t, sp, bus, NewStageHost())
+
+	evt, respCh := textEvent("cli", "你好")
+	if _, out := a.runInputTask(evt, nil); out != outcomeDone {
+		t.Fatalf("runInputTask=%v，期望 outcomeDone", out)
+	}
+
+	select {
+	case r := <-respCh:
+		if got, _ := r.Payload["content"].(string); got != "答复" {
+			t.Fatalf("回执内容=%q，期望 答复", got)
+		}
+		if !r.Done {
+			t.Fatal("回执必须带 Done=true")
+		}
+	default:
+		t.Fatal("同步回执缺失：finish 段必须写 ResponseCh")
+	}
+
+	if outputs != 1 {
+		t.Fatalf("agent_output 事件=%d，期望恰好 1（每任务一次终态）", outputs)
+	}
+	if rawInputs != 1 {
+		t.Fatalf("raw_input 事件=%d，期望 1", rawInputs)
+	}
+	if a.context.Len() != 2 {
+		t.Fatalf("上下文事件=%d，期望 2（输入事件 + 本轮事件）", a.context.Len())
+	}
+}
+
+// X2：被去重的输入以 skipped 终态结束——不提交、不回执、不发输出事件。
+func TestLifecycle_DuplicateSkippedHasTerminal(t *testing.T) {
+	bus := events.NewBus()
+	outputs := 0
+	bus.Subscribe(events.EventAgentOutput, func(*events.Event) { outputs++ })
+
+	sp := &scriptProvider{script: []*agentAPI.CompletionResponse{
+		{Content: "第一次"}, {Content: "第二次"},
+	}}
+	a := newLifecycleAgent(t, sp, bus, NewStageHost())
+
+	e1, _ := textEvent("webui", "同样的消息")
+	if _, out := a.runInputTask(e1, nil); out != outcomeDone {
+		t.Fatalf("首次输入=%v，期望 outcomeDone", out)
+	}
+	after1, outputs1 := a.context.Len(), outputs
+
+	e2, ch2 := textEvent("webui", "同样的消息")
+	if _, out := a.runInputTask(e2, nil); out != outcomeDone {
+		t.Fatalf("去重输入应正常返回（不挂起），实际 %v", out)
+	}
+	if a.context.Len() != after1 {
+		t.Fatalf("去重命中不得提交上下文：%d → %d", after1, a.context.Len())
+	}
+	if len(ch2) != 0 {
+		t.Fatal("去重命中不得回执（原实现静默 return）")
+	}
+	if outputs != outputs1 {
+		t.Fatalf("去重命中不得发输出事件：%d → %d", outputs1, outputs)
+	}
+}
+
+// 被 on_input 阶段短路：回执阶段给的响应，且不提交上下文（与原实现一致）。
+func TestLifecycle_OnInputShortCircuit(t *testing.T) {
+	sh := NewStageHost()
+	reply := "被插件短路"
+	sh.RegisterStage(sdk.StageOnInput, func(ctx *sdk.StageContext) error {
+		ctx.Response = &reply
+		return nil
+	})
+
+	sp := &scriptProvider{} // 不应被调用到
+	a := newLifecycleAgent(t, sp, events.NewBus(), sh)
+
+	evt, respCh := textEvent("cli", "任意")
+	if _, out := a.runInputTask(evt, nil); out != outcomeDone {
+		t.Fatalf("短路任务=%v，期望 outcomeDone", out)
+	}
+	select {
+	case r := <-respCh:
+		if got, _ := r.Payload["content"].(string); got != reply {
+			t.Fatalf("短路响应=%q，期望 %q", got, reply)
+		}
+	default:
+		t.Fatal("短路路径必须回执")
+	}
+	if a.context.Len() != 0 {
+		t.Fatalf("短路路径不得提交上下文，实际 %d 条", a.context.Len())
+	}
+	if len(sp.reqs) != 0 {
+		t.Fatal("短路路径不得调用 LLM")
+	}
+}
+
+// 错误路径：以 error 终态结束，回执错误文本，且提交的是**错误事件**（无 turn 事件）。
+func TestLifecycle_ErrorPathTerminal(t *testing.T) {
+	bus := events.NewBus()
+	outputs := 0
+	bus.Subscribe(events.EventAgentOutput, func(*events.Event) { outputs++ })
+
+	sp := &scriptProvider{err: &agentAPI.ProviderError{StatusCode: 401, Message: "bad key"}}
+	a := newLifecycleAgent(t, sp, bus, NewStageHost())
+
+	evt, respCh := textEvent("cli", "会失败")
+	if _, out := a.runInputTask(evt, nil); out != outcomeFailed {
+		t.Fatalf("runInputTask=%v，期望 outcomeFailed", out)
+	}
+	select {
+	case r := <-respCh:
+		got, _ := r.Payload["content"].(string)
+		if !strings.HasPrefix(got, "处理错误:") {
+			t.Fatalf("错误回执=%q，期望以 处理错误: 开头", got)
+		}
+	default:
+		t.Fatal("错误路径必须回执（否则同步调用方永久挂起）")
+	}
+	if outputs != 1 {
+		t.Fatalf("错误路径的 agent_output 事件=%d，期望 1", outputs)
+	}
+	// 输入事件 + 错误事件 = 2；不得出现带 ToolsUsed 的 turn 事件。
+	if a.context.Len() != 2 {
+		t.Fatalf("错误路径上下文事件=%d，期望 2", a.context.Len())
+	}
+	recent := a.context.Recent(10)
+	last := recent[len(recent)-1]
+	if last.Response == "" {
+		t.Fatal("错误事件必须带 Response")
+	}
+}
+
+// _consolidation_ 走记忆整理专用路径：不回执、不提交上下文。
+func TestLifecycle_ConsolidationRouted(t *testing.T) {
+	bus := events.NewBus()
+	outputs := 0
+	bus.Subscribe(events.EventAgentOutput, func(*events.Event) { outputs++ })
+
+	sp := &scriptProvider{script: []*agentAPI.CompletionResponse{{Content: "整理完毕"}}}
+	a := newLifecycleAgent(t, sp, bus, NewStageHost())
+
+	evt, respCh := textEvent("system", "整理任务")
+	evt.OutputChannel = channelConsolidation
+	if _, out := a.runInputTask(evt, nil); out != outcomeDone {
+		t.Fatalf("consolidation=%v，期望 outcomeDone", out)
+	}
+	if len(respCh) != 0 {
+		t.Fatal("consolidation 路径不得回执")
+	}
+	if outputs != 0 {
+		t.Fatalf("consolidation 路径不得发输出事件，实际 %d", outputs)
+	}
+	if a.context.Len() != 0 {
+		t.Fatalf("consolidation 路径不得写用户上下文，实际 %d", a.context.Len())
+	}
+}

internal/agent/core/task_test.go

@@ -26,11 +26,17 @@ type scriptProvider struct {
 	script []*agentAPI.CompletionResponse
 	idx    int
 	reqs   []*agentAPI.CompletionRequest
+	// err 非空时 Chat 直接返回它（用于错误路径测试）。
+	// 配合 ProviderError(401) 可跳过 2s 瞬时重试，让测试保持快速。
+	err error
 }
 
 func (s *scriptProvider) Name() string { return "script" }
 func (s *scriptProvider) Chat(ctx context.Context, req *agentAPI.CompletionRequest) (*agentAPI.CompletionResponse, error) {
 	s.reqs = append(s.reqs, req)
+	if s.err != nil {
+		return nil, s.err
+	}
 	if s.idx >= len(s.script) {
 		return &agentAPI.CompletionResponse{Content: ""}, nil
 	}