feat(obs): 抢占日志说出「受害者是谁」——suspend/resume 此前完全不落日志

排查「我的任务怎么被莫名打断了」时撞上的观测缺口。

## 缺口

`executeNewTask` 里挂起、`resumeTask` 里恢复,两处都**只发事件、不写日志**:

    a.sched.suspend(t, f)
    a.publishEvent(events.EventScheduler, map[string]any{"action": "suspend", ...})

于是生产日志里只有两行:`interrupt from X` 与 `LLM request cancelled by
preemption` —— **看不到受害者是谁、被谁挤下去、后来有没有恢复**。后果是实测过的:
按时间先后猜凶手,把时间上相邻的输入误认成抢占者。

## 改动

- `sourceOf(task, frame)`:取可辨识来源(`evt.Source` 优先,回退 OutputChannel,
  自循环任务给 `self:<channel>`)。取 Source 而**不是** OutputChannel:
  前者回答「谁送来的」(qq / homeagent-mail-bridge / timer / child/xxx),
  后者只回答投递到哪个通道;多数场景同名,但因果链上要的是前者。
- `describeTask(task)`:`task#N class=queued|interrupt level=L`。
- `suspendDepth()`:日志专用,走锁而不是让日志点直接摸 `suspendStack`。
- 三个日志点:抢占开始(含 victim)、挂起(含来源与栈深)、恢复。

输出形状:

    [agent] preempt start: task#2 class=interrupt level=4 from cli -> victim task#1 class=queued level=0 (qq)
    [agent] suspend: task#1 class=queued level=0 (qq) yields to an interrupt; suspendStack=0
    [agent] resume: task#1 class=queued level=0 (qq) resumes after the interrupt finished

## 验证

- 新增 preempt_logging_test.go:抢占后栈深 0→1、sourceOf 取到 qq、self/nil 不 panic。
- 实测日志(TestPreempt_HigherPreemptsAndResumes)三条齐全,能一眼看出
  是 `cli` 的 L4 挤掉了 `qq` 的排队任务、随后 qq 恢复。
- `go test ./internal/... ./cmd/...` 全绿。
This commit is contained in:
JianFeeeee
2026-09-19 11:31:04 +08:00
parent 545d3b45a1
commit fcaf3ffdb7
2 changed files with 132 additions and 0 deletions

View File

@ -0,0 +1,64 @@
package core
import (
"testing"
"time"
agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io"
)
// 抢占日志必须能说出**受害者是谁**(来源 + 类别 + 级别)。
// 此前 suspend/resume 不落日志,生产上无法回答"我的任务被谁打断了"。
func TestPreemptLogging_NamesVictimAndSource(t *testing.T) {
sp := newPreemptProvider("intr-done", "low-done")
a := newPreemptAgent(t, sp)
lowEvt, _ := textEvent("qq", "低优先级任务")
lowTask := newInputTask(lowEvt)
if !a.sched.enqueue(lowTask) {
t.Fatal("入队失败")
}
lt, _, _ := a.sched.nextRef()
done := make(chan struct{})
go func() { a.executeNewTask(lt); close(done) }()
select {
case <-sp.entered:
case <-time.After(3 * time.Second):
t.Fatal("provider 未被调用")
}
if a.sched.suspendDepth() != 0 {
t.Fatalf("初始栈深应为 0,实际 %d", a.sched.suspendDepth())
}
intrEvt := &agentIO.InputEvent{
Source: "cli", OutputChannel: "cli", Type: "text",
Payload: map[string]interface{}{"content": "紧急打断", "interrupt": true},
}
a.sched.requestPreempt(intrEvt, LevelInteractive)
a.cancelCurrentLLM()
select {
case <-done:
case <-time.After(3 * time.Second):
t.Fatal("未挂起")
}
if d := a.sched.suspendDepth(); d != 1 {
t.Fatalf("挂起后栈深应为 1,实际 %d", d)
}
// sourceOf 要能从任务/帧里取出可辨识来源(qq),而不是空串或 task#N。
if got := sourceOf(lt, nil); got != "qq" {
t.Fatalf("sourceOf(lt) = %q,期望 qq", got)
}
}
func TestSourceOf_SelfAndNilAreSafe(t *testing.T) {
if got := sourceOf(nil, nil); got != "?" {
t.Fatalf("nil -> %q", got)
}
st := newSelfTask(selfInputMsg{text: "x", channel: "cli"})
if got := sourceOf(st, nil); got != "self:cli" {
t.Fatalf("self task -> %q", got)
}
}

