refactor: use WaitGroup.Go to simplify code (#16539)

### Summary

Adopt sync.WaitGroup.Go (Go 1.25) to simplify tracked goroutine
spawning. This replaces the error-prone trio of wg.Add(1), go func(),
and defer wg.Done() with a single, self-contained call.

More info: https://github.com/golang/go/issues/63796

Signed-off-by: grandpig <grandpig@outlook.com>

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

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

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)

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

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

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

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

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

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

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

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)

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)

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

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

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.

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

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

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

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)

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

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

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