From 01113664b4830bdada1c091b1414d91f9f410c89 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sat, 19 Sep 2026 11:57:50 +0800 Subject: [PATCH] =?UTF-8?q?fix(obs):=20=E6=8A=A2=E5=8D=A0=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E6=94=B9=E5=9C=A8=E5=88=A4=E5=86=B3=E7=82=B9=E6=89=93?= =?UTF-8?q?=EF=BC=88=E4=BF=AE=E8=87=AA=E4=BC=A4=EF=BC=89+=20scheduler=20?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E6=8E=A5=E8=BF=9B=20SSE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 修我上一版的自伤 上一版把 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 | 25 ++++++++++++++++++ internal/agent/core/scheduler.go | 29 +++++++++++++-------- internal/plugins/webui/handler_chat.go | 8 +++++- 3 files changed, 50 insertions(+), 12 deletions(-) diff --git a/internal/agent/core/preempt_logging_test.go b/internal/agent/core/preempt_logging_test.go index 09583c9..f5dc3bc 100644 --- a/internal/agent/core/preempt_logging_test.go +++ b/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("排队输入永远不得抢占另一个排队任务") + } +} diff --git a/internal/agent/core/scheduler.go b/internal/agent/core/scheduler.go index e79e449..62eed2e 100644 --- a/internal/agent/core/scheduler.go +++ b/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 { diff --git a/internal/plugins/webui/handler_chat.go b/internal/plugins/webui/handler_chat.go index 49c5db3..9a0288a 100644 --- a/internal/plugins/webui/handler_chat.go +++ b/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 与聚合内容重复追加。