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) {