diff --git a/internal/agent/canvas/state_test.go b/internal/agent/canvas/state_test.go index d85f657ae..0eb97ef61 100644 --- a/internal/agent/canvas/state_test.go +++ b/internal/agent/canvas/state_test.go @@ -145,14 +145,12 @@ func TestCanvasState_ConcurrentReadWrite(t *testing.T) { } var wg sync.WaitGroup for g := 0; g < 8; g++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { for i := 0; i < 200; i++ { _, _ = s.GetVar(cpnID(i%50) + "@v") s.SetVar(cpnID(i%50), "v", i) } - }() + }) } wg.Wait() } diff --git a/internal/agent/component/stagehand_runtime_test.go b/internal/agent/component/stagehand_runtime_test.go index 38e6e5edc..3a237fa7e 100644 --- a/internal/agent/component/stagehand_runtime_test.go +++ b/internal/agent/component/stagehand_runtime_test.go @@ -191,12 +191,10 @@ func TestStagehandRuntime_Cache_ConcurrentSameKey_NoDuplicateBuild(t *testing.T) var wg sync.WaitGroup var calls atomic.Int32 for i := 0; i < N; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, _ = r.clientFor(req) calls.Add(1) - }() + }) } wg.Wait() if calls.Load() != N { @@ -217,15 +215,13 @@ func TestStagehandRuntime_Cache_ConcurrentDifferentKeys(t *testing.T) { var wg sync.WaitGroup for i := 0; i < N; i++ { i := i - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, _ = r.clientFor(RunTaskRequest{ ModelName: "openai/gpt-4o", BaseURL: "https://api.openai.com/v1", APIKey: "sk-" + string(rune('a'+i%26)) + string(rune('A'+(i/26)%26)), }) - }() + }) } wg.Wait() // 32 goroutines, but the apiKey pattern has 26*2 = 52 unique diff --git a/internal/common/timer_test.go b/internal/common/timer_test.go index 042412232..c2f6ff5d6 100644 --- a/internal/common/timer_test.go +++ b/internal/common/timer_test.go @@ -114,13 +114,11 @@ func TestTimer_ConcurrentAccess(t *testing.T) { tm.Start() var wg sync.WaitGroup for i := 0; i < 10; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { tm.Enter(PhaseRetrieval) time.Sleep(time.Millisecond) tm.Exit(PhaseRetrieval) - }() + }) } wg.Wait() got := tm.Phase(PhaseRetrieval) diff --git a/internal/deepdoc/parser/pdf/parser_test.go b/internal/deepdoc/parser/pdf/parser_test.go index 0c2fe026a..6deb03f6c 100644 --- a/internal/deepdoc/parser/pdf/parser_test.go +++ b/internal/deepdoc/parser/pdf/parser_test.go @@ -565,16 +565,14 @@ func TestParser_ConcurrentSafety(t *testing.T) { var wg sync.WaitGroup n := 8 for range n { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { for range 5 { eng := &MockEngine{NumPages: 2} if _, err := p.ParseRaw(context.Background(), eng, mockDLA); err != nil { t.Errorf("ParseRaw: %v", err) } } - }() + }) } wg.Wait() } diff --git a/internal/harness/core/agent_loop_test.go b/internal/harness/core/agent_loop_test.go index 7c552d9f3..28e05808a 100644 --- a/internal/harness/core/agent_loop_test.go +++ b/internal/harness/core/agent_loop_test.go @@ -403,8 +403,7 @@ func TestTurnLoop_ConcurrentPush(t *testing.T) { tl := simpleTurnLoop(nil) var wg sync.WaitGroup for i := 0; i < 50; i++ { - wg.Add(1) - go func() { defer wg.Done(); tl.Push("c") }() + wg.Go(func() { ; tl.Push("c") }) } wg.Wait() tl.Stop() @@ -686,11 +685,9 @@ func TestTurnLoop_WaitMultipleGoroutines(t *testing.T) { for i := 0; i < 3; i++ { i := i - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { results[i] = loop.Wait() - }() + }) } wg.Wait() diff --git a/internal/harness/core/agentcore_full_test.go b/internal/harness/core/agentcore_full_test.go index c6a123709..62538471c 100644 --- a/internal/harness/core/agentcore_full_test.go +++ b/internal/harness/core/agentcore_full_test.go @@ -706,15 +706,13 @@ func TestConcurrentCancel(t *testing.T) { func TestConcurrentIterators(t *testing.T) { var wg sync.WaitGroup for i := 0; i < 3; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { model := &mockModel{} model.addResp("conc") agent := reActAgentSetup(model, nil) iter := agent.Run(context.Background(), &AgentInput{Messages: []Message{schema.UserMessage("hi")}}) drainAgentEvents(t, iter) - }() + }) } wg.Wait() } diff --git a/internal/harness/core/cancel_full_test.go b/internal/harness/core/cancel_full_test.go index 519680741..ff7a0f726 100644 --- a/internal/harness/core/cancel_full_test.go +++ b/internal/harness/core/cancel_full_test.go @@ -267,8 +267,7 @@ func TestBuildCancelFunc_CASFailStateDone(t *testing.T) { cf := cc.buildCancelFunc() var wg sync.WaitGroup for j := 0; j < 100; j++ { - wg.Add(1) - go func() { defer wg.Done(); _, _ = cf() }() + wg.Go(func() { ; _, _ = cf() }) } wg.Wait() cc.markHandled() @@ -1304,8 +1303,7 @@ func TestDeriveAgentToolCancelContext_ConcurrentSetRecursive(t *testing.T) { child := parent.deriveAgentToolCancelContext(ctx) var wg sync.WaitGroup for j := 0; j < 10; j++ { - wg.Add(1) - go func() { defer wg.Done(); parent.setRecursive(true) }() + wg.Go(func() { ; parent.setRecursive(true) }) } wg.Wait() child.markDone() diff --git a/internal/harness/core/concurrency_test.go b/internal/harness/core/concurrency_test.go index 1ab90440a..1161b4a87 100644 --- a/internal/harness/core/concurrency_test.go +++ b/internal/harness/core/concurrency_test.go @@ -353,11 +353,9 @@ func TestCancel_ConcurrentTrigger(t *testing.T) { // Trigger cancel from multiple goroutines var wg sync.WaitGroup for i := 0; i < 3; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { cancelFunc(WithCancelMode(CancelImmediate)) - }() + }) } wg.Wait() diff --git a/internal/harness/core/fault_injection_test.go b/internal/harness/core/fault_injection_test.go index 3d1ef046e..4db54cfdb 100644 --- a/internal/harness/core/fault_injection_test.go +++ b/internal/harness/core/fault_injection_test.go @@ -191,15 +191,13 @@ func TestFault_ConcurrentModelCalls(t *testing.T) { var wg sync.WaitGroup for i := 0; i < 5; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { ctx := context.Background() _, err := model.Generate(ctx, []Message{schema.UserMessage("conc")}) if err != nil { t.Errorf("concurrent call: %v", err) } - }() + }) } wg.Wait() } diff --git a/internal/harness/core/production_stress_test.go b/internal/harness/core/production_stress_test.go index 714357a0f..797480c8e 100644 --- a/internal/harness/core/production_stress_test.go +++ b/internal/harness/core/production_stress_test.go @@ -690,15 +690,13 @@ func TestProduction_BinOpChannelUnderLoad(t *testing.T) { var wg sync.WaitGroup for i := 0; i < writerCount; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { for j := 0; j < opsPerWriter; j++ { mu.Lock() binop.Update([]interface{}{1}) mu.Unlock() } - }() + }) } wg.Wait() diff --git a/internal/harness/graph/graph/graph_channel_test.go b/internal/harness/graph/graph/graph_channel_test.go index 2a53d2ee2..cd1a65d85 100644 --- a/internal/harness/graph/graph/graph_channel_test.go +++ b/internal/harness/graph/graph/graph_channel_test.go @@ -466,14 +466,12 @@ func TestGraphChannel_Race_ConcurrentGraphInvocations(t *testing.T) { errs := make(chan error, concurrency) for i := 0; i < concurrency; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, err := cg.Invoke(context.Background(), map[string]interface{}{"val": ""}) if err != nil { errs <- err } - }() + }) } wg.Wait() close(errs) diff --git a/internal/harness/graph/graph/graph_concurrency_test.go b/internal/harness/graph/graph/graph_concurrency_test.go index 1aaa15ec2..8b569e644 100644 --- a/internal/harness/graph/graph/graph_concurrency_test.go +++ b/internal/harness/graph/graph/graph_concurrency_test.go @@ -108,14 +108,12 @@ func TestGraph_ConcurrentInvoke_ComplexGraph(t *testing.T) { errs := make(chan error, concurrency) for i := 0; i < concurrency; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, err := cg.Invoke(context.Background(), map[string]interface{}{"idx": 0, "path": ""}) if err != nil { errs <- err } - }() + }) } wg.Wait() close(errs) @@ -172,9 +170,7 @@ func TestGraph_ConcurrentInvoke_LoopGraph(t *testing.T) { errs := make(chan error, concurrency) for i := 0; i < concurrency; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { result, err := cg.Invoke(context.Background(), map[string]interface{}{}) if err != nil { errs <- err @@ -188,7 +184,7 @@ func TestGraph_ConcurrentInvoke_LoopGraph(t *testing.T) { if v != "done" { errs <- fmt.Errorf("expected done, got %s", v) } - }() + }) } wg.Wait() close(errs) @@ -244,9 +240,7 @@ func TestGraph_ConcurrentInvoke_DAGGraph(t *testing.T) { errs := make(chan error, concurrency) for i := 0; i < concurrency; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { result, err := cg.Invoke(context.Background(), map[string]interface{}{}) if err != nil { errs <- err @@ -260,7 +254,7 @@ func TestGraph_ConcurrentInvoke_DAGGraph(t *testing.T) { if v != "joined" { errs <- fmt.Errorf("expected joined, got %s", v) } - }() + }) } wg.Wait() close(errs) @@ -311,9 +305,7 @@ func TestGraph_ConcurrentInvoke_MixedGraph(t *testing.T) { errs := make(chan error, concurrency) for i := 0; i < concurrency; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { result, err := cg.Invoke(context.Background(), map[string]interface{}{"result": "", "order": ""}) if err != nil { errs <- err @@ -327,7 +319,7 @@ func TestGraph_ConcurrentInvoke_MixedGraph(t *testing.T) { if r != "merged" { errs <- fmt.Errorf("expected merged, got %s", r) } - }() + }) } wg.Wait() close(errs) @@ -365,9 +357,7 @@ func TestGraph_ConcurrentInvoke_WithChannels(t *testing.T) { errs := make(chan error, concurrency) for i := 0; i < concurrency; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { result, err := cg.Invoke(context.Background(), map[string]interface{}{"val": "start"}) if err != nil { errs <- err @@ -381,7 +371,7 @@ func TestGraph_ConcurrentInvoke_WithChannels(t *testing.T) { if v != "done" { errs <- fmt.Errorf("expected done, got %s", v) } - }() + }) } wg.Wait() close(errs) @@ -425,15 +415,13 @@ func TestGraph_ConcurrentInvoke_InterruptRace(t *testing.T) { errs := make(chan error, concurrency) for i := 0; i < concurrency; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, err := cg.Invoke(context.Background(), map[string]interface{}{"val": ""}) // Interrupt is expected — not a failure if err != nil { errs <- fmt.Errorf("interrupt (expected): %v", err) } - }() + }) } wg.Wait() close(errs) @@ -470,14 +458,12 @@ func TestGraph_ConcurrentInvoke_SharedNodeClosure(t *testing.T) { errs := make(chan error, concurrency) for i := 0; i < concurrency; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, err := cg.Invoke(context.Background(), map[string]interface{}{"val": ""}) if err != nil { errs <- err } - }() + }) } wg.Wait() close(errs) @@ -507,16 +493,14 @@ func TestGraph_ConcurrentStream_SharedGraph(t *testing.T) { errs := make(chan error, concurrency) for i := 0; i < concurrency; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { outCh, errCh := cg.Stream(context.Background(), map[string]interface{}{"val": "test"}, types.StreamModeValues) for range outCh { } if err := <-errCh; err != nil { errs <- err } - }() + }) } wg.Wait() close(errs) @@ -594,14 +578,12 @@ func TestGraph_ConcurrentInvoke_HighContention(t *testing.T) { errs := make(chan error, concurrency) for i := 0; i < concurrency; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, err := cg.Invoke(context.Background(), map[string]interface{}{"val": ""}) if err != nil { errs <- err } - }() + }) } wg.Wait() close(errs) @@ -683,14 +665,12 @@ func TestGraph_ConcurrentInvoke_ErrorPropagation(t *testing.T) { errs := make(chan error, concurrency) for i := 0; i < concurrency; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, err := cg.Invoke(context.Background(), map[string]interface{}{"val": ""}) if err == nil { errs <- fmt.Errorf("expected error, got nil") } - }() + }) } wg.Wait() close(errs) @@ -728,28 +708,24 @@ func TestGraph_ConcurrentInvoke_GetNodesRace(t *testing.T) { // Goroutines that invoke for i := 0; i < 25; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, err := cg.Invoke(context.Background(), map[string]interface{}{"val": ""}) if err != nil { errs <- err } - }() + }) } // Goroutines that read graph metadata for i := 0; i < 25; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { g := cg.GetGraph() _ = g.GetNodes() _ = g.GetEdges() _ = g.GetChannels() _ = g.GetEntryPoint() _ = g.GetConditionalEdges() - }() + }) } wg.Wait() close(errs) diff --git a/internal/harness/graph/graph/graph_enterprise_integration_test.go b/internal/harness/graph/graph/graph_enterprise_integration_test.go index 9fc61a5c2..72a98ce26 100644 --- a/internal/harness/graph/graph/graph_enterprise_integration_test.go +++ b/internal/harness/graph/graph/graph_enterprise_integration_test.go @@ -396,9 +396,7 @@ func TestEnterprise_ConcurrentStream(t *testing.T) { var wg sync.WaitGroup for i := 0; i < numStreams; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() outputCh, errCh := cg.Stream(ctx, map[string]any{"value": "concurrent"}, types.StreamModeValues) @@ -407,7 +405,7 @@ func TestEnterprise_ConcurrentStream(t *testing.T) { if err := <-errCh; err != nil { t.Errorf("Stream error: %v", err) } - }() + }) } wg.Wait() } diff --git a/internal/harness/graph/graph/graph_topology_checkpoint_edge_test.go b/internal/harness/graph/graph/graph_topology_checkpoint_edge_test.go index 6ae36b2bb..2f98963a0 100644 --- a/internal/harness/graph/graph/graph_topology_checkpoint_edge_test.go +++ b/internal/harness/graph/graph/graph_topology_checkpoint_edge_test.go @@ -376,14 +376,12 @@ func TestCheckpoint_100ConcurrentReaders(t *testing.T) { var wg sync.WaitGroup for i := 0; i < 100; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { _, err := ms.Get(ctx, cfg) if err != nil { t.Errorf("Get: %v", err) } - }() + }) } wg.Wait() } diff --git a/internal/harness/graph/pregel/engine.go b/internal/harness/graph/pregel/engine.go index dc22769d6..aed46fc1e 100644 --- a/internal/harness/graph/pregel/engine.go +++ b/internal/harness/graph/pregel/engine.go @@ -204,11 +204,9 @@ func (e *Engine) Run(ctx context.Context, input any, mode types.StreamMode) (<-c // WaitGroup ensures the forward goroutine exits before we close outputCh, // preventing a data race between close(outputCh) and outputCh <- event. var fwWg sync.WaitGroup - fwWg.Add(1) // Forward stream events to output channel - go func() { - defer fwWg.Done() + fwWg.Go(func() { for event := range streamManager.Events() { select { case outputCh <- event: @@ -216,7 +214,7 @@ func (e *Engine) Run(ctx context.Context, input any, mode types.StreamMode) (<-c return } } - }() + }) // Deferred cleanup: close streamManager first (unblocks forward goroutine), // then wait for forward goroutine to exit, then close outputCh. diff --git a/internal/harness/graph/pregel/pregel_durability_timetravel_test.go b/internal/harness/graph/pregel/pregel_durability_timetravel_test.go index eecf2b049..219fd2133 100644 --- a/internal/harness/graph/pregel/pregel_durability_timetravel_test.go +++ b/internal/harness/graph/pregel/pregel_durability_timetravel_test.go @@ -442,15 +442,13 @@ func TestFaultInjection_ConcurrentCheckpointConflict(t *testing.T) { var wg sync.WaitGroup for g := 0; g < goroutines; g++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { engine := NewEngine(sg, WithRecursionLimit(10)) _, err := engine.RunSync(context.Background(), map[string]any{"value": "conc"}) if err != nil { t.Errorf("engine error: %v", err) } - }() + }) } wg.Wait() } diff --git a/internal/harness/graph/pregel/pregel_engine_subgraph_edge_test.go b/internal/harness/graph/pregel/pregel_engine_subgraph_edge_test.go index 2a6ed84e6..04b56ef0c 100644 --- a/internal/harness/graph/pregel/pregel_engine_subgraph_edge_test.go +++ b/internal/harness/graph/pregel/pregel_engine_subgraph_edge_test.go @@ -177,15 +177,13 @@ func TestEngine_ReuseDiffConfig(t *testing.T) { func TestEngine_ManyParallelRuns(t *testing.T) { var wg sync.WaitGroup for i := 0; i < 30; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { engine := NewEngine(newSimpleGraph(t), WithRecursionLimit(10)) _, err := engine.RunSync(context.Background(), map[string]any{"value": "par"}) if err != nil { t.Errorf("RunSync: %v", err) } - }() + }) } wg.Wait() } diff --git a/internal/harness/graph/pregel/pregel_fault_edge_test.go b/internal/harness/graph/pregel/pregel_fault_edge_test.go index d5c260204..d21a0afb1 100644 --- a/internal/harness/graph/pregel/pregel_fault_edge_test.go +++ b/internal/harness/graph/pregel/pregel_fault_edge_test.go @@ -285,14 +285,12 @@ func TestFault_ContextCancelledBeforeRun(t *testing.T) { func TestFault_RapidCreateCancel(t *testing.T) { var wg sync.WaitGroup for i := 0; i < 20; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { engine := NewEngine(newSimpleGraph(t), WithRecursionLimit(5)) ctx, cancel := context.WithTimeout(context.Background(), time.Microsecond) defer cancel() _, _ = engine.RunSync(ctx, map[string]any{"value": "x"}) - }() + }) } wg.Wait() } diff --git a/internal/harness/graph/pregel/pregel_fault_production_test.go b/internal/harness/graph/pregel/pregel_fault_production_test.go index 4ada163cf..f6acdecd4 100644 --- a/internal/harness/graph/pregel/pregel_fault_production_test.go +++ b/internal/harness/graph/pregel/pregel_fault_production_test.go @@ -239,14 +239,12 @@ func TestFault_CancelStorm(t *testing.T) { before := runtime.NumGoroutine() var wg sync.WaitGroup for i := 0; i < 50; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { engine := NewEngine(newSimpleGraph(t), WithRecursionLimit(10)) ctx, cancel := context.WithCancel(context.Background()) cancel() _, _ = engine.RunSync(ctx, map[string]any{"value": "storm"}) - }() + }) } wg.Wait() time.Sleep(10 * time.Millisecond) diff --git a/internal/harness/graph/pregel/pregel_interrupt_nested_test.go b/internal/harness/graph/pregel/pregel_interrupt_nested_test.go index 494561d6e..eec327a74 100644 --- a/internal/harness/graph/pregel/pregel_interrupt_nested_test.go +++ b/internal/harness/graph/pregel/pregel_interrupt_nested_test.go @@ -162,9 +162,7 @@ func TestInterrupt_NoCheckpointer(t *testing.T) { func TestInterrupt_Concurrent(t *testing.T) { var wg sync.WaitGroup for i := 0; i < 10; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { engine := NewEngine(newSimpleGraph(t), WithRecursionLimit(10), WithInterrupts("node_a"), @@ -173,7 +171,7 @@ func TestInterrupt_Concurrent(t *testing.T) { if err == nil { t.Errorf("expected interrupt") } - }() + }) } wg.Wait() } diff --git a/internal/harness/graph/pregel/pregel_stream_retry_integration_test.go b/internal/harness/graph/pregel/pregel_stream_retry_integration_test.go index 98f283c3a..3a21fa7c9 100644 --- a/internal/harness/graph/pregel/pregel_stream_retry_integration_test.go +++ b/internal/harness/graph/pregel/pregel_stream_retry_integration_test.go @@ -153,15 +153,13 @@ func TestStream_ConcurrentConsumers(t *testing.T) { // Multiple consumers read from the same channel. for i := 0; i < 5; i++ { - wg.Add(1) - go func() { - defer wg.Done() + wg.Go(func() { for result := range outputCh { if _, ok := result.(*StreamEvent); ok { eventCount.Add(1) } } - }() + }) } // Wait for all consumers. wg.Wait() diff --git a/internal/server/variable.go b/internal/server/variable.go index 1a6ee7da2..94e229fdd 100644 --- a/internal/server/variable.go +++ b/internal/server/variable.go @@ -212,9 +212,7 @@ func NewVariableWatcher(store VariableStore) *VariableWatcher { // Start starts watching for variable changes func (w *VariableWatcher) Start(interval time.Duration) { - w.wg.Add(1) - go func() { - defer w.wg.Done() + w.wg.Go(func() { ticker := time.NewTicker(interval) defer ticker.Stop() @@ -228,7 +226,7 @@ func (w *VariableWatcher) Start(interval time.Duration) { return } } - }() + }) common.Info("Variable watcher started", zap.Duration("interval", interval)) }