View File

@ -810,6 +810,46 @@ func (s *scheduler) suspend(t *Task, f *TaskFrame) {
s.preemptLevel = 0
}
// sourceOf 从任务/帧里取一个**可辨识来源**,用于抢占日志与故障定位。
//
// 为什么必须记这个:此前 suspend/resume 只发事件不落日志(见
// executeNewTask/resumeTask),于是生产上「我的任务被谁打断了」完全不可查——
// 日志里只有 `interrupt from X` 和 `LLM request cancelled by preemption` 两行,
// **看不到受害者是谁**。排查时只能按时间猜,把相邻的输入误认成凶手。
//
// 取静态注入源(evt.Source)而不是 evt.OutputChannel:前者回答「谁送来的」
// (qq / homeagent-mail-bridge / timer / child/xxx),后者是回答要投到哪个通道,
// 两者在多数场景下同名,但前者才是因果链上的那一环。
func sourceOf(t *Task, f *TaskFrame) string {
if f != nil && f.Evt != nil {
if f.Evt.Source != "" {
return f.Evt.Source
}
return f.Evt.OutputChannel
}
if t != nil && t.Event != nil {
if t.Event.Source != "" {
return t.Event.Source
}
return t.Event.OutputChannel
}
if t != nil && t.Kind == TaskKindSelf {
if t.Self.channel != "" {
return "self:" + t.Self.channel
}
return "self"
}
return "?"
}
// describeTask 把任务的类别/级别拼成一段可读标签(日志用)。
func describeTask(t *Task) string {
if t == nil {
return "<nil>"
}
return fmt.Sprintf("task#%d class=%s level=%d", t.ID, t.Class, t.Level)
}
// canSuspend 报告还有下潜余量(安全点用它决定是否真的让位)。
func (s *scheduler) canSuspend() bool {
s.mu.Lock()
@ -817,6 +857,16 @@ func (s *scheduler) canSuspend() bool {
return len(s.suspendStack) < s.maxInterruptFrames
}
// suspendDepth 返回当前中断栈深度(仅用于日志)。
//
// 单独一个方法而不是让调用方直接读 s.suspendStack:那个字段只允许在 s.mu 下访问,
// 而日志点不应该自己摸调度器内部状态(也不应该为了打一行日志多持一次锁的窗口)。
func (s *scheduler) suspendDepth() int {
s.mu.Lock()
defer s.mu.Unlock()
return len(s.suspendStack)
}
// pendingEvents 收集**尚未执行**(排队队列 / 四条中断队列 / immediate)与
// **已挂起**(中断栈)任务所携带的、且带同步回执通道的输入事件。
//
@ -1114,6 +1164,17 @@ 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 {
@ -1131,6 +1192,11 @@ func (a *Agent) executeNewTask(t *Task) {
}()
if out == outcomeSuspended && f != nil {
// 挂起必须落日志:此前只发事件不落盘,导致生产日志里
// 「谁把谁挤下去了」完全查不到(只有 interrupt from / cancelled 两行)。
// 曾因此把时间上相邻的输入误判成凶手。
log.Printf("[agent] suspend: %s (%s) yields to an interrupt; suspendStack=%d",
describeTask(t), sourceOf(t, f), a.sched.suspendDepth())
a.sched.suspend(t, f)
a.publishEvent(events.EventScheduler, map[string]interface{}{
"action": "suspend", "task": t.ID, "level": int(t.Level),
@ -1151,6 +1217,8 @@ func (a *Agent) resumeTask(t *Task, f *TaskFrame) {
a.sched.mu.Lock()
a.sched.stats.Resumed++
a.sched.mu.Unlock()
log.Printf("[agent] resume: %s (%s) resumes after the interrupt finished",
describeTask(t), sourceOf(t, f))
a.publishEvent(events.EventScheduler, map[string]interface{}{
"action": "resume", "task": t.ID, "level": int(t.Level),
})