diff --git a/agent/robot/api/e2e_interact_test.go b/agent/robot/api/e2e_interact_test.go new file mode 100644 index 00000000..98cf2cd8 --- /dev/null +++ b/agent/robot/api/e2e_interact_test.go @@ -0,0 +1,361 @@ +package api_test + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/yaoapp/gou/model" + "github.com/yaoapp/xun/capsule" + "github.com/yaoapp/yao/agent/robot/api" + "github.com/yaoapp/yao/agent/robot/executor/standard" + "github.com/yaoapp/yao/agent/robot/types" + "github.com/yaoapp/yao/agent/testutils" +) + +// TestE2EInteractNewAssignment tests the full Interact flow for a new task assignment. +// With the conversational Host Agent, the first turn may return natural language +// (waiting_for_more) or an action decision depending on request clarity. +func TestE2EInteractNewAssignment(t *testing.T) { + if testing.Short() { + t.Skip("Skipping E2E test - requires real LLM calls") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + cleanupInteractRobots(t) + cleanupInteractExecutions(t) + defer cleanupInteractRobots(t) + defer cleanupInteractExecutions(t) + + t.Run("assign_via_interact_creates_execution_and_gets_host_reply", func(t *testing.T) { + memberID := "robot_e2e_interact_assign" + setupInteractRobot(t, memberID, "team_e2e_interact") + + err := api.Start() + require.NoError(t, err) + defer api.Stop() + + ctx := types.NewContext(context.Background(), testAuth()) + + robot, err := api.GetRobot(ctx, memberID) + require.NoError(t, err) + require.NotNil(t, robot) + + result, err := api.Interact(ctx, memberID, &api.InteractRequest{ + Source: types.InteractSourceUI, + Message: "Please write a short greeting email for our team meeting tomorrow morning.", + }) + require.NoError(t, err) + require.NotNil(t, result) + + t.Logf("Interact result: status=%s, message=%s, reply=%s, exec_id=%s, wait_for_more=%v", + result.Status, result.Message, result.Reply, result.ExecutionID, result.WaitForMore) + + assert.NotEmpty(t, result.ExecutionID, "should create an execution") + assert.NotEmpty(t, result.ChatID, "should have a chat session") + assert.NotEmpty(t, result.Reply, "Host Agent should provide a reply") + + validStatuses := []string{"confirmed", "waiting_for_more", "adjusted", "acknowledged"} + assert.Contains(t, validStatuses, result.Status, + "status should be one of the valid Host Agent action outcomes") + + if result.Status == "confirmed" { + time.Sleep(2 * time.Second) + executions, err := api.ListExecutions(ctx, memberID, &api.ExecutionQuery{Page: 1, PageSize: 5}) + require.NoError(t, err) + assert.Greater(t, len(executions.Data), 0, "confirmed execution should exist in store") + } + }) +} + +// TestE2EInteractStream tests the streaming version end-to-end. +func TestE2EInteractStream(t *testing.T) { + if testing.Short() { + t.Skip("Skipping E2E test - requires real LLM calls") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + cleanupInteractRobots(t) + cleanupInteractExecutions(t) + defer cleanupInteractRobots(t) + defer cleanupInteractExecutions(t) + + t.Run("stream_assign_returns_chunks_and_valid_result", func(t *testing.T) { + memberID := "robot_e2e_interact_stream" + setupInteractRobot(t, memberID, "team_e2e_interact") + + err := api.Start() + require.NoError(t, err) + defer api.Stop() + + ctx := types.NewContext(context.Background(), testAuth()) + + var mu sync.Mutex + var chunks []*standard.StreamChunk + + streamFn := func(chunk *standard.StreamChunk) int { + mu.Lock() + defer mu.Unlock() + chunks = append(chunks, chunk) + return 0 + } + + result, err := api.InteractStream(ctx, memberID, &api.InteractRequest{ + Source: types.InteractSourceUI, + Message: "Help me draft a brief status update email about completing the Q4 report.", + }, streamFn) + require.NoError(t, err) + require.NotNil(t, result) + + mu.Lock() + chunkCount := len(chunks) + var textChunks []string + for _, c := range chunks { + if c.Type == "text" && c.Delta { + textChunks = append(textChunks, c.Content) + } + } + mu.Unlock() + + combined := strings.Join(textChunks, "") + + t.Logf("Stream received %d total chunks, %d text chunks, combined length: %d", + chunkCount, len(textChunks), len(combined)) + t.Logf("Result: status=%s, exec_id=%s, reply_len=%d, wait_for_more=%v", + result.Status, result.ExecutionID, len(result.Reply), result.WaitForMore) + + assert.Greater(t, len(textChunks), 0, "should receive streaming text chunks from Host Agent") + assert.NotEmpty(t, combined, "combined text should not be empty") + assert.NotEmpty(t, result.ExecutionID, "should create an execution") + assert.NotEmpty(t, result.Reply, "final result should contain reply") + + validStatuses := []string{"confirmed", "waiting_for_more", "adjusted"} + assert.Contains(t, validStatuses, result.Status) + }) +} + +// TestE2EInteractMultiTurn tests a multi-turn conversation: +// Turn 1: Send vague message -> Host Agent replies conversationally (waiting_for_more) +// Turn 2: Send clear confirmation -> Host Agent returns action JSON (confirmed or other action) +func TestE2EInteractMultiTurn(t *testing.T) { + if testing.Short() { + t.Skip("Skipping E2E test - requires real LLM calls") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + cleanupInteractRobots(t) + cleanupInteractExecutions(t) + defer cleanupInteractRobots(t) + defer cleanupInteractExecutions(t) + + t.Run("multi_turn_assign_conversation", func(t *testing.T) { + memberID := "robot_e2e_interact_multiturn" + setupInteractRobot(t, memberID, "team_e2e_interact") + + err := api.Start() + require.NoError(t, err) + defer api.Stop() + + ctx := types.NewContext(context.Background(), testAuth()) + + // Turn 1: Send vague message — expect conversational reply + result1, err := api.Interact(ctx, memberID, &api.InteractRequest{ + Source: types.InteractSourceUI, + Message: "Do something with emails.", + }) + require.NoError(t, err) + require.NotNil(t, result1) + + t.Logf("Turn 1: status=%s, reply=%s, exec_id=%s, wait_for_more=%v", + result1.Status, result1.Reply, result1.ExecutionID, result1.WaitForMore) + + assert.NotEmpty(t, result1.ExecutionID) + assert.NotEmpty(t, result1.Reply) + + // Turn 2: Clarify/confirm with the same execution_id + result2, err := api.Interact(ctx, memberID, &api.InteractRequest{ + ExecutionID: result1.ExecutionID, + Source: types.InteractSourceUI, + Message: "Yes, please write a brief thank-you email to the design team for their Q4 work. Go ahead and confirm.", + }) + require.NoError(t, err) + require.NotNil(t, result2) + + t.Logf("Turn 2: status=%s, reply=%s, exec_id=%s, wait_for_more=%v", + result2.Status, result2.Reply, result2.ExecutionID, result2.WaitForMore) + + assert.NotEmpty(t, result2.Reply) + assert.Equal(t, result1.ExecutionID, result2.ExecutionID, "should be same execution") + + validStatuses := []string{"confirmed", "waiting_for_more", "adjusted", "acknowledged"} + assert.Contains(t, validStatuses, result2.Status, + "second turn should produce a valid outcome") + }) +} + +// TestE2EInteractStreamMultiTurn tests multi-turn with streaming. +func TestE2EInteractStreamMultiTurn(t *testing.T) { + if testing.Short() { + t.Skip("Skipping E2E test - requires real LLM calls") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + cleanupInteractRobots(t) + cleanupInteractExecutions(t) + defer cleanupInteractRobots(t) + defer cleanupInteractExecutions(t) + + t.Run("stream_multi_turn", func(t *testing.T) { + memberID := "robot_e2e_interact_stream_mt" + setupInteractRobot(t, memberID, "team_e2e_interact") + + err := api.Start() + require.NoError(t, err) + defer api.Stop() + + ctx := types.NewContext(context.Background(), testAuth()) + + // Turn 1 + var mu1 sync.Mutex + var chunks1 []*standard.StreamChunk + result1, err := api.InteractStream(ctx, memberID, &api.InteractRequest{ + Source: types.InteractSourceUI, + Message: "I need help with something.", + }, func(chunk *standard.StreamChunk) int { + mu1.Lock() + chunks1 = append(chunks1, chunk) + mu1.Unlock() + return 0 + }) + require.NoError(t, err) + require.NotNil(t, result1) + + mu1.Lock() + t.Logf("Turn 1 stream: %d chunks, status=%s, reply=%s, wait_for_more=%v", + len(chunks1), result1.Status, result1.Reply, result1.WaitForMore) + mu1.Unlock() + + assert.NotEmpty(t, result1.ExecutionID) + assert.NotEmpty(t, result1.Reply) + + // Turn 2: Clarify with same execution_id + var mu2 sync.Mutex + var chunks2 []*standard.StreamChunk + result2, err := api.InteractStream(ctx, memberID, &api.InteractRequest{ + ExecutionID: result1.ExecutionID, + Source: types.InteractSourceUI, + Message: "Please compose a short farewell message for a colleague leaving the team. Yes, go ahead.", + }, func(chunk *standard.StreamChunk) int { + mu2.Lock() + chunks2 = append(chunks2, chunk) + mu2.Unlock() + return 0 + }) + require.NoError(t, err) + require.NotNil(t, result2) + + mu2.Lock() + t.Logf("Turn 2 stream: %d chunks, status=%s, reply=%s, wait_for_more=%v", + len(chunks2), result2.Status, result2.Reply, result2.WaitForMore) + mu2.Unlock() + + assert.NotEmpty(t, result2.Reply) + assert.Equal(t, result1.ExecutionID, result2.ExecutionID) + }) +} + +// ==================== Helper Functions ==================== + +func setupInteractRobot(t *testing.T, memberID, teamID string) { + m := model.Select("__yao.member") + tableName := m.MetaData.Table.Name + qb := capsule.Query() + + robotConfig := map[string]interface{}{ + "identity": map[string]interface{}{ + "role": "Email Assistant", + "duties": []string{"Write and manage emails"}, + "rules": []string{"Always confirm before sending", "Keep emails professional"}, + }, + "quota": map[string]interface{}{ + "max": 5, + "queue": 20, + "priority": 5, + }, + "triggers": map[string]interface{}{ + "intervene": map[string]interface{}{"enabled": true}, + }, + "resources": map[string]interface{}{ + "phases": map[string]interface{}{ + "inspiration": "robot.inspiration", + "goals": "robot.goals", + "tasks": "robot.tasks", + "run": "robot.validation", + "validation": "robot.validation", + "delivery": "robot.delivery", + "learning": "robot.learning", + "host": "robot.host", + }, + "agents": []string{}, + }, + } + configJSON, _ := json.Marshal(robotConfig) + + systemPrompt := `You are an email assistant for E2E testing of the Interact API. +When asked to write an email, confirm the task and generate a brief email draft.` + + err := qb.Table(tableName).Insert([]map[string]interface{}{ + { + "member_id": memberID, + "team_id": teamID, + "member_type": "robot", + "display_name": "E2E Interact Test Robot " + memberID, + "system_prompt": systemPrompt, + "status": "active", + "role_id": "member", + "autonomous_mode": false, + "robot_status": "idle", + "robot_config": string(configJSON), + }, + }) + if err != nil { + t.Fatalf("Failed to insert interact robot %s: %v", memberID, err) + } +} + +func cleanupInteractRobots(t *testing.T) { + m := model.Select("__yao.member") + if m == nil { + return + } + qb := capsule.Query() + _, err := qb.Table(m.MetaData.Table.Name).Where("member_id", "like", "robot_e2e_interact%").Delete() + if err != nil { + t.Logf("Warning: cleanup interact robots: %v", err) + } +} + +func cleanupInteractExecutions(t *testing.T) { + m := model.Select("__yao.agent.execution") + if m == nil { + return + } + qb := capsule.Query() + _, err := qb.Table(m.MetaData.Table.Name).Where("member_id", "like", "robot_e2e_interact%").Delete() + if err != nil { + t.Logf("Warning: cleanup interact executions: %v", err) + } +} diff --git a/agent/robot/api/execution.go b/agent/robot/api/execution.go index 627e99f7..2fa1aab5 100644 --- a/agent/robot/api/execution.go +++ b/agent/robot/api/execution.go @@ -62,57 +62,38 @@ func ListExecutions(ctx *types.Context, memberID string, query *ExecutionQuery) } query.applyDefaults() - // Build list options opts := &store.ListOptions{ MemberID: memberID, - Limit: query.PageSize, - Offset: (query.Page - 1) * query.PageSize, + Page: query.Page, + PageSize: query.PageSize, OrderBy: "start_time desc", } if query.Status != "" { opts.Status = query.Status } + if len(query.ExcludeStatuses) > 0 { + opts.ExcludeStatuses = query.ExcludeStatuses + } if query.Trigger != "" { opts.TriggerType = query.Trigger } - // Query from store - records, err := getExecutionStore().List(context.Background(), opts) + result, err := getExecutionStore().List(context.Background(), opts) if err != nil { return nil, fmt.Errorf("failed to list executions: %w", err) } - // Convert to Execution slice - executions := make([]*types.Execution, 0, len(records)) - for _, record := range records { + executions := make([]*types.Execution, 0, len(result.Data)) + for _, record := range result.Data { executions = append(executions, record.ToExecution()) } - // Get total count - // Note: For accurate total, ExecutionStore.List should return total count - // Current implementation returns estimated total based on returned records - total := len(records) - if total >= query.PageSize { - // Has more records, need to query total count - // For now, indicate there might be more by setting total to -1 - // UI should handle this as "has more" - countOpts := &store.ListOptions{MemberID: memberID} - if query.Status != "" { - countOpts.Status = query.Status - } - if query.Trigger != "" { - countOpts.TriggerType = query.Trigger - } - allRecords, _ := getExecutionStore().List(context.Background(), countOpts) - total = len(allRecords) - } - return &ExecutionResult{ Data: executions, - Total: total, - Page: query.Page, - PageSize: query.PageSize, + Total: result.Total, + Page: result.Page, + PageSize: result.PageSize, }, nil } diff --git a/agent/robot/api/interact.go b/agent/robot/api/interact.go index dafd33db..471c937d 100644 --- a/agent/robot/api/interact.go +++ b/agent/robot/api/interact.go @@ -3,6 +3,7 @@ package api import ( "fmt" + agentcontext "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/robot/executor/standard" "github.com/yaoapp/yao/agent/robot/manager" "github.com/yaoapp/yao/agent/robot/types" @@ -121,6 +122,84 @@ func Confirm(ctx *types.Context, memberID string, execID string, message string) }) } +// InteractStream is the streaming version of Interact. +// It streams Host Agent text tokens via streamFn while still returning the final InteractResult. +// V1 fallback does not support streaming and returns an error. +func InteractStream(ctx *types.Context, memberID string, req *InteractRequest, streamFn standard.StreamCallback) (*InteractResult, error) { + if memberID == "" { + return nil, fmt.Errorf("member_id is required") + } + if req == nil { + return nil, fmt.Errorf("interact request is required") + } + + mgr, err := getManager() + if err != nil || mgr == nil { + return nil, fmt.Errorf("streaming requires V2 manager (not available)") + } + + mgrReq := &manager.InteractRequest{ + ExecutionID: req.ExecutionID, + TaskID: req.TaskID, + Source: req.Source, + Message: req.Message, + Action: req.Action, + } + + resp, err := mgr.HandleInteractStream(ctx, memberID, mgrReq, streamFn) + if err != nil { + return nil, err + } + + return &InteractResult{ + ExecutionID: resp.ExecutionID, + Status: resp.Status, + Message: resp.Message, + ChatID: resp.ChatID, + Reply: resp.Reply, + WaitForMore: resp.WaitForMore, + }, nil +} + +// InteractStreamRaw is the CUI-protocol-aligned streaming version of Interact. +// It passes raw message.Message objects to the onMessage callback, preserving all CUI +// protocol fields for direct SSE passthrough to the frontend. +func InteractStreamRaw(ctx *types.Context, memberID string, req *InteractRequest, onMessage agentcontext.OnMessageFunc) (*InteractResult, error) { + if memberID == "" { + return nil, fmt.Errorf("member_id is required") + } + if req == nil { + return nil, fmt.Errorf("interact request is required") + } + + mgr, err := getManager() + if err != nil || mgr == nil { + return nil, fmt.Errorf("raw streaming requires V2 manager (not available)") + } + + mgrReq := &manager.InteractRequest{ + ExecutionID: req.ExecutionID, + TaskID: req.TaskID, + Source: req.Source, + Message: req.Message, + Action: req.Action, + } + + resp, err := mgr.HandleInteractStreamRaw(ctx, memberID, mgrReq, onMessage) + if err != nil { + return nil, err + } + + return &InteractResult{ + ExecutionID: resp.ExecutionID, + Status: resp.Status, + Message: resp.Message, + ChatID: resp.ChatID, + Reply: resp.Reply, + WaitForMore: resp.WaitForMore, + }, nil +} + // CancelExecution cancels a waiting/confirming execution via the manager. func CancelExecution(ctx *types.Context, execID string) error { mgr, err := getManager() diff --git a/agent/robot/api/results.go b/agent/robot/api/results.go index 3bd06824..78df920a 100644 --- a/agent/robot/api/results.go +++ b/agent/robot/api/results.go @@ -65,11 +65,10 @@ func ListResults(ctx *types.Context, memberID string, query *ResultQuery) (*Resu } query.applyDefaults() - // Build store options opts := &store.ResultListOptions{ MemberID: memberID, - Limit: query.PageSize, - Offset: (query.Page - 1) * query.PageSize, + Page: query.Page, + PageSize: query.PageSize, } if query.TriggerType != "" { diff --git a/agent/robot/api/robot.go b/agent/robot/api/robot.go index cc9cddb8..422f3d69 100644 --- a/agent/robot/api/robot.go +++ b/agent/robot/api/robot.go @@ -113,15 +113,15 @@ func GetRobotStatus(ctx *types.Context, memberID string) (*RobotState, error) { // Get running execution IDs from ExecutionStore (more reliable than in-memory) // This ensures we get accurate status even when robot is loaded from database - runningExecs, err := executionStore.List(context.Background(), &store.ListOptions{ + runningResult, err := executionStore.List(context.Background(), &store.ListOptions{ MemberID: memberID, Status: types.ExecRunning, - Limit: 100, + PageSize: 100, }) - if err == nil && len(runningExecs) > 0 { - state.Running = len(runningExecs) - state.RunningIDs = make([]string, 0, len(runningExecs)) - for _, exec := range runningExecs { + if err == nil && runningResult != nil && len(runningResult.Data) > 0 { + state.Running = len(runningResult.Data) + state.RunningIDs = make([]string, 0, len(runningResult.Data)) + for _, exec := range runningResult.Data { state.RunningIDs = append(state.RunningIDs, exec.ExecutionID) } // Update status based on running count diff --git a/agent/robot/api/types.go b/agent/robot/api/types.go index 0f6d55fb..3e7557a6 100644 --- a/agent/robot/api/types.go +++ b/agent/robot/api/types.go @@ -96,10 +96,11 @@ type TriggerResult struct { // ExecutionQuery - query options for GetExecutions() type ExecutionQuery struct { - Status types.ExecStatus `json:"status,omitempty"` - Trigger types.TriggerType `json:"trigger,omitempty"` - Page int `json:"page,omitempty"` - PageSize int `json:"pagesize,omitempty"` + Status types.ExecStatus `json:"status,omitempty"` + ExcludeStatuses []types.ExecStatus `json:"exclude_statuses,omitempty"` + Trigger types.TriggerType `json:"trigger,omitempty"` + Page int `json:"page,omitempty"` + PageSize int `json:"pagesize,omitempty"` } // ExecutionResult - result of GetExecutions() diff --git a/agent/robot/executor/standard/agent.go b/agent/robot/executor/standard/agent.go index 52cadc89..dff37b35 100644 --- a/agent/robot/executor/standard/agent.go +++ b/agent/robot/executor/standard/agent.go @@ -6,10 +6,22 @@ import ( "github.com/yaoapp/gou/text" "github.com/yaoapp/yao/agent/assistant" agentcontext "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output/message" robottypes "github.com/yaoapp/yao/agent/robot/types" oauthtypes "github.com/yaoapp/yao/openapi/oauth/types" ) +// StreamCallback receives text chunks during streaming agent calls. +// Return 0 to continue, non-zero to stop. +type StreamCallback func(chunk *StreamChunk) int + +// StreamChunk represents a single chunk in a streaming response. +type StreamChunk struct { + Type string // "text", "thinking", "event" + Content string + Delta bool +} + // AgentCaller provides unified interface for calling AI assistants // It wraps the Yao Assistant framework and handles: // - Getting assistant by ID @@ -238,6 +250,145 @@ func (c *AgentCaller) CallWithSystemAndUser(ctx *robottypes.Context, assistantID return c.Call(ctx, assistantID, messages) } +// CallStream calls an assistant with messages and streams text chunks via callback. +// The callback receives each text delta in real-time while the response is being generated. +// After streaming completes, the full CallResult is returned. +func (c *AgentCaller) CallStream(ctx *robottypes.Context, assistantID string, messages []agentcontext.Message, streamFn StreamCallback) (*CallResult, error) { + ast, err := assistant.Get(assistantID) + if err != nil { + return nil, fmt.Errorf("assistant not found: %s: %w", assistantID, err) + } + + opts := &agentcontext.Options{ + Skip: &agentcontext.Skip{ + Output: c.SkipOutput, + History: c.SkipHistory, + Search: c.SkipSearch, + }, + Connector: c.Connector, + } + + // Hook OnMessage to intercept streaming chunks and forward to callback + if streamFn != nil { + opts.OnMessage = func(msg *message.Message) int { + if msg == nil { + return 0 + } + switch msg.Type { + case message.TypeText: + if msg.Delta { + content, _ := msg.Props["content"].(string) + if content != "" { + return streamFn(&StreamChunk{Type: "text", Content: content, Delta: true}) + } + } + case message.TypeThinking: + if msg.Delta { + content, _ := msg.Props["content"].(string) + if content != "" { + return streamFn(&StreamChunk{Type: "thinking", Content: content, Delta: true}) + } + } + } + return 0 + } + } + + agentCtx := c.buildAgentContext(ctx) + defer agentCtx.Release() + + response, err := ast.Stream(agentCtx, messages, opts) + if err != nil { + return nil, fmt.Errorf("assistant call failed: %w", err) + } + + result := &CallResult{Response: response} + if response.Next != nil { + result.Next = response.Next + } + if response.Completion != nil { + if content, ok := response.Completion.Content.(string); ok { + result.Content = content + } + } + + if c.log != nil { + c.log.logAgentCall(assistantID, result) + } + + return result, nil +} + +// CallWithMessagesStream is a convenience method that streams a single user input. +func (c *AgentCaller) CallWithMessagesStream(ctx *robottypes.Context, assistantID string, userContent string, streamFn StreamCallback) (*CallResult, error) { + messages := []agentcontext.Message{ + { + Role: agentcontext.RoleUser, + Content: userContent, + }, + } + return c.CallStream(ctx, assistantID, messages, streamFn) +} + +// CallStreamRaw calls an assistant with streaming, passing raw message.Message objects +// to the callback without any type filtering or degradation. This preserves all CUI +// message protocol fields (chunk_id, message_id, block_id, delta_path, etc.) +// for direct SSE passthrough to the frontend. +func (c *AgentCaller) CallStreamRaw(ctx *robottypes.Context, assistantID string, messages []agentcontext.Message, onMessage agentcontext.OnMessageFunc) (*CallResult, error) { + ast, err := assistant.Get(assistantID) + if err != nil { + return nil, fmt.Errorf("assistant not found: %s: %w", assistantID, err) + } + + opts := &agentcontext.Options{ + Skip: &agentcontext.Skip{ + Output: c.SkipOutput, + History: c.SkipHistory, + Search: c.SkipSearch, + }, + Connector: c.Connector, + } + + if onMessage != nil { + opts.OnMessage = onMessage + } + + agentCtx := c.buildAgentContext(ctx) + defer agentCtx.Release() + + response, err := ast.Stream(agentCtx, messages, opts) + if err != nil { + return nil, fmt.Errorf("assistant call failed: %w", err) + } + + result := &CallResult{Response: response} + if response.Next != nil { + result.Next = response.Next + } + if response.Completion != nil { + if content, ok := response.Completion.Content.(string); ok { + result.Content = content + } + } + + if c.log != nil { + c.log.logAgentCall(assistantID, result) + } + + return result, nil +} + +// CallWithMessagesStreamRaw is a convenience method that streams raw messages for a single user input. +func (c *AgentCaller) CallWithMessagesStreamRaw(ctx *robottypes.Context, assistantID string, userContent string, onMessage agentcontext.OnMessageFunc) (*CallResult, error) { + messages := []agentcontext.Message{ + { + Role: agentcontext.RoleUser, + Content: userContent, + }, + } + return c.CallStreamRaw(ctx, assistantID, messages, onMessage) +} + // buildAgentContext converts robot context to agent context func (c *AgentCaller) buildAgentContext(ctx *robottypes.Context) *agentcontext.Context { // Build authorized info for agent context diff --git a/agent/robot/executor/standard/agent_stream_test.go b/agent/robot/executor/standard/agent_stream_test.go new file mode 100644 index 00000000..70c1ae96 --- /dev/null +++ b/agent/robot/executor/standard/agent_stream_test.go @@ -0,0 +1,102 @@ +package standard_test + +import ( + "context" + "strings" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + agentcontext "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/robot/executor/standard" + "github.com/yaoapp/yao/agent/robot/types" + "github.com/yaoapp/yao/agent/testutils" +) + +func TestAgentCallerCallStream(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test (requires LLM)") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + caller := standard.NewAgentCaller() + ctx := types.NewContext(context.Background(), testAuth()) + + t.Run("streams text chunks and returns result", func(t *testing.T) { + var mu sync.Mutex + var chunks []string + + streamFn := func(chunk *standard.StreamChunk) int { + mu.Lock() + defer mu.Unlock() + if chunk.Type == "text" && chunk.Delta { + chunks = append(chunks, chunk.Content) + } + return 0 + } + + result, err := caller.CallWithMessagesStream(ctx, "tests.robot-single", "Hello, test message", streamFn) + + require.NoError(t, err) + require.NotNil(t, result) + assert.False(t, result.IsEmpty()) + + mu.Lock() + combined := strings.Join(chunks, "") + chunkCount := len(chunks) + mu.Unlock() + + t.Logf("Received %d text chunks, total length: %d", chunkCount, len(combined)) + assert.Greater(t, chunkCount, 0, "should have received at least one text chunk") + assert.NotEmpty(t, combined, "combined chunks should not be empty") + }) + + t.Run("nil callback works like non-stream call", func(t *testing.T) { + result, err := caller.CallStream(ctx, "tests.robot-single", + []agentcontext.Message{{Role: "user", Content: "Hello"}}, + nil, + ) + require.NoError(t, err) + require.NotNil(t, result) + assert.False(t, result.IsEmpty()) + }) + + t.Run("stream returns parseable JSON", func(t *testing.T) { + var mu sync.Mutex + var chunks []string + + streamFn := func(chunk *standard.StreamChunk) int { + mu.Lock() + defer mu.Unlock() + if chunk.Type == "text" && chunk.Delta { + chunks = append(chunks, chunk.Content) + } + return 0 + } + + result, err := caller.CallWithMessagesStream(ctx, "tests.robot-single", "Generate inspiration report", streamFn) + + require.NoError(t, err) + require.NotNil(t, result) + + data, err := result.GetJSON() + require.NoError(t, err) + assert.NotNil(t, data) + assert.Contains(t, data, "type") + + mu.Lock() + chunkCount := len(chunks) + mu.Unlock() + t.Logf("Received %d chunks for JSON response", chunkCount) + }) + + t.Run("assistant not found returns error", func(t *testing.T) { + result, err := caller.CallWithMessagesStream(ctx, "non.existent", "hello", nil) + assert.Error(t, err) + assert.Nil(t, result) + assert.Contains(t, err.Error(), "assistant not found") + }) +} diff --git a/agent/robot/manager/interact.go b/agent/robot/manager/interact.go index d85934ab..08fcfd03 100644 --- a/agent/robot/manager/interact.go +++ b/agent/robot/manager/interact.go @@ -3,10 +3,12 @@ package manager import ( "encoding/json" "fmt" + "strings" "time" "github.com/yaoapp/kun/log" agentcontext "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/output/message" robotevents "github.com/yaoapp/yao/agent/robot/events" "github.com/yaoapp/yao/agent/robot/executor/standard" "github.com/yaoapp/yao/agent/robot/pool" @@ -129,6 +131,13 @@ func (m *Manager) HandleInteract(ctx *types.Context, memberID string, req *Inter case types.ExecWaiting: return m.handleWaitingInteraction(ctx, robot, record, req, execStore) case types.ExecRunning: + if record.WaitingTaskID == "" { + return &InteractResponse{ + ExecutionID: record.ExecutionID, + Status: "rejected", + Message: "Execution is running and not waiting for input", + }, nil + } return m.handleRunningInteraction(ctx, robot, record, req, execStore) default: return nil, fmt.Errorf("execution %s is in status %s, cannot interact", req.ExecutionID, record.Status) @@ -333,26 +342,25 @@ func (m *Manager) callHostAgent(ctx *types.Context, agentID string, input *types return nil, fmt.Errorf("host agent (%s) call failed: %w", agentID, err) } - // Parse Host Agent response as JSON + return m.parseHostAgentResult(result) +} + +// parseHostAgentResult inspects the agent result to determine if it is an action +// decision (JSON with "action" field) or a conversational reply (natural language). +func (m *Manager) parseHostAgentResult(result *standard.CallResult) (*types.HostOutput, error) { data, err := result.GetJSON() - if err != nil { - text := result.GetText() - return &types.HostOutput{ - Reply: text, - Action: types.HostActionConfirm, - }, nil + if err == nil { + output := &types.HostOutput{} + raw, _ := json.Marshal(data) + if err := json.Unmarshal(raw, output); err == nil && output.Action != "" { + return output, nil + } } - output := &types.HostOutput{} - raw, _ := json.Marshal(data) - if err := json.Unmarshal(raw, output); err != nil { - return &types.HostOutput{ - Reply: result.GetText(), - Action: types.HostActionConfirm, - }, nil - } - - return output, nil + return &types.HostOutput{ + Reply: result.GetText(), + WaitForMore: true, + }, nil } // processHostAction processes the output from Host Agent and takes the appropriate action. @@ -571,3 +579,409 @@ func (m *Manager) directResume(ctx *types.Context, record *store.ExecutionRecord ChatID: record.ChatID, }, nil } + +// ==================== Streaming Interact ==================== + +// HandleInteractStream is the streaming version of HandleInteract. +// It streams Host Agent text tokens via streamFn while still returning the final InteractResponse. +func (m *Manager) HandleInteractStream(ctx *types.Context, memberID string, req *InteractRequest, streamFn standard.StreamCallback) (*InteractResponse, error) { + m.mu.RLock() + if !m.started { + m.mu.RUnlock() + return nil, fmt.Errorf("manager not started") + } + m.mu.RUnlock() + + if memberID == "" { + return nil, fmt.Errorf("member_id is required") + } + if req == nil || req.Message == "" { + return nil, fmt.Errorf("message is required") + } + + robot, _, err := m.getOrLoadRobot(ctx, memberID) + if err != nil { + return nil, fmt.Errorf("robot not found: %w", err) + } + + execStore := store.NewExecutionStore() + + if req.ExecutionID == "" { + return m.handleNewInteractionStream(ctx, robot, req, execStore, streamFn) + } + + record, err := execStore.Get(ctx.Context, req.ExecutionID) + if err != nil { + return nil, fmt.Errorf("execution not found: %s", req.ExecutionID) + } + + switch record.Status { + case types.ExecConfirming: + return m.handleConfirmingInteractionStream(ctx, robot, record, req, execStore, streamFn) + case types.ExecWaiting: + return m.handleWaitingInteractionStream(ctx, robot, record, req, execStore, streamFn) + case types.ExecRunning: + if record.WaitingTaskID == "" { + return &InteractResponse{ + ExecutionID: record.ExecutionID, + Status: "rejected", + Message: "Execution is running and not waiting for input", + }, nil + } + return m.handleRunningInteractionStream(ctx, robot, record, req, execStore, streamFn) + default: + return nil, fmt.Errorf("execution %s is in status %s, cannot interact", req.ExecutionID, record.Status) + } +} + +func (m *Manager) handleNewInteractionStream(ctx *types.Context, robot *types.Robot, req *InteractRequest, execStore *store.ExecutionStore, streamFn standard.StreamCallback) (*InteractResponse, error) { + exec, chatID, err := m.createConfirmingExecution(ctx, robot, req, execStore) + if err != nil { + return nil, fmt.Errorf("failed to create confirming execution: %w", err) + } + + hostOutput, err := m.callHostAgentForScenarioStream(ctx, robot, "assign", req.Message, nil, chatID, streamFn) + if err != nil { + log.Warn("Host Agent call failed, using direct assign: %v", err) + return m.directAssign(ctx, robot, exec, req, execStore) + } + + resp, err := m.processHostAction(ctx, robot, exec, hostOutput, execStore) + if err != nil { + return nil, err + } + resp.ExecutionID = exec.ExecutionID + resp.ChatID = chatID + return resp, nil +} + +func (m *Manager) handleConfirmingInteractionStream(ctx *types.Context, robot *types.Robot, record *store.ExecutionRecord, req *InteractRequest, execStore *store.ExecutionStore, streamFn standard.StreamCallback) (*InteractResponse, error) { + hostCtx := m.buildHostContext(robot, record, nil) + hostOutput, err := m.callHostAgentForScenarioStream(ctx, robot, "assign", req.Message, hostCtx, record.ChatID, streamFn) + if err != nil { + log.Warn("Host Agent call failed during confirming: %v", err) + return &InteractResponse{ + ExecutionID: record.ExecutionID, + Status: "error", + Message: fmt.Sprintf("Host Agent failed: %v", err), + }, nil + } + + resp, err := m.processHostAction(ctx, robot, record, hostOutput, execStore) + if err != nil { + return nil, err + } + resp.ExecutionID = record.ExecutionID + resp.ChatID = record.ChatID + return resp, nil +} + +func (m *Manager) handleWaitingInteractionStream(ctx *types.Context, robot *types.Robot, record *store.ExecutionRecord, req *InteractRequest, execStore *store.ExecutionStore, streamFn standard.StreamCallback) (*InteractResponse, error) { + waitingTask := m.findWaitingTask(record) + hostCtx := m.buildHostContext(robot, record, waitingTask) + + hostOutput, err := m.callHostAgentForScenarioStream(ctx, robot, "clarify", req.Message, hostCtx, record.ChatID, streamFn) + if err != nil { + log.Warn("Host Agent call failed during clarify, falling back to direct resume: %v", err) + return m.directResume(ctx, record, req) + } + + resp, err := m.processHostAction(ctx, robot, record, hostOutput, execStore) + if err != nil { + return nil, err + } + resp.ExecutionID = record.ExecutionID + resp.ChatID = record.ChatID + return resp, nil +} + +func (m *Manager) handleRunningInteractionStream(ctx *types.Context, robot *types.Robot, record *store.ExecutionRecord, req *InteractRequest, execStore *store.ExecutionStore, streamFn standard.StreamCallback) (*InteractResponse, error) { + hostCtx := m.buildHostContext(robot, record, nil) + hostOutput, err := m.callHostAgentForScenarioStream(ctx, robot, "guide", req.Message, hostCtx, record.ChatID, streamFn) + if err != nil { + return &InteractResponse{ + ExecutionID: record.ExecutionID, + Status: "acknowledged", + Message: "Guidance noted (Host Agent unavailable)", + }, nil + } + + resp, err := m.processHostAction(ctx, robot, record, hostOutput, execStore) + if err != nil { + return nil, err + } + resp.ExecutionID = record.ExecutionID + resp.ChatID = record.ChatID + return resp, nil +} + +func (m *Manager) callHostAgentForScenarioStream(ctx *types.Context, robot *types.Robot, scenario string, msg string, hostCtx *types.HostContext, chatID string, streamFn standard.StreamCallback) (*types.HostOutput, error) { + agentID := "" + if robot.Config != nil && robot.Config.Resources != nil { + agentID = robot.Config.Resources.GetPhaseAgent(types.PhaseHost) + } + if agentID == "" { + return nil, fmt.Errorf("no Host Agent configured for robot %s", robot.MemberID) + } + + return m.callHostAgentStream(ctx, agentID, &types.HostInput{ + Scenario: scenario, + Messages: []agentcontext.Message{{Role: "user", Content: msg}}, + Context: hostCtx, + }, chatID, streamFn) +} + +func (m *Manager) callHostAgentStream(ctx *types.Context, agentID string, input *types.HostInput, chatID string, streamFn standard.StreamCallback) (*types.HostOutput, error) { + inputJSON, err := json.Marshal(input) + if err != nil { + return nil, fmt.Errorf("failed to marshal host input: %w", err) + } + + caller := standard.NewConversationCaller(chatID) + result, err := caller.CallWithMessagesStream(ctx, agentID, string(inputJSON), streamFn) + if err != nil { + return nil, fmt.Errorf("host agent (%s) call failed: %w", agentID, err) + } + + return m.parseHostAgentResult(result) +} + +// ==================== Raw Message Streaming (CUI Protocol) ==================== + +// HandleInteractStreamRaw is the CUI-protocol-aligned streaming version of HandleInteract. +// It passes raw message.Message objects directly to the onMessage callback, preserving all +// CUI protocol fields for direct SSE passthrough to the frontend. +func (m *Manager) HandleInteractStreamRaw(ctx *types.Context, memberID string, req *InteractRequest, onMessage agentcontext.OnMessageFunc) (*InteractResponse, error) { + m.mu.RLock() + if !m.started { + m.mu.RUnlock() + return nil, fmt.Errorf("manager not started") + } + m.mu.RUnlock() + + if memberID == "" { + return nil, fmt.Errorf("member_id is required") + } + if req == nil || req.Message == "" { + return nil, fmt.Errorf("message is required") + } + + robot, _, err := m.getOrLoadRobot(ctx, memberID) + if err != nil { + return nil, fmt.Errorf("robot not found: %w", err) + } + + execStore := store.NewExecutionStore() + + if req.ExecutionID == "" { + return m.handleNewInteractionStreamRaw(ctx, robot, req, execStore, onMessage) + } + + record, err := execStore.Get(ctx.Context, req.ExecutionID) + if err != nil { + return nil, fmt.Errorf("execution not found: %s", req.ExecutionID) + } + + switch record.Status { + case types.ExecConfirming: + return m.handleConfirmingInteractionStreamRaw(ctx, robot, record, req, execStore, onMessage) + case types.ExecWaiting: + return m.handleWaitingInteractionStreamRaw(ctx, robot, record, req, execStore, onMessage) + case types.ExecRunning: + if record.WaitingTaskID == "" { + return &InteractResponse{ + ExecutionID: record.ExecutionID, + Status: "rejected", + Message: "Execution is running and not waiting for input", + }, nil + } + return m.handleRunningInteractionStreamRaw(ctx, robot, record, req, execStore, onMessage) + default: + return nil, fmt.Errorf("execution %s is in status %s, cannot interact", req.ExecutionID, record.Status) + } +} + +func (m *Manager) handleNewInteractionStreamRaw(ctx *types.Context, robot *types.Robot, req *InteractRequest, execStore *store.ExecutionStore, onMessage agentcontext.OnMessageFunc) (*InteractResponse, error) { + exec, chatID, err := m.createConfirmingExecution(ctx, robot, req, execStore) + if err != nil { + return nil, fmt.Errorf("failed to create confirming execution: %w", err) + } + + hostOutput, err := m.callHostAgentForScenarioStreamRaw(ctx, robot, "assign", req.Message, nil, chatID, onMessage) + if err != nil { + log.Warn("Host Agent call failed, using direct assign: %v", err) + return m.directAssign(ctx, robot, exec, req, execStore) + } + + resp, err := m.processHostAction(ctx, robot, exec, hostOutput, execStore) + if err != nil { + return nil, err + } + resp.ExecutionID = exec.ExecutionID + resp.ChatID = chatID + return resp, nil +} + +func (m *Manager) handleConfirmingInteractionStreamRaw(ctx *types.Context, robot *types.Robot, record *store.ExecutionRecord, req *InteractRequest, execStore *store.ExecutionStore, onMessage agentcontext.OnMessageFunc) (*InteractResponse, error) { + hostCtx := m.buildHostContext(robot, record, nil) + hostOutput, err := m.callHostAgentForScenarioStreamRaw(ctx, robot, "assign", req.Message, hostCtx, record.ChatID, onMessage) + if err != nil { + log.Warn("Host Agent call failed during confirming: %v", err) + return &InteractResponse{ + ExecutionID: record.ExecutionID, + Status: "error", + Message: fmt.Sprintf("Host Agent failed: %v", err), + }, nil + } + + resp, err := m.processHostAction(ctx, robot, record, hostOutput, execStore) + if err != nil { + return nil, err + } + resp.ExecutionID = record.ExecutionID + resp.ChatID = record.ChatID + return resp, nil +} + +func (m *Manager) handleWaitingInteractionStreamRaw(ctx *types.Context, robot *types.Robot, record *store.ExecutionRecord, req *InteractRequest, execStore *store.ExecutionStore, onMessage agentcontext.OnMessageFunc) (*InteractResponse, error) { + waitingTask := m.findWaitingTask(record) + hostCtx := m.buildHostContext(robot, record, waitingTask) + + hostOutput, err := m.callHostAgentForScenarioStreamRaw(ctx, robot, "clarify", req.Message, hostCtx, record.ChatID, onMessage) + if err != nil { + log.Warn("Host Agent call failed during clarify, falling back to direct resume: %v", err) + return m.directResume(ctx, record, req) + } + + resp, err := m.processHostAction(ctx, robot, record, hostOutput, execStore) + if err != nil { + return nil, err + } + resp.ExecutionID = record.ExecutionID + resp.ChatID = record.ChatID + return resp, nil +} + +func (m *Manager) handleRunningInteractionStreamRaw(ctx *types.Context, robot *types.Robot, record *store.ExecutionRecord, req *InteractRequest, execStore *store.ExecutionStore, onMessage agentcontext.OnMessageFunc) (*InteractResponse, error) { + hostCtx := m.buildHostContext(robot, record, nil) + hostOutput, err := m.callHostAgentForScenarioStreamRaw(ctx, robot, "guide", req.Message, hostCtx, record.ChatID, onMessage) + if err != nil { + return &InteractResponse{ + ExecutionID: record.ExecutionID, + Status: "acknowledged", + Message: "Guidance noted (Host Agent unavailable)", + }, nil + } + + resp, err := m.processHostAction(ctx, robot, record, hostOutput, execStore) + if err != nil { + return nil, err + } + resp.ExecutionID = record.ExecutionID + resp.ChatID = record.ChatID + return resp, nil +} + +func (m *Manager) callHostAgentForScenarioStreamRaw(ctx *types.Context, robot *types.Robot, scenario string, msg string, hostCtx *types.HostContext, chatID string, onMessage agentcontext.OnMessageFunc) (*types.HostOutput, error) { + agentID := "" + if robot.Config != nil && robot.Config.Resources != nil { + agentID = robot.Config.Resources.GetPhaseAgent(types.PhaseHost) + } + if agentID == "" { + return nil, fmt.Errorf("no Host Agent configured for robot %s", robot.MemberID) + } + + return m.callHostAgentStreamRaw(ctx, agentID, &types.HostInput{ + Scenario: scenario, + Messages: []agentcontext.Message{{Role: "user", Content: msg}}, + Context: hostCtx, + }, chatID, onMessage) +} + +// callHostAgentStreamRaw calls the Host Agent with CUI raw message streaming. +// It buffers text chunks that look like JSON output (starting with "{" or "```json") +// so the frontend never sees raw decision JSON. If the final result is a decision, +// the buffered chunks are discarded and a clean reply is sent instead. If the +// result is a normal conversation turn, buffered chunks are flushed through. +func (m *Manager) callHostAgentStreamRaw(ctx *types.Context, agentID string, input *types.HostInput, chatID string, onMessage agentcontext.OnMessageFunc) (*types.HostOutput, error) { + inputJSON, err := json.Marshal(input) + if err != nil { + return nil, fmt.Errorf("failed to marshal host input: %w", err) + } + + var ( + bufferedChunks []*message.Message + buffering bool + accumulatedText string + lastTextMsgID string + ) + + wrappedOnMessage := func(msg *message.Message) int { + if msg == nil { + return onMessage(msg) + } + + // Only intercept text type messages with delta content + if msg.Type != message.TypeText || !msg.Delta { + return onMessage(msg) + } + + if msg.MessageID != "" { + lastTextMsgID = msg.MessageID + } + + // Extract the text content from this chunk + chunkText := "" + if msg.Props != nil { + if c, ok := msg.Props["content"].(string); ok { + chunkText = c + } + } + accumulatedText += chunkText + + // Decide whether to buffer: check accumulated text so far + trimmed := strings.TrimSpace(accumulatedText) + if !buffering && len(trimmed) > 0 { + if trimmed[0] == '{' || strings.HasPrefix(trimmed, "```") { + buffering = true + } + } + + if buffering { + bufferedChunks = append(bufferedChunks, msg) + return 0 + } + + return onMessage(msg) + } + + caller := standard.NewConversationCaller(chatID) + result, err := caller.CallWithMessagesStreamRaw(ctx, agentID, string(inputJSON), wrappedOnMessage) + if err != nil { + return nil, fmt.Errorf("host agent (%s) call failed: %w", agentID, err) + } + + output, err := m.parseHostAgentResult(result) + if err != nil { + return nil, err + } + + if output.Action != "" && lastTextMsgID != "" { + // Decision detected — discard buffered JSON chunks, send reply text + onMessage(&message.Message{ + Type: message.TypeText, + MessageID: lastTextMsgID, + Props: map[string]interface{}{"content": output.Reply}, + Delta: false, + }) + } else if len(bufferedChunks) > 0 { + // Not a decision — flush all buffered chunks to the frontend + for _, chunk := range bufferedChunks { + if onMessage(chunk) != 0 { + break + } + } + } + + return output, nil +} diff --git a/agent/robot/manager/interact_test.go b/agent/robot/manager/interact_test.go index 74a9606c..98483893 100644 --- a/agent/robot/manager/interact_test.go +++ b/agent/robot/manager/interact_test.go @@ -5,6 +5,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/yaoapp/yao/agent/robot/executor/standard" "github.com/yaoapp/yao/agent/robot/store" "github.com/yaoapp/yao/agent/robot/types" ) @@ -186,3 +187,63 @@ func TestCancelExecutionValidation(t *testing.T) { assert.Contains(t, err.Error(), "manager not started") }) } + +func TestParseHostAgentResult(t *testing.T) { + m := &Manager{} + + t.Run("plain text returns WaitForMore", func(t *testing.T) { + result := &standard.CallResult{Content: "I understand your request. Shall I proceed?"} + output, err := m.parseHostAgentResult(result) + require.NoError(t, err) + assert.True(t, output.WaitForMore, "plain text should set WaitForMore=true") + assert.Equal(t, "I understand your request. Shall I proceed?", output.Reply) + assert.Empty(t, string(output.Action), "plain text should have no action") + }) + + t.Run("JSON with action returns action", func(t *testing.T) { + result := &standard.CallResult{ + Content: `{"reply":"Task confirmed","action":"confirm","wait_for_more":false}`, + } + output, err := m.parseHostAgentResult(result) + require.NoError(t, err) + assert.False(t, output.WaitForMore) + assert.Equal(t, types.HostActionConfirm, output.Action) + assert.Equal(t, "Task confirmed", output.Reply) + }) + + t.Run("JSON without action returns WaitForMore", func(t *testing.T) { + result := &standard.CallResult{ + Content: `{"reply":"Let me think about this","some_field":"value"}`, + } + output, err := m.parseHostAgentResult(result) + require.NoError(t, err) + assert.True(t, output.WaitForMore, "JSON without action should set WaitForMore=true") + assert.NotEmpty(t, output.Reply) + }) + + t.Run("JSON with adjust action and action_data", func(t *testing.T) { + result := &standard.CallResult{ + Content: `{"reply":"Plan adjusted","action":"adjust","action_data":{"goals":"new goals"}}`, + } + output, err := m.parseHostAgentResult(result) + require.NoError(t, err) + assert.False(t, output.WaitForMore) + assert.Equal(t, types.HostActionAdjust, output.Action) + assert.NotNil(t, output.ActionData) + }) + + t.Run("malformed JSON returns WaitForMore", func(t *testing.T) { + result := &standard.CallResult{Content: `{invalid json`} + output, err := m.parseHostAgentResult(result) + require.NoError(t, err) + assert.True(t, output.WaitForMore) + assert.Equal(t, `{invalid json`, output.Reply) + }) + + t.Run("empty content returns WaitForMore", func(t *testing.T) { + result := &standard.CallResult{Content: ""} + output, err := m.parseHostAgentResult(result) + require.NoError(t, err) + assert.True(t, output.WaitForMore) + }) +} diff --git a/agent/robot/store/execution.go b/agent/robot/store/execution.go index d4411ee7..c65526f0 100644 --- a/agent/robot/store/execution.go +++ b/agent/robot/store/execution.go @@ -62,13 +62,22 @@ type CurrentState struct { // ListOptions - options for listing execution records type ListOptions struct { - MemberID string `json:"member_id,omitempty"` // Filter by robot member ID - TeamID string `json:"team_id,omitempty"` - Status types.ExecStatus `json:"status,omitempty"` - TriggerType types.TriggerType `json:"trigger_type,omitempty"` - Limit int `json:"limit,omitempty"` - Offset int `json:"offset,omitempty"` - OrderBy string `json:"order_by,omitempty"` // e.g., "start_time desc" + MemberID string `json:"member_id,omitempty"` + TeamID string `json:"team_id,omitempty"` + Status types.ExecStatus `json:"status,omitempty"` + ExcludeStatuses []types.ExecStatus `json:"exclude_statuses,omitempty"` + TriggerType types.TriggerType `json:"trigger_type,omitempty"` + Page int `json:"page,omitempty"` + PageSize int `json:"pagesize,omitempty"` + OrderBy string `json:"order_by,omitempty"` +} + +// ListResult wraps paginated list results +type ListResult struct { + Data []*ExecutionRecord + Total int + Page int + PageSize int } // ExecutionStore - persistent storage for robot execution records @@ -142,17 +151,19 @@ func (s *ExecutionStore) Get(ctx context.Context, executionID string) (*Executio return s.mapToRecord(rows[0]) } -// List retrieves execution records with filters -func (s *ExecutionStore) List(ctx context.Context, opts *ListOptions) ([]*ExecutionRecord, error) { +// List retrieves execution records with pagination using mod.Paginate +func (s *ExecutionStore) List(ctx context.Context, opts *ListOptions) (*ListResult, error) { mod := model.Select(s.modelID) if mod == nil { return nil, fmt.Errorf("model %s not found", s.modelID) } params := model.QueryParam{} - - // Build where conditions var wheres []model.QueryWhere + + page := 1 + pageSize := 20 + if opts != nil { if opts.MemberID != "" { wheres = append(wheres, model.QueryWhere{Column: "member_id", Value: opts.MemberID}) @@ -163,49 +174,62 @@ func (s *ExecutionStore) List(ctx context.Context, opts *ListOptions) ([]*Execut if opts.Status != "" { wheres = append(wheres, model.QueryWhere{Column: "status", Value: string(opts.Status)}) } + for _, es := range opts.ExcludeStatuses { + wheres = append(wheres, model.QueryWhere{Column: "status", Value: string(es), OP: "ne"}) + } if opts.TriggerType != "" { wheres = append(wheres, model.QueryWhere{Column: "trigger_type", Value: string(opts.TriggerType)}) } - params.Limit = opts.Limit - if params.Limit == 0 { - params.Limit = 100 // default limit + if opts.Page > 0 { + page = opts.Page } - - // Note: model.QueryParam doesn't have Offset, use Page instead - if opts.Offset > 0 && opts.Limit > 0 { - params.Page = (opts.Offset / opts.Limit) + 1 + if opts.PageSize > 0 { + pageSize = opts.PageSize + if pageSize > 100 { + pageSize = 100 + } } if opts.OrderBy != "" { - // Parse OrderBy: "column desc" or "column asc" or just "column" parts := splitOrderBy(opts.OrderBy) params.Orders = []model.QueryOrder{{Column: parts[0], Option: parts[1]}} } else { params.Orders = []model.QueryOrder{{Column: "start_time", Option: "desc"}} } } else { - params.Limit = 100 params.Orders = []model.QueryOrder{{Column: "start_time", Option: "desc"}} } params.Wheres = wheres - rows, err := mod.Get(params) + res, err := mod.Paginate(params, page, pageSize) if err != nil { return nil, fmt.Errorf("failed to list execution records: %w", err) } - records := make([]*ExecutionRecord, 0, len(rows)) - for _, row := range rows { + total := 0 + if v, ok := res["total"].(int64); ok { + total = int(v) + } else if v, ok := res["total"].(int); ok { + total = v + } + + records := make([]*ExecutionRecord, 0) + for _, row := range toRows(res["data"]) { record, err := s.mapToRecord(row) if err != nil { - continue // skip invalid records + continue } records = append(records, record) } - return records, nil + return &ListResult{ + Data: records, + Total: total, + Page: page, + PageSize: pageSize, + }, nil } // UpdatePhase updates the current phase and its data @@ -767,6 +791,23 @@ func (s *ExecutionStore) toJSON(v interface{}) ([]byte, error) { // splitOrderBy parses "column desc" or "column asc" or just "column" // Returns [column, option] where option defaults to "desc" +// toRows converts Paginate result data to []map[string]interface{} +// handles type aliases like maps.MapStrAny via JSON round-trip +func toRows(data interface{}) []map[string]interface{} { + if data == nil { + return nil + } + raw, err := json.Marshal(data) + if err != nil { + return nil + } + var rows []map[string]interface{} + if err := json.Unmarshal(raw, &rows); err != nil { + return nil + } + return rows +} + func splitOrderBy(orderBy string) [2]string { parts := [2]string{"", "desc"} if orderBy == "" { @@ -819,12 +860,12 @@ func (s *ExecutionStore) parseTime(v interface{}) *time.Time { // ResultListOptions - options for listing execution results (deliveries) type ResultListOptions struct { - MemberID string `json:"member_id,omitempty"` // Filter by robot member ID - TeamID string `json:"team_id,omitempty"` // Filter by team ID - TriggerType types.TriggerType `json:"trigger_type,omitempty"` // Filter by trigger type - Keyword string `json:"keyword,omitempty"` // Search in delivery.content.summary - Limit int `json:"limit,omitempty"` - Offset int `json:"offset,omitempty"` + MemberID string `json:"member_id,omitempty"` + TeamID string `json:"team_id,omitempty"` + TriggerType types.TriggerType `json:"trigger_type,omitempty"` + Keyword string `json:"keyword,omitempty"` + Page int `json:"page,omitempty"` + PageSize int `json:"pagesize,omitempty"` } // ResultListResponse - paginated result list response @@ -867,52 +908,43 @@ func (s *ExecutionStore) ListResults(ctx context.Context, opts *ResultListOption } } - // Get total count first - total, err := s.countWithWheres(wheres) - if err != nil { - return nil, fmt.Errorf("failed to count results: %w", err) - } - - // Set pagination defaults - limit := 20 - offset := 0 + page := 1 + pageSize := 20 if opts != nil { - if opts.Limit > 0 { - limit = opts.Limit - if limit > 100 { - limit = 100 + if opts.Page > 0 { + page = opts.Page + } + if opts.PageSize > 0 { + pageSize = opts.PageSize + if pageSize > 100 { + pageSize = 100 } } - if opts.Offset > 0 { - offset = opts.Offset - } - } - - // Calculate page from offset - page := 1 - if limit > 0 && offset > 0 { - page = (offset / limit) + 1 } params := model.QueryParam{ Wheres: wheres, - Limit: limit, - Page: page, Orders: []model.QueryOrder{{Column: "end_time", Option: "desc"}}, } - rows, err := mod.Get(params) + res, err := mod.Paginate(params, page, pageSize) if err != nil { return nil, fmt.Errorf("failed to list results: %w", err) } - records := make([]*ExecutionRecord, 0, len(rows)) - for _, row := range rows { + total := 0 + if v, ok := res["total"].(int64); ok { + total = int(v) + } else if v, ok := res["total"].(int); ok { + total = v + } + + records := make([]*ExecutionRecord, 0) + for _, row := range toRows(res["data"]) { record, err := s.mapToRecord(row) if err != nil { - continue // skip invalid records + continue } - // Double check delivery content exists if record.Delivery != nil && record.Delivery.Content != nil { records = append(records, record) } @@ -922,7 +954,7 @@ func (s *ExecutionStore) ListResults(ctx context.Context, opts *ResultListOption Data: records, Total: total, Page: page, - PageSize: limit, + PageSize: pageSize, }, nil } diff --git a/agent/robot/store/execution_test.go b/agent/robot/store/execution_test.go index 9fac7d46..033f77f2 100644 --- a/agent/robot/store/execution_test.go +++ b/agent/robot/store/execution_test.go @@ -162,71 +162,71 @@ func TestExecutionStoreList(t *testing.T) { setupTestExecutionsForList(t, s, ctx) t.Run("lists_all_records_without_filters", func(t *testing.T) { - records, err := s.List(ctx, nil) + result, err := s.List(ctx, nil) require.NoError(t, err) - assert.GreaterOrEqual(t, len(records), 4) + assert.GreaterOrEqual(t, len(result.Data), 4) }) t.Run("filters_by_member_id", func(t *testing.T) { - records, err := s.List(ctx, &store.ListOptions{ + result, err := s.List(ctx, &store.ListOptions{ MemberID: "member_list_001", }) require.NoError(t, err) - assert.Equal(t, 2, len(records)) - for _, r := range records { + assert.Equal(t, 2, len(result.Data)) + for _, r := range result.Data { assert.Equal(t, "member_list_001", r.MemberID) } }) t.Run("filters_by_team_id", func(t *testing.T) { - records, err := s.List(ctx, &store.ListOptions{ + result, err := s.List(ctx, &store.ListOptions{ TeamID: "team_list_001", }) require.NoError(t, err) - assert.Equal(t, 3, len(records)) - for _, r := range records { + assert.Equal(t, 3, len(result.Data)) + for _, r := range result.Data { assert.Equal(t, "team_list_001", r.TeamID) } }) t.Run("filters_by_status", func(t *testing.T) { - records, err := s.List(ctx, &store.ListOptions{ + result, err := s.List(ctx, &store.ListOptions{ Status: types.ExecCompleted, }) require.NoError(t, err) - assert.GreaterOrEqual(t, len(records), 2) - for _, r := range records { + assert.GreaterOrEqual(t, len(result.Data), 2) + for _, r := range result.Data { assert.Equal(t, types.ExecCompleted, r.Status) } }) t.Run("filters_by_trigger_type", func(t *testing.T) { - records, err := s.List(ctx, &store.ListOptions{ + result, err := s.List(ctx, &store.ListOptions{ TriggerType: types.TriggerHuman, }) require.NoError(t, err) - assert.GreaterOrEqual(t, len(records), 1) - for _, r := range records { + assert.GreaterOrEqual(t, len(result.Data), 1) + for _, r := range result.Data { assert.Equal(t, types.TriggerHuman, r.TriggerType) } }) - t.Run("respects_limit", func(t *testing.T) { - records, err := s.List(ctx, &store.ListOptions{ - Limit: 2, + t.Run("respects_pagesize", func(t *testing.T) { + result, err := s.List(ctx, &store.ListOptions{ + PageSize: 2, }) require.NoError(t, err) - assert.Equal(t, 2, len(records)) + assert.Equal(t, 2, len(result.Data)) }) t.Run("combines_multiple_filters", func(t *testing.T) { - records, err := s.List(ctx, &store.ListOptions{ + result, err := s.List(ctx, &store.ListOptions{ TeamID: "team_list_001", Status: types.ExecCompleted, }) require.NoError(t, err) - assert.Equal(t, 2, len(records)) - for _, r := range records { + assert.Equal(t, 2, len(result.Data)) + for _, r := range result.Data { assert.Equal(t, "team_list_001", r.TeamID) assert.Equal(t, types.ExecCompleted, r.Status) } @@ -1085,8 +1085,8 @@ func TestExecutionStoreListResults(t *testing.T) { t.Run("respects_pagination", func(t *testing.T) { result, err := s.ListResults(ctx, &store.ResultListOptions{ MemberID: "member_result_001", - Limit: 1, - Offset: 0, + PageSize: 1, + Page: 1, }) require.NoError(t, err) require.NotNil(t, result) diff --git a/openapi/agent/robot/execution.go b/openapi/agent/robot/execution.go index ca603f1e..e857fb0e 100644 --- a/openapi/agent/robot/execution.go +++ b/openapi/agent/robot/execution.go @@ -95,6 +95,14 @@ func ListExecutions(c *gin.Context) { if filter.Status != "" { query.Status = robottypes.ExecStatus(filter.Status) } + if filter.ExcludeStatus != "" { + for _, s := range strings.Split(filter.ExcludeStatus, ",") { + s = strings.TrimSpace(s) + if s != "" { + query.ExcludeStatuses = append(query.ExcludeStatuses, robottypes.ExecStatus(s)) + } + } + } if filter.TriggerType != "" { query.Trigger = robottypes.TriggerType(filter.TriggerType) } diff --git a/openapi/agent/robot/interact.go b/openapi/agent/robot/interact.go index 98abc4af..4ea368e1 100644 --- a/openapi/agent/robot/interact.go +++ b/openapi/agent/robot/interact.go @@ -1,10 +1,13 @@ package robot import ( + "encoding/json" "errors" + "fmt" "github.com/gin-gonic/gin" "github.com/yaoapp/kun/log" + "github.com/yaoapp/yao/agent/output/message" robotapi "github.com/yaoapp/yao/agent/robot/api" robottypes "github.com/yaoapp/yao/agent/robot/types" "github.com/yaoapp/yao/openapi/oauth/authorized" @@ -18,6 +21,7 @@ type InteractRequest struct { Source string `json:"source,omitempty"` Message string `json:"message" binding:"required"` Action string `json:"action,omitempty"` + Stream bool `json:"stream,omitempty"` } // InteractResponse - HTTP response for interaction @@ -108,6 +112,14 @@ func InteractRobot(c *gin.Context) { Action: req.Action, } + // Detect SSE mode: request body stream=true or Accept header + wantSSE := req.Stream || c.GetHeader("Accept") == "text/event-stream" + + if wantSSE { + interactSSE(c, ctx, robotID, apiReq) + return + } + result, err := robotapi.Interact(ctx, robotID, apiReq) if err != nil { log.Error("Failed to interact with robot %s: %v", robotID, err) @@ -130,6 +142,78 @@ func InteractRobot(c *gin.Context) { response.RespondWithSuccess(c, response.StatusOK, resp) } +// interactSSE handles the SSE streaming mode for robot interaction. +// Outputs standard CUI Message protocol (data: {json}\n\n) for direct frontend consumption, +// plus a final "interact_done" event with execution metadata. +func interactSSE(c *gin.Context, ctx *robottypes.Context, robotID string, apiReq *robotapi.InteractRequest) { + c.Header("Content-Type", "text/event-stream;charset=utf-8") + c.Header("Cache-Control", "no-cache") + c.Header("Connection", "keep-alive") + c.Header("X-Accel-Buffering", "no") + + w := c.Writer + flusher, ok := w.(interface{ Flush() }) + if !ok { + log.Error("ResponseWriter does not support Flush") + return + } + + writeData := func(data interface{}) { + raw, err := json.Marshal(data) + if err != nil { + return + } + fmt.Fprintf(w, "data: %s\n\n", raw) + flusher.Flush() + } + + onMessage := func(msg *message.Message) int { + if msg == nil { + return 0 + } + writeData(msg) + return 0 + } + + result, err := robotapi.InteractStreamRaw(ctx, robotID, apiReq, onMessage) + if err != nil { + writeData(&message.Message{ + Type: message.TypeError, + Props: map[string]interface{}{ + "message": err.Error(), + }, + }) + writeData(&message.Message{ + Type: message.TypeEvent, + Props: map[string]interface{}{ + "event": "interact_done", + "message": "error", + "data": map[string]interface{}{ + "status": "error", + "error": err.Error(), + }, + }, + }) + return + } + + writeData(&message.Message{ + Type: message.TypeEvent, + Props: map[string]interface{}{ + "event": "interact_done", + "message": result.Message, + "data": map[string]interface{}{ + "execution_id": result.ExecutionID, + "status": result.Status, + "message": result.Message, + "chat_id": result.ChatID, + "reply": result.Reply, + "wait_for_more": result.WaitForMore, + }, + }, + }) +} + // ReplyToTask handles replying to a specific waiting task // POST /v1/agent/robots/:id/executions/:exec_id/tasks/:task_id/reply func ReplyToTask(c *gin.Context) { diff --git a/openapi/agent/robot/types.go b/openapi/agent/robot/types.go index a5129196..26990677 100644 --- a/openapi/agent/robot/types.go +++ b/openapi/agent/robot/types.go @@ -266,11 +266,12 @@ func NewStatusResponse(s *robotapi.RobotState) *StatusResponse { // ExecutionFilter - query params for listing executions type ExecutionFilter struct { - Status string `form:"status"` // pending | running | paused | completed | failed | cancelled - TriggerType string `form:"trigger_type"` // clock | human | event - Keyword string `form:"keyword"` // search in execution details - Page int `form:"page"` - PageSize int `form:"pagesize"` + Status string `form:"status"` // pending | running | paused | completed | failed | cancelled + ExcludeStatus string `form:"exclude_status"` // comma-separated statuses to exclude, e.g. "confirming,waiting" + TriggerType string `form:"trigger_type"` // clock | human | event + Keyword string `form:"keyword"` // search in execution details + Page int `form:"page"` + PageSize int `form:"pagesize"` } // ExecutionResponse - single execution response diff --git a/openapi/tests/agent/robot_interact_test.go b/openapi/tests/agent/robot_interact_test.go new file mode 100644 index 00000000..1e8f1700 --- /dev/null +++ b/openapi/tests/agent/robot_interact_test.go @@ -0,0 +1,351 @@ +package openapi_test + +import ( + "bufio" + "bytes" + "encoding/json" + "fmt" + "net/http" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + robotapi "github.com/yaoapp/yao/agent/robot/api" + "github.com/yaoapp/yao/openapi" + "github.com/yaoapp/yao/openapi/tests/testutils" +) + +func TestInteractRobot(t *testing.T) { + if testing.Short() { + t.Skip("Skipping interact tests in short mode (requires AI/manager)") + } + + serverURL := testutils.Prepare(t) + defer testutils.Clean() + + baseURL := "" + if openapi.Server != nil && openapi.Server.Config != nil { + baseURL = openapi.Server.Config.BaseURL + } + + client := testutils.RegisterTestClient(t, "Interact Test Client", []string{"https://localhost/callback"}) + defer testutils.CleanupTestClient(t, client.ClientID) + tokenInfo := testutils.ObtainAccessToken(t, serverURL, client.ClientID, client.ClientSecret, "https://localhost/callback", "openid profile") + + err := robotapi.Start() + require.NoError(t, err, "Manager must start for Interact tests") + defer robotapi.Stop() + + robotID := fmt.Sprintf("test_interact_%d", time.Now().UnixNano()) + createRobotForInteract(t, serverURL, baseURL, tokenInfo.AccessToken, robotID, "Interact Test Robot") + defer deleteRobotForInteract(t, serverURL, baseURL, tokenInfo.AccessToken, robotID) + + t.Run("InteractSync_FullFlow", func(t *testing.T) { + interactData := map[string]interface{}{ + "message": "Please write a short greeting email for our Monday standup.", + } + + body, _ := json.Marshal(interactData) + req, err := http.NewRequest("POST", serverURL+baseURL+"/agent/robots/"+robotID+"/interact", bytes.NewBuffer(body)) + require.NoError(t, err) + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+tokenInfo.AccessToken) + + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + require.NotNil(t, resp) + defer resp.Body.Close() + + var response map[string]interface{} + err = json.NewDecoder(resp.Body).Decode(&response) + require.NoError(t, err) + + t.Logf("Sync interact: status_code=%d, response=%+v", resp.StatusCode, response) + + if resp.StatusCode == http.StatusOK { + data, _ := response["data"].(map[string]interface{}) + if data != nil { + assert.NotEmpty(t, data["execution_id"], "should have execution_id") + assert.NotEmpty(t, data["reply"], "Host Agent should provide a reply") + assert.NotEmpty(t, data["status"], "should have a status") + + validStatuses := []string{"confirmed", "waiting_for_more", "adjusted", "acknowledged"} + status, _ := data["status"].(string) + assert.Contains(t, validStatuses, status, + "status should reflect Host Agent action outcome, got: %s", status) + t.Logf("Sync result: exec_id=%v, status=%v, reply=%v, wait_for_more=%v", + data["execution_id"], data["status"], data["reply"], data["wait_for_more"]) + } + } else { + t.Logf("Sync interact returned %d: %v (may indicate Manager routing issue)", resp.StatusCode, response) + } + }) + + t.Run("InteractSSE_FullFlow", func(t *testing.T) { + interactData := map[string]interface{}{ + "message": "Draft a brief thank-you note for the design team.", + "stream": true, + } + + body, _ := json.Marshal(interactData) + req, err := http.NewRequest("POST", serverURL+baseURL+"/agent/robots/"+robotID+"/interact", bytes.NewBuffer(body)) + require.NoError(t, err) + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + req.Header.Set("Authorization", "Bearer "+tokenInfo.AccessToken) + + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + require.NotNil(t, resp) + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + var errResp map[string]interface{} + json.NewDecoder(resp.Body).Decode(&errResp) + t.Fatalf("SSE interact failed: status=%d, error=%v", resp.StatusCode, errResp) + } + + contentType := resp.Header.Get("Content-Type") + assert.Contains(t, contentType, "text/event-stream") + + messages := parseCUISSEMessages(t, resp) + require.NotEmpty(t, messages, "should receive CUI message events") + + var textMessages []map[string]interface{} + var interactDone map[string]interface{} + for _, msg := range messages { + msgType, _ := msg["type"].(string) + if msgType == "text" { + textMessages = append(textMessages, msg) + } + if msgType == "event" { + props, _ := msg["props"].(map[string]interface{}) + if props != nil { + if evt, _ := props["event"].(string); evt == "interact_done" { + interactDone = props + } + } + } + } + + t.Logf("SSE: %d total messages, %d text messages, interact_done=%v", + len(messages), len(textMessages), interactDone != nil) + + assert.NotNil(t, interactDone, "should have an interact_done event") + if interactDone != nil { + doneData, _ := interactDone["data"].(map[string]interface{}) + if doneData != nil { + if status, ok := doneData["status"].(string); ok { + validStatuses := []string{"confirmed", "waiting_for_more", "adjusted", "acknowledged", "error"} + assert.Contains(t, validStatuses, status, + "final status should be a valid outcome") + } + if execID, ok := doneData["execution_id"].(string); ok { + assert.NotEmpty(t, execID, "done event should carry execution_id") + } + t.Logf("SSE done data: %+v", doneData) + } + } + }) + + t.Run("InteractSSE_MultiTurn", func(t *testing.T) { + // Turn 1: vague request + body1, _ := json.Marshal(map[string]interface{}{ + "message": "Do something with emails.", + "stream": true, + }) + req1, _ := http.NewRequest("POST", serverURL+baseURL+"/agent/robots/"+robotID+"/interact", bytes.NewBuffer(body1)) + req1.Header.Set("Content-Type", "application/json") + req1.Header.Set("Accept", "text/event-stream") + req1.Header.Set("Authorization", "Bearer "+tokenInfo.AccessToken) + + resp1, err := http.DefaultClient.Do(req1) + require.NoError(t, err) + defer resp1.Body.Close() + + if resp1.StatusCode != http.StatusOK { + var errResp map[string]interface{} + json.NewDecoder(resp1.Body).Decode(&errResp) + t.Fatalf("Turn 1 SSE failed: status=%d, error=%v", resp1.StatusCode, errResp) + } + + turn1Messages := parseCUISSEMessages(t, resp1) + require.NotEmpty(t, turn1Messages) + + turn1Done := findInteractDone(turn1Messages) + require.NotNil(t, turn1Done, "Turn 1 should have interact_done event") + + doneData1, _ := turn1Done["data"].(map[string]interface{}) + require.NotNil(t, doneData1) + execID, _ := doneData1["execution_id"].(string) + t.Logf("Turn 1: exec_id=%s, status=%v, wait_for_more=%v", + execID, doneData1["status"], doneData1["wait_for_more"]) + assert.NotEmpty(t, execID, "Turn 1 should create an execution") + + // Turn 2: clarify with same execution_id + body2, _ := json.Marshal(map[string]interface{}{ + "execution_id": execID, + "message": "Please write a congratulations email for the team hitting Q4 targets. Yes, proceed.", + "stream": true, + }) + req2, _ := http.NewRequest("POST", serverURL+baseURL+"/agent/robots/"+robotID+"/interact", bytes.NewBuffer(body2)) + req2.Header.Set("Content-Type", "application/json") + req2.Header.Set("Accept", "text/event-stream") + req2.Header.Set("Authorization", "Bearer "+tokenInfo.AccessToken) + + resp2, err := http.DefaultClient.Do(req2) + require.NoError(t, err) + defer resp2.Body.Close() + + if resp2.StatusCode != http.StatusOK { + var errResp map[string]interface{} + json.NewDecoder(resp2.Body).Decode(&errResp) + t.Fatalf("Turn 2 SSE failed: status=%d, error=%v", resp2.StatusCode, errResp) + } + + turn2Messages := parseCUISSEMessages(t, resp2) + require.NotEmpty(t, turn2Messages) + + turn2Done := findInteractDone(turn2Messages) + require.NotNil(t, turn2Done, "Turn 2 should have interact_done event") + + doneData2, _ := turn2Done["data"].(map[string]interface{}) + require.NotNil(t, doneData2) + execID2, _ := doneData2["execution_id"].(string) + t.Logf("Turn 2: exec_id=%s, status=%v, wait_for_more=%v", + execID2, doneData2["status"], doneData2["wait_for_more"]) + assert.Equal(t, execID, execID2, "Turn 2 should reference same execution") + }) + + t.Run("InteractMissingMessage", func(t *testing.T) { + body, _ := json.Marshal(map[string]interface{}{}) + req, err := http.NewRequest("POST", serverURL+baseURL+"/agent/robots/"+robotID+"/interact", bytes.NewBuffer(body)) + require.NoError(t, err) + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+tokenInfo.AccessToken) + + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + assert.Equal(t, http.StatusBadRequest, resp.StatusCode) + }) + + t.Run("InteractNotFound", func(t *testing.T) { + body, _ := json.Marshal(map[string]interface{}{"message": "test"}) + req, err := http.NewRequest("POST", serverURL+baseURL+"/agent/robots/non_existent_robot/interact", bytes.NewBuffer(body)) + require.NoError(t, err) + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+tokenInfo.AccessToken) + + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + assert.Equal(t, http.StatusNotFound, resp.StatusCode) + }) + + t.Run("InteractUnauthorized", func(t *testing.T) { + body, _ := json.Marshal(map[string]interface{}{"message": "test"}) + req, err := http.NewRequest("POST", serverURL+baseURL+"/agent/robots/"+robotID+"/interact", bytes.NewBuffer(body)) + require.NoError(t, err) + + req.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + assert.Equal(t, http.StatusUnauthorized, resp.StatusCode) + }) +} + +// parseCUISSEMessages parses the SSE stream using CUI Message protocol format: +// each line is "data: {json}\n\n" where the JSON is a message.Message object. +func parseCUISSEMessages(t *testing.T, resp *http.Response) []map[string]interface{} { + scanner := bufio.NewScanner(resp.Body) + var messages []map[string]interface{} + + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "data: ") { + data := strings.TrimPrefix(line, "data: ") + var parsed map[string]interface{} + if err := json.Unmarshal([]byte(data), &parsed); err == nil { + messages = append(messages, parsed) + } + } + } + return messages +} + +// findInteractDone finds the interact_done event from CUI messages. +func findInteractDone(messages []map[string]interface{}) map[string]interface{} { + for _, msg := range messages { + msgType, _ := msg["type"].(string) + if msgType == "event" { + props, _ := msg["props"].(map[string]interface{}) + if props != nil { + if evt, _ := props["event"].(string); evt == "interact_done" { + return props + } + } + } + } + return nil +} + +func createRobotForInteract(t *testing.T, serverURL, baseURL, token, robotID, displayName string) { + createData := map[string]interface{}{ + "member_id": robotID, + "team_id": "test_team_001", + "display_name": displayName, + "robot_config": map[string]interface{}{ + "identity": map[string]interface{}{ + "role": "Email Assistant", + "duties": []string{"Write and manage emails"}, + }, + "quota": map[string]interface{}{ + "max": 5, + "queue": 20, + }, + "triggers": map[string]interface{}{ + "intervene": map[string]interface{}{"enabled": true}, + }, + "resources": map[string]interface{}{ + "phases": map[string]interface{}{ + "host": "robot.host", + }, + }, + }, + } + + body, _ := json.Marshal(createData) + req, err := http.NewRequest("POST", serverURL+baseURL+"/agent/robots", bytes.NewBuffer(body)) + require.NoError(t, err) + + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+token) + + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + var errBody map[string]interface{} + json.NewDecoder(resp.Body).Decode(&errBody) + t.Logf("Create robot response: %d %v", resp.StatusCode, errBody) + } +} + +func deleteRobotForInteract(t *testing.T, serverURL, baseURL, token, robotID string) { + req, _ := http.NewRequest("DELETE", serverURL+baseURL+"/agent/robots/"+robotID, nil) + req.Header.Set("Authorization", "Bearer "+token) + resp, _ := http.DefaultClient.Do(req) + if resp != nil { + resp.Body.Close() + } +}