diff --git a/agent/robot/TODO.md b/agent/robot/TODO.md index cb26818e..5f37fa98 100644 --- a/agent/robot/TODO.md +++ b/agent/robot/TODO.md @@ -307,15 +307,29 @@ Trigger → Manager → Cache → Dedup → Pool → Worker → Executor(stub) - [x] `job/log_test.go` - 24 test cases - [x] All tests passing with real database -### 3.6 Executor Stub Enhancement +### ✅ 3.6 Executor Stub Enhancement (COMPLETE) -- [ ] `executor/executor.go` - enhance stub implementation - - [ ] `Execute()` - simulate full execution with Job integration - 1. Create Execution record + Job +- [x] `executor/executor.go` - enhanced stub implementation + - [x] `Execute()` - simulate full execution with Job integration + 1. Create Execution record + Job (via job package) 2. Update phase: P0 → P1 → P2 → P3 → P4 → P5 3. Log phase transitions 4. Return success with mock data -- [ ] Test: verify stub called, verify phase progression, verify job logs + - [x] `Config` struct with `SkipJobIntegration`, `OnPhaseStart`, `OnPhaseEnd` + - [x] `NewWithDelay()`, `NewWithCallback()` for testing + - [x] Quota check with `robot.TryAcquireSlot()` + - [x] Clock trigger: P0→P5, Human/Event trigger: P1→P5 +- [x] Phase-specific files (modular design for Phase 4+ replacement): + - [x] `executor/inspiration.go` - `RunInspiration()` P0 mock + - [x] `executor/goals.go` - `RunGoals()` P1 mock + - [x] `executor/tasks.go` - `RunTasks()` P2 mock + - [x] `executor/run.go` - `RunExecution()` P3 mock + - [x] `executor/delivery.go` - `RunDelivery()` P4 mock + - [x] `executor/learning.go` - `RunLearning()` P5 mock +- [x] `simulateStreamDelay()` - 50ms hardcoded delay per phase +- [x] Test: smoke tests for basic flow verification + - [x] `executor/executor_test.go` - 6 test cases + - [x] Clock/Human/Event triggers, nil robot, simulated failure, counters ### 3.7 Integration Test (End-to-End Scheduling) @@ -649,19 +663,19 @@ func TestWithLLM(t *testing.T) { ## Progress Tracking -| Phase | Status | Description | -| --------------------- | ------ | ------------------------------------------------- | -| 1. Types & Interfaces | ✅ | All types, enums, interfaces | -| 2. Skeleton | ✅ | Empty stubs, code compiles | -| 3. Scheduling System | 🟡 | Cache + Pool + Trigger + Job ✅, Executor stub 🟡 | -| 4. P0 Inspiration | ⬜ | Inspiration Agent integration | -| 5. P1 Goals | ⬜ | Goal Generation Agent integration | -| 6. P2 Tasks | ⬜ | Task Planning Agent integration | -| 7. P3 Run | ⬜ | Task execution (assistant/mcp/process) | -| 8. P4 Delivery | ⬜ | Output delivery (email/file/webhook/notify) | -| 9. P5 Learning | ⬜ | Learning Agent + KB save | -| 10. API & Integration | ⬜ | Complete API, end-to-end tests | -| 11. Advanced | ⬜ | Semantic dedup, plan queue | +| Phase | Status | Description | +| --------------------- | ------ | -------------------------------------------------------------------- | +| 1. Types & Interfaces | ✅ | All types, enums, interfaces | +| 2. Skeleton | ✅ | Empty stubs, code compiles | +| 3. Scheduling System | 🟡 | Cache + Pool + Trigger + Job + Executor stub ✅, Integration test 🟡 | +| 4. P0 Inspiration | ⬜ | Inspiration Agent integration | +| 5. P1 Goals | ⬜ | Goal Generation Agent integration | +| 6. P2 Tasks | ⬜ | Task Planning Agent integration | +| 7. P3 Run | ⬜ | Task execution (assistant/mcp/process) | +| 8. P4 Delivery | ⬜ | Output delivery (email/file/webhook/notify) | +| 9. P5 Learning | ⬜ | Learning Agent + KB save | +| 10. API & Integration | ⬜ | Complete API, end-to-end tests | +| 11. Advanced | ⬜ | Semantic dedup, plan queue | Legend: ⬜ Not started | 🟡 In progress | ✅ Complete diff --git a/agent/robot/executor/delivery.go b/agent/robot/executor/delivery.go new file mode 100644 index 00000000..72910584 --- /dev/null +++ b/agent/robot/executor/delivery.go @@ -0,0 +1,48 @@ +package executor + +import ( + "time" + + "github.com/yaoapp/yao/agent/robot/types" +) + +// RunDelivery executes P4: Delivery phase +// +// Sends execution output via configured delivery channel. +// Supports: email, file, webhook, notify. +// +// Implementation (TODO Phase 8): +// 1. Build delivery content from execution results +// 2. Call Delivery Agent via Assistant.Stream() to format output +// 3. Send via configured channel (email/file/webhook/notify) +func (e *Executor) RunDelivery(_ *types.Context, exec *types.Execution, _ interface{}) error { + // TODO (Phase 8): Replace with real delivery + // agentID := robot.Config.Resources.GetPhaseAgent(types.PhaseDelivery) + // messages := buildDeliveryMessages(exec.Results, robot) + // response, err := callAgentStream(ctx, agentID, messages) + // if err != nil { + // return err + // } + // deliveryContent := parseDeliveryContent(response) + // err = sendDelivery(ctx, robot.Config.Delivery, deliveryContent) + // if err != nil { + // return err + // } + + // Simulate Agent Stream delay + e.simulateStreamDelay() + + // Generate mock delivery result + exec.Delivery = &types.DeliveryResult{ + Type: types.DeliveryNotify, + Success: true, + Details: map[string]interface{}{ + "message": "Mock delivery completed successfully", + "channel": "notify", + "recipient": "test-user", + "timestamp": time.Now().Format(time.RFC3339), + }, + } + + return nil +} diff --git a/agent/robot/executor/executor.go b/agent/robot/executor/executor.go index c3033c81..1b783aa5 100644 --- a/agent/robot/executor/executor.go +++ b/agent/robot/executor/executor.go @@ -1,21 +1,47 @@ package executor import ( + "fmt" "sync/atomic" "time" + "github.com/yaoapp/yao/agent/robot/job" "github.com/yaoapp/yao/agent/robot/types" - "github.com/yaoapp/yao/agent/robot/utils" ) +// Config holds executor configuration +type Config struct { + // SkipJobIntegration skips job system integration (for unit tests) + SkipJobIntegration bool + + // OnPhaseStart callback when a phase starts (for testing) + OnPhaseStart func(phase types.Phase) + + // OnPhaseEnd callback when a phase ends (for testing) + OnPhaseEnd func(phase types.Phase) +} + // Executor implements types.Executor interface -// This is a stub implementation for Phase 2 +// This is a stub implementation that simulates full execution with Job integration +// +// Phase Implementation Strategy: +// Each phase has a dedicated file and method: +// - inspiration.go: RunInspiration() - P0 +// - goals.go: RunGoals() - P1 +// - tasks.go: RunTasks() - P2 +// - run.go: RunExecution() - P3 +// - delivery.go: RunDelivery() - P4 +// - learning.go: RunLearning() - P5 +// +// Currently all methods return mock data with simulated delay. +// When implementing real phases (Phase 4+), replace the method body with +// actual Agent Stream calls (Assistant.Stream()). type Executor struct { - delay time.Duration // simulated execution delay - execCount atomic.Int32 // total execution count - currentCount atomic.Int32 // currently running count - onStart func() // callback on execution start (for testing) - onEnd func() // callback on execution end (for testing) + config Config + execCount atomic.Int32 // total execution count + currentCount atomic.Int32 // currently running count + onStart func() // callback on execution start (for testing) + onEnd func() // callback on execution end (for testing) } // New creates a new executor instance @@ -23,43 +49,92 @@ func New() *Executor { return &Executor{} } -// NewWithDelay creates a new executor with simulated delay (for testing) -func NewWithDelay(delay time.Duration) *Executor { +// NewWithConfig creates a new executor with custom configuration +func NewWithConfig(config Config) *Executor { return &Executor{ - delay: delay, + config: config, + } +} + +// NewWithDelay creates a new executor with simulated delay (for testing) +// Note: delay parameter is kept for API compatibility but not used internally +// Real delay comes from simulateStreamDelay() which simulates Agent Stream latency +func NewWithDelay(_ time.Duration) *Executor { + return &Executor{ + config: Config{ + SkipJobIntegration: true, // Skip job integration for simple delay tests + }, } } // NewWithCallback creates a new executor with callbacks (for testing concurrency) -func NewWithCallback(delay time.Duration, onStart, onEnd func()) *Executor { +func NewWithCallback(_ time.Duration, onStart, onEnd func()) *Executor { return &Executor{ - delay: delay, + config: Config{ + SkipJobIntegration: true, // Skip job integration for callback tests + }, onStart: onStart, onEnd: onEnd, } } // Execute executes a robot through all phases -// Stub: returns empty execution (will be implemented in Phase 3+) +// This stub implementation: +// 1. Creates Execution record + Job (via job package) +// 2. Updates phase: P0 → P1 → P2 → P3 → P4 → P5 +// 3. Logs phase transitions +// 4. Returns success with mock data func (e *Executor) Execute(ctx *types.Context, robot *types.Robot, trigger types.TriggerType, data interface{}) (*types.Execution, error) { - // Create execution record first - execID := utils.NewID() - exec := &types.Execution{ - ID: execID, - MemberID: robot.MemberID, - TeamID: robot.TeamID, - TriggerType: trigger, - Status: types.ExecRunning, - Phase: types.PhaseInspiration, + if robot == nil { + return nil, fmt.Errorf("robot cannot be nil") + } + + var exec *types.Execution + var err error + + // Determine starting phase based on trigger type + // Clock trigger starts from P0 (Inspiration) + // Human/Event triggers skip P0 and start from P1 (Goals) + startPhaseIndex := 0 + if trigger == types.TriggerHuman || trigger == types.TriggerEvent { + startPhaseIndex = 1 // Skip P0 (Inspiration) + } + + // Create execution with Job integration + if !e.config.SkipJobIntegration { + exec, err = job.CreateExecution(ctx, &job.CreateOptions{ + Robot: robot, + TriggerType: trigger, + Input: buildTriggerInput(trigger, data), + }) + if err != nil { + return nil, fmt.Errorf("failed to create execution: %w", err) + } + } else { + // Simple execution for tests without job integration + exec = &types.Execution{ + ID: fmt.Sprintf("exec_%d", time.Now().UnixNano()), + MemberID: robot.MemberID, + TeamID: robot.TeamID, + TriggerType: trigger, + StartTime: time.Now(), + Status: types.ExecPending, + Phase: types.AllPhases[startPhaseIndex], + Input: buildTriggerInput(trigger, data), + } } // Atomically check quota and acquire slot // This prevents race condition where multiple workers pass CanRun() check // but then all add executions, exceeding the quota if !robot.TryAcquireSlot(exec) { + // If job was created, mark it as failed + if !e.config.SkipJobIntegration && exec.JobID != "" { + _ = job.FailExecution(ctx, exec, types.ErrQuotaExceeded) + } return nil, types.ErrQuotaExceeded } - defer robot.RemoveExecution(execID) + defer robot.RemoveExecution(exec.ID) // Track execution count (after successful slot acquisition) e.execCount.Add(1) @@ -75,24 +150,140 @@ func (e *Executor) Execute(ctx *types.Context, robot *types.Robot, trigger types defer e.onEnd() } - // Simulate execution delay - if e.delay > 0 { - time.Sleep(e.delay) + // Update status to running + exec.Status = types.ExecRunning + if !e.config.SkipJobIntegration { + if err := job.UpdateStatus(ctx, exec, types.ExecRunning); err != nil { + // Log error but continue execution + _ = job.LogWarn(ctx, exec, fmt.Sprintf("Failed to update status to running: %v", err)) + } } // Check for simulated failure if dataStr, ok := data.(string); ok && dataStr == "simulate_failure" { exec.Status = types.ExecFailed - return exec, nil // return error is optional, we track status + exec.Error = "simulated failure" + if !e.config.SkipJobIntegration { + _ = job.FailExecution(ctx, exec, fmt.Errorf("simulated failure")) + } + return exec, nil } - // Update execution status + // Execute phases + phases := types.AllPhases[startPhaseIndex:] + for _, phase := range phases { + // Run phase with common pre/post processing + if err := e.runPhase(ctx, exec, phase, data); err != nil { + exec.Status = types.ExecFailed + exec.Error = err.Error() + if !e.config.SkipJobIntegration { + _ = job.FailExecution(ctx, exec, err) + } + return exec, nil + } + } + + // Mark execution as completed exec.Status = types.ExecCompleted - exec.Phase = types.PhaseLearning + now := time.Now() + exec.EndTime = &now + + if !e.config.SkipJobIntegration { + if err := job.CompleteExecution(ctx, exec); err != nil { + // Log error but return success since execution completed + _ = job.LogWarn(ctx, exec, fmt.Sprintf("Failed to mark execution as completed: %v", err)) + } + } return exec, nil } +// runPhase executes a single phase with common pre/post processing +func (e *Executor) runPhase(ctx *types.Context, exec *types.Execution, phase types.Phase, data interface{}) error { + // Update phase + exec.Phase = phase + + // Log phase start + if !e.config.SkipJobIntegration { + if err := job.UpdatePhase(ctx, exec, phase); err != nil { + // Log error but continue + _ = job.LogWarn(ctx, exec, fmt.Sprintf("Failed to update phase to %s: %v", phase, err)) + } + } + + // Call phase start callback + if e.config.OnPhaseStart != nil { + e.config.OnPhaseStart(phase) + } + + phaseStart := time.Now() + + // Execute phase-specific logic + // Each phase method calls the corresponding Agent via Assistant.Stream() + // Currently returns mock data; replace with real Agent calls in Phase 4+ + var err error + switch phase { + case types.PhaseInspiration: + err = e.RunInspiration(ctx, exec, data) + case types.PhaseGoals: + err = e.RunGoals(ctx, exec, data) + case types.PhaseTasks: + err = e.RunTasks(ctx, exec, data) + case types.PhaseRun: + err = e.RunExecution(ctx, exec, data) + case types.PhaseDelivery: + err = e.RunDelivery(ctx, exec, data) + case types.PhaseLearning: + err = e.RunLearning(ctx, exec, data) + } + + if err != nil { + // Log phase error + if !e.config.SkipJobIntegration { + _ = job.LogPhaseError(ctx, exec, phase, err) + } + return err + } + + // Call phase end callback + if e.config.OnPhaseEnd != nil { + e.config.OnPhaseEnd(phase) + } + + // Log phase end + if !e.config.SkipJobIntegration { + phaseDuration := time.Since(phaseStart).Milliseconds() + _ = job.LogPhaseEnd(ctx, exec, phase, phaseDuration) + } + + return nil +} + +// buildTriggerInput builds TriggerInput from trigger data +func buildTriggerInput(trigger types.TriggerType, data interface{}) *types.TriggerInput { + input := &types.TriggerInput{} + + switch trigger { + case types.TriggerClock: + input.Clock = types.NewClockContext(time.Now(), "") + + case types.TriggerHuman: + if req, ok := data.(*types.InterveneRequest); ok { + input.Action = req.Action + input.Messages = req.Messages + } + + case types.TriggerEvent: + if req, ok := data.(*types.EventRequest); ok { + input.Source = types.EventSource(req.Source) + input.EventType = req.EventType + input.Data = req.Data + } + } + + return input +} + // ExecCount returns total execution count func (e *Executor) ExecCount() int { return int(e.execCount.Load()) @@ -108,3 +299,13 @@ func (e *Executor) Reset() { e.execCount.Store(0) e.currentCount.Store(0) } + +// DefaultStreamDelay is the simulated delay for Agent Stream calls +// This will be removed when real Agent calls are implemented +const DefaultStreamDelay = 50 * time.Millisecond + +// simulateStreamDelay simulates the delay of an Agent Stream call +// This will be removed when real Agent calls are implemented in Phase 4+ +func (e *Executor) simulateStreamDelay() { + time.Sleep(DefaultStreamDelay) +} diff --git a/agent/robot/executor/executor_test.go b/agent/robot/executor/executor_test.go new file mode 100644 index 00000000..f56fc0ef --- /dev/null +++ b/agent/robot/executor/executor_test.go @@ -0,0 +1,126 @@ +package executor + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/yaoapp/yao/agent/robot/types" +) + +// Smoke tests to verify basic flow works +// These tests use SkipJobIntegration=true to avoid DB dependencies +// Real integration tests are in manager_test.go and job_test.go + +func TestExecutorSmoke(t *testing.T) { + exec := NewWithDelay(0) + robot := &types.Robot{ + MemberID: "test-smoke", + TeamID: "team-1", + Config: &types.Config{Quota: &types.Quota{Max: 1}}, + } + ctx := types.NewContext(context.Background(), nil) + + result, err := exec.Execute(ctx, robot, types.TriggerClock, nil) + + assert.NoError(t, err) + assert.NotNil(t, result) + assert.Equal(t, types.ExecCompleted, result.Status) + assert.Equal(t, types.TriggerClock, result.TriggerType) + + // Clock trigger executes all phases (P0-P5) + assert.NotNil(t, result.Inspiration, "P0 should be executed for clock trigger") + assert.NotNil(t, result.Goals, "P1 should be executed") + assert.NotEmpty(t, result.Tasks, "P2 should generate tasks") + assert.NotEmpty(t, result.Results, "P3 should generate results") + assert.NotNil(t, result.Delivery, "P4 should be executed") + assert.NotEmpty(t, result.Learning, "P5 should be executed") +} + +func TestExecutorHumanTriggerSkipsP0(t *testing.T) { + exec := NewWithDelay(0) + robot := &types.Robot{ + MemberID: "test-human", + TeamID: "team-1", + Config: &types.Config{Quota: &types.Quota{Max: 1}}, + } + ctx := types.NewContext(context.Background(), nil) + + result, err := exec.Execute(ctx, robot, types.TriggerHuman, nil) + + assert.NoError(t, err) + assert.NotNil(t, result) + assert.Equal(t, types.ExecCompleted, result.Status) + + // Human trigger skips P0 (Inspiration) + assert.Nil(t, result.Inspiration, "P0 should be skipped for human trigger") + assert.NotNil(t, result.Goals, "P1 should be executed") +} + +func TestExecutorEventTriggerSkipsP0(t *testing.T) { + exec := NewWithDelay(0) + robot := &types.Robot{ + MemberID: "test-event", + TeamID: "team-1", + Config: &types.Config{Quota: &types.Quota{Max: 1}}, + } + ctx := types.NewContext(context.Background(), nil) + + result, err := exec.Execute(ctx, robot, types.TriggerEvent, nil) + + assert.NoError(t, err) + assert.Nil(t, result.Inspiration, "P0 should be skipped for event trigger") + assert.NotNil(t, result.Goals) +} + +func TestExecutorNilRobot(t *testing.T) { + exec := NewWithDelay(0) + ctx := types.NewContext(context.Background(), nil) + + result, err := exec.Execute(ctx, nil, types.TriggerClock, nil) + + assert.Error(t, err) + assert.Nil(t, result) + assert.Contains(t, err.Error(), "robot cannot be nil") +} + +func TestExecutorSimulatedFailure(t *testing.T) { + exec := NewWithDelay(0) + robot := &types.Robot{ + MemberID: "test-fail", + TeamID: "team-1", + Config: &types.Config{Quota: &types.Quota{Max: 1}}, + } + ctx := types.NewContext(context.Background(), nil) + + // Pass "simulate_failure" to trigger simulated failure + result, err := exec.Execute(ctx, robot, types.TriggerClock, "simulate_failure") + + assert.NoError(t, err) // Execute returns nil error, failure is in result + assert.NotNil(t, result) + assert.Equal(t, types.ExecFailed, result.Status) + assert.Equal(t, "simulated failure", result.Error) +} + +func TestExecutorCounters(t *testing.T) { + exec := NewWithDelay(0) + robot := &types.Robot{ + MemberID: "test-counter", + TeamID: "team-1", + Config: &types.Config{Quota: &types.Quota{Max: 10}}, + } + ctx := types.NewContext(context.Background(), nil) + + assert.Equal(t, 0, exec.ExecCount()) + assert.Equal(t, 0, exec.CurrentCount()) + + _, _ = exec.Execute(ctx, robot, types.TriggerClock, nil) + assert.Equal(t, 1, exec.ExecCount()) + assert.Equal(t, 0, exec.CurrentCount()) // Completed, so 0 + + _, _ = exec.Execute(ctx, robot, types.TriggerClock, nil) + assert.Equal(t, 2, exec.ExecCount()) + + exec.Reset() + assert.Equal(t, 0, exec.ExecCount()) +} diff --git a/agent/robot/executor/goals.go b/agent/robot/executor/goals.go new file mode 100644 index 00000000..cd4d20e3 --- /dev/null +++ b/agent/robot/executor/goals.go @@ -0,0 +1,47 @@ +package executor + +import ( + "github.com/yaoapp/yao/agent/robot/types" +) + +// RunGoals executes P1: Goals phase +// +// For Clock trigger: Uses InspirationReport to generate goals +// For Human/Event: Uses TriggerInput directly as goals or to generate goals +// +// Implementation (TODO Phase 5): +// 1. Build prompt with InspirationReport (or TriggerInput for Human/Event) +// 2. Call Goal Generation Agent via Assistant.Stream() +// 3. Parse response to Goals (markdown) +func (e *Executor) RunGoals(_ *types.Context, exec *types.Execution, _ interface{}) error { + // TODO (Phase 5): Replace with real Agent call + // agentID := robot.Config.Resources.GetPhaseAgent(types.PhaseGoals) + // messages := buildGoalsMessages(exec.Inspiration, exec.Input, robot) + // response, err := callAgentStream(ctx, agentID, messages) + // if err != nil { + // return err + // } + // exec.Goals = parseGoals(response) + + // Simulate Agent Stream delay + e.simulateStreamDelay() + + // Generate mock goals + exec.Goals = &types.Goals{ + Content: `## Goals + +1. [High] Complete primary objective + - Reason: Critical for business success + - Expected outcome: Measurable improvement + +2. [Normal] Review and validate results + - Reason: Quality assurance required + - Expected outcome: Verified deliverables + +3. [Low] Document learnings + - Reason: Future reference and improvement + - Expected outcome: Knowledge base update`, + } + + return nil +} diff --git a/agent/robot/executor/inspiration.go b/agent/robot/executor/inspiration.go new file mode 100644 index 00000000..515c9487 --- /dev/null +++ b/agent/robot/executor/inspiration.go @@ -0,0 +1,55 @@ +package executor + +import ( + "time" + + "github.com/yaoapp/yao/agent/robot/types" +) + +// RunInspiration executes P0: Inspiration phase (Clock trigger only) +// +// This phase gathers information to help make good goals. +// ClockContext is the key input - Agent knows what time it is and can decide +// what to do (e.g., 5pm Friday → write weekly report). +// +// Implementation (TODO Phase 4): +// 1. Build prompt with ClockContext + data sources (KB, DB, web search) +// 2. Call Inspiration Agent via Assistant.Stream() +// 3. Parse response to InspirationReport (markdown) +func (e *Executor) RunInspiration(_ *types.Context, exec *types.Execution, _ interface{}) error { + // TODO (Phase 4): Replace with real Agent call + // agentID := robot.Config.Resources.GetPhaseAgent(types.PhaseInspiration) + // messages := buildInspirationMessages(exec.Input.Clock, robot) + // response, err := callAgentStream(ctx, agentID, messages) + // if err != nil { + // return err + // } + // exec.Inspiration = parseInspirationReport(response) + + // Simulate Agent Stream delay + e.simulateStreamDelay() + + // Generate mock inspiration report + exec.Inspiration = &types.InspirationReport{ + Clock: types.NewClockContext(time.Now(), ""), + Content: `## Summary +Mock inspiration report for testing. + +## Highlights +- [High] Test item 1 - Critical business metric changed +- [Normal] Test item 2 - Regular update available + +## Opportunities +- Market growth potential identified +- New customer segment emerging + +## Risks +- None identified in current period + +## Pending +- 2 tasks from previous execution +- 1 scheduled report due`, + } + + return nil +} diff --git a/agent/robot/executor/learning.go b/agent/robot/executor/learning.go new file mode 100644 index 00000000..aa564dc1 --- /dev/null +++ b/agent/robot/executor/learning.go @@ -0,0 +1,48 @@ +package executor + +import ( + "github.com/yaoapp/yao/agent/robot/types" +) + +// RunLearning executes P5: Learning phase +// +// Extracts learnings from execution and saves to private KB. +// Learning types: execution (what worked), feedback (errors), insight (patterns). +// +// Implementation (TODO Phase 9): +// 1. Build prompt with execution summary +// 2. Call Learning Agent via Assistant.Stream() to extract learnings +// 3. Save learning entries to private KB +func (e *Executor) RunLearning(_ *types.Context, exec *types.Execution, _ interface{}) error { + // TODO (Phase 9): Replace with real learning + // agentID := robot.Config.Resources.GetPhaseAgent(types.PhaseLearning) + // messages := buildLearningMessages(exec, robot) + // response, err := callAgentStream(ctx, agentID, messages) + // if err != nil { + // return err + // } + // exec.Learning = parseLearningEntries(response) + // err = saveLearningToKB(ctx, robot, exec.Learning) + // if err != nil { + // return err + // } + + // Simulate Agent Stream delay + e.simulateStreamDelay() + + // Generate mock learning entries + exec.Learning = []types.LearningEntry{ + { + Type: types.LearnExecution, + Content: "Execution completed successfully with all tasks passing. Total duration within expected range.", + Tags: []string{"success", "performance"}, + }, + { + Type: types.LearnInsight, + Content: "Task execution order optimization: Running data analysis before report generation improves efficiency.", + Tags: []string{"optimization", "workflow"}, + }, + } + + return nil +} diff --git a/agent/robot/executor/run.go b/agent/robot/executor/run.go new file mode 100644 index 00000000..82d55f3a --- /dev/null +++ b/agent/robot/executor/run.go @@ -0,0 +1,78 @@ +package executor + +import ( + "fmt" + "time" + + "github.com/yaoapp/yao/agent/robot/types" +) + +// RunExecution executes P3: Run phase +// +// Iterates through tasks and executes each one using the specified executor. +// Supports three executor types: assistant, mcp, process. +// +// Implementation (TODO Phase 7): +// 1. Iterate tasks +// 2. For each task, call executor (assistant/mcp/process) +// 3. Validate results +// 4. Collect results +func (e *Executor) RunExecution(_ *types.Context, exec *types.Execution, _ interface{}) error { + // TODO (Phase 7): Replace with real task execution + // for i, task := range exec.Tasks { + // exec.Current = &types.CurrentState{Task: &task, TaskIndex: i} + // result, err := executeTask(ctx, task, robot) + // if err != nil { + // return err + // } + // exec.Results = append(exec.Results, result) + // } + + // Handle empty tasks case + if len(exec.Tasks) == 0 { + exec.Current = &types.CurrentState{ + TaskIndex: 0, + Progress: "0/0 tasks", + } + exec.Results = []types.TaskResult{} + return nil + } + + // Set current state (will be updated as tasks complete) + exec.Current = &types.CurrentState{ + TaskIndex: 0, + Progress: fmt.Sprintf("0/%d tasks", len(exec.Tasks)), + } + + // Simulate execution of each task + exec.Results = make([]types.TaskResult, len(exec.Tasks)) + for i := range exec.Tasks { + // Update current state + exec.Current.TaskIndex = i + exec.Current.Task = &exec.Tasks[i] + exec.Current.Progress = fmt.Sprintf("%d/%d tasks", i+1, len(exec.Tasks)) + + // Mark task start time + startTime := time.Now() + exec.Tasks[i].StartTime = &startTime + + // Simulate Agent Stream delay for each task + e.simulateStreamDelay() + + // Mark task as completed + exec.Tasks[i].Status = types.TaskCompleted + endTime := time.Now() + exec.Tasks[i].EndTime = &endTime + + // Generate mock result with actual duration + exec.Results[i] = types.TaskResult{ + TaskID: exec.Tasks[i].ID, + Success: true, + Output: fmt.Sprintf("Mock output for %s: Task completed successfully", exec.Tasks[i].ID), + Duration: endTime.Sub(startTime).Milliseconds(), + Validated: true, + } + } + + return nil +} diff --git a/agent/robot/executor/tasks.go b/agent/robot/executor/tasks.go new file mode 100644 index 00000000..1533febf --- /dev/null +++ b/agent/robot/executor/tasks.go @@ -0,0 +1,52 @@ +package executor + +import ( + "github.com/yaoapp/yao/agent/robot/types" +) + +// RunTasks executes P2: Tasks phase +// +// Reads Goals markdown and breaks into executable tasks. +// Each task specifies executor type (assistant/mcp/process) and arguments. +// +// Implementation (TODO Phase 6): +// 1. Build prompt with Goals +// 2. Call Task Planning Agent via Assistant.Stream() +// 3. Parse response to []Task (structured) +func (e *Executor) RunTasks(_ *types.Context, exec *types.Execution, _ interface{}) error { + // TODO (Phase 6): Replace with real Agent call + // agentID := robot.Config.Resources.GetPhaseAgent(types.PhaseTasks) + // messages := buildTasksMessages(exec.Goals, robot) + // response, err := callAgentStream(ctx, agentID, messages) + // if err != nil { + // return err + // } + // exec.Tasks = parseTasks(response) + + // Simulate Agent Stream delay + e.simulateStreamDelay() + + // Generate mock tasks + exec.Tasks = []types.Task{ + { + ID: "task_1", + GoalRef: "Goal 1", + Source: types.TaskSourceAuto, + ExecutorType: types.ExecutorAssistant, + ExecutorID: "data-analyst", + Status: types.TaskPending, + Order: 0, + }, + { + ID: "task_2", + GoalRef: "Goal 2", + Source: types.TaskSourceAuto, + ExecutorType: types.ExecutorAssistant, + ExecutorID: "report-writer", + Status: types.TaskPending, + Order: 1, + }, + } + + return nil +}