fix(obs): 抢占日志改在判决点打（修自伤）+ scheduler 事件接进 SSE

## 修我上一版的自伤

上一版把 preempt 日志打在 `executeNewTask`，但那时 `nextRef` 已经把
`s.running` 换成了抢占者自己，于是输出成了

    preempt start: task#2 ... -> victim task#2 (cli)

victim 打印的是入侵者本人。判据必须落在 `registerInterrupt`——那一刻
running 还是真正的受害者。改为在抢占判决点打：

    [agent] preempt: task#2 class=interrupt level=3 from cli (L3) preempts task#1 class=queued level=0 (qq)

## 补上「入队而非抢占」的日志

中断到了却没生效，此前完全不可解释。现在两种成因分开写：

    [agent] interrupt queued: ... vs ... (qq) — running in critical section; queue=N
    [agent] interrupt queued: ... vs ... (qq) — preempt cooldown; queue=N
    [agent] interrupt queued: ... vs ... (qq) — level insufficient; queue=N

没有这条，`interrupt from X` 打过之后任务为什么没让位就只能猜。

## scheduler 事件接进 SSE

`EventScheduler` 此前既不在 `handler_chat.go` 的 subTypes、也没有任何订阅者
（全仓 grep 零命中）——内核里 suspend/resume 只 publishEvent，于是事件发出来
就掉地上，对内对外都不可见。加进 subTypes 后前端/客户端能看到抢占链。

## 验证

- preempt_logging_test.go 增一条：victim 与入侵者必须是不同来源（qq vs cli），
  且排队输入对排队任务 `canPreempt` 必为假。
- 实测输出含 `cli (L3) preempts task#1 class=queued level=0 (qq)`。
- 全量 `go test ./internal/... ./cmd/...` 与 `go vet ./internal/...` 全绿。

internal/agent/core/preempt_logging_test.go

@@ -62,3 +62,28 @@ func TestSourceOf_SelfAndNilAreSafe(t *testing.T) {
 		t.Fatalf("self task -> %q", got)
 	}
 }
+
+// 抢占日志必须把「入侵者」与「受害者」分开写，且受害者是**真正在跑的那个**。
+//
+// 镇的是一个自伤：第一版把日志打在 executeNewTask 里，而那时 nextRef 已经把
+// s.running 换成了抢占者自己，于是日志写成 "victim = 入侵者"（实测输出过
+// `preempt start: task#2 ... -> victim task#2 (cli)`）。判据必须落在
+// registerInterrupt —— 那一刻 running 还没被换。
+func TestSourceOf_DistinctVictimAndIntruder(t *testing.T) {
+	if sourceOf(newInputTask(&agentIO.InputEvent{Source: "qq"}), nil) != "qq" {
+		t.Fatal("queued qq 任务的来源应为 qq")
+	}
+	it := newInterruptTask(&agentIO.InputEvent{Source: "homeagent-mail-bridge"}, LevelMessage)
+	if got := sourceOf(it, nil); got != "homeagent-mail-bridge" {
+		t.Fatalf("中断来源应为 homeagent-mail-bridge，实际 %q", got)
+	}
+	if it.Class != TaskInterrupt || it.Level != LevelMessage {
+		t.Fatalf("中断任务应 class=interrupt level=2，实际 %v/%v", it.Class, it.Level)
+	}
+	// 邮件若走排队注入则会变成 TaskQueued —— 那种情况下 canPreempt 必为假。
+	q := &Task{Class: TaskQueued, Kind: TaskKindInput,
+		Event: &agentIO.InputEvent{Source: "homeagent-mail-bridge"}}
+	if canPreempt(q, newInputTask(&agentIO.InputEvent{Source: "qq"})) {
+		t.Fatal("排队输入永远不得抢占另一个排队任务")
+	}
+}

internal/agent/core/scheduler.go

@@ -706,11 +706,29 @@ func (s *scheduler) registerInterrupt(t *Task) bool {
 				s.preemptArmed = true
 				s.preemptLevel = t.Level
 				s.stats.bumpInterruptLevel(&s.stats.PreemptsByLevel, t.Level)
+				// 抢占判决点：此刻 running 还是真正的受害者，记下来才能回答
+				// 「我的任务被谁打断了」。放在 executeNewTask 里打是错的——
+				// 那时 nextRef 已把 running 换成抢占者自己，日志会写成
+				// "victim = 入侵者自己"（实测过这个错误输出）。
+				log.Printf("[agent] preempt: %s from %s (L%d) preempts %s (%s)",
+					describeTask(t), sourceOf(t, nil), int(t.Level),
+					describeTask(running), sourceOf(running, nil))
 			}
 		}
 	}
 	if !arm {
 		s.enqueueInterruptLocked(t)
+		// 入队而不是抢占：这也是要能看见的（否则「中断明明到了却没生效」无从解释）。
+		// 两种成因分开写：临界区 vs 级别不够/冷却期。
+		reason := "level insufficient"
+		if critical {
+			reason = "running in critical section"
+		} else if canPreempt(t, running) && running != nil && !running.LastPreemptAt.IsZero() && time.Since(running.LastPreemptAt) < preemptCooldown {
+			reason = "preempt cooldown"
+		}
+		log.Printf("[agent] interrupt queued: %s from %s (L%d) vs %s (%s) — %s; queue=%d",
+			describeTask(t), sourceOf(t, nil), int(t.Level),
+			describeTask(running), sourceOf(running, nil), reason, s.interruptCountLocked())
 	}
 	s.mu.Unlock()
 
@@ -1164,17 +1182,6 @@ func (a *Agent) executeNewTask(t *Task) {
 	var f *TaskFrame
 	var out stepOutcome = outcomeDone
 
-	if t.Class == TaskInterrupt {
-		// 让位之前先记一笔「谁将要被打断」——这是排查「任务被莫名打断」的锚点。
-		// victim 从调度器取：此刻 s.running 还是被抢占者本身（suspend 里才清）。
-		a.sched.mu.Lock()
-		victim := describeTask(a.sched.running)
-		victimSrc := sourceOf(a.sched.running, nil)
-		a.sched.mu.Unlock()
-		log.Printf("[agent] preempt start: %s from %s (level=%d) -> victim %s (%s)",
-			describeTask(t), sourceOf(t, nil), int(t.Level), victim, victimSrc)
-	}
-
 	func() {
 		defer func() {
 			if r := recover(); r != nil {

internal/plugins/webui/handler_chat.go

@@ -728,7 +728,13 @@ func (h *Handler) handleChatEvents(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	subTypes := []string{"agent_output", "reasoning", "agent_error", "tool_call", "stage", "agent_llm_chain", "terminal_output"}
+	// scheduler：输入调度器的状态变更（抢占 / 挂起 / 恢复）。
+	//
+	// 为何必须转发：内核里 suspend/resume 只发事件（见 core/scheduler.go 的
+	// executeNewTask/resumeTask），而 `scheduler` 曾既不在本列表里、也没有任何
+	// 订阅方—— 事件发出来就掉地上。后果是「我的任务被谁打断了」对内对外都
+	// 不可见，排查时只能按时序猜。这里补上对外那一半。
+	subTypes := []string{"agent_output", "reasoning", "agent_error", "tool_call", "stage", "agent_llm_chain", "terminal_output", "scheduler"}
 	// token 级流式增量事件：实时转发给浏览器做逐 token 渲染。
 	// 不进 sseEventRing —— 断线重连只重放聚合事件（最终真相），
 	// 避免重放 delta 与聚合内容重复追加。