fix(adk): forward opts in taskTool.InvokableRun to emit internal events (#654)

Author: shentongmartin

.gitignore

@@ -42,3 +42,4 @@ output/*
 # Trae files
 .trae
 
+.DS_Store

adk/prebuilt/deep/deep_test.go

@@ -25,6 +25,7 @@ import (
 	"go.uber.org/mock/gomock"
 
 	"github.com/cloudwego/eino/adk"
+	"github.com/cloudwego/eino/adk/prebuilt/planexecute"
 	"github.com/cloudwego/eino/components/model"
 	"github.com/cloudwego/eino/components/tool"
 	mockModel "github.com/cloudwego/eino/internal/mock/components/model"
@@ -105,3 +106,186 @@ func (s *spySubAgent) Run(ctx context.Context, _ *adk.AgentInput, _ ...adk.Agent
 	gen.Close()
 	return it
 }
+
+func TestDeepAgentWithPlanExecuteSubAgent_InternalEventsEmitted(t *testing.T) {
+	ctx := context.Background()
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	deepModel := mockModel.NewMockToolCallingChatModel(ctrl)
+	plannerModel := mockModel.NewMockToolCallingChatModel(ctrl)
+	executorModel := mockModel.NewMockToolCallingChatModel(ctrl)
+	replannerModel := mockModel.NewMockToolCallingChatModel(ctrl)
+
+	deepModel.EXPECT().WithTools(gomock.Any()).Return(deepModel, nil).AnyTimes()
+	plannerModel.EXPECT().WithTools(gomock.Any()).Return(plannerModel, nil).AnyTimes()
+	executorModel.EXPECT().WithTools(gomock.Any()).Return(executorModel, nil).AnyTimes()
+	replannerModel.EXPECT().WithTools(gomock.Any()).Return(replannerModel, nil).AnyTimes()
+
+	plannerModel.EXPECT().Stream(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, input []*schema.Message, opts ...interface{}) (*schema.StreamReader[*schema.Message], error) {
+			sr, sw := schema.Pipe[*schema.Message](1)
+			go func() {
+				defer sw.Close()
+				planJSON := `{"steps":["step1"]}`
+				msg := schema.AssistantMessage("", []schema.ToolCall{
+					{
+						ID:   "plan_call_1",
+						Type: "function",
+						Function: schema.FunctionCall{
+							Name:      "plan",
+							Arguments: planJSON,
+						},
+					},
+				})
+				sw.Send(msg, nil)
+			}()
+			return sr, nil
+		},
+	).Times(1)
+
+	executorModel.EXPECT().Generate(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, msgs []*schema.Message, opts ...model.Option) (*schema.Message, error) {
+			return schema.AssistantMessage("executed step1", nil), nil
+		},
+	).Times(1)
+
+	replannerModel.EXPECT().Stream(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, input []*schema.Message, opts ...interface{}) (*schema.StreamReader[*schema.Message], error) {
+			sr, sw := schema.Pipe[*schema.Message](1)
+			go func() {
+				defer sw.Close()
+				responseJSON := `{"response":"final response"}`
+				msg := schema.AssistantMessage("", []schema.ToolCall{
+					{
+						ID:   "respond_call_1",
+						Type: "function",
+						Function: schema.FunctionCall{
+							Name:      "respond",
+							Arguments: responseJSON,
+						},
+					},
+				})
+				sw.Send(msg, nil)
+			}()
+			return sr, nil
+		},
+	).Times(1)
+
+	planner, err := planexecute.NewPlanner(ctx, &planexecute.PlannerConfig{
+		ToolCallingChatModel: plannerModel,
+	})
+	assert.NoError(t, err)
+
+	executor, err := planexecute.NewExecutor(ctx, &planexecute.ExecutorConfig{
+		Model: executorModel,
+	})
+	assert.NoError(t, err)
+
+	replanner, err := planexecute.NewReplanner(ctx, &planexecute.ReplannerConfig{
+		ChatModel: replannerModel,
+	})
+	assert.NoError(t, err)
+
+	planExecuteAgent, err := planexecute.New(ctx, &planexecute.Config{
+		Planner:   planner,
+		Executor:  executor,
+		Replanner: replanner,
+	})
+	assert.NoError(t, err)
+
+	namedPlanExecuteAgent := &namedPlanExecuteAgent{
+		ResumableAgent: planExecuteAgent,
+		name:           "plan_execute_subagent",
+		description:    "a plan execute subagent",
+	}
+
+	deepModelCalls := 0
+	deepModel.EXPECT().Generate(gomock.Any(), gomock.Any(), gomock.Any()).
+		DoAndReturn(func(ctx context.Context, msgs []*schema.Message, opts ...model.Option) (*schema.Message, error) {
+			deepModelCalls++
+			if deepModelCalls == 1 {
+				c := schema.ToolCall{ID: "id-1", Type: "function"}
+				c.Function.Name = taskToolName
+				c.Function.Arguments = fmt.Sprintf(`{"subagent_type":"%s","description":"execute the plan"}`, namedPlanExecuteAgent.name)
+				return schema.AssistantMessage("", []schema.ToolCall{c}), nil
+			}
+			return schema.AssistantMessage("done", nil), nil
+		}).AnyTimes()
+
+	deepAgent, err := New(ctx, &Config{
+		Name:                   "deep",
+		Description:            "deep agent",
+		ChatModel:              deepModel,
+		Instruction:            "you are deep agent",
+		SubAgents:              []adk.Agent{namedPlanExecuteAgent},
+		ToolsConfig:            adk.ToolsConfig{EmitInternalEvents: true},
+		MaxIteration:           5,
+		WithoutWriteTodos:      true,
+		WithoutGeneralSubAgent: true,
+	})
+	assert.NoError(t, err)
+
+	r := adk.NewRunner(ctx, adk.RunnerConfig{Agent: deepAgent})
+	it := r.Run(ctx, []adk.Message{schema.UserMessage("hi")})
+
+	var events []*adk.AgentEvent
+	for {
+		event, ok := it.Next()
+		if !ok {
+			break
+		}
+		events = append(events, event)
+	}
+
+	assert.Greater(t, len(events), 0, "should have at least one event")
+
+	var deepAgentEvents []*adk.AgentEvent
+	var plannerEvents []*adk.AgentEvent
+	var executorEvents []*adk.AgentEvent
+	var replannerEvents []*adk.AgentEvent
+	var planExecuteEvents []*adk.AgentEvent
+
+	for _, event := range events {
+		switch event.AgentName {
+		case "deep":
+			deepAgentEvents = append(deepAgentEvents, event)
+		case "planner":
+			plannerEvents = append(plannerEvents, event)
+		case "executor":
+			executorEvents = append(executorEvents, event)
+		case "replanner":
+			replannerEvents = append(replannerEvents, event)
+		case "plan_execute_replan", "execute_replan":
+			planExecuteEvents = append(planExecuteEvents, event)
+		}
+	}
+
+	assert.Greater(t, len(deepAgentEvents), 0, "should have events from deep agent")
+
+	assert.Greater(t, len(plannerEvents), 0, "planner internal events should be emitted when EmitInternalEvents is true")
+	assert.Greater(t, len(executorEvents), 0, "executor internal events should be emitted when EmitInternalEvents is true")
+	assert.Greater(t, len(replannerEvents), 0, "replanner internal events should be emitted when EmitInternalEvents is true")
+
+	t.Logf("Total events: %d", len(events))
+	t.Logf("Deep agent events: %d", len(deepAgentEvents))
+	t.Logf("Planner events: %d", len(plannerEvents))
+	t.Logf("Executor events: %d", len(executorEvents))
+	t.Logf("Replanner events: %d", len(replannerEvents))
+	t.Logf("PlanExecute events: %d", len(planExecuteEvents))
+}
+
+type namedPlanExecuteAgent struct {
+	adk.ResumableAgent
+	name        string
+	description string
+}
+
+func (n *namedPlanExecuteAgent) Name(_ context.Context) string {
+	return n.name
+}
+
+func (n *namedPlanExecuteAgent) Description(_ context.Context) string {
+	return n.description
+}

adk/prebuilt/deep/task_tool.go

@@ -157,7 +157,7 @@ func (t *taskTool) InvokableRun(ctx context.Context, argumentsInJSON string, opt
 		return "", err
 	}
 
-	return a.InvokableRun(ctx, params)
+	return a.InvokableRun(ctx, params, opts...)
 }
 
 func defaultTaskToolDescription(ctx context.Context, subAgents []adk.Agent) (string, error) {