Merge pull request #1477 from trheyi/main

Refactor delivery event handling and payload structure
This commit is contained in:
Max 2026-02-27 19:11:18 +08:00 committed by GitHub
commit b49ce2f926
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
27 changed files with 2412 additions and 1746 deletions

View file

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

View file

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

View file

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

View file

@ -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 != "" {

View file

@ -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

View file

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

View file

@ -6,6 +6,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
// EP1: ExecPayload with all execution statuses
@ -76,18 +77,14 @@ func TestTaskPayloadErrorSerialization(t *testing.T) {
// EP4: DeliveryPayload with nested content
func TestDeliveryPayloadNestedContent(t *testing.T) {
content := map[string]interface{}{
"report": map[string]interface{}{
"title": "Daily Summary",
"sections": []interface{}{"intro", "body", "conclusion"},
},
}
payload := DeliveryPayload{
ExecutionID: "exec-ep4",
MemberID: "member-ep4",
TeamID: "team-ep4",
Content: content,
Content: &robottypes.DeliveryContent{
Summary: "Daily Summary",
Body: "Full body with sections: intro, body, conclusion",
},
}
data, err := json.Marshal(payload)
@ -97,11 +94,9 @@ func TestDeliveryPayloadNestedContent(t *testing.T) {
err = json.Unmarshal(data, &parsed)
require.NoError(t, err)
contentMap, ok := parsed.Content.(map[string]interface{})
require.True(t, ok)
report, ok := contentMap["report"].(map[string]interface{})
require.True(t, ok)
assert.Equal(t, "Daily Summary", report["title"])
require.NotNil(t, parsed.Content)
assert.Equal(t, "Daily Summary", parsed.Content.Summary)
assert.Contains(t, parsed.Content.Body, "sections")
}
// EP5: Event constants follow naming convention

View file

@ -1,5 +1,7 @@
package events
import robottypes "github.com/yaoapp/yao/agent/robot/types"
// Robot event type constants for event.Push integration.
// Events are fire-and-forget; handlers are registered via event.Register().
const (
@ -46,11 +48,10 @@ type TaskPayload struct {
// DeliveryPayload is the event payload for Delivery events.
type DeliveryPayload struct {
ExecutionID string `json:"execution_id"`
MemberID string `json:"member_id"`
TeamID string `json:"team_id"`
ChatID string `json:"chat_id,omitempty"`
Result interface{} `json:"result,omitempty"`
Content interface{} `json:"content,omitempty"` // DeliveryContent from agent
Preferences interface{} `json:"preferences,omitempty"` // DeliveryPreferences for routing
ExecutionID string `json:"execution_id"`
MemberID string `json:"member_id"`
TeamID string `json:"team_id"`
ChatID string `json:"chat_id,omitempty"`
Content *robottypes.DeliveryContent `json:"content,omitempty"`
Preferences *robottypes.DeliveryPreferences `json:"preferences,omitempty"`
}

View file

@ -6,6 +6,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
robottypes "github.com/yaoapp/yao/agent/robot/types"
)
func TestEventConstants(t *testing.T) {
@ -121,8 +122,16 @@ func TestDeliveryPayloadMarshalling(t *testing.T) {
MemberID: "member-d1",
TeamID: "team-d1",
ChatID: "chat-d1",
Content: map[string]interface{}{"summary": "done"},
Preferences: map[string]interface{}{"channel": "email"},
Content: &robottypes.DeliveryContent{
Summary: "done",
Body: "full report",
},
Preferences: &robottypes.DeliveryPreferences{
Email: &robottypes.EmailPreference{
Enabled: true,
Targets: []robottypes.EmailTarget{{To: []string{"a@b.com"}}},
},
},
}
data, err := json.Marshal(payload)
@ -134,13 +143,7 @@ func TestDeliveryPayloadMarshalling(t *testing.T) {
assert.Equal(t, "exec-d1", parsed.ExecutionID)
assert.Equal(t, "member-d1", parsed.MemberID)
assert.NotNil(t, parsed.Content)
assert.Equal(t, "done", parsed.Content.Summary)
assert.NotNil(t, parsed.Preferences)
contentMap, ok := parsed.Content.(map[string]interface{})
require.True(t, ok)
assert.Equal(t, "done", contentMap["summary"])
prefMap, ok := parsed.Preferences.(map[string]interface{})
require.True(t, ok)
assert.Equal(t, "email", prefMap["channel"])
assert.NotNil(t, parsed.Preferences.Email)
}

View file

@ -1,54 +1,451 @@
package events
import (
"bytes"
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"time"
"github.com/yaoapp/gou/process"
"github.com/yaoapp/gou/text"
"github.com/yaoapp/kun/log"
robottypes "github.com/yaoapp/yao/agent/robot/types"
"github.com/yaoapp/yao/attachment"
"github.com/yaoapp/yao/event"
eventtypes "github.com/yaoapp/yao/event/types"
"github.com/yaoapp/yao/messenger"
messengerTypes "github.com/yaoapp/yao/messenger/types"
)
// DeliveryHandler processes robot.delivery events asynchronously.
// It routes delivery content to configured channels (email, webhook, process).
type DeliveryHandler struct{}
func init() {
event.Register("robot", &robotHandler{
httpClient: &http.Client{Timeout: 30 * time.Second},
})
}
// Handle processes a delivery event from the event bus.
func (h *DeliveryHandler) Handle(ctx context.Context, ev *eventtypes.Event, resp chan<- eventtypes.Result) {
var payload DeliveryPayload
if err := ev.Should(&payload); err != nil {
log.Error("delivery handler: invalid payload: %v", err)
return
}
// robotHandler processes all robot.* events.
type robotHandler struct {
httpClient *http.Client
}
log.Info(
"delivery handler: processing delivery for execution=%s member=%s",
payload.ExecutionID, payload.MemberID,
)
// Log delivery content summary for observability
if payload.Content != nil {
if data, err := json.Marshal(payload.Content); err == nil {
log.Debug("delivery handler: content=%s", string(data))
}
}
// Actual delivery routing is deferred to registered channel handlers.
// In the current implementation, the DeliveryCenter logic in delivery.go
// can be invoked here if needed. For now, this handler serves as the
// event-driven entry point for future channel-specific handlers.
if ev.IsCall {
resp <- eventtypes.Result{Data: fmt.Sprintf("delivery processed for %s", payload.ExecutionID)}
// Handle dispatches robot events by type.
func (h *robotHandler) Handle(ctx context.Context, ev *eventtypes.Event, resp chan<- eventtypes.Result) {
switch ev.Type {
case Delivery:
h.handleDelivery(ctx, ev, resp)
default:
log.Debug("robot handler: unhandled event type=%s id=%s", ev.Type, ev.ID)
}
}
// Shutdown gracefully shuts down the delivery handler.
func (h *DeliveryHandler) Shutdown(ctx context.Context) error {
// Shutdown gracefully shuts down the robot handler.
func (h *robotHandler) Shutdown(ctx context.Context) error {
return nil
}
// NewDeliveryHandler creates a new DeliveryHandler.
func NewDeliveryHandler() *DeliveryHandler {
return &DeliveryHandler{}
// handleDelivery routes delivery content to configured channels (email, webhook, process).
func (h *robotHandler) handleDelivery(ctx context.Context, ev *eventtypes.Event, resp chan<- eventtypes.Result) {
var payload DeliveryPayload
if err := ev.Should(&payload); err != nil {
log.Error("delivery handler: invalid payload: %v", err)
if ev.IsCall {
resp <- eventtypes.Result{Err: err}
}
return
}
log.Info("delivery handler: execution=%s member=%s", payload.ExecutionID, payload.MemberID)
content := payload.Content
prefs := payload.Preferences
if content == nil {
log.Warn("delivery handler: nil content for execution=%s", payload.ExecutionID)
if ev.IsCall {
resp <- eventtypes.Result{Data: "no content"}
}
return
}
if prefs == nil {
if ev.IsCall {
resp <- eventtypes.Result{Data: "no preferences, skipped"}
}
return
}
deliveryCtx := &robottypes.DeliveryContext{
MemberID: payload.MemberID,
ExecutionID: payload.ExecutionID,
TeamID: payload.TeamID,
}
var results []robottypes.ChannelResult
var lastErr error
if prefs.Email != nil && prefs.Email.Enabled {
for _, target := range prefs.Email.Targets {
r := h.sendEmail(ctx, content, target, deliveryCtx)
results = append(results, r)
if !r.Success && lastErr == nil {
lastErr = fmt.Errorf("email delivery failed: %s", r.Error)
}
}
}
if prefs.Webhook != nil && prefs.Webhook.Enabled {
for _, target := range prefs.Webhook.Targets {
r := h.postWebhook(ctx, content, target, deliveryCtx)
results = append(results, r)
if !r.Success && lastErr == nil {
lastErr = fmt.Errorf("webhook delivery failed: %s", r.Error)
}
}
}
if prefs.Process != nil && prefs.Process.Enabled {
for _, target := range prefs.Process.Targets {
r := h.callProcess(ctx, content, target, deliveryCtx)
results = append(results, r)
if !r.Success && lastErr == nil {
lastErr = fmt.Errorf("process delivery failed: %s", r.Error)
}
}
}
if lastErr != nil {
log.Error("delivery handler: partial failure execution=%s: %v", payload.ExecutionID, lastErr)
}
if ev.IsCall {
resp <- eventtypes.Result{
Data: map[string]interface{}{
"execution_id": payload.ExecutionID,
"results": results,
},
Err: lastErr,
}
}
}
// ============================================================================
// Email
// ============================================================================
func (h *robotHandler) sendEmail(
ctx context.Context,
content *robottypes.DeliveryContent,
target robottypes.EmailTarget,
deliveryCtx *robottypes.DeliveryContext,
) robottypes.ChannelResult {
now := time.Now()
targetID := strings.Join(target.To, ",")
if targetID == "" {
targetID = "no-recipients"
}
result := robottypes.ChannelResult{
Type: robottypes.DeliveryEmail,
Target: targetID,
SentAt: &now,
}
svc := messenger.Instance
if svc == nil {
result.Error = "messenger service not available"
return result
}
htmlBody, plainBody := buildEmailBody(target.Template, content)
msg := &messengerTypes.Message{
To: target.To,
Subject: buildEmailSubject(target.Subject, target.Template, content, deliveryCtx),
Body: plainBody,
HTML: htmlBody,
Type: messengerTypes.MessageTypeEmail,
}
attachments := convertAttachments(ctx, content.Attachments)
if len(attachments) > 0 {
msg.Attachments = attachments
}
channel := robottypes.DefaultEmailChannel()
if err := svc.Send(ctx, channel, msg); err != nil {
result.Error = err.Error()
return result
}
result.Success = true
result.Recipients = target.To
return result
}
// ============================================================================
// Webhook
// ============================================================================
func (h *robotHandler) postWebhook(
ctx context.Context,
content *robottypes.DeliveryContent,
target robottypes.WebhookTarget,
deliveryCtx *robottypes.DeliveryContext,
) robottypes.ChannelResult {
now := time.Now()
result := robottypes.ChannelResult{
Type: robottypes.DeliveryWebhook,
Target: target.URL,
SentAt: &now,
}
payload := map[string]interface{}{
"event": "robot.delivery",
"timestamp": now.Format(time.RFC3339),
"execution_id": deliveryCtx.ExecutionID,
"member_id": deliveryCtx.MemberID,
"team_id": deliveryCtx.TeamID,
"trigger_type": deliveryCtx.TriggerType,
"content": map[string]interface{}{
"summary": content.Summary,
"body": content.Body,
},
}
if len(content.Attachments) > 0 {
info := make([]map[string]interface{}, 0, len(content.Attachments))
for _, att := range content.Attachments {
info = append(info, map[string]interface{}{
"title": att.Title,
"description": att.Description,
"task_id": att.TaskID,
"file": att.File,
})
}
payload["attachments"] = info
}
payloadBytes, err := json.Marshal(payload)
if err != nil {
result.Error = fmt.Sprintf("failed to marshal payload: %v", err)
return result
}
method := target.Method
if method == "" {
method = "POST"
}
req, err := http.NewRequestWithContext(ctx, method, target.URL, bytes.NewReader(payloadBytes))
if err != nil {
result.Error = fmt.Sprintf("failed to create request: %v", err)
return result
}
req.Header.Set("Content-Type", "application/json")
for key, value := range target.Headers {
req.Header.Set(key, value)
}
if target.Secret != "" {
signature := ComputeHMACSignature(payloadBytes, target.Secret)
req.Header.Set("X-Yao-Signature", signature)
req.Header.Set("X-Yao-Signature-Algorithm", "HMAC-SHA256")
}
resp, err := h.httpClient.Do(req)
if err != nil {
result.Error = fmt.Sprintf("request failed: %v", err)
return result
}
defer resp.Body.Close()
body, _ := io.ReadAll(resp.Body)
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
result.Error = fmt.Sprintf("webhook returned status %d: %s", resp.StatusCode, string(body))
return result
}
result.Success = true
result.Details = map[string]interface{}{
"status_code": resp.StatusCode,
"response": string(body),
}
return result
}
// ============================================================================
// Process
// ============================================================================
func (h *robotHandler) callProcess(
ctx context.Context,
content *robottypes.DeliveryContent,
target robottypes.ProcessTarget,
deliveryCtx *robottypes.DeliveryContext,
) robottypes.ChannelResult {
now := time.Now()
result := robottypes.ChannelResult{
Type: robottypes.DeliveryProcess,
Target: target.Process,
SentAt: &now,
}
args := make([]interface{}, 0, 1+len(target.Args))
args = append(args, map[string]interface{}{
"content": map[string]interface{}{
"summary": content.Summary,
"body": content.Body,
"attachments": content.Attachments,
},
"context": map[string]interface{}{
"execution_id": deliveryCtx.ExecutionID,
"member_id": deliveryCtx.MemberID,
"team_id": deliveryCtx.TeamID,
"trigger_type": deliveryCtx.TriggerType,
},
})
args = append(args, target.Args...)
proc, err := process.Of(target.Process, args...)
if err != nil {
result.Error = fmt.Sprintf("failed to create process: %v", err)
return result
}
proc.Context = ctx
if err = proc.Execute(); err != nil {
result.Error = err.Error()
return result
}
result.Success = true
result.Details = toJSONSerializable(proc.Value)
return result
}
// ============================================================================
// Helpers
// ============================================================================
func toJSONSerializable(v interface{}) interface{} {
if v == nil {
return nil
}
if _, err := json.Marshal(v); err != nil {
return fmt.Sprintf("%v", v)
}
return v
}
func buildEmailSubject(subject, template string, content *robottypes.DeliveryContent, ctx *robottypes.DeliveryContext) string {
if subject != "" {
return subject
}
if content.Summary != "" {
return content.Summary
}
return fmt.Sprintf("Execution %s Complete", ctx.ExecutionID)
}
func buildEmailBody(template string, content *robottypes.DeliveryContent) (string, string) {
markdown := content.Body
if markdown == "" {
markdown = content.Summary
}
html, err := text.MarkdownToHTML(markdown)
if err != nil {
return markdown, markdown
}
return html, markdown
}
func convertAttachments(ctx context.Context, attachments []robottypes.DeliveryAttachment) []messengerTypes.Attachment {
if len(attachments) == 0 {
return nil
}
result := make([]messengerTypes.Attachment, 0, len(attachments))
for _, att := range attachments {
uploader, fileID, isWrapper := attachment.Parse(att.File)
if !isWrapper {
log.Warn("convertAttachments: skipping non-wrapper file value=%q title=%q", att.File, att.Title)
continue
}
manager, ok := attachment.Managers[uploader]
if !ok {
log.Warn("convertAttachments: manager not found uploader=%q file=%q title=%q (available: %v)",
uploader, att.File, att.Title, attachmentManagerKeys())
continue
}
info, err := manager.Info(ctx, fileID)
if err != nil {
log.Warn("convertAttachments: failed to get file info fileID=%q uploader=%q: %v", fileID, uploader, err)
continue
}
content, err := manager.Read(ctx, fileID)
if err != nil {
log.Warn("convertAttachments: failed to read file fileID=%q uploader=%q: %v", fileID, uploader, err)
continue
}
// Prefer the semantic title from the delivery agent over the raw storage filename.
// The storage filename may be an auto-generated zip name (e.g. output_xxx.zip),
// while att.Title is the human-readable name set by the delivery agent.
filename := info.Filename
if att.Title != "" {
// Keep the original file extension from storage so the email client
// knows how to open it, but use the human-readable title as the base name.
ext := ""
if idx := strings.LastIndex(info.Filename, "."); idx >= 0 {
ext = info.Filename[idx:]
}
titleExt := ""
if idx := strings.LastIndex(att.Title, "."); idx >= 0 {
titleExt = att.Title[idx:]
}
if titleExt != "" {
// Title already has an extension — use it as-is.
filename = att.Title
} else {
// Title has no extension — append the storage extension.
filename = att.Title + ext
}
}
log.Info("convertAttachments: added attachment filename=%q contentType=%q size=%d", filename, info.ContentType, len(content))
result = append(result, messengerTypes.Attachment{
Filename: filename,
ContentType: info.ContentType,
Content: content,
})
}
return result
}
// attachmentManagerKeys returns registered attachment manager names for debug logging.
func attachmentManagerKeys() []string {
keys := make([]string, 0, len(attachment.Managers))
for k := range attachment.Managers {
keys = append(keys, k)
}
return keys
}
// ComputeHMACSignature computes HMAC-SHA256 signature for webhook payload.
func ComputeHMACSignature(payload []byte, secret string) string {
mac := hmac.New(sha256.New, []byte(secret))
mac.Write(payload)
return hex.EncodeToString(mac.Sum(nil))
}
// VerifyHMACSignature verifies the HMAC-SHA256 signature of a webhook payload.
func VerifyHMACSignature(payload []byte, secret, signature string) bool {
expected := ComputeHMACSignature(payload, secret)
return hmac.Equal([]byte(expected), []byte(signature))
}

View file

@ -2,62 +2,155 @@ package events
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
robottypes "github.com/yaoapp/yao/agent/robot/types"
eventtypes "github.com/yaoapp/yao/event/types"
)
func TestDeliveryHandler_Handle(t *testing.T) {
handler := NewDeliveryHandler()
t.Run("processes valid delivery payload", func(t *testing.T) {
ev := &eventtypes.Event{
Type: Delivery,
ID: "test-event-1",
Payload: DeliveryPayload{
ExecutionID: "exec-1",
MemberID: "member-1",
TeamID: "team-1",
Content: map[string]interface{}{"summary": "test"},
},
}
resp := make(chan eventtypes.Result, 1)
handler.Handle(context.Background(), ev, resp)
// Fire-and-forget: no response expected for Push
})
t.Run("handles call mode with response", func(t *testing.T) {
ev := &eventtypes.Event{
Type: Delivery,
ID: "test-event-2",
IsCall: true,
Payload: DeliveryPayload{
ExecutionID: "exec-2",
MemberID: "member-2",
},
}
resp := make(chan eventtypes.Result, 1)
handler.Handle(context.Background(), ev, resp)
result := <-resp
require.NotNil(t, result.Data)
assert.Contains(t, result.Data.(string), "exec-2")
})
t.Run("handles invalid payload gracefully", func(t *testing.T) {
ev := &eventtypes.Event{
Type: Delivery,
Payload: "invalid",
}
resp := make(chan eventtypes.Result, 1)
handler.Handle(context.Background(), ev, resp)
})
func newTestHandler() *robotHandler {
return &robotHandler{
httpClient: http.DefaultClient,
}
}
func TestDeliveryHandler_Shutdown(t *testing.T) {
handler := NewDeliveryHandler()
func TestRobotHandler_DeliveryWebhook(t *testing.T) {
var received map[string]interface{}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
decoder := json.NewDecoder(r.Body)
_ = decoder.Decode(&received)
w.WriteHeader(http.StatusOK)
w.Write([]byte(`{"ok":true}`))
}))
defer server.Close()
handler := newTestHandler()
ev := &eventtypes.Event{
Type: Delivery,
ID: "test-ev-1",
IsCall: true,
Payload: DeliveryPayload{
ExecutionID: "exec-1",
MemberID: "member-1",
TeamID: "team-1",
Content: &robottypes.DeliveryContent{
Summary: "test summary",
Body: "test body",
},
Preferences: &robottypes.DeliveryPreferences{
Webhook: &robottypes.WebhookPreference{
Enabled: true,
Targets: []robottypes.WebhookTarget{
{URL: server.URL},
},
},
},
},
}
resp := make(chan eventtypes.Result, 1)
handler.Handle(context.Background(), ev, resp)
result := <-resp
require.NotNil(t, result.Data)
assert.NoError(t, result.Err)
data, ok := result.Data.(map[string]interface{})
require.True(t, ok)
assert.Equal(t, "exec-1", data["execution_id"])
require.NotNil(t, received)
assert.Equal(t, "robot.delivery", received["event"])
}
func TestRobotHandler_DeliveryNoContent(t *testing.T) {
handler := newTestHandler()
ev := &eventtypes.Event{
Type: Delivery,
ID: "test-ev-2",
IsCall: true,
Payload: DeliveryPayload{
ExecutionID: "exec-2",
MemberID: "member-2",
TeamID: "team-2",
},
}
resp := make(chan eventtypes.Result, 1)
handler.Handle(context.Background(), ev, resp)
result := <-resp
assert.Equal(t, "no content", result.Data)
}
func TestRobotHandler_DeliveryNoPreferences(t *testing.T) {
handler := newTestHandler()
ev := &eventtypes.Event{
Type: Delivery,
ID: "test-ev-3",
IsCall: true,
Payload: DeliveryPayload{
ExecutionID: "exec-3",
MemberID: "member-3",
TeamID: "team-3",
Content: &robottypes.DeliveryContent{
Summary: "test",
Body: "body",
},
},
}
resp := make(chan eventtypes.Result, 1)
handler.Handle(context.Background(), ev, resp)
result := <-resp
assert.Equal(t, "no preferences, skipped", result.Data)
}
func TestRobotHandler_InvalidPayload(t *testing.T) {
handler := newTestHandler()
ev := &eventtypes.Event{
Type: Delivery,
ID: "test-ev-4",
IsCall: true,
Payload: "invalid",
}
resp := make(chan eventtypes.Result, 1)
handler.Handle(context.Background(), ev, resp)
result := <-resp
assert.Error(t, result.Err)
}
func TestRobotHandler_UnhandledEvent(t *testing.T) {
handler := newTestHandler()
ev := &eventtypes.Event{
Type: "robot.unknown",
ID: "test-ev-5",
}
resp := make(chan eventtypes.Result, 1)
handler.Handle(context.Background(), ev, resp)
// Fire-and-forget, no response expected
}
func TestRobotHandler_Shutdown(t *testing.T) {
handler := newTestHandler()
err := handler.Shutdown(context.Background())
assert.NoError(t, err)
}
func TestVerifyHMACSignature(t *testing.T) {
payload := []byte(`{"event":"robot.delivery"}`)
secret := "test-secret"
sig := ComputeHMACSignature(payload, secret)
assert.True(t, VerifyHMACSignature(payload, secret, sig))
assert.False(t, VerifyHMACSignature(payload, "wrong-secret", sig))
}

View file

@ -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

View file

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

View file

@ -7,43 +7,32 @@ import (
"time"
"github.com/yaoapp/gou/model"
"github.com/yaoapp/kun/log"
robotevents "github.com/yaoapp/yao/agent/robot/events"
robottypes "github.com/yaoapp/yao/agent/robot/types"
"github.com/yaoapp/yao/event"
)
// RunDelivery executes P4: Delivery phase
// Calls the Delivery Agent to generate content, then routes to Delivery Center
//
// Input:
// - Full execution context (P0-P3)
// - Robot config
//
// Output:
// - DeliveryResult with content and channel results
//
// Process:
// 1. Call Delivery Agent with full execution context
// 2. Agent generates DeliveryContent (summary, body, attachments)
// 3. Route content to Delivery Center for actual delivery
// 3. Push delivery event for asynchronous routing via handlers
func (e *Executor) RunDelivery(ctx *robottypes.Context, exec *robottypes.Execution, _ interface{}) error {
// Get robot for identity and resources
robot := exec.GetRobot()
if robot == nil {
return fmt.Errorf("robot not found in execution")
}
// Update UI field with i18n
locale := getEffectiveLocale(robot, exec.Input)
e.updateUIFields(ctx, exec, "", getLocalizedMessage(locale, "generating_delivery"))
// Get agent ID for delivery phase
agentID := "__yao.delivery" // default
agentID := "__yao.delivery"
if robot.Config != nil && robot.Config.Resources != nil {
agentID = robot.Config.Resources.GetPhaseAgent(robottypes.PhaseDelivery)
}
// Build input for Delivery Agent
formatter := NewInputFormatter()
userContent := formatter.FormatDeliveryInput(exec, robot)
@ -51,7 +40,6 @@ func (e *Executor) RunDelivery(ctx *robottypes.Context, exec *robottypes.Executi
return fmt.Errorf("no content available for delivery generation")
}
// Call Delivery Agent
caller := NewAgentCaller()
caller.Connector = robot.LanguageModel
result, err := caller.CallWithMessages(ctx, agentID, userContent)
@ -59,11 +47,8 @@ func (e *Executor) RunDelivery(ctx *robottypes.Context, exec *robottypes.Executi
return fmt.Errorf("delivery agent (%s) call failed: %w", agentID, err)
}
// Parse response as JSON
// Delivery Agent returns: { "content": { "summary": "...", "body": "...", "attachments": [...] } }
data, err := result.GetJSON()
if err != nil {
// Fallback: if not JSON, create minimal content from raw text
content := result.GetText()
if content == "" {
return fmt.Errorf("delivery agent returned empty response")
@ -79,20 +64,17 @@ func (e *Executor) RunDelivery(ctx *robottypes.Context, exec *robottypes.Executi
return e.pushDeliveryEvent(ctx, exec, robot)
}
// Build DeliveryContent from JSON
content := parseDeliveryContent(data)
if content == nil {
return fmt.Errorf("delivery agent (%s) returned invalid content", agentID)
}
// Build DeliveryResult
exec.Delivery = &robottypes.DeliveryResult{
RequestID: generateRequestID(exec.ID),
Content: content,
Success: true,
}
// Push delivery event for asynchronous routing via handlers
return e.pushDeliveryEvent(ctx, exec, robot)
}
@ -100,7 +82,7 @@ func (e *Executor) RunDelivery(ctx *robottypes.Context, exec *robottypes.Executi
// Registered handlers (see events/handlers.go) route to email/webhook/process channels.
func (e *Executor) pushDeliveryEvent(ctx *robottypes.Context, exec *robottypes.Execution, robot *robottypes.Robot) error {
prefs := buildDeliveryPreferences(robot)
event.Push(ctx.Context, robotevents.Delivery, robotevents.DeliveryPayload{
_, err := event.Push(ctx.Context, robotevents.Delivery, robotevents.DeliveryPayload{
ExecutionID: exec.ID,
MemberID: exec.MemberID,
TeamID: exec.TeamID,
@ -108,61 +90,9 @@ func (e *Executor) pushDeliveryEvent(ctx *robottypes.Context, exec *robottypes.E
Content: exec.Delivery.Content,
Preferences: prefs,
})
return nil
}
// routeToDeliveryCenter sends content to the Delivery Center for actual delivery
// The Delivery Center decides which channels to use based on robot/user preferences
//
// Delivery logic:
// 1. Manager email: ALWAYS send to manager if manager_id is set (mandatory)
// 2. Additional targets: Append configured email/webhook/process targets
func (e *Executor) routeToDeliveryCenter(ctx *robottypes.Context, exec *robottypes.Execution, robot *robottypes.Robot) error {
if exec.Delivery == nil || exec.Delivery.Content == nil {
return fmt.Errorf("no delivery content to route")
}
// Build final delivery preferences by merging manager email + configured targets
prefs := buildDeliveryPreferences(robot)
if prefs == nil || !hasActiveChannels(prefs) {
exec.Delivery.Success = true
return nil
}
// Update UI field to show delivery is in progress
locale := getEffectiveLocale(robot, exec.Input)
e.updateUIFields(ctx, exec, "", getLocalizedMessage(locale, "sending_delivery"))
// Create Delivery Center and execute
center := NewDeliveryCenter()
results, err := center.Deliver(ctx, exec.Delivery.Content, &robottypes.DeliveryContext{
MemberID: exec.MemberID,
ExecutionID: exec.ID,
TriggerType: exec.TriggerType,
TeamID: exec.TeamID,
}, prefs, robot)
// Update delivery result
exec.Delivery.Results = results
now := time.Now()
exec.Delivery.SentAt = &now
if err != nil {
exec.Delivery.Success = false
exec.Delivery.Error = err.Error()
return err
log.Error("delivery event push failed: execution=%s error=%v", exec.ID, err)
}
// Check if all channels succeeded
allSuccess := true
for _, r := range results {
if !r.Success {
allSuccess = false
break
}
}
exec.Delivery.Success = allSuccess
return nil
}
@ -172,26 +102,20 @@ func parseDeliveryContent(data map[string]interface{}) *robottypes.DeliveryConte
return nil
}
// Try to get content object
contentData, ok := data["content"].(map[string]interface{})
if !ok {
// Fallback: maybe the data itself is the content
contentData = data
}
content := &robottypes.DeliveryContent{}
// Parse summary
if summary, ok := contentData["summary"].(string); ok {
content.Summary = summary
}
// Parse body
if body, ok := contentData["body"].(string); ok {
content.Body = body
}
// Parse attachments
if attachments, ok := contentData["attachments"].([]interface{}); ok {
for _, att := range attachments {
if attMap, ok := att.(map[string]interface{}); ok {
@ -203,7 +127,6 @@ func parseDeliveryContent(data map[string]interface{}) *robottypes.DeliveryConte
}
}
// Validate: at least summary or body should be present
if content.Summary == "" && content.Body == "" {
return nil
}
@ -211,7 +134,6 @@ func parseDeliveryContent(data map[string]interface{}) *robottypes.DeliveryConte
return content
}
// parseDeliveryAttachment parses a single attachment from the agent response
func parseDeliveryAttachment(data map[string]interface{}) *robottypes.DeliveryAttachment {
if data == nil {
return nil
@ -232,7 +154,6 @@ func parseDeliveryAttachment(data map[string]interface{}) *robottypes.DeliveryAt
att.File = file
}
// At minimum, need title and file
if att.Title == "" || att.File == "" {
return nil
}
@ -240,21 +161,17 @@ func parseDeliveryAttachment(data map[string]interface{}) *robottypes.DeliveryAt
return att
}
// generateRequestID generates a unique request ID for delivery tracking
func generateRequestID(execID string) string {
return fmt.Sprintf("dlv-%s-%d", execID, time.Now().UnixNano()%1000000)
}
// getTaskDescription extracts a description from task messages
func getTaskDescription(task robottypes.Task) string {
if len(task.Messages) == 0 {
return task.GoalRef
}
// Try to get text from first message
for _, msg := range task.Messages {
if content, ok := msg.Content.(string); ok && content != "" {
// Truncate if too long
if len(content) > 100 {
return content[:97] + "..."
}
@ -262,7 +179,6 @@ func getTaskDescription(task robottypes.Task) string {
}
}
// Fallback to goal reference
if task.GoalRef != "" {
return task.GoalRef
}
@ -270,12 +186,10 @@ func getTaskDescription(task robottypes.Task) string {
return "Task " + task.ID
}
// truncateSummary truncates text to maxLen characters
func truncateSummary(text string, maxLen int) string {
if len(text) <= maxLen {
return text
}
// Find last space before maxLen to avoid cutting words
truncated := text[:maxLen]
if idx := strings.LastIndex(truncated, " "); idx > maxLen/2 {
return truncated[:idx] + "..."
@ -283,26 +197,6 @@ func truncateSummary(text string, maxLen int) string {
return truncated + "..."
}
// hasActiveChannels checks if any delivery channel is configured
func hasActiveChannels(prefs *robottypes.DeliveryPreferences) bool {
if prefs == nil {
return false
}
if prefs.Email != nil && prefs.Email.Enabled && len(prefs.Email.Targets) > 0 {
return true
}
if prefs.Webhook != nil && prefs.Webhook.Enabled && len(prefs.Webhook.Targets) > 0 {
return true
}
if prefs.Process != nil && prefs.Process.Enabled && len(prefs.Process.Targets) > 0 {
return true
}
return false
}
// buildDeliveryPreferences builds the final delivery preferences by:
// 1. Always including manager email if manager_id is set (mandatory)
// 2. Appending all configured email/webhook/process targets
func buildDeliveryPreferences(robot *robottypes.Robot) *robottypes.DeliveryPreferences {
if robot == nil {
return nil
@ -310,26 +204,22 @@ func buildDeliveryPreferences(robot *robottypes.Robot) *robottypes.DeliveryPrefe
prefs := &robottypes.DeliveryPreferences{}
// Step 1: Get manager email (mandatory if manager_id is set)
managerEmail := robot.ManagerEmail
if managerEmail == "" && robot.ManagerID != "" {
managerEmail = getManagerEmail(robot.ManagerID)
if managerEmail != "" {
robot.ManagerEmail = managerEmail // Cache for future use
robot.ManagerEmail = managerEmail
}
}
// Step 2: Build email targets (manager first, then configured targets)
var emailTargets []robottypes.EmailTarget
// Add manager email as first target (mandatory)
if managerEmail != "" {
emailTargets = append(emailTargets, robottypes.EmailTarget{
To: []string{managerEmail},
})
}
// Append configured email targets
if robot.Config != nil && robot.Config.Delivery != nil && robot.Config.Delivery.Email != nil {
for _, target := range robot.Config.Delivery.Email.Targets {
if len(target.To) > 0 {
@ -338,7 +228,6 @@ func buildDeliveryPreferences(robot *robottypes.Robot) *robottypes.DeliveryPrefe
}
}
// Set email preference if we have any targets
if len(emailTargets) > 0 {
prefs.Email = &robottypes.EmailPreference{
Enabled: true,
@ -346,14 +235,12 @@ func buildDeliveryPreferences(robot *robottypes.Robot) *robottypes.DeliveryPrefe
}
}
// Step 3: Copy webhook preferences from config (if enabled)
if robot.Config != nil && robot.Config.Delivery != nil && robot.Config.Delivery.Webhook != nil {
if robot.Config.Delivery.Webhook.Enabled && len(robot.Config.Delivery.Webhook.Targets) > 0 {
prefs.Webhook = robot.Config.Delivery.Webhook
}
}
// Step 4: Copy process preferences from config (if enabled)
if robot.Config != nil && robot.Config.Delivery != nil && robot.Config.Delivery.Process != nil {
if robot.Config.Delivery.Process.Enabled && len(robot.Config.Delivery.Process.Targets) > 0 {
prefs.Process = robot.Config.Delivery.Process
@ -363,8 +250,6 @@ func buildDeliveryPreferences(robot *robottypes.Robot) *robottypes.DeliveryPrefe
return prefs
}
// getManagerEmail retrieves the manager's email from __yao.member table by member_id
// manager_id in Robot refers to a member_id in __yao.member table
func getManagerEmail(managerID string) string {
if managerID == "" {
return ""
@ -400,7 +285,6 @@ func (f *InputFormatter) FormatDeliveryInput(exec *robottypes.Execution, robot *
var sb strings.Builder
// Robot identity
if robot != nil && robot.Config != nil && robot.Config.Identity != nil {
sb.WriteString("## Robot Identity\n\n")
sb.WriteString(fmt.Sprintf("- **Role**: %s\n", robot.Config.Identity.Role))
@ -412,7 +296,6 @@ func (f *InputFormatter) FormatDeliveryInput(exec *robottypes.Execution, robot *
sb.WriteString("\n")
}
// Trigger type
sb.WriteString("## Execution Context\n\n")
sb.WriteString(fmt.Sprintf("- **Trigger**: %s\n", exec.TriggerType))
sb.WriteString(fmt.Sprintf("- **Status**: %s\n", exec.Status))
@ -423,25 +306,21 @@ func (f *InputFormatter) FormatDeliveryInput(exec *robottypes.Execution, robot *
}
sb.WriteString("\n")
// Inspiration (P0) - for clock trigger
if exec.Inspiration != nil && exec.Inspiration.Content != "" {
sb.WriteString("## Inspiration (P0)\n\n")
sb.WriteString(exec.Inspiration.Content)
sb.WriteString("\n\n")
}
// Goals (P1)
if exec.Goals != nil && exec.Goals.Content != "" {
sb.WriteString("## Goals (P1)\n\n")
sb.WriteString(exec.Goals.Content)
sb.WriteString("\n\n")
}
// Tasks (P2)
if len(exec.Tasks) > 0 {
sb.WriteString("## Tasks (P2)\n\n")
for i, task := range exec.Tasks {
// Extract task description from messages if available
taskDesc := getTaskDescription(task)
sb.WriteString(fmt.Sprintf("%d. **%s** - %s\n", i+1, task.ID, taskDesc))
sb.WriteString(fmt.Sprintf(" - Executor: %s (%s)\n", task.ExecutorID, task.ExecutorType))
@ -453,7 +332,6 @@ func (f *InputFormatter) FormatDeliveryInput(exec *robottypes.Execution, robot *
sb.WriteString("\n")
}
// Results (P3) - detailed
if len(exec.Results) > 0 {
sb.WriteString("## Results (P3)\n\n")
@ -471,7 +349,6 @@ func (f *InputFormatter) FormatDeliveryInput(exec *robottypes.Execution, robot *
sb.WriteString(fmt.Sprintf("- **Duration**: %dms\n", result.Duration))
// Validation
if result.Validation != nil {
if result.Validation.Passed {
sb.WriteString(fmt.Sprintf("- **Validation**: ✓ Passed (score: %.2f)\n", result.Validation.Score))
@ -485,7 +362,6 @@ func (f *InputFormatter) FormatDeliveryInput(exec *robottypes.Execution, robot *
}
}
// Output
if result.Output != nil {
sb.WriteString("\n**Output**:\n")
if output, err := json.MarshalIndent(result.Output, "", " "); err == nil {
@ -497,7 +373,6 @@ func (f *InputFormatter) FormatDeliveryInput(exec *robottypes.Execution, robot *
}
}
// Error
if result.Error != "" {
sb.WriteString(fmt.Sprintf("\n**Error**: %s\n", result.Error))
}
@ -505,7 +380,6 @@ func (f *InputFormatter) FormatDeliveryInput(exec *robottypes.Execution, robot *
sb.WriteString("\n")
}
// Summary
sb.WriteString(fmt.Sprintf("### Summary\n\n- **Total Tasks**: %d\n- **Succeeded**: %d\n- **Failed**: %d\n\n",
len(exec.Results), successCount, failCount))
}

View file

@ -1,444 +0,0 @@
package standard
import (
"bytes"
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"net/http"
"strings"
"time"
"github.com/yaoapp/gou/process"
"github.com/yaoapp/gou/text"
robottypes "github.com/yaoapp/yao/agent/robot/types"
"github.com/yaoapp/yao/attachment"
"github.com/yaoapp/yao/messenger"
messengerTypes "github.com/yaoapp/yao/messenger/types"
)
// DeliveryCenter handles routing delivery content to configured channels
// It decides which channels to use based on robot/user preferences and executes the delivery
type DeliveryCenter struct {
httpClient *http.Client
}
// NewDeliveryCenter creates a new DeliveryCenter instance
func NewDeliveryCenter() *DeliveryCenter {
return &DeliveryCenter{
httpClient: &http.Client{
Timeout: 30 * time.Second,
},
}
}
// Deliver sends content to all configured channels based on preferences
// Returns results for each channel target and any error
func (dc *DeliveryCenter) Deliver(
ctx *robottypes.Context,
content *robottypes.DeliveryContent,
deliveryCtx *robottypes.DeliveryContext,
prefs *robottypes.DeliveryPreferences,
robotInstance *robottypes.Robot,
) ([]robottypes.ChannelResult, error) {
if content == nil {
return nil, fmt.Errorf("delivery content is nil")
}
if prefs == nil {
return nil, nil // No preferences = no delivery
}
var results []robottypes.ChannelResult
var lastErr error
// Process email targets
if prefs.Email != nil && prefs.Email.Enabled {
for _, target := range prefs.Email.Targets {
result := dc.sendEmail(ctx.Context, content, target, deliveryCtx, robotInstance)
results = append(results, result)
if !result.Success && lastErr == nil {
lastErr = fmt.Errorf("email delivery failed: %s", result.Error)
}
}
}
// Process webhook targets
if prefs.Webhook != nil && prefs.Webhook.Enabled {
for _, target := range prefs.Webhook.Targets {
result := dc.postWebhook(ctx.Context, content, target, deliveryCtx)
results = append(results, result)
if !result.Success && lastErr == nil {
lastErr = fmt.Errorf("webhook delivery failed: %s", result.Error)
}
}
}
// Process process targets
if prefs.Process != nil && prefs.Process.Enabled {
for _, target := range prefs.Process.Targets {
result := dc.callProcess(ctx.Context, content, target, deliveryCtx)
results = append(results, result)
if !result.Success && lastErr == nil {
lastErr = fmt.Errorf("process delivery failed: %s", result.Error)
}
}
}
return results, lastErr
}
// sendEmail sends delivery content to a single email target
func (dc *DeliveryCenter) sendEmail(
ctx context.Context,
content *robottypes.DeliveryContent,
target robottypes.EmailTarget,
deliveryCtx *robottypes.DeliveryContext,
robotInstance *robottypes.Robot,
) robottypes.ChannelResult {
now := time.Now()
// Build target identifier from recipients
targetID := strings.Join(target.To, ",")
if targetID == "" {
targetID = "no-recipients"
}
result := robottypes.ChannelResult{
Type: robottypes.DeliveryEmail,
Target: targetID,
SentAt: &now,
}
// Get messenger service
svc := messenger.Instance
if svc == nil {
result.Error = "messenger service not available"
return result
}
// Build email message with HTML content
htmlBody, plainBody := buildEmailBody(target.Template, content)
msg := &messengerTypes.Message{
To: target.To,
Subject: buildEmailSubject(target.Subject, target.Template, content, deliveryCtx, robotInstance),
Body: plainBody, // Plain text fallback
HTML: htmlBody, // HTML content for rich email display
Type: messengerTypes.MessageTypeEmail,
}
// Set From address from Robot's email (if configured)
if robotInstance != nil && robotInstance.RobotEmail != "" {
msg.From = robotInstance.RobotEmail
}
// Convert attachments
attachments := convertAttachments(ctx, content.Attachments)
if len(attachments) > 0 {
msg.Attachments = attachments
}
// Send email using global default channel
channel := robottypes.DefaultEmailChannel()
err := svc.Send(ctx, channel, msg)
if err != nil {
result.Error = err.Error()
return result
}
result.Success = true
result.Recipients = target.To
return result
}
// postWebhook posts delivery content to a single webhook target
func (dc *DeliveryCenter) postWebhook(
ctx context.Context,
content *robottypes.DeliveryContent,
target robottypes.WebhookTarget,
deliveryCtx *robottypes.DeliveryContext,
) robottypes.ChannelResult {
now := time.Now()
result := robottypes.ChannelResult{
Type: robottypes.DeliveryWebhook,
Target: target.URL,
SentAt: &now,
}
// Build webhook payload
payload := map[string]interface{}{
"event": "robot.delivery",
"timestamp": now.Format(time.RFC3339),
"execution_id": deliveryCtx.ExecutionID,
"member_id": deliveryCtx.MemberID,
"team_id": deliveryCtx.TeamID,
"trigger_type": deliveryCtx.TriggerType,
"content": map[string]interface{}{
"summary": content.Summary,
"body": content.Body,
},
}
// Add attachments info (not the actual files)
if len(content.Attachments) > 0 {
attachmentInfo := make([]map[string]interface{}, 0, len(content.Attachments))
for _, att := range content.Attachments {
attachmentInfo = append(attachmentInfo, map[string]interface{}{
"title": att.Title,
"description": att.Description,
"task_id": att.TaskID,
"file": att.File,
})
}
payload["attachments"] = attachmentInfo
}
// Marshal payload
payloadBytes, err := json.Marshal(payload)
if err != nil {
result.Error = fmt.Sprintf("failed to marshal payload: %v", err)
return result
}
// Build request
method := target.Method
if method == "" {
method = "POST"
}
req, err := http.NewRequestWithContext(ctx, method, target.URL, bytes.NewReader(payloadBytes))
if err != nil {
result.Error = fmt.Sprintf("failed to create request: %v", err)
return result
}
req.Header.Set("Content-Type", "application/json")
// Add custom headers
for key, value := range target.Headers {
req.Header.Set(key, value)
}
// Add HMAC signature if secret is configured
if target.Secret != "" {
signature := computeHMACSignature(payloadBytes, target.Secret)
req.Header.Set("X-Yao-Signature", signature)
req.Header.Set("X-Yao-Signature-Algorithm", "HMAC-SHA256")
}
// Send request
resp, err := dc.httpClient.Do(req)
if err != nil {
result.Error = fmt.Sprintf("request failed: %v", err)
return result
}
defer resp.Body.Close()
// Read response body
body, _ := io.ReadAll(resp.Body)
// Check status code
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
result.Error = fmt.Sprintf("webhook returned status %d: %s", resp.StatusCode, string(body))
return result
}
result.Success = true
result.Details = map[string]interface{}{
"status_code": resp.StatusCode,
"response": string(body),
}
return result
}
// callProcess calls a Yao Process with delivery content
func (dc *DeliveryCenter) callProcess(
ctx context.Context,
content *robottypes.DeliveryContent,
target robottypes.ProcessTarget,
deliveryCtx *robottypes.DeliveryContext,
) robottypes.ChannelResult {
now := time.Now()
result := robottypes.ChannelResult{
Type: robottypes.DeliveryProcess,
Target: target.Process,
SentAt: &now,
}
// Build args: DeliveryContent as first arg, then additional args
args := make([]interface{}, 0, 1+len(target.Args))
args = append(args, map[string]interface{}{
"content": map[string]interface{}{
"summary": content.Summary,
"body": content.Body,
"attachments": content.Attachments,
},
"context": map[string]interface{}{
"execution_id": deliveryCtx.ExecutionID,
"member_id": deliveryCtx.MemberID,
"team_id": deliveryCtx.TeamID,
"trigger_type": deliveryCtx.TriggerType,
},
})
args = append(args, target.Args...)
// Create and execute process
proc, err := process.Of(target.Process, args...)
if err != nil {
result.Error = fmt.Sprintf("failed to create process: %v", err)
return result
}
proc.Context = ctx
err = proc.Execute()
if err != nil {
result.Error = err.Error()
return result
}
result.Success = true
// Convert proc.Value to JSON-serializable format to avoid func type issues
result.Details = toJSONSerializable(proc.Value)
return result
}
// toJSONSerializable ensures the value can be JSON serialized
// Returns the original value if serializable, or a string fallback if not
func toJSONSerializable(v interface{}) interface{} {
if v == nil {
return nil
}
// Try to marshal to check if it's JSON serializable
_, err := json.Marshal(v)
if err != nil {
// If it can't be serialized (e.g., contains func), return a string representation
return fmt.Sprintf("%v", v)
}
// Return original value if it's serializable
return v
}
// buildEmailSubject builds the email subject line
func buildEmailSubject(subject, template string, content *robottypes.DeliveryContent, ctx *robottypes.DeliveryContext, robot *robottypes.Robot) string {
// Use explicit subject if provided
if subject != "" {
return subject
}
// Get robot display name for subject prefix
robotName := "Robot"
if robot != nil && robot.DisplayName != "" {
robotName = robot.DisplayName
}
// Use template-based subject if template is specified
// TODO: Implement template rendering
if template != "" {
return fmt.Sprintf("[%s] %s", robotName, content.Summary)
}
// Default: use summary
if content.Summary != "" {
return fmt.Sprintf("[%s] %s", robotName, content.Summary)
}
return fmt.Sprintf("[%s] Execution %s Complete", robotName, ctx.ExecutionID)
}
// buildEmailBody builds the email body content
// buildEmailBody returns HTML and plain text versions of the email body
// Returns: (htmlBody, plainBody)
func buildEmailBody(template string, content *robottypes.DeliveryContent) (string, string) {
// TODO: Implement template rendering
// Get markdown content (used as plain text fallback)
markdown := content.Body
if markdown == "" {
markdown = content.Summary
}
// Convert Markdown to HTML for rich email display
html, err := text.MarkdownToHTML(markdown)
if err != nil {
// Fallback: use markdown as both HTML and plain text
return markdown, markdown
}
return html, markdown
}
// convertAttachments converts DeliveryAttachment to messenger Attachment format
func convertAttachments(ctx context.Context, attachments []robottypes.DeliveryAttachment) []messengerTypes.Attachment {
if len(attachments) == 0 {
return nil
}
result := make([]messengerTypes.Attachment, 0, len(attachments))
for _, att := range attachments {
// Parse file wrapper: __<uploader>://<fileID>
uploader, fileID, isWrapper := attachment.Parse(att.File)
if !isWrapper {
// Skip non-wrapper attachments
continue
}
// Get file info from attachment manager
manager, ok := attachment.Managers[uploader]
if !ok {
continue
}
info, err := manager.Info(ctx, fileID)
if err != nil {
continue
}
// Read file content
content, err := manager.Read(ctx, fileID)
if err != nil {
continue
}
// Build messenger attachment
msgAtt := messengerTypes.Attachment{
Filename: info.Filename,
ContentType: info.ContentType,
Content: content,
}
result = append(result, msgAtt)
}
return result
}
// ============================================================================
// Webhook Signature
// ============================================================================
// computeHMACSignature computes HMAC-SHA256 signature for webhook payload
// Returns hex-encoded signature string
func computeHMACSignature(payload []byte, secret string) string {
mac := hmac.New(sha256.New, []byte(secret))
mac.Write(payload)
return hex.EncodeToString(mac.Sum(nil))
}
// VerifyHMACSignature verifies the HMAC-SHA256 signature of a webhook payload
// Headers:
// - X-Yao-Signature: hex-encoded HMAC-SHA256 signature
// - X-Yao-Signature-Algorithm: "HMAC-SHA256"
//
// Returns true if the signature is valid
func VerifyHMACSignature(payload []byte, secret, signature string) bool {
expected := computeHMACSignature(payload, secret)
return hmac.Equal([]byte(expected), []byte(signature))
}

View file

@ -2,9 +2,6 @@ package standard_test
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
@ -34,7 +31,6 @@ func TestRunDeliveryBasic(t *testing.T) {
robot := createDeliveryTestRobot(t, "robot.delivery")
exec := createDeliveryTestExecution(robot)
// Set up execution context with P0-P3 results
exec.Inspiration = &types.InspirationReport{
Content: "Morning analysis suggests focus on Q4 review.",
}
@ -50,7 +46,6 @@ func TestRunDeliveryBasic(t *testing.T) {
{TaskID: "task-002", Success: true, Duration: 800, Output: "Q4 sales exceeded expectations by 15%."},
}
// Run delivery phase
e := standard.New()
err := e.RunDelivery(ctx, exec, nil)
@ -85,7 +80,6 @@ func TestRunDeliveryBasic(t *testing.T) {
require.NotNil(t, exec.Delivery)
require.NotNil(t, exec.Delivery.Content)
// Content should mention the failure
body := strings.ToLower(exec.Delivery.Content.Body)
hasFailureInfo := strings.Contains(body, "fail") ||
strings.Contains(body, "error") ||
@ -111,7 +105,6 @@ func TestRunDeliveryErrorHandling(t *testing.T) {
ID: "test-exec-1",
TriggerType: types.TriggerClock,
}
// Don't set robot
e := standard.New()
err := e.RunDelivery(ctx, exec, nil)
@ -147,314 +140,23 @@ func TestRunDeliveryErrorHandling(t *testing.T) {
}
// ============================================================================
// Delivery Center Tests
// Email Channel Config Tests
// ============================================================================
func TestDeliveryCenterWebhook(t *testing.T) {
t.Run("posts to webhook successfully", func(t *testing.T) {
// Create mock webhook server
var receivedPayload map[string]interface{}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
assert.Equal(t, "POST", r.Method)
assert.Equal(t, "application/json", r.Header.Get("Content-Type"))
decoder := json.NewDecoder(r.Body)
err := decoder.Decode(&receivedPayload)
assert.NoError(t, err)
w.WriteHeader(http.StatusOK)
w.Write([]byte(`{"status": "received"}`))
}))
defer server.Close()
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Test delivery completed",
Body: "# Test Report\n\nThis is a test.",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-001",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Webhook: &types.WebhookPreference{
Enabled: true,
Targets: []types.WebhookTarget{
{URL: server.URL},
},
},
}
results, err := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
require.NoError(t, err)
require.Len(t, results, 1)
assert.True(t, results[0].Success)
assert.Equal(t, types.DeliveryWebhook, results[0].Type)
assert.Equal(t, server.URL, results[0].Target)
// Verify payload structure
assert.Equal(t, "robot.delivery", receivedPayload["event"])
assert.Equal(t, "exec-001", receivedPayload["execution_id"])
assert.Equal(t, "member-001", receivedPayload["member_id"])
contentMap := receivedPayload["content"].(map[string]interface{})
assert.Equal(t, "Test delivery completed", contentMap["summary"])
})
t.Run("handles webhook failure", func(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(`{"error": "internal error"}`))
}))
defer server.Close()
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Test",
Body: "Test body",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-001",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Webhook: &types.WebhookPreference{
Enabled: true,
Targets: []types.WebhookTarget{
{URL: server.URL},
},
},
}
results, err := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
assert.Error(t, err) // Should return error for failed delivery
require.Len(t, results, 1)
assert.False(t, results[0].Success)
assert.Contains(t, results[0].Error, "500")
})
t.Run("supports multiple webhook targets", func(t *testing.T) {
callCount := 0
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
callCount++
w.WriteHeader(http.StatusOK)
w.Write([]byte(`{"ok": true}`))
}))
defer server.Close()
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Test",
Body: "Test body",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-001",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Webhook: &types.WebhookPreference{
Enabled: true,
Targets: []types.WebhookTarget{
{URL: server.URL + "/hook1"},
{URL: server.URL + "/hook2"},
{URL: server.URL + "/hook3"},
},
},
}
results, err := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
require.NoError(t, err)
require.Len(t, results, 3)
assert.Equal(t, 3, callCount)
for _, r := range results {
assert.True(t, r.Success)
}
})
t.Run("includes custom headers", func(t *testing.T) {
var receivedHeaders http.Header
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
receivedHeaders = r.Header
w.WriteHeader(http.StatusOK)
}))
defer server.Close()
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Test",
Body: "Test body",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-001",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Webhook: &types.WebhookPreference{
Enabled: true,
Targets: []types.WebhookTarget{
{
URL: server.URL,
Headers: map[string]string{
"X-Custom-Header": "custom-value",
"Authorization": "Bearer test-token",
},
},
},
},
}
results, _ := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
require.Len(t, results, 1)
assert.True(t, results[0].Success)
assert.Equal(t, "custom-value", receivedHeaders.Get("X-Custom-Header"))
assert.Equal(t, "Bearer test-token", receivedHeaders.Get("Authorization"))
})
}
func TestDeliveryCenterNoChannels(t *testing.T) {
t.Run("succeeds with no channels configured", func(t *testing.T) {
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Test",
Body: "Test body",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-001",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
// No preferences
results, err := center.Deliver(ctx, content, deliveryCtx, nil, nil)
assert.NoError(t, err)
assert.Empty(t, results)
})
t.Run("succeeds with disabled channels", func(t *testing.T) {
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Test",
Body: "Test body",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-001",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Webhook: &types.WebhookPreference{
Enabled: false, // Disabled
Targets: []types.WebhookTarget{
{URL: "http://example.com"},
},
},
}
results, err := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
assert.NoError(t, err)
assert.Empty(t, results)
})
}
func TestDeliveryCenterMixedChannels(t *testing.T) {
t.Run("delivers to multiple channel types", func(t *testing.T) {
webhookCalled := false
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
webhookCalled = true
w.WriteHeader(http.StatusOK)
}))
defer server.Close()
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Test",
Body: "Test body",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-001",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Webhook: &types.WebhookPreference{
Enabled: true,
Targets: []types.WebhookTarget{
{URL: server.URL},
},
},
// Email would fail without messenger setup, but webhook should succeed
}
results, _ := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
assert.True(t, webhookCalled)
require.Len(t, results, 1)
assert.True(t, results[0].Success)
})
}
func TestDefaultEmailChannel(t *testing.T) {
t.Run("returns default email channel", func(t *testing.T) {
// Default should be "default"
assert.Equal(t, "default", types.DefaultEmailChannel())
})
t.Run("can set custom email channel", func(t *testing.T) {
// Save original
original := types.DefaultEmailChannel()
defer types.SetDefaultEmailChannel(original)
// Set custom channel
types.SetDefaultEmailChannel("custom-email")
assert.Equal(t, "custom-email", types.DefaultEmailChannel())
})
t.Run("ignores empty channel", func(t *testing.T) {
// Save original and restore after test
original := types.DefaultEmailChannel()
defer types.SetDefaultEmailChannel(original)
@ -464,53 +166,6 @@ func TestDefaultEmailChannel(t *testing.T) {
}
func TestRobotEmailInDelivery(t *testing.T) {
t.Run("robot email is passed to delivery center", func(t *testing.T) {
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Test",
Body: "Test body",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-001",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
// Robot with email configured
robot := &types.Robot{
MemberID: "robot-001",
RobotEmail: "robot@example.com",
}
// Webhook to verify robot is passed (email would fail without messenger)
webhookCalled := false
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
webhookCalled = true
w.WriteHeader(http.StatusOK)
}))
defer server.Close()
prefs := &types.DeliveryPreferences{
Webhook: &types.WebhookPreference{
Enabled: true,
Targets: []types.WebhookTarget{
{URL: server.URL},
},
},
}
// Deliver with robot
results, _ := center.Deliver(ctx, content, deliveryCtx, prefs, robot)
assert.True(t, webhookCalled)
require.Len(t, results, 1)
assert.True(t, results[0].Success)
})
t.Run("robot email field is loaded from map", func(t *testing.T) {
data := map[string]interface{}{
"member_id": "robot-001",
@ -527,7 +182,6 @@ func TestRobotEmailInDelivery(t *testing.T) {
data := map[string]interface{}{
"member_id": "robot-001",
"team_id": "team-001",
// robot_email not set
}
robot, err := types.NewRobotFromMap(data)
@ -536,6 +190,10 @@ func TestRobotEmailInDelivery(t *testing.T) {
})
}
// ============================================================================
// FormatDeliveryInput Tests
// ============================================================================
func TestFormatDeliveryInput(t *testing.T) {
formatter := standard.NewInputFormatter()
@ -603,538 +261,6 @@ func TestFormatDeliveryInput(t *testing.T) {
})
}
// ============================================================================
// Email Delivery Tests (requires messenger setup)
// ============================================================================
func TestDeliveryCenterEmail(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test")
}
testutils.Prepare(t)
defer testutils.Clean(t)
// Set robot channel as default for tests
original := types.DefaultEmailChannel()
types.SetDefaultEmailChannel("robot")
defer types.SetDefaultEmailChannel(original)
t.Run("sends email to single target", func(t *testing.T) {
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Test Delivery Report",
Body: "This is a test delivery from Robot Agent.\n\n## Results\n- Task 1: Completed\n- Task 2: Completed",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-001",
TriggerType: types.TriggerClock,
TeamID: "team-001",
}
robot := &types.Robot{
MemberID: "robot-001",
RobotEmail: "robot@example.com",
}
prefs := &types.DeliveryPreferences{
Email: &types.EmailPreference{
Enabled: true,
Targets: []types.EmailTarget{
{
To: []string{"test@example.com"},
Subject: "Robot Delivery Test",
},
},
},
}
// Send email - ignore billing/API errors, just verify the call was made
results, _ := center.Deliver(ctx, content, deliveryCtx, prefs, robot)
require.Len(t, results, 1)
assert.Equal(t, types.DeliveryEmail, results[0].Type)
assert.Equal(t, "test@example.com", results[0].Target)
// Note: Success depends on messenger configuration
})
t.Run("sends email to multiple targets", func(t *testing.T) {
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Multi-target Test",
Body: "Test body for multiple recipients",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-002",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
robot := &types.Robot{
MemberID: "robot-001",
RobotEmail: "robot@example.com",
}
prefs := &types.DeliveryPreferences{
Email: &types.EmailPreference{
Enabled: true,
Targets: []types.EmailTarget{
{
To: []string{"user1@example.com"},
Subject: "Report for User 1",
},
{
To: []string{"user2@example.com", "user3@example.com"},
Subject: "Report for Team",
},
},
},
}
results, _ := center.Deliver(ctx, content, deliveryCtx, prefs, robot)
require.Len(t, results, 2)
assert.Equal(t, types.DeliveryEmail, results[0].Type)
assert.Equal(t, types.DeliveryEmail, results[1].Type)
assert.Equal(t, "user1@example.com", results[0].Target)
assert.Equal(t, "user2@example.com,user3@example.com", results[1].Target)
})
t.Run("sends email with attachments", func(t *testing.T) {
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Report with Attachments",
Body: "Please find the attached report.",
Attachments: []types.DeliveryAttachment{
{
Title: "Q4 Report.pdf",
Description: "Quarterly sales report",
TaskID: "task-001",
File: "__local://reports/q4-2024.pdf",
},
{
Title: "Data Export.csv",
Description: "Raw data export",
TaskID: "task-002",
File: "__local://exports/data.csv",
},
},
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-003",
TriggerType: types.TriggerClock,
TeamID: "team-001",
}
robot := &types.Robot{
MemberID: "robot-001",
RobotEmail: "robot@example.com",
}
prefs := &types.DeliveryPreferences{
Email: &types.EmailPreference{
Enabled: true,
Targets: []types.EmailTarget{
{
To: []string{"manager@example.com"},
Subject: "Weekly Report with Attachments",
},
},
},
}
// Send - attachment conversion may fail if files don't exist, but structure is tested
results, _ := center.Deliver(ctx, content, deliveryCtx, prefs, robot)
require.Len(t, results, 1)
assert.Equal(t, types.DeliveryEmail, results[0].Type)
})
}
// ============================================================================
// Process Delivery Tests (requires Yao process setup)
// ============================================================================
func TestDeliveryCenterProcess(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test")
}
testutils.Prepare(t)
defer testutils.Clean(t)
t.Run("calls process with delivery content", func(t *testing.T) {
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Process Test Summary",
Body: "This is the body content for process testing.",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-process-001",
TriggerType: types.TriggerEvent,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Process: &types.ProcessPreference{
Enabled: true,
Targets: []types.ProcessTarget{
{
Process: "scripts.tests.delivery.Handle",
},
},
},
}
results, err := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
require.NoError(t, err)
require.Len(t, results, 1)
assert.Equal(t, types.DeliveryProcess, results[0].Type)
assert.Equal(t, "scripts.tests.delivery.Handle", results[0].Target)
assert.True(t, results[0].Success)
// Verify process received correct data (Details structure depends on process return)
assert.NotNil(t, results[0].Details, "process should return details")
})
t.Run("calls process with additional args", func(t *testing.T) {
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Args Test",
Body: "Testing additional arguments",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-process-002",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Process: &types.ProcessPreference{
Enabled: true,
Targets: []types.ProcessTarget{
{
Process: "scripts.tests.delivery.Handle",
Args: []interface{}{"custom-arg-1", "custom-arg-2"},
},
},
},
}
results, err := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
require.NoError(t, err)
require.Len(t, results, 1)
assert.True(t, results[0].Success)
assert.NotNil(t, results[0].Details, "process should return details with args")
})
t.Run("calls multiple process targets", func(t *testing.T) {
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Multi-process Test",
Body: "Testing multiple process targets",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-process-003",
TriggerType: types.TriggerClock,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Process: &types.ProcessPreference{
Enabled: true,
Targets: []types.ProcessTarget{
{
Process: "scripts.tests.delivery.Handle",
},
{
Process: "scripts.tests.delivery.Notify",
Args: []interface{}{"user-123", "push"},
},
},
},
}
results, err := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
require.NoError(t, err)
require.Len(t, results, 2)
// First process: Handle
assert.Equal(t, "scripts.tests.delivery.Handle", results[0].Target)
assert.True(t, results[0].Success)
// Second process: Notify
assert.Equal(t, "scripts.tests.delivery.Notify", results[1].Target)
assert.True(t, results[1].Success)
})
t.Run("handles process failure gracefully", func(t *testing.T) {
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Failure Test",
Body: "Testing process failure handling",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-process-004",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Process: &types.ProcessPreference{
Enabled: true,
Targets: []types.ProcessTarget{
{
Process: "scripts.tests.delivery.HandleWithFailure",
Args: []interface{}{true}, // shouldFail = true
},
},
},
}
results, err := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
assert.Error(t, err)
require.Len(t, results, 1)
assert.False(t, results[0].Success)
assert.Contains(t, results[0].Error, "Simulated process failure")
})
t.Run("handles process with attachments", func(t *testing.T) {
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Attachment Process Test",
Body: "Testing process with attachments",
Attachments: []types.DeliveryAttachment{
{
Title: "Report.pdf",
Description: "Test report",
TaskID: "task-001",
File: "__local://test/report.pdf",
},
},
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-process-005",
TriggerType: types.TriggerClock,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Process: &types.ProcessPreference{
Enabled: true,
Targets: []types.ProcessTarget{
{
Process: "scripts.tests.delivery.HandleAttachments",
},
},
},
}
results, err := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
require.NoError(t, err)
require.Len(t, results, 1)
assert.True(t, results[0].Success)
assert.NotNil(t, results[0].Details, "process should return details with attachments info")
})
}
// ============================================================================
// Mixed Channel Tests
// ============================================================================
func TestDeliveryCenterAllChannels(t *testing.T) {
if testing.Short() {
t.Skip("Skipping integration test")
}
testutils.Prepare(t)
defer testutils.Clean(t)
// Set robot channel as default
original := types.DefaultEmailChannel()
types.SetDefaultEmailChannel("robot")
defer types.SetDefaultEmailChannel(original)
t.Run("delivers to email, webhook, and process simultaneously", func(t *testing.T) {
// Setup webhook server
webhookCalled := false
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
webhookCalled = true
w.WriteHeader(http.StatusOK)
w.Write([]byte(`{"ok": true}`))
}))
defer server.Close()
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Full Channel Test",
Body: "Testing all delivery channels together",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-full-001",
TriggerType: types.TriggerClock,
TeamID: "team-001",
}
robot := &types.Robot{
MemberID: "robot-001",
RobotEmail: "robot@example.com",
}
prefs := &types.DeliveryPreferences{
Email: &types.EmailPreference{
Enabled: true,
Targets: []types.EmailTarget{
{To: []string{"user@example.com"}, Subject: "Test"},
},
},
Webhook: &types.WebhookPreference{
Enabled: true,
Targets: []types.WebhookTarget{
{URL: server.URL},
},
},
Process: &types.ProcessPreference{
Enabled: true,
Targets: []types.ProcessTarget{
{Process: "scripts.tests.delivery.Handle"},
},
},
}
results, _ := center.Deliver(ctx, content, deliveryCtx, prefs, robot)
// Should have 3 results (1 email + 1 webhook + 1 process)
require.Len(t, results, 3)
// Verify each channel type
var emailResult, webhookResult, processResult *types.ChannelResult
for i := range results {
switch results[i].Type {
case types.DeliveryEmail:
emailResult = &results[i]
case types.DeliveryWebhook:
webhookResult = &results[i]
case types.DeliveryProcess:
processResult = &results[i]
}
}
assert.NotNil(t, emailResult, "should have email result")
assert.NotNil(t, webhookResult, "should have webhook result")
assert.NotNil(t, processResult, "should have process result")
// Webhook and process should succeed
assert.True(t, webhookCalled, "webhook should be called")
assert.True(t, webhookResult.Success, "webhook should succeed")
assert.True(t, processResult.Success, "process should succeed")
})
t.Run("partial failure does not stop other channels", func(t *testing.T) {
// Webhook that fails
failServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
}))
defer failServer.Close()
// Webhook that succeeds
successCalled := false
successServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
successCalled = true
w.WriteHeader(http.StatusOK)
}))
defer successServer.Close()
center := standard.NewDeliveryCenter()
ctx := types.NewContext(context.Background(), nil)
content := &types.DeliveryContent{
Summary: "Partial Failure Test",
Body: "Testing partial failure handling",
}
deliveryCtx := &types.DeliveryContext{
MemberID: "member-001",
ExecutionID: "exec-partial-001",
TriggerType: types.TriggerHuman,
TeamID: "team-001",
}
prefs := &types.DeliveryPreferences{
Webhook: &types.WebhookPreference{
Enabled: true,
Targets: []types.WebhookTarget{
{URL: failServer.URL}, // This will fail
{URL: successServer.URL}, // This should still be called
},
},
Process: &types.ProcessPreference{
Enabled: true,
Targets: []types.ProcessTarget{
{Process: "scripts.tests.delivery.Handle"}, // This should succeed
},
},
}
results, err := center.Deliver(ctx, content, deliveryCtx, prefs, nil)
// Should have error (from first webhook failure)
assert.Error(t, err)
// But all targets should be attempted
require.Len(t, results, 3)
// First webhook failed
assert.False(t, results[0].Success)
// Second webhook and process should succeed
assert.True(t, successCalled, "second webhook should be called despite first failure")
assert.True(t, results[1].Success, "second webhook should succeed")
assert.True(t, results[2].Success, "process should succeed")
})
}
// ============================================================================
// Helper Functions
// ============================================================================

