From f33ebee01b2c7893d0850105f29acb19f3559e3c Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Tue, 6 Oct 2026 09:22:12 +0800 Subject: [PATCH] =?UTF-8?q?test(agent/core):=20=E6=B6=88=E9=99=A4=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E5=AF=B9=20scheduler.queue=20=E7=9A=84=E6=97=A0?= =?UTF-8?q?=E9=94=81=E7=9B=B4=E8=AF=BB=EF=BC=8C=E5=B9=B6=E5=8A=A0=E5=88=A4?= =?UTF-8?q?=E6=8D=AE=E9=98=B2=E5=9B=9E=E5=BD=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 起因 offload_test.go 等 4 个文件里有 18 处 `len(s.queue)` / `range s.queue`， 全在锁外。当时**安全**——那些测试用 newRootWithoutSchedulerLoop， 刻意不启 schedulerLoop，所以没有并发写者。 但这个安全性 ① 不由测试自身保证，只由「别改构造函数」这个口头约定保证； ② 一旦有人换成 newRootWith 就变成真竞争，症状是「今天全绿、改一次就炸」。 先确认生产代码没问题：逐处核对 scheduler.go 里 6 个读 s.queue 的函数， 全部在 mu.Lock/RLock 内（takeQueuedInputs/hasRoom/nextRef/armStop/ pendingEvents/takeAllPendingEvents）。所以这是**测试的隐患**，不是核心缺陷。 ## 改动 - scheduler.go 新增两个带锁只读访问器： · queueLen() —— 替代 len(s.queue) · queueSnapshot() —— 返回浅拷贝，替代 range/索引（拷贝保证 len 与 range 不会在遍历中途变化） - 4 个测试文件 18 处直读全部改走访问器 - 新增 queue_access_test.go：AST 静态判据，扫所有 *_test.go， 同一函数内出现过 Lock/RLock 即视为持锁（Go 的 defer Unlock 静态难还原， 按函数近似足够保守）；命中无锁直读即红 ## 判据自检 静态判据最大的失败模式是「根本没抓到东西」，所以带双向自检： - 正向：注入一处无锁直读 → 必须被抓到 - 反向：注入一处持锁直读 → 必须**不**被报（否则噪声大没人看） - 真变异：把一处访问器改回 `len(s.queue)` → 红 ## 验证 - go build ./... / go vet / go test ./internal/... 全绿 - go test -race ./internal/agent/core/ 无数据竞争 - 部署后：39 插件加载、3 个 multimodal 工具注册、0 panic/fatal - graph.db integrity ok，blocks 2787 未动 Co-Authored-By: Claude Opus 4.8 (1M context) --- internal/agent/core/offload_test.go | 47 ++++--- internal/agent/core/queue_access_test.go | 169 +++++++++++++++++++++++ internal/agent/core/scheduler.go | 28 ++++ internal/agent/core/scheduler_test.go | 6 +- 4 files changed, 224 insertions(+), 26 deletions(-) create mode 100644 internal/agent/core/queue_access_test.go diff --git a/internal/agent/core/offload_test.go b/internal/agent/core/offload_test.go index 1c46a182..01173854 100644 --- a/internal/agent/core/offload_test.go +++ b/internal/agent/core/offload_test.go @@ -61,8 +61,8 @@ func TestTakeQueuedInputsOnlyTakesQueuedInputs(t *testing.T) { s.enqueue(makeQueuedInput(3)) s.enqueue(makeQueuedInput(4)) - if len(s.queue) != 6 { - t.Fatalf("就绪队列应有 6 条（4 排队输入 + 1 中断 + 1 self），实际 %d", len(s.queue)) + if n := s.queueLen(); n != 6 { + t.Fatalf("就绪队列应有 6 条（4 排队输入 + 1 中断 + 1 self），实际 %d", n) } got := s.takeQueuedInputs(3) if len(got) != 3 { @@ -75,7 +75,7 @@ func TestTakeQueuedInputsOnlyTakesQueuedInputs(t *testing.T) { } // self 与中断必须还在 var hasSelf, hasInterrupt bool - for _, tt := range s.queue { + for _, tt := range s.queueSnapshot() { if tt.Kind == TaskKindSelf { hasSelf = true } @@ -101,8 +101,8 @@ func TestTakeQueuedInputsIsAllOrNothing(t *testing.T) { if got := s.takeQueuedInputs(3); got != nil { t.Fatalf("不足 3 条时不应取走任何任务，实际取走 %d", len(got)) } - if len(s.queue) != 2 { - t.Errorf("队列不应被改动，实际剩 %d", len(s.queue)) + if n := s.queueLen(); n != 2 { + t.Errorf("队列不应被改动，实际剩 %d", n) } } @@ -118,17 +118,18 @@ func TestRequeueFrontKeepsAllTasks(t *testing.T) { if len(taken) != 3 { t.Fatalf("应取走 3 条，实际 %d", len(taken)) } - if len(s.queue) != 0 { - t.Fatalf("取走后队列应空，实际 %d", len(s.queue)) + if n := s.queueLen(); n != 0 { + t.Fatalf("取走后队列应空，实际 %d", n) } s.requeueFront(taken) - if len(s.queue) != 3 { - t.Fatalf("★ 放回后必须一条不少：期望 3，实际 %d", len(s.queue)) + if n := s.queueLen(); n != 3 { + t.Fatalf("★ 放回后必须一条不少：期望 3，实际 %d", n) } // 放回的是**前端**：它们比队列里原有的一切都早 s.enqueue(makeQueuedInput(4)) - if s.queue[len(s.queue)-1].Event.Payload["content"] != "hello" { + snap := s.queueSnapshot() + if snap[len(snap)-1].Event.Payload["content"] != "hello" { t.Error("放回的任务应在队列前端") } } @@ -160,8 +161,8 @@ func TestOffloadDisabledByDefault(t *testing.T) { if n := a.offloadPendingTasks(opts); n != 0 { t.Errorf("关闭时不得转投，实际转了 %d", n) } - if len(s.queue) != 3 { - t.Errorf("关闭时队列不得被改动，实际 %d", len(s.queue)) + if n := s.queueLen(); n != 3 { + t.Errorf("关闭时队列不得被改动，实际 %d", n) } } @@ -225,10 +226,10 @@ func TestOffloadMovesTasksToResidentEndToEnd(t *testing.T) { t.Logf("驻留子 %s: inputch=%v outputs=%v", resident.ID, resident.InputChs, resident.AllowedOutputs) // ③ 原队列里留下说明（且说明是内核发的） - if len(root.sched.queue) != 1 { - t.Fatalf("原队列应只剩 1 条说明，实际 %d", len(root.sched.queue)) + if root.sched.queueLen() != 1 { + t.Fatalf("原队列应只剩 1 条说明，实际 %d", root.sched.queueLen()) } - notice := root.sched.queue[0] + notice := root.sched.queueSnapshot()[0] if notice.Event == nil || notice.Event.Source != "kernel" { t.Fatalf("留下的应是内核说明，实际 %+v", notice.Event) } @@ -330,8 +331,8 @@ func TestOffloadSeesInputsStuckInChannel(t *testing.T) { root.io.InjectInputTo("webui", "webui", "text", map[string]interface{}{"content": "stuck"}) } - if len(root.sched.queue) != 0 { - t.Fatalf("前置条件：此时 sched.queue 应为 0（输入还没被搬运），实际 %d", len(root.sched.queue)) + if root.sched.queueLen() != 0 { + t.Fatalf("前置条件：此时 sched.queue 应为 0（输入还没被搬运），实际 %d", root.sched.queueLen()) } if root.io.PendingInputs() != 3 { t.Fatalf("前置条件：输入应堆在 channel 里，实际 %d", root.io.PendingInputs()) @@ -502,11 +503,11 @@ func TestResidualKeepReturnsTasksToParent(t *testing.T) { t.Fatalf("应处置 2 条，实际 %d（msg=%s）", n, msg) } // keep = 转回父自己：两条都要出现在父的队列里，且**带 ResponseCh 的那条仍带** - if len(root.sched.queue) != 2 { - t.Fatalf("两条残余任务应转回父队列，实际 %d", len(root.sched.queue)) + if root.sched.queueLen() != 2 { + t.Fatalf("两条残余任务应转回父队列，实际 %d", root.sched.queueLen()) } var hasResponseCh bool - for _, task := range root.sched.queue { + for _, task := range root.sched.queueSnapshot() { if task.Event != nil && task.Event.ResponseCh != nil { hasResponseCh = true } @@ -557,8 +558,8 @@ func TestResidualDropNotifiesSyncCaller(t *testing.T) { t.Fatal("★ drop 未通知同步调用方：cli/a2a 会永久挂起") } // drop 后父队列不应多出东西 - if len(root.sched.queue) != 0 { - t.Errorf("drop 不应把任务转回父队列，实际 %d", len(root.sched.queue)) + if root.sched.queueLen() != 0 { + t.Errorf("drop 不应把任务转回父队列，实际 %d", root.sched.queueLen()) } } @@ -657,7 +658,7 @@ func TestKernelNoticeIsNeverOffloaded(t *testing.T) { } // 说明必须还在队列里（留给主 agent 看），不能被搬走 var noticeLeft bool - for _, task := range root.sched.queue { + for _, task := range root.sched.queueSnapshot() { if task.Event != nil && isKernelNotice(task.Event) { noticeLeft = true } diff --git a/internal/agent/core/queue_access_test.go b/internal/agent/core/queue_access_test.go new file mode 100644 index 00000000..fb46f65e --- /dev/null +++ b/internal/agent/core/queue_access_test.go @@ -0,0 +1,169 @@ +package core + +import ( + "fmt" + "go/ast" + "go/parser" + "go/token" + "os" + "path/filepath" + "strings" + "testing" +) + +// 判据：测试代码不得在锁外直读 scheduler 的内部队列字段。 +// +// ## 起因（2026-10-06） +// +// offload_test.go 曾有 6 处 `len(root.sched.queue)` / `range root.sched.queue`， +// 全在锁外。当时**安全**——那些测试用 newRootWithoutSchedulerLoop， +// 刻意不启 schedulerLoop，所以没有并发写者。但这个安全性： +// +// ① 不由测试自身保证，只由「别改构造函数」这个口头约定保证； +// ② 一旦有人把 WithoutSchedulerLoop 换成 newRootWith，就变成真竞争， +// 而症状是「今天全绿、改一次就炸」。 +// +// 这正是本项目反复记录的形态：**隐患不在当前代码里，在下一次修改里。** +// +// ## 判据怎么做到不误报 +// +// · 只扫 `*_test.go`（生产代码直读是另一回事，且已全部在锁内） +// · 跳过 queueLen()/queueSnapshot()（带锁访问器） +// · 同一函数内**出现过** Lock/RLock 即视为整函数持锁 +// （Go 的 defer Unlock 很难静态还原，按函数近似足够保守） +// +// 判据不可达 = 判据描述的状态不存在。所以本判据自带自检： +// TestQueueAccessCriterion_SelfCheck 注入一处无锁直读，要求被抓到。 + +func TestNoUnlockedQueueReadInTests(t *testing.T) { + ents, err := os.ReadDir(".") + if err != nil { + t.Fatalf("读包目录: %v", err) + } + var violations []string + for _, e := range ents { + name := e.Name() + if !strings.HasSuffix(name, "_test.go") { + continue + } + violations = append(violations, unlockedQueueReadsInFile(t, filepath.Join(".", name))...) + } + if len(violations) > 0 { + t.Errorf("测试里不得在锁外直读调度器队列（请用 queueLen()/queueSnapshot()）：\n\t%s", + strings.Join(violations, "\n\t")) + } +} + +// unlockedQueueReadsInFile 解析一个文件并返回所有「无锁直读 .queue」的位置。 +func unlockedQueueReadsInFile(t *testing.T, path string) []string { + t.Helper() + fset := token.NewFileSet() + f, err := parser.ParseFile(fset, path, nil, 0) + if err != nil { + t.Fatalf("解析 %s: %v", path, err) + } + parent := buildParentMap(f) + // 先标记「函数体内出现过 Lock/RLock」 + lockedFn := map[*ast.FuncDecl]bool{} + ast.Inspect(f, func(n ast.Node) bool { + sel, ok := n.(*ast.SelectorExpr) + if !ok || (sel.Sel.Name != "Lock" && sel.Sel.Name != "RLock") { + return true + } + if fn := enclosingFunc(parent, n); fn != nil { + lockedFn[fn] = true + } + return true + }) + + var out []string + ast.Inspect(f, func(n ast.Node) bool { + sel, ok := n.(*ast.SelectorExpr) + if !ok || sel.Sel.Name != "queue" { + return true + } + if fn := enclosingFunc(parent, n); fn != nil && lockedFn[fn] { + return true // 该函数整体持锁 + } + pos := fset.Position(sel.Pos()) + out = append(out, fmt.Sprintf("%s:%d: 直接访问 .queue（无锁）", path, pos.Line)) + return true + }) + return out +} + +// buildParentMap 建立 AST 父指针表（ast.Inspect 的回调拿不到父节点）。 +func buildParentMap(f *ast.File) map[ast.Node]ast.Node { + parent := map[ast.Node]ast.Node{} + var stack []ast.Node + ast.Inspect(f, func(n ast.Node) bool { + if n == nil { + stack = stack[:len(stack)-1] + return false + } + if len(stack) > 0 { + parent[n] = stack[len(stack)-1] + } + stack = append(stack, n) + return true + }) + return parent +} + +// enclosingFunc 沿父指针上溯到最近的 *ast.FuncDecl。 +func enclosingFunc(parent map[ast.Node]ast.Node, n ast.Node) *ast.FuncDecl { + for cur := n; cur != nil; cur = parent[cur] { + if fn, ok := cur.(*ast.FuncDecl); ok { + return fn + } + } + return nil +} + +// TestQueueAccessCriterion_SelfCheck 证明判据能红。 +// +// ★ 没有自检的静态判据，无法区分「没问题」与「判据根本没抓到东西」。 +// +// 本项目踩过：`-run` 正则被 shell 转义破坏，打印 "[no tests to run]" 并 +// exit 0 —— 绿色 + 没测任何东西，最坏的组合。 +func TestQueueAccessCriterion_SelfCheck(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "injected_test.go") + src := `package core + +import "testing" + +func injectedUnlockedRead(a *Agent) int { + return len(a.sched.queue) +} + +var _ = injectedUnlockedRead +` + if err := os.WriteFile(path, []byte(src), 0o644); err != nil { + t.Fatal(err) + } + got := unlockedQueueReadsInFile(t, path) + if len(got) == 0 { + t.Fatal("★ 判据失效：注入的无锁直读没被抓到 —— 它对真实回归也会失效") + } + // 反向自检：持锁的写法**不得**被报（否则判据噪声太大，没人看） + path2 := filepath.Join(dir, "locked_test.go") + src2 := `package core + +import "testing" + +func injectedLockedRead(a *Agent) int { + a.sched.mu.Lock() + defer a.sched.mu.Unlock() + return len(a.sched.queue) +} + +var _ = injectedLockedRead +` + if err := os.WriteFile(path2, []byte(src2), 0o644); err != nil { + t.Fatal(err) + } + if got := unlockedQueueReadsInFile(t, path2); len(got) != 0 { + t.Fatalf("★ 判据误报：持锁的直读被报出来 %v", got) + } +} diff --git a/internal/agent/core/scheduler.go b/internal/agent/core/scheduler.go index b7133b95..564ea9d0 100644 --- a/internal/agent/core/scheduler.go +++ b/internal/agent/core/scheduler.go @@ -582,6 +582,34 @@ func (s *scheduler) next() *Task { return t } +// queueLen 返回当前排队任务数（**带锁**）。 +// +// ★ 2026-10-06 新增：offload_test.go 有 6 处直读 s.queue 取 len。 +// +// 那在「无并发写者」时安全（测试用 newRootWithoutSchedulerLoop， +// 刻意不启 schedulerLoop），但**只要有人改成 newRootWith 就变成真竞争**， +// 而且是「今天绿的、改一次就炸」的那种。 +// +// ⇒ 生产代码本来就走 takeQueuedInputs/hasRoom 等带锁访问器； +// 这里补一个对称的只读访问器，让测试不必直读内部字段。 +func (s *scheduler) queueLen() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.queue) +} + +// queueSnapshot 返回排队任务的浅拷贝，供测试断言内容（**带锁**）。 +// +// 返回拷贝而非内部切片：调用方拿到的就是某个时刻的快照， +// 后续并发 enqueue/dequeue 不会让它的 len 或 range 在中途变化。 +func (s *scheduler) queueSnapshot() []*Task { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]*Task, len(s.queue)) + copy(out, s.queue) + return out +} + // nextRef 选出下一个任务。优先顺序： // // 1. immediate —— 刚抢占成功的中断（抢占必须立即生效） diff --git a/internal/agent/core/scheduler_test.go b/internal/agent/core/scheduler_test.go index 2b6add8f..2b3994ff 100644 --- a/internal/agent/core/scheduler_test.go +++ b/internal/agent/core/scheduler_test.go @@ -155,11 +155,11 @@ func TestScheduler_Lifecycle(t *testing.T) { if t1 == nil || s.running != t1 { t.Fatal("next 应取出任务并置为 running") } - if len(s.queue) != 1 { - t.Fatalf("取出后队列长度=%d，期望 1", len(s.queue)) + if n := s.queueLen(); n != 1 { + t.Fatalf("取出后队列长度=%d，期望 1", n) } // 队列内不得同时出现 running（O1：三集合互不重叠）。 - for _, q := range s.queue { + for _, q := range s.queueSnapshot() { if q == t1 { t.Fatal("running 任务不得同时留在就绪队列") }