Enhance execution listing and interaction with pagination and streaming support

- Refactor execution listing to support pagination with `Page` and `PageSize` options, replacing previous `Limit` and `Offset` parameters.
- Introduce `ExcludeStatuses` in execution queries to filter out specific execution statuses.
- Implement streaming interaction methods in the manager, allowing real-time responses from the host agent during interactions.
- Update API endpoints to accommodate new query parameters and enhance interaction capabilities with streaming support.
- Modify tests to ensure coverage for new pagination and streaming functionalities.

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

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,
+		Page:     query.Page,
-		Offset:   (query.Page - 1) * query.PageSize,
+		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
+	result, err := getExecutionStore().List(context.Background(), opts)
-	records, 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(result.Data))
-	executions := make([]*types.Execution, 0, len(records))
+	for _, record := range result.Data {
-	for _, record := range records {
 		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,
+		Total:    result.Total,
-		Page:     query.Page,
+		Page:     result.Page,
-		PageSize: query.PageSize,
+		PageSize: result.PageSize,
 	}, nil
 }
 

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

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,
+		Page:     query.Page,
-		Offset:   (query.Page - 1) * query.PageSize,
+		PageSize: query.PageSize,
 	}
 
 	if query.TriggerType != "" {

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 {
+	if err == nil && runningResult != nil && len(runningResult.Data) > 0 {
-		state.Running = len(runningExecs)
+		state.Running = len(runningResult.Data)
-		state.RunningIDs = make([]string, 0, len(runningExecs))
+		state.RunningIDs = make([]string, 0, len(runningResult.Data))
-		for _, exec := range runningExecs {
+		for _, exec := range runningResult.Data {
 			state.RunningIDs = append(state.RunningIDs, exec.ExecutionID)
 		}
 		// Update status based on running count

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"`
+	Status          types.ExecStatus   `json:"status,omitempty"`
-	Trigger  types.TriggerType `json:"trigger,omitempty"`
+	ExcludeStatuses []types.ExecStatus `json:"exclude_statuses,omitempty"`
-	Page     int               `json:"page,omitempty"`
+	Trigger         types.TriggerType  `json:"trigger,omitempty"`
-	PageSize int               `json:"pagesize,omitempty"`
+	Page            int                `json:"page,omitempty"`
+	PageSize        int                `json:"pagesize,omitempty"`
 }
 
 // ExecutionResult - result of GetExecutions()

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

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

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 {
+	if err == nil {
-		text := result.GetText()
+		output := &types.HostOutput{}
-		return &types.HostOutput{
+		raw, _ := json.Marshal(data)
-			Reply:  text,
+		if err := json.Unmarshal(raw, output); err == nil && output.Action != "" {
-			Action: types.HostActionConfirm,
+			return output, nil
-		}, nil
+		}
 	}
 
-	output := &types.HostOutput{}
+	return &types.HostOutput{
-	raw, _ := json.Marshal(data)
+		Reply:       result.GetText(),
-	if err := json.Unmarshal(raw, output); err != nil {
+		WaitForMore: true,
-		return &types.HostOutput{
+	}, nil
-			Reply:  result.GetText(),
-			Action: types.HostActionConfirm,
-		}, nil
-	}
-
-	return output, 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
+}

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

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
+	MemberID        string             `json:"member_id,omitempty"`
-	TeamID      string            `json:"team_id,omitempty"`
+	TeamID          string             `json:"team_id,omitempty"`
-	Status      types.ExecStatus  `json:"status,omitempty"`
+	Status          types.ExecStatus   `json:"status,omitempty"`
-	TriggerType types.TriggerType `json:"trigger_type,omitempty"`
+	ExcludeStatuses []types.ExecStatus `json:"exclude_statuses,omitempty"`
-	Limit       int               `json:"limit,omitempty"`
+	TriggerType     types.TriggerType  `json:"trigger_type,omitempty"`
-	Offset      int               `json:"offset,omitempty"`
+	Page            int                `json:"page,omitempty"`
-	OrderBy     string            `json:"order_by,omitempty"` // e.g., "start_time desc"
+	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
+// List retrieves execution records with pagination using mod.Paginate
-func (s *ExecutionStore) List(ctx context.Context, opts *ListOptions) ([]*ExecutionRecord, error) {
+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 opts.Page > 0 {
-		if params.Limit == 0 {
+			page = opts.Page
-			params.Limit = 100 // default limit
 		}
-
+		if opts.PageSize > 0 {
-		// Note: model.QueryParam doesn't have Offset, use Page instead
+			pageSize = opts.PageSize
-		if opts.Offset > 0 && opts.Limit > 0 {
+			if pageSize > 100 {
-			params.Page = (opts.Offset / opts.Limit) + 1
+				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))
+	total := 0
-	for _, row := range rows {
+	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
+	MemberID    string            `json:"member_id,omitempty"`
-	TeamID      string            `json:"team_id,omitempty"`      // Filter by team ID
+	TeamID      string            `json:"team_id,omitempty"`
-	TriggerType types.TriggerType `json:"trigger_type,omitempty"` // Filter by trigger type
+	TriggerType types.TriggerType `json:"trigger_type,omitempty"`
-	Keyword     string            `json:"keyword,omitempty"`      // Search in delivery.content.summary
+	Keyword     string            `json:"keyword,omitempty"`
-	Limit       int               `json:"limit,omitempty"`
+	Page        int               `json:"page,omitempty"`
-	Offset      int               `json:"offset,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
+	page := 1
-	total, err := s.countWithWheres(wheres)
+	pageSize := 20
-	if err != nil {
-		return nil, fmt.Errorf("failed to count results: %w", err)
-	}
-
-	// Set pagination defaults
-	limit := 20
-	offset := 0
 	if opts != nil {
-		if opts.Limit > 0 {
+		if opts.Page > 0 {
-			limit = opts.Limit
+			page = opts.Page
-			if limit > 100 {
+		}
-				limit = 100
+		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))
+	total := 0
-	for _, row := range rows {
+	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
 }
 

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))
+		assert.Equal(t, 2, len(result.Data))
-		for _, r := range records {
+		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))
+		assert.Equal(t, 3, len(result.Data))
-		for _, r := range records {
+		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)
+		assert.GreaterOrEqual(t, len(result.Data), 2)
-		for _, r := range records {
+		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)
+		assert.GreaterOrEqual(t, len(result.Data), 1)
-		for _, r := range records {
+		for _, r := range result.Data {
 			assert.Equal(t, types.TriggerHuman, r.TriggerType)
 		}
 	})
 
-	t.Run("respects_limit", func(t *testing.T) {
+	t.Run("respects_pagesize", func(t *testing.T) {
-		records, err := s.List(ctx, &store.ListOptions{
+		result, err := s.List(ctx, &store.ListOptions{
-			Limit: 2,
+			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))
+		assert.Equal(t, 2, len(result.Data))
-		for _, r := range records {
+		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,
+			PageSize: 1,
-			Offset:   0,
+			Page:     1,
 		})
 		require.NoError(t, err)
 		require.NotNil(t, result)

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

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

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
+	Status        string `form:"status"`         // pending | running | paused | completed | failed | cancelled
-	TriggerType string `form:"trigger_type"` // clock | human | event
+	ExcludeStatus string `form:"exclude_status"` // comma-separated statuses to exclude, e.g. "confirming,waiting"
-	Keyword     string `form:"keyword"`      // search in execution details
+	TriggerType   string `form:"trigger_type"`   // clock | human | event
-	Page        int    `form:"page"`
+	Keyword       string `form:"keyword"`        // search in execution details
-	PageSize    int    `form:"pagesize"`
+	Page          int    `form:"page"`
+	PageSize      int    `form:"pagesize"`
 }
 
 // ExecutionResponse - single execution response

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