View file

@ -70,9 +70,9 @@ func TestIntegrationSchedulingFlow(t *testing.T) {
assert.Equal(t, "robot_integ_flow_clock", robot.MemberID)
assert.Equal(t, types.RobotIdle, robot.Status)
// Simulate clock trigger at matching time (09:00 on Wednesday)
// Simulate clock trigger at matching time (03:33 on Wednesday)
loc, _ := time.LoadLocation("Asia/Shanghai")
triggerTime := time.Date(2025, 1, 15, 9, 0, 0, 0, loc) // Wednesday 09:00
triggerTime := time.Date(2025, 1, 15, 3, 33, 0, 0, loc) // Wednesday 03:33
ctx := types.NewContext(context.Background(), nil)
err = m.Tick(ctx, triggerTime)
@ -229,6 +229,7 @@ func TestIntegrationPhaseProgression(t *testing.T) {
defer cleanupIntegrationRobots(t)
t.Run("clock trigger executes all phases P0-P5", func(t *testing.T) {
cleanupIntegrationRobots(t)
setupIntegrationRobotTimes(t, "robot_integ_phases_clock", "team_integ_phases")
// Track phases executed
@ -250,7 +251,6 @@ func TestIntegrationPhaseProgression(t *testing.T) {
err := m.Start()
require.NoError(t, err)
defer m.Stop()
// Trigger execution
ctx := types.NewContext(context.Background(), nil)
@ -260,6 +260,9 @@ func TestIntegrationPhaseProgression(t *testing.T) {
// Wait for execution
time.Sleep(500 * time.Millisecond)
// Stop manager before asserting to prevent ticker from triggering extra executions
m.Stop()
// Verify all 6 phases executed (P0-P5)
assert.Len(t, phasesExecuted, 6, "Should execute all 6 phases for clock trigger")
assert.Equal(t, types.PhaseInspiration, phasesExecuted[0], "Should start with P0")
@ -267,6 +270,7 @@ func TestIntegrationPhaseProgression(t *testing.T) {
})
t.Run("human trigger skips P0 and executes P1-P5", func(t *testing.T) {
cleanupIntegrationRobots(t)
setupIntegrationRobotIntervene(t, "robot_integ_phases_human", "team_integ_phases")
// Track phases executed
@ -288,7 +292,6 @@ func TestIntegrationPhaseProgression(t *testing.T) {
err := m.Start()
require.NoError(t, err)
defer m.Stop()
// Trigger execution via human trigger
ctx := types.NewContext(context.Background(), nil)
@ -298,6 +301,9 @@ func TestIntegrationPhaseProgression(t *testing.T) {
// Wait for execution
time.Sleep(500 * time.Millisecond)
// Stop manager before asserting to prevent ticker from triggering extra executions
m.Stop()
// Verify 5 phases executed (P1-P5, skipping P0)
assert.Len(t, phasesExecuted, 5, "Should execute 5 phases for human trigger")
assert.Equal(t, types.PhaseGoals, phasesExecuted[0], "Should start with P1 (Goals)")
@ -369,7 +375,7 @@ func setupIntegrationRobotTimes(t *testing.T, memberID, teamID string) {
},
"clock": map[string]interface{}{
"mode": "times",
"times": []string{"09:00", "14:00", "17:00"},
"times": []string{"03:33"},
"days": []string{"Mon", "Tue", "Wed", "Thu", "Fri"},
"tz": "Asia/Shanghai",
"timeout": "30m",
@ -460,7 +466,7 @@ func setupIntegrationRobotHighQuota(t *testing.T, memberID, teamID string) {
},
"clock": map[string]interface{}{
"mode": "times",
"times": []string{"09:00"},
"times": []string{"03:33"},
"tz": "Asia/Shanghai",
},
}

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -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

View file

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

View file

@ -24,23 +24,24 @@ func (s *Session) SetContext(ctx *AgentContext) {
// Close closes the session and cleans up resources
func (s *Session) Close() error {
if s.cancel != nil {
s.cancel()
}
if s.Conn != nil {
s.Conn.Close()
}
if s.Listener != nil {
s.Listener.Close()
}
// Remove socket file
os.Remove(s.SocketPath)
s.closeOnce.Do(func() {
if s.cancel != nil {
s.cancel()
}
if s.Conn != nil {
s.Conn.Close()
}
if s.Listener != nil {
s.Listener.Close()
}
os.Remove(s.SocketPath)
})
return nil
}
// serve handles incoming connections
func (s *Session) serve(ctx context.Context) {
defer s.cleanup()
defer s.Close()
for {
select {
@ -49,10 +50,8 @@ func (s *Session) serve(ctx context.Context) {
default:
}
// Accept connection with deadline to allow context cancellation check
conn, err := s.Listener.Accept()
if err != nil {
// Check if context was cancelled
select {
case <-ctx.Done():
return
@ -66,17 +65,6 @@ func (s *Session) serve(ctx context.Context) {
}
}
// cleanup cleans up session resources
func (s *Session) cleanup() {
if s.Conn != nil {
s.Conn.Close()
}
if s.Listener != nil {
s.Listener.Close()
}
os.Remove(s.SocketPath)
}
// handleConnection handles a single connection
func (s *Session) handleConnection(ctx context.Context, conn net.Conn) {
defer conn.Close()

View file

@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"net"
"sync"
)
// Session represents an IPC session for a sandbox container
@ -15,6 +16,7 @@ type Session struct {
Context *AgentContext // Agent context
MCPTools map[string]*MCPTool // Authorized MCP tools
cancel context.CancelFunc // Cancel function for cleanup
closeOnce sync.Once // Ensures cleanup runs exactly once
}
// AgentContext holds context information for the agent