Function TestFanout

internal/controller/executor/executor_test.go:104–176

(t *testing.T)

}

func TestFanout(t *testing.T) {
	ctx := context.Background()

	var executions atomic.Int32
	registry := map[string]agent.Agent{
		"root": AgentFunc(func(inputs []*proto.Message, tm agent.Executor, o agent.OutputHandler) {
			var g errgroup.Group
			for i := range 50 {
				i := i // Capture loop variable.
				g.Go(func() error {
					_, err := tm.Exec(ctx, "test-conv", fmt.Sprintf("child-%d", i), &proto.AgentStart{
						AgentId:  "child",
						Messages: inputs,
					}, nil)
					return err
				})
			}
			if err := g.Wait(); err != nil {
				t.Fatal(err)
			}
			if o != nil {
				o(&proto.AgentOutputs{
					Messages: []*proto.Message{text("assistant", "root done")},
				})
			}
		}),
		"child": AgentFunc(func(inputs []*proto.Message, tm agent.Executor, o agent.OutputHandler) {
			executions.Add(1)
			time.Sleep(100 * time.Millisecond)

			var g errgroup.Group
			for i := range 2 {
				i := i // Capture loop variable.
				g.Go(func() error {
					_, err := tm.Exec(ctx, "test-conv", fmt.Sprintf("child2-%d", i), &proto.AgentStart{
						AgentId:  "child2",
						Messages: inputs,
					}, nil)
					return err
				})
			}
			if err := g.Wait(); err != nil {
				t.Fatal(err)
			}
			if o != nil {
				o(&proto.AgentOutputs{
					Messages: []*proto.Message{text("assistant", "child done")},
				})
			}
		}),
		"child2": AgentFunc(func(inputs []*proto.Message, tm agent.Executor, o agent.OutputHandler) {
			executions.Add(1)
			time.Sleep(100 * time.Millisecond)
			if o != nil {
				o(&proto.AgentOutputs{
					Messages: []*proto.Message{text("assistant", "child2 done")},
				})
			}