test(agent/core): 消除测试对 scheduler.queue 的无锁直读,并加判据防回归

## 起因

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) <noreply@anthropic.com>
This commit is contained in:
JianFeeeee
2026-10-06 09:22:12 +08:00
parent a7d048c922
commit f33ebee01b
4 changed files with 224 additions and 26 deletions

View File

@ -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
}

View File

@ -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)
}
}

View File

@ -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 —— 刚抢占成功的中断(抢占必须立即生效)

View File

@ -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 任务不得同时留在就绪队列")
}