From b1f9b11d6e564d49c5f9eed3028ced8fe22ffc2c Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 18:51:25 +0800 Subject: [PATCH] =?UTF-8?q?test(flake):=204=20=E6=9D=A1=E8=B4=9F=E8=BD=BD?= =?UTF-8?q?=E6=95=8F=E6=84=9F=E7=94=A8=E4=BE=8B=E7=A1=AE=E5=AE=9A=E6=80=A7?= =?UTF-8?q?=E5=8C=96=EF=BC=88=E5=8F=AA=E6=94=B9=E6=B5=8B=E8=AF=95=EF=BC=8C?= =?UTF-8?q?=E7=94=9F=E4=BA=A7=E4=BB=A3=E7=A0=81=E9=9B=B6=E6=94=B9=E5=8A=A8?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - #1 TestSpawnTimeoutFromToolDefault：墙钟竞态（Spawn→Wait 的调度抖动 vs 120ms 工具时限）⇒ 用「托底子代理占住唯一槽位 + gate 放行」把 running 变成由门控保证；并新增 DeadlineExceeded 断言。 - #2 TestParallelLimit：真实成因**不是**「同进程迭代状态泄漏」（任务书推测错了）—— 是 transitionRunning 与 runner 内 active++ 之间的**抢占窗口**（复现：-count=200 六次红、三进程并行 1/9）⇒ 改等 runner 内部 activeCount()>=2 这一**稳定态**，并新增「整轮排空峰值==2」。 - #3 TestBackthink_List_NewestFirst：真实成因**不是**时间戳并列（SQL 是 ORDER BY id DESC）—— 是**后台心跳 writer 在两次 List 之间插入新行**（决定性证据：心跳周期缩到 1ms ⇒ 21/200 红，且第二次读到的 id 更大）⇒ 不再 Start() 心跳，改同步 think 写 3 条。 - #4 TestClient_OnConnectErrorTriggersRedial：固定 60ms 睡眠 vs 每周期耗时不定的墙钟竞态（改动前基线 5/9 红）⇒ 改为等第 2 次拨号事件；顺带改正与内容不符的旧注释。 - 证据：逐条 -count=20 绿；整树连跑 3 次 59 ok/0 FAIL；**三进程并行整树 9/9 绿（改动前基线 5/9 红）**；GOMAXPROCS=1 ×4 进程加压；-count=200/500/500/300；-race 三包绿。 - 主会话独立复验：隔离 worktree build/vet 绿、四条 -count=20 全绿、两进程并行整树双 59 ok/0 FAIL；**变异独立复现**（TestParallelLimit 的 MaxParallel 2→3 ⇒ max concurrent runs = 3, want 2 红）。 - 顺带发现第 5 条既有 flake：internal/runtime/exec/TestTerminal_Timeout_Killed（改动前基线 2/9 红）；以及 audit/00-总览.md:321/372 对 #2/#3 成因的记载有误（DevDocs 由主会话更正）。 --- .../channel/adapter/wsutil/client_test.go | 37 +++++++++++++++---- .../runtime/orchestration/scheduler_test.go | 12 +++++- .../runtime/orchestration/spawntask_test.go | 25 ++++++++++++- internal/services/backthink/backthink_test.go | 29 +++++++++++---- 4 files changed, 84 insertions(+), 19 deletions(-) diff --git a/internal/channel/adapter/wsutil/client_test.go b/internal/channel/adapter/wsutil/client_test.go index 0bc3b63..b0d70c0 100644 --- a/internal/channel/adapter/wsutil/client_test.go +++ b/internal/channel/adapter/wsutil/client_test.go @@ -101,10 +101,21 @@ func TestClient_DialReadReconnect(t *testing.T) { } } -// TestClient_HeartbeatVerifiesPing sends an s=2-like heartbeat every 20ms and -// asserts the dial URL is still the one provided. +// TestClient_OnConnectErrorTriggersRedial 断言连接钩子失败会触发重拨：等到第二次 +// 拨号事件发生后才取消上下文，而不是先睡固定的 60ms 再看拨了几次——后者在并行 +// 负载下可能只够拨一次，等于把断言交给机器负载决定。 func TestClient_OnConnectErrorTriggersRedial(t *testing.T) { + const ( + // redialTarget 是断言「发生过重拨」所需的拨号次数：1 次首拨 + 1 次重拨。 + redialTarget = 2 + // dialEvents 是拨号事件通道的容量，保证 Run 在测试取消之前不会阻塞在投递上。 + dialEvents = 8 + ) + var dials atomic.Int64 + // 每次拨号投递一个事件；测试等满 redialTarget 次再取消，因此断言的是 + // 「重拨确实发生了」这件事本身，与墙钟无关。 + dialed := make(chan struct{}, dialEvents) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { conn, err := wsu.Upgrade(w, r, nil) if err != nil { @@ -120,10 +131,13 @@ func TestClient_OnConnectErrorTriggersRedial(t *testing.T) { defer srv.Close() url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + "/testws" - // OnConnect always errors so the client keeps rejecting each session and the - // dial fires repeatedly. + // OnConnect 恒错，因此每次会话都被客户端拒绝、拨号反复发生。 client := New( - func(context.Context) (string, error) { dials.Add(1); return url, nil }, + func(context.Context) (string, error) { + dials.Add(1) + dialed <- struct{}{} + return url, nil + }, Handler{ OnConnect: func(ctx context.Context, c *Conn) error { return context.DeadlineExceeded }, }, @@ -131,17 +145,24 @@ func TestClient_OnConnectErrorTriggersRedial(t *testing.T) { WithLogger(logger.New(logger.Options{Out: io.Discard})), ) ctx, cancel := context.WithCancel(context.Background()) + defer cancel() done := make(chan error, 1) go func() { done <- client.Run(ctx) }() - time.Sleep(60 * time.Millisecond) + for i := 0; i < redialTarget; i++ { + select { + case <-dialed: + case <-time.After(3 * time.Second): + t.Fatalf("timed out waiting for dial %d", i+1) + } + } cancel() select { case <-done: case <-time.After(2 * time.Second): t.Fatal("Run did not exit") } - if dials.Load() < 2 { - t.Errorf("dials = %d, want >= 2 (reconnect on failed connect hook)", dials.Load()) + if dials.Load() < redialTarget { + t.Errorf("dials = %d, want >= %d (reconnect on failed connect hook)", dials.Load(), redialTarget) } } diff --git a/internal/runtime/orchestration/scheduler_test.go b/internal/runtime/orchestration/scheduler_test.go index 3bcf63b..46960b4 100644 --- a/internal/runtime/orchestration/scheduler_test.go +++ b/internal/runtime/orchestration/scheduler_test.go @@ -224,8 +224,11 @@ func TestParallelLimit(t *testing.T) { subs = append(subs, sub) } - // 等到两个并行槽位都被占用，再确认并发运行数从未超过 MaxParallel。 - waitFor(t, func() bool { return countStatus(subs, StatusRunning) == 2 }, waitTimeout) + // 等到两个并行槽位都被占用，再确认并发运行数从未超过 MaxParallel。这里等的是 + // runner 内部的活跃计数，而不是「状态已翻成 running」：状态是在进入 runner 之前 + // 置位的，两者之间有一个调度窗口，用墙钟轮询去撞它必然偶发（仅看状态时读到过 + // maxActive = 0/1）。gate 未开，因此 active == 2 是一个稳定状态。 + waitFor(t, func() bool { return m.activeCount() >= 2 }, waitTimeout) if got := m.maxActiveCount(); got != 2 { t.Fatalf("max concurrent runs = %d, want 2 while gated", got) } @@ -246,6 +249,11 @@ func TestParallelLimit(t *testing.T) { if res.Total != 5 || res.Succeeded != 5 { t.Fatalf("merged = %d total / %d succeeded, want 5/5", res.Total, res.Succeeded) } + // 峰值在整轮排空中都不得越界：这条覆盖放行后逐个进出槽位的阶段， + // 因此它也校验每个运行体结束时确实归还了槽位。 + if got := m.maxActiveCount(); got != 2 { + t.Fatalf("max concurrent runs = %d, want 2 across the whole drain", got) + } } func TestPerAgentTimeout(t *testing.T) { diff --git a/internal/runtime/orchestration/spawntask_test.go b/internal/runtime/orchestration/spawntask_test.go index 5b3f407..27cf0c0 100644 --- a/internal/runtime/orchestration/spawntask_test.go +++ b/internal/runtime/orchestration/spawntask_test.go @@ -2,6 +2,7 @@ package orchestration import ( "context" + "errors" "strings" "sync" "testing" @@ -172,12 +173,25 @@ func TestSpawnInjectsDataParagraph(t *testing.T) { // TestSpawnTimeoutFromToolDefault 断言 WithSpawnTimeout（生产经 // SUBAGENT_TIMEOUT_SEC 填充）覆盖调度器默认时限（02.4.4）。 +// +// 「批次在 30ms 前台窗口内没有结算」这一条由门控保证：唯一的并行槽位先被一个 +// 托底子代理占住（它的时限是调度器默认的 1h，测试放行之前绝不结束），因此无论 +// 机器多忙，被派发的子代理在窗口内都不可能开始跑，更不可能结算。原先这条是靠 +// 「30ms 窗口 < 120ms 时限」的墙钟巧合成立的，负载一高就会先等到已结算的 "done"。 func TestSpawnTimeoutFromToolDefault(t *testing.T) { t.Parallel() s := newScheduler(t, Options{MaxParallel: 1, PerAgentTimeout: time.Hour}) - s.runner = &mockRunner{gate: make(chan struct{})} // 阻塞直到 ctx 超时 - sp := NewSpawnTool(s, WithSpawnWait(30*time.Millisecond), WithSpawnTimeout(120*time.Millisecond)) + gate := make(chan struct{}) + s.runner = &mockRunner{gate: gate} // 阻塞直到测试放行或 ctx 超时 + // 托底者占住唯一槽位；它在测试放行之前不会结束。 + holder, err := s.Spawn(context.Background(), AgentSpec{Goal: "hold the only slot"}) + if err != nil { + t.Fatalf("Spawn holder: %v", err) + } + waitFor(t, func() bool { return statusOf(s, holder.ID) == StatusRunning }, waitTimeout) + + sp := NewSpawnTool(s, WithSpawnWait(30*time.Millisecond), WithSpawnTimeout(120*time.Millisecond)) raw, err := sp.Execute(context.Background(), map[string]any{"goal": "slow"}) if err != nil { t.Fatalf("Execute: %v", err) @@ -187,11 +201,18 @@ func TestSpawnTimeoutFromToolDefault(t *testing.T) { t.Fatalf("Status = %q, want running", got.Status) } id := got.Agents[0].ID + + // 放行托底者：槽位归还，被派发的子代理这才开始跑并在 120ms 上中止。 + // 该发送与托底者的接收配对（此时还没有别的接收者），因此它同样是确定性的。 + gate <- struct{}{} waitFor(t, func() bool { return statusOf(s, id) == StatusAborted }, waitTimeout) sub := s.pickOrFail(t, id) if !strings.Contains(sub.Summary, "120ms") { t.Fatalf("Summary = %q, want the 120ms tool-default deadline named", sub.Summary) } + if !errors.Is(sub.Err, context.DeadlineExceeded) { + t.Fatalf("Err = %v, want DeadlineExceeded", sub.Err) + } } // TestSpawnTimeoutPerCallOverride 断言按调用覆盖（timeout_seconds）优先于 diff --git a/internal/services/backthink/backthink_test.go b/internal/services/backthink/backthink_test.go index 87d1b07..8d14b07 100644 --- a/internal/services/backthink/backthink_test.go +++ b/internal/services/backthink/backthink_test.go @@ -266,18 +266,23 @@ func TestBackthink_SystemEventHighValueThinks(t *testing.T) { eventually(t, 2*time.Second, func() bool { return thinker.count() >= before+1 }) } +// TestBackthink_List_NewestFirst 断言 List 严格由新到旧，且单行列表的头部与完整 +// 列表的头部是同一条记录。记录全部由测试同步写入，因此两次读取之间不存在并发 +// 写入者（见下方注释）。 func TestBackthink_List_NewestFirst(t *testing.T) { t.Parallel() thinker := &mockThinker{text: "nothing to report"} - b, _ := newBackthink(t, Config{Interval: 15 * time.Millisecond}, Deps{Thinker: thinker}) + b, _ := newBackthink(t, Config{}, Deps{Thinker: thinker}) - b.Start(context.Background()) - defer func() { _ = b.Stop(context.Background()) }() + // 三条可区分的记录，写入顺序即 sources 的顺序。 + sources := []string{"first", "second", "third"} + // 记录由测试同步写入（直接调用 think），刻意不启动心跳：后台 ticker 会在 + // 两次 List 之间插入一条新记录，使「单行列表的头部 == 完整列表的头部」这个 + // 断言变成跟调度赛跑。此前偶红时读到的正是「刚插入的更新的 id」。 + for _, src := range sources { + b.think(context.Background(), Trigger{Source: src, Summary: src, Import: ImportanceLow, Kind: TriggerHeartbeat}) + } - eventually(t, 2*time.Second, func() bool { - recs, _ := b.List(context.Background(), 10) - return len(recs) >= 2 - }) top, err := b.List(context.Background(), 1) if err != nil { t.Fatalf("List: %v", err) @@ -294,11 +299,21 @@ func TestBackthink_List_NewestFirst(t *testing.T) { if top[0].ID != all[0].ID { t.Fatalf("List(1)[0].ID = %d, want %d (== newest)", top[0].ID, all[0].ID) } + if len(all) != len(sources) { + t.Fatalf("len(all) = %d, want %d (no writer ran between the two reads)", len(all), len(sources)) + } for i := 1; i < len(all); i++ { if all[i].ID >= all[i-1].ID { t.Fatalf("records not newest-first: %d before %d", all[i-1].ID, all[i].ID) } } + // 最新在前不仅体现在 id 上，也体现在内容上：列表是写入顺序的逆序。 + for i := range sources { + want := sources[len(sources)-1-i] + if all[i].Trigger != want { + t.Fatalf("all[%d].Trigger = %q, want %q (newest first)", i, all[i].Trigger, want) + } + } } func TestBackthink_Stop_NoLeak(t *testing.T) {