fix(remotedevice): 心跳 pong 忘了 Flush —— 修「设备通道每 60 秒掉线重连」

真因（实测定位）：服务端 writePong 只调 writeFrameHeader，**不 Flush**。
pong 只有两个字节，且设备空闲时没有任何别的写会顺带把 bufio 缓冲刷出去 ——
于是 pong 永远留在服务端缓冲里。

链路：客户端每 30s 发一个 ping（pingLoop）→ 服务端算出 pong 却没发出 →
客户端的读循环设的是「2 倍 ping 间隔」读超时（默认 60s）→ 每 60 秒准点
i/o timeout → 桥断开 → 3s 后重连 → 服务端 markOffline 注销 outputch，
重连后再注册。

生产日志就是这个指纹（online :20 → offline 下一分钟 :20 → 重连 :23，
连续数小时无一次例外）；面板上表现为设备通道/工具凭空消失又出现，
/devices 列表跟着闪。

改法：writePong 复用 writeFrame（它 Flush）。另把客户端读循环退出时的
静默 return 改成带错误与 opcode 的日志 —— 此前断线真因在设备侧完全不可见，
只能靠对端日志倒推，正是这次排查一开始卡住的地方。

回归用例 TestWSPingGetsPongWhileIdle：只发一个 ping，随后什么都不发，
要求 2s 内必须收到 pong。**反向验证过**：把修复改回 writeFrameHeader，
用例即以 `read tcp ...: i/o timeout` 失败（与生产症状一致）。

internal/agent/core/agent.go

@@ -404,9 +404,30 @@ func (a *Agent) Start() {
 func (a *Agent) Stop() {
 	// 父退出**必须**销毁全部驻留子（设计 §10 硬约束：子不得比父活得久、不留孤儿）。
 	a.StopResidents()
+	// 停机前给待办任务补终态。运行中的任务会经 cancel → LLM 失败 → emitResponse
+	// 自然拿到终态，但**从未运行**（排队/待处理）与**已挂起**的任务不会有任何人
+	// 回它们；带 ResponseCh 的同步注入方（cli / clawhubadapter 均无超时）会永久挂起
+	// （设计 §7 I5、§11.3 X2/X4）。必须在 cancel 之前做：cancel 会让调度器直接 return。
+	a.drainPendingInterrupts("agent_stopped")
 	a.cancel()
 }
 
+// drainPendingInterrupts 给排队/待处理/已挂起任务中带同步回执通道的调用方补一条
+// skipped 终态（复用 emitSkippedReply：非阻塞写，不对外发 agent_output 事件）。
+func (a *Agent) drainPendingInterrupts(reason string) {
+	if a.sched == nil {
+		return
+	}
+	pending := a.sched.pendingEvents()
+	if len(pending) == 0 {
+		return
+	}
+	for _, evt := range pending {
+		a.emitSkippedReply(evt, reason)
+	}
+	log.Printf("[agent] %s: 停机，%d 条待办任务已补 skipped 终态", a.id, len(pending))
+}
+
 // graphMemoryOf 决定本 agent 的图记忆共同面实现。
 //
 //   - 轻量内核（给了 LightMemory）：用 LightMemory，**整理面保持 nil**；

internal/agent/core/eventloop.go

@@ -331,13 +331,20 @@ func (a *Agent) emitResponse(evt *agentIO.InputEvent, response string) {
 		payload["usage"] = stageCtx.TokenUsage
 	}
 	if evt.ResponseCh != nil {
-		evt.ResponseCh <- &agentIO.OutputEvent{
+		// 非阻塞写：ResponseCh 由同步调用方以 cap=1 创建。按不变量 I5（每任务恰一次
+		// 终态）这里永远写得进去；但一旦哪天写出第二次，阻塞会卡死**调度器 goroutine**
+		// （整个 agent 停摆），而丢弃只是丢一条回执——与 emitSkippedReply 对称。
+		select {
+		case evt.ResponseCh <- &agentIO.OutputEvent{
 			RequestID:     evt.RequestID,
 			Target:        evt.Source,
 			Type:          "text",
 			Payload:       payload,
 			Done:          true,
 			OutputChannel: ch,
+		}:
+		default:
+			log.Printf("[agent] ResponseCh 已满，终态回执被丢弃（request=%s，可能违反不变量 I5）", evt.RequestID)
 		}
 	}
 

internal/agent/core/scheduler.go

@@ -229,6 +229,10 @@ type SchedulerStats struct {
 	Executed uint64
 	// Rejected 是因队列满（或深度超限）而未被接纳的次数。
 	Rejected uint64
+	// Backpressure 是就绪队列满、输入被挡回 channel 的次数
+	// （设计 §4.4 / §11.4 Q4：满时阻塞发送方，**必须计数并打日志**）。
+	// 与 Rejected 的区别：Rejected 是「丢了」，Backpressure 是「暂时不收、发送方在等」。
+	Backpressure uint64
 	// Suspended / Resumed 是挂起与恢复的次数。
 	// 不变量：系统排空后 Suspended == Resumed（挂起必然被恢复），
 	// 因此两者各自只在**一处**计数（suspend / resumeTask）。
@@ -284,7 +288,13 @@ func (a *Agent) schedulerStatus() sdk.SchedulerStatus {
 		Rejected:          snap.Stats.Rejected,
 		Suspended:         snap.Stats.Suspended,
 		Resumed:           snap.Stats.Resumed,
-		Preempted:         snap.Stats.Suspended,
+		Backpressure:      snap.Stats.Backpressure,
+	}
+	// Preempted 是「各级抢占成功次数之和」，**不是** Suspended：受害者可能在
+	// 让位信号生效前就自行结束，此时有抢占而没有挂起（见 PreemptsByLevel 注释）。
+	// 此前这里直接拿 Suspended 顶替，导致 DTO 里 preempted 与 preempts_by_level 自相矛盾。
+	for lv := LevelBackground; lv <= LevelCritical; lv++ {
+		out.Preempted += snap.Stats.PreemptsByLevel[lv]
 	}
 	if snap.Running != nil {
 		out.Running = &sdk.SchedulerTask{
@@ -320,6 +330,9 @@ type scheduler struct {
 	// critical 报告运行任务是否在不可抢占临界区（如记忆整理）。
 	// 由于 interceptLoop 要读它，必须是原子的：帧仍只由调度器读写。
 	critical atomic.Bool
+	// backpressured 记录「就绪队列满」这一状态的翻转，用于只打一次日志。
+	// 满着的时候 pumpInbox 每轮都会走到，逐轮打日志会把日志刷爆。
+	backpressured bool
 	// wake 用于把空闲的调度器叫醒：pendingInterrupts 不是 channel，
 	// 没有这个信号时“空闲时到达的中断”会一直等下一次输入（设计 §5.1 ③）。
 	wake chan struct{}
@@ -381,6 +394,28 @@ func (s *scheduler) hasRoom() bool {
 	return len(s.queue) < s.maxQueue
 }
 
+// noteBackpressure 记一次背压，并报告这是否是「从有空间 → 满」的翻转。
+//
+// 为什么需要翻转信息：满的时候每轮泵入都会调用本函数，逐轮打日志会刷爆；
+// 而设计 §4.4 要求「必须计数并打日志」——两者靠这个布尔量同时满足。
+func (s *scheduler) noteBackpressure() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.stats.Backpressure++
+	if s.backpressured {
+		return false
+	}
+	s.backpressured = true
+	return true
+}
+
+// clearBackpressure 在就绪队列重新可收（泵空）时复位翻转标记。
+func (s *scheduler) clearBackpressure() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.backpressured = false
+}
+
 // allocateIDLocked 分配任务 ID 与入队时刻（调用方持锁）。
 func (s *scheduler) allocateIDLocked(t *Task) {
 	s.seq++
@@ -490,10 +525,22 @@ func (s *scheduler) interruptCountLocked() int {
 // setImmediateLocked 登记一个应“立即运行”的抢占者。
 //
 // 槽只有一格：若已有抢占者且新的级别更高，旧的降级入队；否则新的入队。
-func (s *scheduler) setImmediateLocked(t *Task) {
+// 返回 true 表示 t **确实占住了 immediate 槽**；false 表示它被降级进了自己的
+// 级别队列（immediate 是单槽，这是设计要求的降级分支，见设计 §2「至多一个」）。
+//
+// 调用方必须用返回值决定是否计入 PreemptsByLevel：那条计数器的语义是
+// 「进入 immediate 的次数」，被降级的中断从未进过 immediate。
+// setImmediateLocked 尝试把 t 放进 immediate 槽。
+//
+// 返回 true：t 已占住 immediate（若原有抢占者被顶掉，它**已被**降级入队）。
+// 返回 false：t 没有进 immediate，且本函数**未动 t** —— 调用方负责按级别入队。
+//
+// 把「降级入队」的责任留给调用方，是为了让「到底入队了几次」只有一个出口：
+// 早先由本函数在返回 false 前自行入队，调用方又照着 false 再入一次，
+// 同一任务就会在队列里出现两份（实测：中断任务被执行两次、Executed 虚高）。
+func (s *scheduler) setImmediateLocked(t *Task) bool {
 	if s.immediate != nil && effectiveLevel(t) <= effectiveLevel(s.immediate) {
-		s.enqueueInterruptLocked(t)
-		return
+		return false
 	}
 	if s.immediate != nil {
 		s.enqueueInterruptLocked(s.immediate)
@@ -501,6 +548,7 @@ func (s *scheduler) setImmediateLocked(t *Task) {
 	s.allocateIDLocked(t)
 	s.stats.Enqueued++
 	s.immediate = t
+	return true
 }
 
 func removeTask(list []*Task, target *Task) []*Task {
@@ -571,11 +619,16 @@ func (s *scheduler) registerInterrupt(t *Task) bool {
 	arm := false
 	if !critical && canPreempt(t, running) {
 		if running.LastPreemptAt.IsZero() || time.Since(running.LastPreemptAt) >= preemptCooldown {
-			arm = true
-			s.preemptArmed = true
-			s.preemptLevel = t.Level
-			s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level)
-			s.setImmediateLocked(t)
+			// 只有**真的占住 immediate 槽**才算一次抢占，才计入 PreemptsByLevel：
+			// immediate 是单槽，若它被另一个更高级的抢占者占着，t 会走上而下的
+			// 「否则入队」分支——那种情况 t 从未进入 immediate（否则同一安全点前
+			// 到达两条同级中断时该计数会高估）。
+			if s.setImmediateLocked(t) {
+				arm = true
+				s.preemptArmed = true
+				s.preemptLevel = t.Level
+				s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level)
+			}
 		}
 	}
 	if !arm {
@@ -606,6 +659,54 @@ func (s *scheduler) clearPreempt() {
 	s.mu.Unlock()
 }
 
+// rearmPending 在**安全点重新求值**中断队列（设计 §4.3 / §5.2）。
+//
+// 为什么必须有这一步：中断只在 registerInterrupt 里被武装一次，而那一刻运行任务
+// 可能正在临界区（S_TOOL_EXEC / ONNX / CAS）或处于抢占冷却期，于是请求只能入队。
+// 若安全点不再回头看队列，它就永远等不到执行——只能等当前任务**自然结束**，
+// 这违背设计承诺的「临界区期间到达的抢占请求……在临界区结束后的第一个安全点
+// 重新求值」。可复现症状：WebUI 终止按钮连按两次，第二次（落在 2s 冷却窗内）
+// 入队后再也不会被求值，「终止」看起来没反应。
+//
+// 判据与 registerInterrupt **完全同一套**（canPreempt + 冷却 + 临界区闸门），
+// 因此不会凭空制造设计之外的抢占。
+func (s *scheduler) rearmPending() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	// 已有让位信号、或 immediate 槽已被占用：下一个安全点的选择已经在路上，
+	// 不必（也不该）重复武装。
+	if s.preemptArmed || s.immediate != nil || s.running == nil || s.critical.Load() {
+		return
+	}
+	// 冷却期内不武装：与 registerInterrupt 同一判据（抗饥饿）。
+	if !s.running.LastPreemptAt.IsZero() && time.Since(s.running.LastPreemptAt) < preemptCooldown {
+		return
+	}
+	// 中断队列本就按级别组织：从最高级往下找第一条能抢占的队头。
+	// （队列里的任务有效级恒等于基础级，故「第一条能抢」= 最高级可抢占者。）
+	for lv := LevelCritical; lv >= LevelBackground; lv-- {
+		q := s.interruptQueues[lv]
+		if len(q) == 0 {
+			continue
+		}
+		t := q[0]
+		if !canPreempt(t, s.running) {
+			continue
+		}
+		s.popInterruptLocked(lv)
+		if s.setImmediateLocked(t) {
+			s.preemptArmed = true
+			s.preemptLevel = t.Level
+			s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level)
+		} else {
+			// immediate 槽没拿到（理论上进不来，顶部已判 immediate == nil）：放回队列，
+			// 否则任务会凭空消失。
+			s.enqueueInterruptLocked(t)
+		}
+		return
+	}
+}
+
 // suspend 保存现场。
 //
 // 深度上界是**结构推论**（= 中断级数），不是配置项：安全点上的 canSuspend 已提前
@@ -638,6 +739,37 @@ func (s *scheduler) canSuspend() bool {
 	return len(s.suspendStack) < s.maxInterruptFrames
 }
 
+// pendingEvents 收集**尚未执行**（排队队列 / 四条中断队列 / immediate）与
+// **已挂起**（中断栈）任务所携带的、且带同步回执通道的输入事件。
+//
+// 用途只有一个：停机收尾。这些任务不会再被调度，若不给它们补终态，
+// 无超时的同步注入方（cli / clawhubadapter）会永久挂起（设计 §7 I5、§11.3 X2/X4）。
+func (s *scheduler) pendingEvents() []*agentIO.InputEvent {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var out []*agentIO.InputEvent
+	add := func(t *Task) {
+		if t != nil && t.Event != nil && t.Event.ResponseCh != nil {
+			out = append(out, t.Event)
+		}
+	}
+	for _, t := range s.queue {
+		add(t)
+	}
+	for lv := LevelBackground; lv <= LevelCritical; lv++ {
+		for _, t := range s.interruptQueues[lv] {
+			add(t)
+		}
+	}
+	add(s.immediate)
+	for _, f := range s.suspendStack {
+		if f != nil {
+			add(f.Task)
+		}
+	}
+	return out
+}
+
 // done 标记任务执行结束。
 func (s *scheduler) done(t *Task) {
 	s.mu.Lock()
@@ -796,9 +928,11 @@ func (a *Agent) schedulerLoop() {
 			// 无待办：阻塞等新输入、新中断（wake）或退出。
 			select {
 			case evt := <-a.io.InputChan():
-				a.sched.enqueue(newInputTask(evt))
+				if !a.sched.enqueue(newInputTask(evt)) {
+					a.emitSkippedReply(evt, "queue_full")
+				}
 			case msg := <-a.selfInputCh:
-				a.sched.enqueue(newSelfTask(msg))
+				_ = a.sched.enqueue(newSelfTask(msg))
 			case <-a.sched.wake:
 				// 中断已入 pendingInterrupts，回到循环顶部重新挑选。
 			case <-a.ctx.Done():
@@ -825,15 +959,28 @@ func (a *Agent) pumpInbox() {
 	for a.sched.hasRoom() {
 		select {
 		case evt := <-a.io.InputChan():
-			a.sched.enqueue(newInputTask(evt))
+			// 返回值必须处理：静默丢弃会让同步调用方永久挂起（回执路径 E）。
+			if !a.sched.enqueue(newInputTask(evt)) {
+				a.sched.noteBackpressure()
+				a.emitSkippedReply(evt, "queue_full")
+			}
 		case msg := <-a.selfInputCh:
-			a.sched.enqueue(newSelfTask(msg))
+			// 自循环输入没有同步调用方，满时记一次背压即可。
+			if !a.sched.enqueue(newSelfTask(msg)) {
+				a.sched.noteBackpressure()
+			}
 		case <-a.ctx.Done():
 			return
 		default:
+			a.sched.clearBackpressure()
 			return
 		}
 	}
+	// 队列满：输入留在 channel 里，发送方阻塞（设计 §4.4「阻塞发送方」）。
+	// 必须计数并打日志——否则运维看到 Rejected=0 会以为没背压，而输入正卡在 channel。
+	if a.sched.noteBackpressure() {
+		log.Printf("[agent] ready queue full (%d), input channel backpressured", a.sched.maxQueue)
+	}
 }
 
 // executeTask 执行一个任务（测试与旧调用方的入口）；见 executeNewTask。

internal/agent/core/scheduler_rearm_test.go

@@ -0,0 +1,186 @@
+package core
+
+// 回归测试：安全点「重新求值」、抢占计数语义、停机补终态、背压计数。
+//
+// 对照设计稿原文修正的四条：
+//
+//  1. §4.3/§5.2 —— 临界区期间到达的抢占请求「不丢失：按级别进入中断队列，
+//     在**临界区结束后的第一个安全点重新求值**」。实现里此前没有这一步：
+//     唯一的武装点是 registerInterrupt，凡被拦成「入队」的中断只能等当前任务
+//     **自然结束**。可复现症状：WebUI 终止按钮连按两次，第二次落在 2s 抢占冷却
+//     窗内 → 入队 → 再也不会被求值，「终止」看起来没反应。
+//  2. SchedulerStats.PreemptsByLevel 的语义是「判定可抢占**并进入 immediate** 的
+//     次数」；此前在 setImmediateLocked 之前就计数，于是同一安全点前到达的两条同级
+//     中断里、被降级入队的那条也被计入（immediate 是单槽，降级是设计要求的路径）。
+//  3. 状态面 Preempted 此前直接拿 Stats.Suspended 顶替，与 preempts_by_level 自相矛盾。
+//  4. §4.4 / §11.4 Q4 —— 就绪队列满必须「阻塞发送方 + **计数并打日志**」。
+
+import (
+	"testing"
+	"time"
+
+	agentAPI "gitcode.com/JianFeeeee/HomeAgent/internal/agent/api"
+	agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
+)
+
+// 被冷却拦成入队的中断，在冷却期满后的第一个安全点必须被重新武装。
+//
+// 这是「终止按钮连按两次」的最小复现：第一次抢占成功（受害者进入 2s 冷却），
+// 第二次在冷却窗内只能入队——修复前它就永远等不到执行了。
+func TestRearm_CooldownExpiryPromotesQueuedInterrupt(t *testing.T) {
+	a := newPreemptAgent(t, newPreemptProvider())
+
+	victim := &Task{ID: 1, Class: TaskInterrupt, Level: LevelBackground, EnqueuedAt: time.Now()}
+	a.sched.immediate = victim
+	a.sched.nextRef() // running = victim
+	// 模拟「刚被抢占过」：冷却起点就在此刻，且抢占提升已生效（有效级 L2）。
+	victim.PreemptCount = 1
+	victim.LastPreemptAt = time.Now()
+
+	evt, _ := textEvent("cli", "第二次终止")
+	evt.Payload["interrupt"] = true
+	if a.sched.requestPreempt(evt, LevelCritical) {
+		t.Fatal("抢占冷却期内不得抢占（应入队）")
+	}
+	if got := a.sched.stats.PreemptsByLevel[LevelCritical]; got != 0 {
+		t.Fatalf("被冷却拦成入队的中断不得计入抢占数，实际 %d", got)
+	}
+	if n := len(a.sched.interruptQueues[LevelCritical]); n != 1 {
+		t.Fatalf("应恰好入队一条，实际 %d（>1 说明入队路径重复）", n)
+	}
+
+	// 冷却期满 → 安全点的「重新求值」必须把它武装起来（修复前缺失的正是这一步）。
+	victim.LastPreemptAt = time.Now().Add(-3 * time.Second)
+	a.sched.rearmPending()
+	if !a.sched.preemptGrantedFor() {
+		t.Fatal("冷却期结束后应重新武装让位信号（设计 §4.3「第一个安全点重新求值」）")
+	}
+	snap := a.DumpScheduler()
+	if snap.Immediate == nil {
+		t.Fatal("重新求值后应把该中断提升进 immediate 槽")
+	}
+	if got := snap.Stats.PreemptsByLevel[LevelCritical]; got != 1 {
+		t.Fatalf("真正占住 immediate 才能计一次抢占，实际 %d", got)
+	}
+	// 重新求值不该把任务复制一份：队列必须空、immediate 恰好一条。
+	if n := len(snap.InterruptQueues[LevelCritical]); n != 0 {
+		t.Fatalf("提升后 L4 队列应空，实际 %d", n)
+	}
+}
+
+// 同一安全点前到达的两条同级中断：immediate 是单槽，第二条只能降级入队；
+// 它**没有**进入 immediate，因此不得计入 PreemptsByLevel，也不得被入队两次。
+func TestRearm_SameLevelSecondPreempterIsQueuedNotCounted(t *testing.T) {
+	a := newPreemptAgent(t, newPreemptProvider())
+
+	victim := &Task{ID: 1, Class: TaskQueued, EnqueuedAt: time.Now()}
+	a.sched.immediate = victim
+	a.sched.nextRef() // running = 排队任务（有效级 0，任何中断都能抢）
+
+	e1, _ := textEvent("cli", "irq-1")
+	if !a.sched.requestPreempt(e1, LevelInteractive) {
+		t.Fatal("第一条 L3 应抢占排队任务")
+	}
+	e2, _ := textEvent("cli", "irq-2")
+	a.sched.requestPreempt(e2, LevelInteractive) // 同级 → 降级入队
+
+	snap := a.DumpScheduler()
+	if got := snap.Stats.PreemptsByLevel[LevelInteractive]; got != 1 {
+		t.Fatalf("被降级的同级第二条不得计入抢占数（期望 1，实际 %d）", got)
+	}
+	if n := len(snap.InterruptQueues[LevelInteractive]); n != 1 {
+		t.Fatalf("被降级的那条应在 L3 队列里**恰好**出现一次，实际 %d", n)
+	}
+	if snap.Immediate == nil {
+		t.Fatal("第一条应留在 immediate 槽，两条都不能丢")
+	}
+}
+
+// 状态面 Preempted 必须是「各级抢占数之和」，不能拿 Suspended 顶替。
+func TestStatus_PreemptedEqualsSumOfLevels(t *testing.T) {
+	a := New(AgentConfig{ID: "rt-sum", ProviderManager: agentAPI.NewProviderManager(), IO: agentIO.NewIOManager()})
+	if a.sched == nil {
+		t.Fatal("agent 应带调度器")
+	}
+	a.sched.mu.Lock()
+	a.sched.stats.PreemptsByLevel[LevelBackground] = 2
+	a.sched.stats.PreemptsByLevel[LevelInteractive] = 3
+	a.sched.stats.Suspended = 99 // 故意与抢占数不等
+	a.sched.mu.Unlock()
+
+	got := a.schedulerStatus()
+	if got.Preempted != 5 {
+		t.Fatalf("Preempted 应为各级抢占数之和 5，实际 %d（拿 Suspended 顶替会得 99）", got.Preempted)
+	}
+	if got.Suspended != 99 {
+		t.Fatalf("Suspended 应原样透传，实际 %d", got.Suspended)
+	}
+}
+
+// 停机必须给「从未运行」与「已挂起」的同步任务补终态，
+// 否则 cli / clawhubadapter 这类无超时的同步注入方会永久挂起（设计 §7 I5、§11.3 X2/X4）。
+func TestStop_DrainsPendingSyncTasks(t *testing.T) {
+	a := newPreemptAgent(t, newPreemptProvider())
+
+	queuedEvt, queuedCh := textEvent("cli", "排队中，永远不会被调度")
+	if !a.sched.enqueue(newInputTask(queuedEvt)) {
+		t.Fatal("入队失败")
+	}
+	suspEvt, suspCh := textEvent("cli", "已挂起，停机时不会恢复")
+	a.sched.suspend(
+		&Task{ID: 2, Class: TaskInterrupt, Level: LevelInteractive, EnqueuedAt: time.Now(), Event: suspEvt},
+		a.newTaskFrame("挂起", a.stageCtxFromInput("挂起", "", "")),
+	)
+
+	a.Stop()
+
+	for name, ch := range map[string]chan *agentIO.OutputEvent{"排队": queuedCh, "挂起": suspCh} {
+		select {
+		case r := <-ch:
+			if r == nil || r.Payload["skipped"] != true {
+				t.Fatalf("%s 任务停机时应补 skipped 终态，实际 %+v", name, r)
+			}
+		case <-time.After(2 * time.Second):
+			t.Fatalf("%s 任务停机未补终态（同步调用方会永久挂起）", name)
+		}
+	}
+}
+
+// 背压计数：持续满只报一次「翻转」不发生；计数本身每次都要累加。
+func TestBackpressure_CounterAndTransition(t *testing.T) {
+	s := newScheduler(1)
+	if !s.noteBackpressure() {
+		t.Fatal("首次背压应报告「翻转」")
+	}
+	if s.noteBackpressure() {
+		t.Fatal("持续背压不得重复报告翻转（否则日志会被刷爆）")
+	}
+	if s.stats.Backpressure != 2 {
+		t.Fatalf("背压计数应为 2，实际 %d", s.stats.Backpressure)
+	}
+	s.clearBackpressure()
+	if !s.noteBackpressure() {
+		t.Fatal("队列恢复后再满应再次报告翻转")
+	}
+}
+
+// 集成：就绪队列满时 pumpInbox 必须计一次背压（Rejected 保持 0——背压不是丢弃）。
+func TestBackpressure_PumpInboxCountsWhenFull(t *testing.T) {
+	a := newPreemptAgent(t, newPreemptProvider())
+	a.sched.maxQueue = 1
+
+	a.io.InjectInput("cli", "text", map[string]interface{}{"content": "第一条"})
+	a.io.InjectInput("cli", "text", map[string]interface{}{"content": "第二条"})
+	a.pumpInbox()
+
+	snap := a.DumpScheduler()
+	if len(snap.Queue) != 1 {
+		t.Fatalf("maxQueue=1 时队列应恰好 1 条，实际 %d", len(snap.Queue))
+	}
+	if snap.Stats.Backpressure == 0 {
+		t.Fatal("队列满必须计一次背压（设计 §4.4/Q4：阻塞发送方 + 计数）")
+	}
+	if snap.Stats.Rejected != 0 {
+		t.Fatalf("背压不是丢弃，Rejected 必须保持 0，实际 %d", snap.Stats.Rejected)
+	}
+}

internal/agent/core/task.go

@@ -178,6 +178,10 @@ func (a *Agent) runTaskSteps(f *TaskFrame) stepOutcome {
 	for i := 0; i < maxSteps; i++ {
 		// 安全点：只在 step 之间检查让位。临界区（StepToolExec）不在此列，
 		// 因为让位信号由 interruptLoop 置位、而本循环是唯一读帧者。
+		//
+		// 先「重新求值」再判让位：临界区（或抢占冷却期）内被拦成入队的中断，
+		// 必须在这里重新武装——否则它只能等当前任务自然结束（设计 §4.3/§5.2）。
+		a.sched.rearmPending()
 		if !isCriticalChannel(f.OutputChannel) && a.sched.preemptGrantedFor() && a.sched.canSuspend() {
 			return outcomeSuspended
 		}

internal/devicebridge/client/bridge.go

@@ -411,7 +411,12 @@ func (b *Bridge) readLoop() {
 				_ = ws.writePong()
 				continue
 			}
-			// 超时或其他错误，退出
+			// 超时或其他错误，退出。
+			//
+			// **必须记日志**：此前这里静默 return，设备断线的真因（读超时 / 对端
+			// 关闭 / 帧错）在设备侧完全不可见，只能靠对端日志倒推。
+			// 2 倍 ping 间隔内的读超时通常是“心跳没人回”——查服务端 writePong 是否真发出。
+			log.Printf("[devicebridge] read loop exit (opcode=%#x, close=%v): %v", opcode, isClose, err)
 			return
 		}
 		if isClose {

internal/plugins/remotedevice/ping_test.go

@@ -0,0 +1,53 @@
+package remotedevice
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+// 心跳回包必须**真的发出去**：pong 只有两个字节，且设备空闲时没有任何别的写
+// 会顺带把 bufio 缓冲刷出去——`writePong` 一旦忘了 Flush，pong 就永远留在
+// 服务端缓冲里。
+//
+// 这就是「device channel 不稳定」的真因（实测）：客户端每 30s 发一个 ping，
+// 服务端算好了 pong 却没发；客户端的读循环设的是 2 倍 ping 间隔（默认 60s）
+// 读超时，于是**每 60 秒准点断开一次**，重连后 outputch 被注销又注册，
+// 模型侧看到的就是工具/通道凭空消失又出现。
+//
+// 本用例只发一个 ping，随后**什么都不发**：pong 必须在无后续流量的情况下到达。
+func TestWSPingGetsPongWhileIdle(t *testing.T) {
+	reg := NewRegistry()
+	token := "test-token-ping"
+	reg.SetAcceptToken(func(provided string) bool { return provided == token })
+
+	srv := httptest.NewServer(http.HandlerFunc(reg.ServeWS))
+	defer srv.Close()
+
+	cli := dialTestWS(t, srv.URL, token)
+	defer cli.close()
+
+	// 先走完 hello + bind（服务端要先把设备登记进 conns，pong 才写得回来）。
+	cli.sendText([]byte(`{"op":"hello","device":{"device_id":"ping-dev","name":"前端机","kind":"computer","caps":["cmd"]}}`))
+	cli.readHelloAckAndBind(t, token)
+
+	cli.sendFrame(0x9, nil) // ping
+
+	if err := cli.conn.SetReadDeadline(time.Now().Add(2 * time.Second)); err != nil {
+		t.Fatalf("set read deadline: %v", err)
+	}
+	payload, isClose, opcode, err := readFrame(cli.rw.Reader)
+	if err != nil {
+		t.Fatalf("2s 内没收到 pong（writePong 忘了 Flush？）: %v", err)
+	}
+	if isClose {
+		t.Fatal("连接被关闭，而不是回了 pong")
+	}
+	if opcode != 0xa {
+		t.Fatalf("期望 pong(0xa)，实际 opcode=%#x payload=%q", opcode, payload)
+	}
+	if len(payload) != 0 {
+		t.Fatalf("pong 不该带负载，实际 %q", payload)
+	}
+}

internal/plugins/remotedevice/registry.go

@@ -606,7 +606,13 @@ func writeFrame(w *bufio.Writer, opcode byte, payload []byte) error {
 }
 
 func writePong(w *bufio.Writer) error {
-	return writeFrameHeader(w, 0xa, 0)
+	// 必须走 writeFrame（它 Flush）。
+	//
+	// 回归的 bug：这里原先是裸的 writeFrameHeader，**不 Flush**。设备空闲时
+	// 没有任何别的写会顺带把 bufio 缓冲刷出去，于是 pong 永远留在服务端缓冲里，
+	// 客户端等 2 倍 ping 间隔（默认 30s×2 = 60s）读超时断开、重连——
+	// 实测表现就是「设备通道每 60 秒掉线一次」，连带着 outputch 反复注销/注册。
+	return writeFrame(w, 0xa, nil)
 }
 
 func writeFrameHeader(w *bufio.Writer, opcode byte, length int) error {

internal/sdk/status.go

@@ -68,7 +68,11 @@ type SchedulerStatus struct {
 	Rejected  uint64 `json:"rejected"`
 	Suspended uint64 `json:"suspended"`
 	Resumed   uint64 `json:"resumed"`
+	// Preempted = Σ PreemptsByLevel[1..4]，即「真正抢占成功」的次数。
+	// 它与 Suspended 不等价（受害者可能先自行结束），因此不是 Suspended 的别名。
 	Preempted uint64 `json:"preempted"`
+	// Backpressure 是就绪队列满、输入被挡回 channel 的次数（暂时不收，不是丢弃）。
+	Backpressure uint64 `json:"backpressure"`
 }
 
 // SchedulerTask 是任务的最小标识（不暴露帧内容）。