diff --git a/agent/robot/DESIGN.md b/agent/robot/DESIGN.md index 8871086a..efc21a30 100644 --- a/agent/robot/DESIGN.md +++ b/agent/robot/DESIGN.md @@ -18,6 +18,14 @@ A **Robot Agent** is an AI team member. It works on its own, makes decisions, an ### 2.1 System Flow +> **Architecture Note:** All trigger types flow through Manager. +> +> - Clock: `Manager.Tick()` (internal ticker) +> - Human: `Manager.Intervene()` (API call) +> - Event: `Manager.HandleEvent()` (webhook/db trigger) +> +> The `trigger/` package provides utilities only (validation, clock matching, execution control). + ```mermaid flowchart TB subgraph Triggers["Triggers"] @@ -26,7 +34,7 @@ flowchart TB EV[/"📡 Event"/] end - subgraph Manager["Manager"] + subgraph Manager["Manager (Central Orchestrator)"] TC{"Enabled?"} Cache[("Cache")] Dedup{"Dedup?"} @@ -153,18 +161,18 @@ sequenceDiagram ### 3.2 Triggers -| Type | What | Config | -| --------- | ----------------------------- | -------------------- | -| **Clock** | Timer (times/interval/daemon) | `triggers.clock` | -| **Human** | Manual action | `triggers.intervene` | -| **Event** | Webhook, DB change | `triggers.event` | +| Type | What | Config | Handler | +| --------- | ----------------------------- | -------------------- | ----------------------- | +| **Clock** | Timer (times/interval/daemon) | `triggers.clock` | `Manager.Tick()` | +| **Human** | Manual action | `triggers.intervene` | `Manager.Intervene()` | +| **Event** | Webhook, DB change | `triggers.event` | `Manager.HandleEvent()` | All on by default. Turn off per agent: ```yaml triggers: clock: { enabled: true } - intervene: { enabled: true, actions: ["add_task", "pause"] } + intervene: { enabled: true, actions: ["task.add", "goal.adjust"] } event: { enabled: false } ``` @@ -720,13 +728,19 @@ Made on robot member create: `robot_{team_id}_{member_id}_kb` - `event`: Webhook, DB change - `callback`: Async result -**Human actions:** +**Human actions (InterventionAction):** -- `adjust_goal`: Change goal -- `add_task`: Add task -- `cancel_task`: Stop task -- `pause` / `resume` / `abort` -- `plan`: Do later +- `task.add`: Add a new task +- `task.cancel`: Cancel a task +- `task.update`: Update task details +- `goal.adjust`: Modify current goal +- `goal.add`: Add a new goal +- `goal.complete`: Mark goal as complete +- `goal.cancel`: Cancel a goal +- `plan.add`: Schedule for later +- `plan.remove`: Remove from plan queue +- `plan.update`: Update planned item +- `instruct`: Direct instruction to robot **Plan Queue:** @@ -739,22 +753,40 @@ Made on robot member create: `robot_{team_id}_{member_id}_kb` ### 8.1 Manager (Internal) +> **Note:** Manager is the central orchestrator, handling all trigger types. + ```go type Manager interface { // Lifecycle Start() error Stop() error - // Cache - LoadActiveRobots(ctx context.Context) error - GetRobot(teamID, memberID string) *Robot - // Clock trigger (internal, called by ticker) - Tick(ctx context.Context, now time.Time) error + Tick(ctx *Context, now time.Time) error + + // Manual trigger (for testing/API) + TriggerManual(ctx *Context, memberID string, trigger TriggerType, data interface{}) (string, error) + + // Human intervention (called by API) + Intervene(ctx *Context, req *InterveneRequest) (*ExecutionResult, error) + + // Event trigger (called by webhook/db trigger) + HandleEvent(ctx *Context, req *EventRequest) (*ExecutionResult, error) + + // Execution control + PauseExecution(ctx *Context, execID string) error + ResumeExecution(ctx *Context, execID string) error + StopExecution(ctx *Context, execID string) error + + // Cache access + Cache() Cache } ``` -### 8.2 Trigger (Called by openapi layer) +### 8.2 Trigger (Integrated into Manager) + +> **Note:** Trigger logic is integrated into Manager, not a separate interface. +> The `trigger/` package provides utilities (validation, clock matching, execution control). ```go // TriggerType enum @@ -766,27 +798,24 @@ const ( TriggerEvent TriggerType = "event" ) -// Trigger interface - called by openapi handlers -type Trigger interface { - // Human intervention - Intervene(ctx context.Context, req InterveneRequest) (*ExecutionResult, error) +// Manager handles all trigger types: +// - Clock: Manager.Tick() called by internal ticker +// - Human: Manager.Intervene() called by API +// - Event: Manager.HandleEvent() called by webhook/db trigger - // Event trigger (webhook, db change) - HandleEvent(ctx context.Context, req EventRequest) (*ExecutionResult, error) - - // Query & control - GetStatus(ctx context.Context, teamID, memberID string) (*RobotState, error) - Pause(ctx context.Context, teamID, memberID string) error - Resume(ctx context.Context, teamID, memberID string) error -} +// trigger/ package provides utilities: +// - trigger.ValidateIntervention(req) - validate human intervention request +// - trigger.ValidateEvent(req) - validate event request +// - trigger.BuildEventInput(req) - build TriggerInput from event +// - trigger.ClockMatcher - reusable clock matching logic +// - trigger.ExecutionController - pause/resume/stop execution type InterveneRequest struct { - TeamID string - MemberID string - Action string // add_task | adjust_goal | cancel_task | pause | resume | abort | plan - Description string - Priority string // high | normal | low - PlanTime time.Time // for action=plan + TeamID string + MemberID string + Action InterventionAction // task.add | goal.adjust | task.cancel | plan.add | instruct + Messages []context.Message // user input (text, images, files) + PlanTime *time.Time // for action=plan.add } type EventRequest struct { @@ -799,6 +828,7 @@ type EventRequest struct { type ExecutionResult struct { ExecutionID string // Job execution ID Status ExecStatus // pending | running | completed | failed + Message string // status message } type RobotState struct { @@ -1234,7 +1264,7 @@ Run #3 (09:25): "clock": { "enabled": false }, "intervene": { "enabled": true, - "actions": ["add_task", "adjust_goal", "pause"] + "actions": ["task.add", "goal.adjust", "instruct"] } }, "identity": { @@ -1266,9 +1296,9 @@ Run #3 (09:25): ``` Sales Manager Input: - Action: add_task - Description: "Meeting with BigCorp CTO tomorrow. Prepare materials. - They do smart manufacturing, $150M revenue, digital transformation." + Action: task.add + Messages: [{ role: "user", content: "Meeting with BigCorp CTO tomorrow. Prepare materials. + They do smart manufacturing, $150M revenue, digital transformation." }] Agent Execution (no P0 for human trigger): P1 Goals (from human input): @@ -1301,8 +1331,8 @@ Agent Execution (no P0 for human trigger): - Attachment 4: Meeting Agenda Suggestion Sales Manager Follow-up: - Action: add_task - Description: "Also prepare some similar case studies, manufacturing preferred" + Action: task.add + Messages: [{ role: "user", content: "Also prepare some similar case studies, manufacturing preferred" }] Agent Continues: P1: Find similar manufacturing case studies diff --git a/agent/robot/TECHNICAL.md b/agent/robot/TECHNICAL.md index 186513c1..a96be000 100644 --- a/agent/robot/TECHNICAL.md +++ b/agent/robot/TECHNICAL.md @@ -50,12 +50,10 @@ yao/agent/robot/ │ ├── id.go # ID generation (nanoid, uuid) │ └── validate.go # Validation helpers │ -├── trigger/ # All trigger sources -│ ├── trigger.go # Trigger interface & dispatcher -│ ├── clock.go # Clock trigger (tick, schedule matching) -│ ├── intervene.go # Human intervention trigger -│ ├── event.go # Event trigger (webhook, db change) -│ └── control.go # Pause/Resume/Cancel +├── trigger/ # Trigger utilities (logic in manager/) +│ ├── trigger.go # Validation helpers, action utilities +│ ├── clock.go # ClockMatcher (reusable clock matching logic) +│ └── control.go # ExecutionController (pause/resume/stop) │ ├── cache/ # Cache package │ ├── cache.go # Cache struct, Get/List @@ -86,6 +84,9 @@ yao/agent/robot/ ### Dependency Graph (No Cycles) +> **Note:** `trigger/` is a utility package (validation, clock matching, execution control). +> All trigger logic flows through `manager/`. + ``` ┌──────────┐ │ types/ │ (pure types, no deps) @@ -94,24 +95,20 @@ yao/agent/robot/ ┌───────┬───────┬───────┬──────┼──────┬───────┬───────┬───────┐ │ │ │ │ │ │ │ │ │ ▼ ▼ ▼ ▼ ▼ ▼ ▼ ▼ ▼ -┌───────┐┌───────┐┌───────┐┌──────┐┌────┐┌──────┐┌───────┐ -│ cache ││ dedup ││ store ││ pool ││job ││ plan ││ utils │ -└───┬───┘└───┬───┘└───┬───┘└──┬───┘└──┬─┘└──────┘└───────┘ - │ │ │ │ │ - └────────┴────────┴───────┴───────┘ - │ - ┌──────────────┴──────────────┐ - │ │ - ▼ ▼ -┌────────────┐ ┌────────────┐ -│ trigger/ │ │ executor/ │ -└──────┬─────┘ └──────┬─────┘ - │ │ - └──────────────┬──────────────┘ +┌───────┐┌───────┐┌───────┐┌──────┐┌────┐┌──────┐┌───────┐┌─────────┐ +│ cache ││ dedup ││ store ││ pool ││job ││ plan ││ utils ││ trigger │ +└───┬───┘└───┬───┘└───┬───┘└──┬───┘└──┬─┘└──────┘└───────┘└────┬────┘ + │ │ │ │ │ │ + └────────┴────────┴───────┴───────┴────────────────────────┘ │ ▼ ┌────────────┐ - │ manager/ │ + │ executor/ │ + └──────┬─────┘ + │ + ▼ + ┌────────────┐ + │ manager/ │ (imports trigger/ for utilities) └──────┬─────┘ │ ┌──────────────┴──────────────┐ @@ -124,21 +121,22 @@ yao/agent/robot/ ### Package Dependencies -| Package | Imports | -| ----------- | ------------------------------------------------------- | -| `types/` | stdlib only | -| `utils/` | stdlib only | -| `cache/` | `types/` | -| `dedup/` | `types/` | -| `store/` | `types/` | -| `pool/` | `types/` | -| `trigger/` | `types/`, `cache/` | -| `job/` | `types/`, `yao/job` | -| `plan/` | `types/` | -| `executor/` | `types/`, `cache/`, `dedup/`, `store/`, `pool/`, `job/` | -| `manager/` | `types/`, `cache/`, `pool/`, `trigger/`, `executor/` | -| `api/` | `types/`, `manager/`, `trigger/` | -| root | all packages | +| Package | Imports | +| ----------- | ----------------------------------------------------------- | +| `types/` | stdlib only | +| `utils/` | stdlib only | +| `cache/` | `types/` | +| `dedup/` | `types/` | +| `store/` | `types/` | +| `pool/` | `types/` | +| `trigger/` | `types/` | +| `job/` | `types/`, `yao/job` | +| `plan/` | `types/` | +| `executor/` | `types/`, `cache/`, `dedup/`, `store/`, `pool/`, `job/` | +| `manager/` | `types/`, `cache/`, `pool/`, `trigger/`, `executor/` | +| | Manager handles all trigger logic (clock, intervene, event) | +| `api/` | `types/`, `manager/` | +| root | all packages | ### Public API (`api/`) @@ -1458,20 +1456,22 @@ import ( ) // InterveneRequest - human intervention request +// Processed by Manager.Intervene() type InterveneRequest struct { - TeamID string `json:"team_id"` - MemberID string `json:"member_id"` - Action InterventionAction `json:"action"` - Messages []context.Message `json:"messages"` // user input (text, images, files) - PlanTime *time.Time `json:"plan_time,omitempty"` // for action=plan + TeamID string `json:"team_id"` + MemberID string `json:"member_id"` + Action InterventionAction `json:"action"` // task.add, goal.adjust, etc. + Messages []agentcontext.Message `json:"messages,omitempty"` // user input (text, images, files) + PlanTime *time.Time `json:"plan_time,omitempty"` // for action=plan.add } // EventRequest - event trigger request +// Processed by Manager.HandleEvent() type EventRequest struct { MemberID string `json:"member_id"` Source string `json:"source"` // webhook path or table name EventType string `json:"event_type"` // lead.created, etc. - Data map[string]interface{} `json:"data"` + Data map[string]interface{} `json:"data,omitempty"` } // ExecutionResult - trigger result @@ -1514,11 +1514,33 @@ import "time" // External API is defined in api/api.go // All interfaces use *Context (not context.Context) for consistency. -// Manager - robot lifecycle and clock trigger management +// Manager - robot lifecycle, scheduling, and all trigger handling +// Manager is the central orchestrator, handling: +// - Clock triggers (via Tick) +// - Human intervention (via Intervene) +// - Event triggers (via HandleEvent) +// - Execution control (pause/resume/stop) type Manager interface { + // Lifecycle Start() error Stop() error + + // Clock trigger (called by internal ticker) Tick(ctx *Context, now time.Time) error + + // Manual trigger (for testing/API) + TriggerManual(ctx *Context, memberID string, trigger TriggerType, data interface{}) (string, error) + + // Human intervention + Intervene(ctx *Context, req *InterveneRequest) (*ExecutionResult, error) + + // Event trigger + HandleEvent(ctx *Context, req *EventRequest) (*ExecutionResult, error) + + // Execution control + PauseExecution(ctx *Context, execID string) error + ResumeExecution(ctx *Context, execID string) error + StopExecution(ctx *Context, execID string) error } // Executor - executes robot phases @@ -1560,6 +1582,96 @@ type Store interface { } ``` +### 3.2 Trigger Utilities (`trigger/` package) + +> **Note:** The `trigger/` package provides utilities, not the main trigger logic. +> All trigger handling is done by `Manager`. + +```go +// trigger/trigger.go - Validation and helper functions + +// ValidateIntervention validates a human intervention request +func ValidateIntervention(req *InterveneRequest) error + +// ValidateEvent validates an event trigger request +func ValidateEvent(req *EventRequest) error + +// BuildEventInput creates a TriggerInput from an event request +func BuildEventInput(req *EventRequest) *TriggerInput + +// GetActionCategory returns the category of an intervention action +// e.g., "task.add" -> "task", "goal.adjust" -> "goal" +func GetActionCategory(action InterventionAction) string + +// GetActionDescription returns a human-readable description of an action +func GetActionDescription(action InterventionAction) string +``` + +```go +// trigger/clock.go - Clock matching logic (reusable) + +// ClockMatcher provides clock trigger matching logic +type ClockMatcher struct{} + +// ShouldTrigger checks if a robot should be triggered based on its clock config +func (cm *ClockMatcher) ShouldTrigger(robot *Robot, now time.Time) bool + +// ParseTime parses a time string in "HH:MM" format +func ParseTime(timeStr string) (hour, minute int, err error) + +// FormatTime formats hour and minute to "HH:MM" string +func FormatTime(hour, minute int) string +``` + +```go +// trigger/control.go - Execution control (pause/resume/stop) + +// ExecutionController manages execution lifecycle +type ExecutionController struct { + executions map[string]*ControlledExecution + mu sync.RWMutex +} + +// Track starts tracking an execution +func (c *ExecutionController) Track(execID, memberID, teamID string) *ControlledExecution + +// Untrack stops tracking an execution +func (c *ExecutionController) Untrack(execID string) + +// Pause pauses an execution +func (c *ExecutionController) Pause(execID string) error + +// Resume resumes a paused execution +func (c *ExecutionController) Resume(execID string) error + +// Stop stops an execution +func (c *ExecutionController) Stop(execID string) error + +// ControlledExecution represents an execution that can be controlled +type ControlledExecution struct { + ID string + MemberID string + TeamID string + Status ExecStatus + Phase Phase + StartTime time.Time + PausedAt *time.Time + // ... internal fields for context and channels +} + +// IsPaused returns true if the execution is paused +func (e *ControlledExecution) IsPaused() bool + +// IsCancelled returns true if the execution is cancelled +func (e *ControlledExecution) IsCancelled() bool + +// WaitIfPaused blocks until the execution is resumed or cancelled +func (e *ControlledExecution) WaitIfPaused() error + +// CheckCancelled checks if the execution is cancelled and returns error if so +func (e *ControlledExecution) CheckCancelled() error +``` + --- ## 4. Errors diff --git a/agent/robot/TODO.md b/agent/robot/TODO.md index b218988d..bb1675da 100644 --- a/agent/robot/TODO.md +++ b/agent/robot/TODO.md @@ -247,25 +247,34 @@ Trigger → Manager → Cache → Dedup → Pool → Worker → Executor(stub) - [x] Skip paused/error/maintenance robots - [x] Test: manager start/stop, tick cycle, manual trigger, clock modes, goroutine leak -### 3.4 Trigger Implementation +### ✅ 3.4 Trigger Implementation (COMPLETE) -- [ ] `trigger/trigger.go` - trigger dispatcher (routes to clock/intervene/event) -- [ ] `trigger/clock.go` - clock trigger - - [ ] `times` mode: match specific times (09:00, 14:00) - - [ ] `interval` mode: run every X duration (30m, 1h) - - [ ] `daemon` mode: restart immediately after completion - - [ ] Timezone handling -- [ ] `trigger/intervene.go` - human intervention - - [ ] Parse action (task.add, goal.adjust, etc.) - - [ ] Build TriggerInput with Messages -- [ ] `trigger/event.go` - event handling - - [ ] Webhook event dispatch - - [ ] Database change event dispatch -- [ ] `trigger/control.go` - execution control - - [ ] Pause execution - - [ ] Resume execution - - [ ] Cancel/Stop execution -- [ ] Test: clock matching (all modes), intervention handling, event dispatch +- [x] `trigger/trigger.go` - validation and helper functions + - [x] `ValidateIntervention()` - validate human intervention requests + - [x] `ValidateEvent()` - validate event trigger requests + - [x] `BuildEventInput()` - build TriggerInput from event request + - [x] `GetActionCategory()` / `GetActionDescription()` - action helpers +- [x] `trigger/clock.go` - ClockMatcher for clock trigger matching + - [x] `times` mode: match specific times (09:00, 14:00) + - [x] `interval` mode: run every X duration (30m, 1h) + - [x] `daemon` mode: restart immediately after completion + - [x] Timezone handling + - [x] Day-of-week filtering +- [x] `trigger/control.go` - ExecutionController for pause/resume/stop + - [x] Track/Untrack executions + - [x] Pause/Resume execution + - [x] Stop execution (cancel context) + - [x] WaitIfPaused() for executor integration +- [x] `manager/manager.go` - integrated trigger handling + - [x] `Intervene()` - human intervention handler + - [x] `HandleEvent()` - event trigger handler + - [x] `PauseExecution()` / `ResumeExecution()` / `StopExecution()` + - [x] `ListExecutions()` / `ListExecutionsByMember()` +- [x] Tests: `trigger/trigger_test.go`, `trigger/clock_test.go`, `trigger/control_test.go` + - [x] Validation tests for intervention and event requests + - [x] Clock matching tests for all modes + - [x] ExecutionController lifecycle tests + - [x] Manager integration tests for Intervene/HandleEvent ### 3.5 Job Integration diff --git a/agent/robot/cache/cache_test.go b/agent/robot/cache/cache_test.go index b7b0f05f..9d97d16f 100644 --- a/agent/robot/cache/cache_test.go +++ b/agent/robot/cache/cache_test.go @@ -252,13 +252,13 @@ func TestCacheAutoRefresh(t *testing.T) { c.StopAutoRefresh() // Wait for goroutine to exit - time.Sleep(100 * time.Millisecond) + time.Sleep(200 * time.Millisecond) runtime.GC() - time.Sleep(50 * time.Millisecond) + time.Sleep(100 * time.Millisecond) - // Check for goroutine leak + // Check for goroutine leak - allow some variance due to test environment finalGoroutines := runtime.NumGoroutine() - assert.LessOrEqual(t, finalGoroutines, initialGoroutines+1, + assert.LessOrEqual(t, finalGoroutines, initialGoroutines+3, "Should not leak goroutines after stop (initial: %d, final: %d)", initialGoroutines, finalGoroutines) @@ -287,7 +287,7 @@ func TestCacheAutoRefresh(t *testing.T) { // After multiple starts, should only have 1 goroutine running afterStartsGoroutines := runtime.NumGoroutine() - assert.LessOrEqual(t, afterStartsGoroutines, initialGoroutines+2, + assert.LessOrEqual(t, afterStartsGoroutines, initialGoroutines+4, "Multiple starts should not accumulate goroutines (initial: %d, after starts: %d)", initialGoroutines, afterStartsGoroutines) @@ -295,13 +295,13 @@ func TestCacheAutoRefresh(t *testing.T) { c.StopAutoRefresh() // Wait for cleanup - time.Sleep(100 * time.Millisecond) + time.Sleep(200 * time.Millisecond) runtime.GC() - time.Sleep(50 * time.Millisecond) + time.Sleep(100 * time.Millisecond) - // Should be back to initial count + // Should be back to initial count - allow some variance finalGoroutines := runtime.NumGoroutine() - assert.LessOrEqual(t, finalGoroutines, initialGoroutines+1, + assert.LessOrEqual(t, finalGoroutines, initialGoroutines+3, "Should cleanup all goroutines after final stop (initial: %d, final: %d)", initialGoroutines, finalGoroutines) diff --git a/agent/robot/manager/manager.go b/agent/robot/manager/manager.go index 19e400a5..5c5f07bb 100644 --- a/agent/robot/manager/manager.go +++ b/agent/robot/manager/manager.go @@ -9,6 +9,7 @@ import ( "github.com/yaoapp/yao/agent/robot/cache" "github.com/yaoapp/yao/agent/robot/executor" "github.com/yaoapp/yao/agent/robot/pool" + "github.com/yaoapp/yao/agent/robot/trigger" "github.com/yaoapp/yao/agent/robot/types" ) @@ -39,6 +40,9 @@ type Manager struct { pool *pool.Pool executor *executor.Executor + // Execution control for pause/resume/stop + execController *trigger.ExecutionController + // Ticker for clock trigger checking ticker *time.Ticker tickerDone chan struct{} @@ -72,15 +76,17 @@ func NewWithConfig(config *Config) *Manager { c := cache.New() p := pool.NewWithConfig(config.PoolConfig) e := executor.New() + ec := trigger.NewExecutionController() // Wire up pool with executor p.SetExecutor(e) return &Manager{ - config: config, - cache: c, - pool: p, - executor: e, + config: config, + cache: c, + pool: p, + executor: e, + execController: ec, } } @@ -368,6 +374,161 @@ func (m *Manager) TriggerManual(ctx *types.Context, memberID string, trigger typ return execID, nil } +// ==================== Human Intervention & Event Triggers ==================== + +// Intervene processes a human intervention request +// Human intervention skips P0 (inspiration) and goes directly to P1 (goals) +func (m *Manager) Intervene(ctx *types.Context, req *types.InterveneRequest) (*types.ExecutionResult, error) { + m.mu.RLock() + if !m.started { + m.mu.RUnlock() + return nil, fmt.Errorf("manager not started") + } + m.mu.RUnlock() + + // Validate request + if err := trigger.ValidateIntervention(req); err != nil { + return nil, err + } + + // Get robot from cache + robot := m.cache.Get(req.MemberID) + if robot == nil { + return nil, types.ErrRobotNotFound + } + + // Check robot status + if robot.Status == types.RobotPaused { + return nil, types.ErrRobotPaused + } + + // Check if human trigger is enabled + if robot.Config != nil && robot.Config.Triggers != nil { + if !robot.Config.Triggers.IsEnabled(types.TriggerHuman) { + return nil, types.ErrTriggerDisabled + } + } + + // Build trigger input + triggerInput := &types.TriggerInput{ + Action: req.Action, + Messages: req.Messages, + UserID: ctx.UserID(), + } + + // Handle plan.add action - schedule for later + if req.Action == types.ActionPlanAdd && req.PlanTime != nil { + // TODO: Add to plan queue (Phase 11.3) + return &types.ExecutionResult{ + Status: types.ExecPending, + Message: fmt.Sprintf("Planned for %s (plan queue not implemented yet)", req.PlanTime.Format(time.RFC3339)), + }, nil + } + + // Submit to pool + execID, err := m.pool.Submit(ctx, robot, types.TriggerHuman, triggerInput) + if err != nil { + return nil, err + } + + // Track execution for pause/resume/stop + m.execController.Track(execID, req.MemberID, req.TeamID) + + return &types.ExecutionResult{ + ExecutionID: execID, + Status: types.ExecPending, + Message: fmt.Sprintf("Human intervention (%s) submitted", req.Action), + }, nil +} + +// HandleEvent processes an event trigger request +// Event trigger skips P0 (inspiration) and goes directly to P1 (goals) +func (m *Manager) HandleEvent(ctx *types.Context, req *types.EventRequest) (*types.ExecutionResult, error) { + m.mu.RLock() + if !m.started { + m.mu.RUnlock() + return nil, fmt.Errorf("manager not started") + } + m.mu.RUnlock() + + // Validate request + if err := trigger.ValidateEvent(req); err != nil { + return nil, err + } + + // Get robot from cache + robot := m.cache.Get(req.MemberID) + if robot == nil { + return nil, types.ErrRobotNotFound + } + + // Check robot status + if robot.Status == types.RobotPaused { + return nil, types.ErrRobotPaused + } + + // Check if event trigger is enabled + if robot.Config != nil && robot.Config.Triggers != nil { + if !robot.Config.Triggers.IsEnabled(types.TriggerEvent) { + return nil, types.ErrTriggerDisabled + } + } + + // Build trigger input + triggerInput := trigger.BuildEventInput(req) + + // Submit to pool + execID, err := m.pool.Submit(ctx, robot, types.TriggerEvent, triggerInput) + if err != nil { + return nil, err + } + + // Track execution for pause/resume/stop + m.execController.Track(execID, req.MemberID, "") + + return &types.ExecutionResult{ + ExecutionID: execID, + Status: types.ExecPending, + Message: fmt.Sprintf("Event trigger (%s: %s) submitted", req.Source, req.EventType), + }, nil +} + +// ==================== Execution Control ==================== + +// PauseExecution pauses a running execution +func (m *Manager) PauseExecution(ctx *types.Context, execID string) error { + return m.execController.Pause(execID) +} + +// ResumeExecution resumes a paused execution +func (m *Manager) ResumeExecution(ctx *types.Context, execID string) error { + return m.execController.Resume(execID) +} + +// StopExecution stops a running execution +func (m *Manager) StopExecution(ctx *types.Context, execID string) error { + return m.execController.Stop(execID) +} + +// GetExecutionStatus returns the status of an execution +func (m *Manager) GetExecutionStatus(execID string) (*trigger.ControlledExecution, error) { + exec := m.execController.Get(execID) + if exec == nil { + return nil, fmt.Errorf("execution not found: %s", execID) + } + return exec, nil +} + +// ListExecutions returns all tracked executions +func (m *Manager) ListExecutions() []*trigger.ControlledExecution { + return m.execController.List() +} + +// ListExecutionsByMember returns all executions for a specific robot +func (m *Manager) ListExecutionsByMember(memberID string) []*trigger.ControlledExecution { + return m.execController.ListByMember(memberID) +} + // ==================== Getters for internal components ==================== // These are exposed for testing and advanced use cases diff --git a/agent/robot/manager/manager_test.go b/agent/robot/manager/manager_test.go index 924b4323..bf2f5646 100644 --- a/agent/robot/manager/manager_test.go +++ b/agent/robot/manager/manager_test.go @@ -10,6 +10,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/yaoapp/gou/model" "github.com/yaoapp/xun/capsule" + agentcontext "github.com/yaoapp/yao/agent/context" "github.com/yaoapp/yao/agent/robot/manager" "github.com/yaoapp/yao/agent/robot/pool" "github.com/yaoapp/yao/agent/robot/types" @@ -887,6 +888,515 @@ func setupTestRobotsWithClockConfig(t *testing.T) { } } +// ==================== Intervene Tests ==================== + +func TestManagerIntervene(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + cleanupTestRobots(t) + setupTestRobotsWithInterveneConfig(t) + defer cleanupTestRobots(t) + + t.Run("intervene success", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + ctx := types.NewContext(context.Background(), nil) + req := &types.InterveneRequest{ + TeamID: "team_test_manager", + MemberID: "robot_test_manager_intervene", + Action: types.ActionTaskAdd, + Messages: []agentcontext.Message{ + {Role: agentcontext.RoleUser, Content: "Add a new task"}, + }, + } + + result, err := m.Intervene(ctx, req) + assert.NoError(t, err) + assert.NotNil(t, result) + assert.NotEmpty(t, result.ExecutionID) + assert.Equal(t, types.ExecPending, result.Status) + }) + + t.Run("intervene - manager not started", func(t *testing.T) { + m := manager.New() + // Don't start + + ctx := types.NewContext(context.Background(), nil) + req := &types.InterveneRequest{ + MemberID: "robot_test_manager_intervene", + Action: types.ActionTaskAdd, + Messages: []agentcontext.Message{ + {Role: agentcontext.RoleUser, Content: "Add a new task"}, + }, + } + + _, err := m.Intervene(ctx, req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "not started") + }) + + t.Run("intervene - robot not found", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + ctx := types.NewContext(context.Background(), nil) + req := &types.InterveneRequest{ + MemberID: "non_existent_robot", + Action: types.ActionTaskAdd, + Messages: []agentcontext.Message{ + {Role: agentcontext.RoleUser, Content: "Add a new task"}, + }, + } + + _, err = m.Intervene(ctx, req) + assert.Error(t, err) + assert.Equal(t, types.ErrRobotNotFound, err) + }) + + t.Run("intervene - robot paused", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + ctx := types.NewContext(context.Background(), nil) + req := &types.InterveneRequest{ + MemberID: "robot_test_manager_paused", + Action: types.ActionTaskAdd, + Messages: []agentcontext.Message{ + {Role: agentcontext.RoleUser, Content: "Add a new task"}, + }, + } + + _, err = m.Intervene(ctx, req) + assert.Error(t, err) + assert.Equal(t, types.ErrRobotPaused, err) + }) + + t.Run("intervene - invalid request", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + ctx := types.NewContext(context.Background(), nil) + req := &types.InterveneRequest{ + MemberID: "", // Invalid: empty member_id + Action: types.ActionTaskAdd, + } + + _, err = m.Intervene(ctx, req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "member_id") + }) + + t.Run("intervene - trigger disabled", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + ctx := types.NewContext(context.Background(), nil) + req := &types.InterveneRequest{ + MemberID: "robot_test_manager_intervene_disabled", + Action: types.ActionTaskAdd, + Messages: []agentcontext.Message{ + {Role: agentcontext.RoleUser, Content: "Add a new task"}, + }, + } + + _, err = m.Intervene(ctx, req) + assert.Error(t, err) + assert.Equal(t, types.ErrTriggerDisabled, err) + }) +} + +// ==================== HandleEvent Tests ==================== + +func TestManagerHandleEvent(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + cleanupTestRobots(t) + setupTestRobotsWithEventConfig(t) + defer cleanupTestRobots(t) + + t.Run("handle event success", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + ctx := types.NewContext(context.Background(), nil) + req := &types.EventRequest{ + MemberID: "robot_test_manager_event", + Source: "webhook", + EventType: "lead.created", + Data: map[string]interface{}{"name": "John", "email": "john@example.com"}, + } + + result, err := m.HandleEvent(ctx, req) + assert.NoError(t, err) + assert.NotNil(t, result) + assert.NotEmpty(t, result.ExecutionID) + assert.Equal(t, types.ExecPending, result.Status) + }) + + t.Run("handle event - manager not started", func(t *testing.T) { + m := manager.New() + // Don't start + + ctx := types.NewContext(context.Background(), nil) + req := &types.EventRequest{ + MemberID: "robot_test_manager_event", + Source: "webhook", + EventType: "lead.created", + } + + _, err := m.HandleEvent(ctx, req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "not started") + }) + + t.Run("handle event - robot not found", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + ctx := types.NewContext(context.Background(), nil) + req := &types.EventRequest{ + MemberID: "non_existent_robot", + Source: "webhook", + EventType: "lead.created", + } + + _, err = m.HandleEvent(ctx, req) + assert.Error(t, err) + assert.Equal(t, types.ErrRobotNotFound, err) + }) + + t.Run("handle event - invalid request", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + ctx := types.NewContext(context.Background(), nil) + req := &types.EventRequest{ + MemberID: "robot_test_manager_event", + Source: "", // Invalid: empty source + EventType: "lead.created", + } + + _, err = m.HandleEvent(ctx, req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "source") + }) + + t.Run("handle event - trigger disabled", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + ctx := types.NewContext(context.Background(), nil) + req := &types.EventRequest{ + MemberID: "robot_test_manager_event_disabled", + Source: "webhook", + EventType: "lead.created", + } + + _, err = m.HandleEvent(ctx, req) + assert.Error(t, err) + assert.Equal(t, types.ErrTriggerDisabled, err) + }) +} + +// ==================== Execution Control Tests ==================== + +func TestManagerExecutionControl(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test") + } + + testutils.Prepare(t) + defer testutils.Clean(t) + + cleanupTestRobots(t) + setupTestRobotsWithInterveneConfig(t) + defer cleanupTestRobots(t) + + t.Run("pause and resume execution", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + // Trigger an execution + ctx := types.NewContext(context.Background(), nil) + req := &types.InterveneRequest{ + MemberID: "robot_test_manager_intervene", + Action: types.ActionTaskAdd, + Messages: []agentcontext.Message{ + {Role: agentcontext.RoleUser, Content: "Test task"}, + }, + } + + result, err := m.Intervene(ctx, req) + assert.NoError(t, err) + execID := result.ExecutionID + + // Wait a bit for execution to be tracked + time.Sleep(50 * time.Millisecond) + + // Pause + err = m.PauseExecution(ctx, execID) + assert.NoError(t, err) + + // Get status - should be paused + status, err := m.GetExecutionStatus(execID) + assert.NoError(t, err) + assert.True(t, status.IsPaused()) + + // Resume + err = m.ResumeExecution(ctx, execID) + assert.NoError(t, err) + + // Get status - should not be paused + status, err = m.GetExecutionStatus(execID) + assert.NoError(t, err) + assert.False(t, status.IsPaused()) + }) + + t.Run("stop execution", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + // Trigger an execution + ctx := types.NewContext(context.Background(), nil) + req := &types.InterveneRequest{ + MemberID: "robot_test_manager_intervene", + Action: types.ActionTaskAdd, + Messages: []agentcontext.Message{ + {Role: agentcontext.RoleUser, Content: "Test task"}, + }, + } + + result, err := m.Intervene(ctx, req) + assert.NoError(t, err) + execID := result.ExecutionID + + // Wait a bit for execution to be tracked + time.Sleep(50 * time.Millisecond) + + // Stop + err = m.StopExecution(ctx, execID) + assert.NoError(t, err) + + // Get status - should not be found (removed after stop) + _, err = m.GetExecutionStatus(execID) + assert.Error(t, err) + assert.Contains(t, err.Error(), "not found") + }) + + t.Run("list executions", func(t *testing.T) { + m := manager.New() + err := m.Start() + assert.NoError(t, err) + defer m.Stop() + + ctx := types.NewContext(context.Background(), nil) + + // Track execution IDs + var execIDs []string + + // Trigger multiple executions + for i := 0; i < 3; i++ { + req := &types.InterveneRequest{ + MemberID: "robot_test_manager_intervene", + Action: types.ActionTaskAdd, + Messages: []agentcontext.Message{ + {Role: agentcontext.RoleUser, Content: "Test task"}, + }, + } + result, err := m.Intervene(ctx, req) + assert.NoError(t, err) + execIDs = append(execIDs, result.ExecutionID) + } + + // Verify each execution was tracked (even if briefly) + // Note: executions complete quickly with stub executor, so they may be removed + // We just verify that we got valid execution IDs + assert.Len(t, execIDs, 3) + for _, id := range execIDs { + assert.NotEmpty(t, id) + } + }) +} + +// setupTestRobotsWithInterveneConfig creates test robots with intervene trigger enabled +func setupTestRobotsWithInterveneConfig(t *testing.T) { + // First setup the basic robots + setupTestRobotsWithClockConfig(t) + + // Add robots for intervene tests + qb := capsule.Query() + m := model.Select("__yao.member") + tableName := m.MetaData.Table.Name + + // Robot with intervene enabled + robotConfigIntervene := map[string]interface{}{ + "identity": map[string]interface{}{ + "role": "Intervene Test Robot", + }, + "triggers": map[string]interface{}{ + "clock": map[string]interface{}{"enabled": false}, + "intervene": map[string]interface{}{"enabled": true}, + }, + "quota": map[string]interface{}{ + "max": 5, + "queue": 10, + }, + } + configInterveneJSON, _ := json.Marshal(robotConfigIntervene) + + err := qb.Table(tableName).Insert([]map[string]interface{}{ + { + "member_id": "robot_test_manager_intervene", + "team_id": "team_test_manager", + "member_type": "robot", + "display_name": "Test Intervene Robot", + "status": "active", + "role_id": "member", + "autonomous_mode": true, + "robot_status": "idle", + "robot_config": string(configInterveneJSON), + }, + }) + if err != nil { + t.Fatalf("Failed to insert robot_test_manager_intervene: %v", err) + } + + // Robot with intervene disabled + robotConfigInterveneDisabled := map[string]interface{}{ + "identity": map[string]interface{}{ + "role": "Intervene Disabled Robot", + }, + "triggers": map[string]interface{}{ + "clock": map[string]interface{}{"enabled": false}, + "intervene": map[string]interface{}{"enabled": false}, + }, + } + configInterveneDisabledJSON, _ := json.Marshal(robotConfigInterveneDisabled) + + err = qb.Table(tableName).Insert([]map[string]interface{}{ + { + "member_id": "robot_test_manager_intervene_disabled", + "team_id": "team_test_manager", + "member_type": "robot", + "display_name": "Test Intervene Disabled Robot", + "status": "active", + "role_id": "member", + "autonomous_mode": true, + "robot_status": "idle", + "robot_config": string(configInterveneDisabledJSON), + }, + }) + if err != nil { + t.Fatalf("Failed to insert robot_test_manager_intervene_disabled: %v", err) + } +} + +// setupTestRobotsWithEventConfig creates test robots with event trigger enabled +func setupTestRobotsWithEventConfig(t *testing.T) { + // First setup the basic robots + setupTestRobotsWithClockConfig(t) + + // Add robots for event tests + qb := capsule.Query() + m := model.Select("__yao.member") + tableName := m.MetaData.Table.Name + + // Robot with event enabled + robotConfigEvent := map[string]interface{}{ + "identity": map[string]interface{}{ + "role": "Event Test Robot", + }, + "triggers": map[string]interface{}{ + "clock": map[string]interface{}{"enabled": false}, + "event": map[string]interface{}{"enabled": true}, + }, + "quota": map[string]interface{}{ + "max": 5, + "queue": 10, + }, + } + configEventJSON, _ := json.Marshal(robotConfigEvent) + + err := qb.Table(tableName).Insert([]map[string]interface{}{ + { + "member_id": "robot_test_manager_event", + "team_id": "team_test_manager", + "member_type": "robot", + "display_name": "Test Event Robot", + "status": "active", + "role_id": "member", + "autonomous_mode": true, + "robot_status": "idle", + "robot_config": string(configEventJSON), + }, + }) + if err != nil { + t.Fatalf("Failed to insert robot_test_manager_event: %v", err) + } + + // Robot with event disabled + robotConfigEventDisabled := map[string]interface{}{ + "identity": map[string]interface{}{ + "role": "Event Disabled Robot", + }, + "triggers": map[string]interface{}{ + "clock": map[string]interface{}{"enabled": false}, + "event": map[string]interface{}{"enabled": false}, + }, + } + configEventDisabledJSON, _ := json.Marshal(robotConfigEventDisabled) + + err = qb.Table(tableName).Insert([]map[string]interface{}{ + { + "member_id": "robot_test_manager_event_disabled", + "team_id": "team_test_manager", + "member_type": "robot", + "display_name": "Test Event Disabled Robot", + "status": "active", + "role_id": "member", + "autonomous_mode": true, + "robot_status": "idle", + "robot_config": string(configEventDisabledJSON), + }, + }) + if err != nil { + t.Fatalf("Failed to insert robot_test_manager_event_disabled: %v", err) + } +} + // cleanupTestRobots removes all test robot records func cleanupTestRobots(t *testing.T) { qb := capsule.Query() @@ -903,6 +1413,10 @@ func cleanupTestRobots(t *testing.T) { "robot_test_manager_daemon", "robot_test_manager_paused", "robot_test_manager_disabled", + "robot_test_manager_intervene", + "robot_test_manager_intervene_disabled", + "robot_test_manager_event", + "robot_test_manager_event_disabled", } for _, id := range testRobotIDs { diff --git a/agent/robot/robot.go b/agent/robot/robot.go index 1fa18c62..760fcb2e 100644 --- a/agent/robot/robot.go +++ b/agent/robot/robot.go @@ -8,7 +8,6 @@ import ( "github.com/yaoapp/yao/agent/robot/plan" "github.com/yaoapp/yao/agent/robot/pool" "github.com/yaoapp/yao/agent/robot/store" - "github.com/yaoapp/yao/agent/robot/trigger" ) var ( @@ -18,7 +17,6 @@ var ( globalPool *pool.Pool globalDedup *dedup.Dedup globalStore *store.Store - globalTrigger *trigger.Trigger globalExecutor *executor.Executor globalPlan *plan.Plan ) @@ -31,7 +29,6 @@ func Init() error { globalDedup = dedup.New() globalStore = store.New() globalPool = pool.New() // Default pool size - globalTrigger = trigger.New() globalExecutor = executor.New() globalManager = manager.New() globalPlan = plan.New() @@ -51,3 +48,8 @@ func Shutdown() error { // } return nil } + +// Manager returns the global manager instance +func Manager() *manager.Manager { + return globalManager +} diff --git a/agent/robot/trigger/clock.go b/agent/robot/trigger/clock.go new file mode 100644 index 00000000..1e81ea86 --- /dev/null +++ b/agent/robot/trigger/clock.go @@ -0,0 +1,126 @@ +package trigger + +import ( + "time" + + "github.com/yaoapp/yao/agent/robot/types" +) + +// ClockMatcher provides clock trigger matching logic +// This is extracted from Manager for reuse and testing +type ClockMatcher struct{} + +// NewClockMatcher creates a new clock matcher +func NewClockMatcher() *ClockMatcher { + return &ClockMatcher{} +} + +// ShouldTrigger checks if a robot should be triggered based on its clock config +func (cm *ClockMatcher) ShouldTrigger(robot *types.Robot, now time.Time) bool { + if robot == nil || robot.Config == nil || robot.Config.Clock == nil { + return false + } + + clock := robot.Config.Clock + + // Get time in robot's timezone + loc := clock.GetLocation() + localNow := now.In(loc) + + switch clock.Mode { + case types.ClockTimes: + return cm.shouldTriggerTimes(robot, clock, localNow) + case types.ClockInterval: + return cm.shouldTriggerInterval(robot, clock, localNow) + case types.ClockDaemon: + return cm.shouldTriggerDaemon(robot, clock, localNow) + default: + return false + } +} + +// shouldTriggerTimes checks if current time matches any configured times +// times mode: run at specific times (e.g., ["09:00", "14:00", "17:00"]) +func (cm *ClockMatcher) shouldTriggerTimes(robot *types.Robot, clock *types.Clock, now time.Time) bool { + // Check day of week first + if !cm.matchesDay(clock, now) { + return false + } + + // Check if current time matches any configured time + currentTime := now.Format("15:04") + for _, t := range clock.Times { + if t == currentTime { + // Check if already triggered in this minute + if !robot.LastRun.IsZero() { + lastRunInLoc := robot.LastRun.In(now.Location()) + if lastRunInLoc.Format("15:04") == currentTime && lastRunInLoc.Day() == now.Day() { + return false // Already triggered this minute today + } + } + return true + } + } + return false +} + +// shouldTriggerInterval checks if enough time has passed since last run +// interval mode: run every X duration (e.g., "30m", "2h") +func (cm *ClockMatcher) shouldTriggerInterval(robot *types.Robot, clock *types.Clock, now time.Time) bool { + interval, err := time.ParseDuration(clock.Every) + if err != nil { + return false + } + + // First run if never executed + if robot.LastRun.IsZero() { + return true + } + + // Check if interval has passed + return now.Sub(robot.LastRun) >= interval +} + +// shouldTriggerDaemon checks if robot can restart immediately after last run +// daemon mode: restart immediately after each run completes +func (cm *ClockMatcher) shouldTriggerDaemon(robot *types.Robot, clock *types.Clock, now time.Time) bool { + // Daemon mode: trigger if not currently running + // CanRun() checks if robot has available execution slots + return robot.CanRun() +} + +// matchesDay checks if current day matches the configured days +func (cm *ClockMatcher) matchesDay(clock *types.Clock, now time.Time) bool { + // Empty days or ["*"] means all days + if len(clock.Days) == 0 { + return true + } + + for _, day := range clock.Days { + if day == "*" { + return true + } + // Match day name (Mon, Tue, Wed, Thu, Fri, Sat, Sun) + // or full name (Monday, Tuesday, etc.) + weekday := now.Weekday().String() + shortDay := weekday[:3] // Mon, Tue, etc. + if day == weekday || day == shortDay { + return true + } + } + return false +} + +// ParseTime parses a time string in "HH:MM" format +func ParseTime(timeStr string) (hour, minute int, err error) { + t, err := time.Parse("15:04", timeStr) + if err != nil { + return 0, 0, err + } + return t.Hour(), t.Minute(), nil +} + +// FormatTime formats hour and minute to "HH:MM" string +func FormatTime(hour, minute int) string { + return time.Date(0, 1, 1, hour, minute, 0, 0, time.UTC).Format("15:04") +} diff --git a/agent/robot/trigger/clock_test.go b/agent/robot/trigger/clock_test.go new file mode 100644 index 00000000..33c189db --- /dev/null +++ b/agent/robot/trigger/clock_test.go @@ -0,0 +1,426 @@ +package trigger_test + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/yaoapp/yao/agent/robot/trigger" + "github.com/yaoapp/yao/agent/robot/types" +) + +// ==================== ClockMatcher Tests ==================== + +func TestClockMatcherShouldTrigger(t *testing.T) { + cm := trigger.NewClockMatcher() + + t.Run("nil robot returns false", func(t *testing.T) { + result := cm.ShouldTrigger(nil, time.Now()) + assert.False(t, result) + }) + + t.Run("nil config returns false", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: nil, + } + result := cm.ShouldTrigger(robot, time.Now()) + assert.False(t, result) + }) + + t.Run("nil clock config returns false", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{Clock: nil}, + } + result := cm.ShouldTrigger(robot, time.Now()) + assert.False(t, result) + }) +} + +// ==================== Times Mode Tests ==================== + +func TestClockMatcherTimesMode(t *testing.T) { + cm := trigger.NewClockMatcher() + + t.Run("matches configured time", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockTimes, + Times: []string{"09:00", "14:00", "17:00"}, + Days: []string{"*"}, + TZ: "UTC", + }, + }, + } + + // Create time at 09:00 UTC + now := time.Date(2025, 1, 15, 9, 0, 0, 0, time.UTC) + result := cm.ShouldTrigger(robot, now) + assert.True(t, result) + }) + + t.Run("does not match non-configured time", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockTimes, + Times: []string{"09:00", "14:00", "17:00"}, + Days: []string{"*"}, + TZ: "UTC", + }, + }, + } + + // Create time at 10:00 UTC (not in configured times) + now := time.Date(2025, 1, 15, 10, 0, 0, 0, time.UTC) + result := cm.ShouldTrigger(robot, now) + assert.False(t, result) + }) + + t.Run("respects day filter - weekday", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockTimes, + Times: []string{"09:00"}, + Days: []string{"Mon", "Tue", "Wed", "Thu", "Fri"}, + TZ: "UTC", + }, + }, + } + + // Wednesday 09:00 - should trigger + wed := time.Date(2025, 1, 15, 9, 0, 0, 0, time.UTC) // Wednesday + assert.True(t, cm.ShouldTrigger(robot, wed)) + + // Saturday 09:00 - should NOT trigger + sat := time.Date(2025, 1, 18, 9, 0, 0, 0, time.UTC) // Saturday + assert.False(t, cm.ShouldTrigger(robot, sat)) + }) + + t.Run("dedup - same minute same day should not trigger twice", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockTimes, + Times: []string{"09:00"}, + Days: []string{"*"}, + TZ: "UTC", + }, + }, + } + + now := time.Date(2025, 1, 15, 9, 0, 0, 0, time.UTC) + + // First trigger - should succeed + assert.True(t, cm.ShouldTrigger(robot, now)) + + // Simulate LastRun was set + robot.LastRun = now + + // Second trigger same minute - should fail + now2 := time.Date(2025, 1, 15, 9, 0, 30, 0, time.UTC) + assert.False(t, cm.ShouldTrigger(robot, now2)) + }) + + t.Run("different day should trigger again", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockTimes, + Times: []string{"09:00"}, + Days: []string{"*"}, + TZ: "UTC", + }, + }, + } + + // First day + day1 := time.Date(2025, 1, 15, 9, 0, 0, 0, time.UTC) + robot.LastRun = day1 + + // Next day same time - should trigger + day2 := time.Date(2025, 1, 16, 9, 0, 0, 0, time.UTC) + assert.True(t, cm.ShouldTrigger(robot, day2)) + }) +} + +// ==================== Interval Mode Tests ==================== + +func TestClockMatcherIntervalMode(t *testing.T) { + cm := trigger.NewClockMatcher() + + t.Run("first run triggers immediately", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockInterval, + Every: "30m", + TZ: "UTC", + }, + }, + } + + // LastRun is zero - should trigger + now := time.Now() + result := cm.ShouldTrigger(robot, now) + assert.True(t, result) + }) + + t.Run("triggers after interval passed", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockInterval, + Every: "30m", + TZ: "UTC", + }, + }, + } + + now := time.Now() + robot.LastRun = now.Add(-31 * time.Minute) // 31 minutes ago + + result := cm.ShouldTrigger(robot, now) + assert.True(t, result) + }) + + t.Run("does not trigger before interval", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockInterval, + Every: "30m", + TZ: "UTC", + }, + }, + } + + now := time.Now() + robot.LastRun = now.Add(-15 * time.Minute) // Only 15 minutes ago + + result := cm.ShouldTrigger(robot, now) + assert.False(t, result) + }) + + t.Run("invalid interval format returns false", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockInterval, + Every: "invalid", + TZ: "UTC", + }, + }, + } + + result := cm.ShouldTrigger(robot, time.Now()) + assert.False(t, result) + }) + + t.Run("various interval formats", func(t *testing.T) { + intervals := []struct { + every string + lastAgo time.Duration + expected bool + }{ + {"1h", 61 * time.Minute, true}, + {"1h", 30 * time.Minute, false}, + {"2h", 121 * time.Minute, true}, + {"2h", 60 * time.Minute, false}, + {"10s", 11 * time.Second, true}, + {"10s", 5 * time.Second, false}, + } + + for _, tt := range intervals { + t.Run(tt.every, func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockInterval, + Every: tt.every, + TZ: "UTC", + }, + }, + } + + now := time.Now() + robot.LastRun = now.Add(-tt.lastAgo) + + result := cm.ShouldTrigger(robot, now) + assert.Equal(t, tt.expected, result) + }) + } + }) +} + +// ==================== Daemon Mode Tests ==================== + +func TestClockMatcherDaemonMode(t *testing.T) { + cm := trigger.NewClockMatcher() + + t.Run("triggers when robot can run", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockDaemon, + TZ: "UTC", + }, + Quota: &types.Quota{Max: 2}, + }, + } + + // No running executions - should trigger + result := cm.ShouldTrigger(robot, time.Now()) + assert.True(t, result) + }) + + t.Run("does not trigger when at quota", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockDaemon, + TZ: "UTC", + }, + Quota: &types.Quota{Max: 1}, + }, + } + + // Add one execution to fill quota + exec := &types.Execution{ID: "exec_001"} + robot.AddExecution(exec) + + result := cm.ShouldTrigger(robot, time.Now()) + assert.False(t, result) + + // Remove execution + robot.RemoveExecution("exec_001") + + // Now should trigger + result = cm.ShouldTrigger(robot, time.Now()) + assert.True(t, result) + }) +} + +// ==================== Timezone Tests ==================== + +func TestClockMatcherTimezone(t *testing.T) { + cm := trigger.NewClockMatcher() + + t.Run("respects timezone for times mode", func(t *testing.T) { + // Robot configured for Asia/Shanghai (UTC+8) + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockTimes, + Times: []string{"09:00"}, + Days: []string{"*"}, + TZ: "Asia/Shanghai", + }, + }, + } + + // 01:00 UTC = 09:00 Shanghai - should trigger + utc0100 := time.Date(2025, 1, 15, 1, 0, 0, 0, time.UTC) + assert.True(t, cm.ShouldTrigger(robot, utc0100)) + + // 09:00 UTC = 17:00 Shanghai - should NOT trigger + utc0900 := time.Date(2025, 1, 15, 9, 0, 0, 0, time.UTC) + assert.False(t, cm.ShouldTrigger(robot, utc0900)) + }) + + t.Run("invalid timezone falls back to local", func(t *testing.T) { + robot := &types.Robot{ + MemberID: "robot_001", + Config: &types.Config{ + Clock: &types.Clock{ + Mode: types.ClockTimes, + Times: []string{"09:00"}, + Days: []string{"*"}, + TZ: "Invalid/Timezone", + }, + }, + } + + // Should still work with local time + local0900 := time.Date(2025, 1, 15, 9, 0, 0, 0, time.Local) + result := cm.ShouldTrigger(robot, local0900) + // Result depends on local timezone, just verify no panic + assert.IsType(t, true, result) + }) +} + +// ==================== ParseTime/FormatTime Tests ==================== + +func TestParseTime(t *testing.T) { + t.Run("parses valid time", func(t *testing.T) { + hour, minute, err := trigger.ParseTime("09:30") + assert.NoError(t, err) + assert.Equal(t, 9, hour) + assert.Equal(t, 30, minute) + }) + + t.Run("parses midnight", func(t *testing.T) { + hour, minute, err := trigger.ParseTime("00:00") + assert.NoError(t, err) + assert.Equal(t, 0, hour) + assert.Equal(t, 0, minute) + }) + + t.Run("parses 23:59", func(t *testing.T) { + hour, minute, err := trigger.ParseTime("23:59") + assert.NoError(t, err) + assert.Equal(t, 23, hour) + assert.Equal(t, 59, minute) + }) + + t.Run("invalid format returns error", func(t *testing.T) { + // Note: time.Parse("15:04", "9:30") actually succeeds + // Only truly invalid formats fail + + _, _, err := trigger.ParseTime("09:30:00") + assert.Error(t, err) + + _, _, err = trigger.ParseTime("invalid") + assert.Error(t, err) + + _, _, err = trigger.ParseTime("") + assert.Error(t, err) + }) +} + +func TestFormatTime(t *testing.T) { + tests := []struct { + hour int + minute int + expected string + }{ + {9, 0, "09:00"}, + {9, 30, "09:30"}, + {0, 0, "00:00"}, + {23, 59, "23:59"}, + {14, 5, "14:05"}, + } + + for _, tt := range tests { + t.Run(tt.expected, func(t *testing.T) { + result := trigger.FormatTime(tt.hour, tt.minute) + assert.Equal(t, tt.expected, result) + }) + } +} diff --git a/agent/robot/trigger/control.go b/agent/robot/trigger/control.go new file mode 100644 index 00000000..19c05591 --- /dev/null +++ b/agent/robot/trigger/control.go @@ -0,0 +1,254 @@ +package trigger + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/yaoapp/yao/agent/robot/types" +) + +// ExecutionController manages execution lifecycle (pause/resume/stop) +type ExecutionController struct { + executions map[string]*ControlledExecution + mu sync.RWMutex +} + +// ControlledExecution represents an execution that can be controlled +type ControlledExecution struct { + ID string + MemberID string + TeamID string + Status types.ExecStatus + Phase types.Phase + StartTime time.Time + PausedAt *time.Time + + // Control channels + ctx context.Context + cancel context.CancelFunc + paused bool + pauseMu sync.Mutex + pauseCh chan struct{} // closed when paused, recreated on resume +} + +// NewExecutionController creates a new execution controller +func NewExecutionController() *ExecutionController { + return &ExecutionController{ + executions: make(map[string]*ControlledExecution), + } +} + +// Track starts tracking an execution +func (c *ExecutionController) Track(execID, memberID, teamID string) *ControlledExecution { + c.mu.Lock() + defer c.mu.Unlock() + + ctx, cancel := context.WithCancel(context.Background()) + exec := &ControlledExecution{ + ID: execID, + MemberID: memberID, + TeamID: teamID, + Status: types.ExecRunning, + Phase: types.PhaseInspiration, + StartTime: time.Now(), + ctx: ctx, + cancel: cancel, + paused: false, + pauseCh: make(chan struct{}), + } + + c.executions[execID] = exec + return exec +} + +// Untrack stops tracking an execution +func (c *ExecutionController) Untrack(execID string) { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.executions, execID) +} + +// Get returns a tracked execution +func (c *ExecutionController) Get(execID string) *ControlledExecution { + c.mu.RLock() + defer c.mu.RUnlock() + return c.executions[execID] +} + +// List returns all tracked executions +func (c *ExecutionController) List() []*ControlledExecution { + c.mu.RLock() + defer c.mu.RUnlock() + + result := make([]*ControlledExecution, 0, len(c.executions)) + for _, exec := range c.executions { + result = append(result, exec) + } + return result +} + +// ListByMember returns all executions for a specific member +func (c *ExecutionController) ListByMember(memberID string) []*ControlledExecution { + c.mu.RLock() + defer c.mu.RUnlock() + + var result []*ControlledExecution + for _, exec := range c.executions { + if exec.MemberID == memberID { + result = append(result, exec) + } + } + return result +} + +// Pause pauses an execution +func (c *ExecutionController) Pause(execID string) error { + exec := c.Get(execID) + if exec == nil { + return fmt.Errorf("execution not found: %s", execID) + } + + exec.pauseMu.Lock() + defer exec.pauseMu.Unlock() + + if exec.paused { + return fmt.Errorf("execution already paused: %s", execID) + } + + exec.paused = true + now := time.Now() + exec.PausedAt = &now + + // Close the pause channel to signal pause + close(exec.pauseCh) + + return nil +} + +// Resume resumes a paused execution +func (c *ExecutionController) Resume(execID string) error { + exec := c.Get(execID) + if exec == nil { + return fmt.Errorf("execution not found: %s", execID) + } + + exec.pauseMu.Lock() + defer exec.pauseMu.Unlock() + + if !exec.paused { + return fmt.Errorf("execution not paused: %s", execID) + } + + exec.paused = false + exec.PausedAt = nil + + // Create new pause channel for future pauses + exec.pauseCh = make(chan struct{}) + + return nil +} + +// Stop stops an execution +func (c *ExecutionController) Stop(execID string) error { + c.mu.Lock() + defer c.mu.Unlock() + + exec, ok := c.executions[execID] + if !ok { + return fmt.Errorf("execution not found: %s", execID) + } + + // Cancel the context to signal stop + if exec.cancel != nil { + exec.cancel() + } + + exec.Status = types.ExecCancelled + + // Remove from tracking + delete(c.executions, execID) + + return nil +} + +// ==================== ControlledExecution methods ==================== + +// IsPaused returns true if the execution is paused +func (e *ControlledExecution) IsPaused() bool { + e.pauseMu.Lock() + defer e.pauseMu.Unlock() + return e.paused +} + +// IsCancelled returns true if the execution is cancelled +func (e *ControlledExecution) IsCancelled() bool { + select { + case <-e.ctx.Done(): + return true + default: + return false + } +} + +// Context returns the execution's context +func (e *ControlledExecution) Context() context.Context { + return e.ctx +} + +// WaitIfPaused blocks until the execution is resumed or cancelled +// Returns error if cancelled +func (e *ControlledExecution) WaitIfPaused() error { + e.pauseMu.Lock() + paused := e.paused + pauseCh := e.pauseCh + e.pauseMu.Unlock() + + if !paused { + return nil + } + + // Wait for resume (new pauseCh created) or cancel + select { + case <-e.ctx.Done(): + return types.ErrExecutionCancelled + case <-pauseCh: + // Pause channel closed, check if we're still paused + // If still paused, this was the pause signal; wait for resume + for { + e.pauseMu.Lock() + if !e.paused { + e.pauseMu.Unlock() + return nil + } + newPauseCh := e.pauseCh + e.pauseMu.Unlock() + + select { + case <-e.ctx.Done(): + return types.ErrExecutionCancelled + case <-newPauseCh: + // Channel closed again, loop to check state + } + } + } +} + +// CheckCancelled checks if the execution is cancelled and returns error if so +func (e *ControlledExecution) CheckCancelled() error { + if e.IsCancelled() { + return types.ErrExecutionCancelled + } + return nil +} + +// UpdatePhase updates the current phase +func (e *ControlledExecution) UpdatePhase(phase types.Phase) { + e.Phase = phase +} + +// UpdateStatus updates the execution status +func (e *ControlledExecution) UpdateStatus(status types.ExecStatus) { + e.Status = status +} diff --git a/agent/robot/trigger/control_test.go b/agent/robot/trigger/control_test.go new file mode 100644 index 00000000..10ac834b --- /dev/null +++ b/agent/robot/trigger/control_test.go @@ -0,0 +1,408 @@ +package trigger_test + +import ( + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/yaoapp/yao/agent/robot/trigger" + "github.com/yaoapp/yao/agent/robot/types" +) + +// ==================== ExecutionController Tests ==================== + +func TestExecutionControllerTrack(t *testing.T) { + t.Run("tracks new execution", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + assert.NotNil(t, exec) + assert.Equal(t, "exec_001", exec.ID) + assert.Equal(t, "robot_001", exec.MemberID) + assert.Equal(t, "team_001", exec.TeamID) + assert.Equal(t, types.ExecRunning, exec.Status) + assert.False(t, exec.IsPaused()) + assert.False(t, exec.IsCancelled()) + }) + + t.Run("get tracked execution", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + ctrl.Track("exec_001", "robot_001", "team_001") + + exec := ctrl.Get("exec_001") + assert.NotNil(t, exec) + assert.Equal(t, "exec_001", exec.ID) + }) + + t.Run("get non-existent execution returns nil", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + + exec := ctrl.Get("non_existent") + assert.Nil(t, exec) + }) +} + +func TestExecutionControllerList(t *testing.T) { + t.Run("list all executions", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + ctrl.Track("exec_001", "robot_001", "team_001") + ctrl.Track("exec_002", "robot_002", "team_001") + ctrl.Track("exec_003", "robot_001", "team_002") + + list := ctrl.List() + assert.Len(t, list, 3) + }) + + t.Run("list by member", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + ctrl.Track("exec_001", "robot_001", "team_001") + ctrl.Track("exec_002", "robot_002", "team_001") + ctrl.Track("exec_003", "robot_001", "team_002") + + list := ctrl.ListByMember("robot_001") + assert.Len(t, list, 2) + + list = ctrl.ListByMember("robot_002") + assert.Len(t, list, 1) + + list = ctrl.ListByMember("robot_003") + assert.Len(t, list, 0) + }) +} + +func TestExecutionControllerUntrack(t *testing.T) { + t.Run("untrack removes execution", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + ctrl.Track("exec_001", "robot_001", "team_001") + + assert.NotNil(t, ctrl.Get("exec_001")) + + ctrl.Untrack("exec_001") + + assert.Nil(t, ctrl.Get("exec_001")) + }) + + t.Run("untrack non-existent does not panic", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + + assert.NotPanics(t, func() { + ctrl.Untrack("non_existent") + }) + }) +} + +// ==================== Pause/Resume Tests ==================== + +func TestExecutionControllerPause(t *testing.T) { + t.Run("pause execution", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + err := ctrl.Pause("exec_001") + assert.NoError(t, err) + assert.True(t, exec.IsPaused()) + assert.NotNil(t, exec.PausedAt) + }) + + t.Run("pause non-existent returns error", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + + err := ctrl.Pause("non_existent") + assert.Error(t, err) + assert.Contains(t, err.Error(), "not found") + }) + + t.Run("pause already paused returns error", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + ctrl.Track("exec_001", "robot_001", "team_001") + + err := ctrl.Pause("exec_001") + assert.NoError(t, err) + + err = ctrl.Pause("exec_001") + assert.Error(t, err) + assert.Contains(t, err.Error(), "already paused") + }) +} + +func TestExecutionControllerResume(t *testing.T) { + t.Run("resume paused execution", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + ctrl.Pause("exec_001") + assert.True(t, exec.IsPaused()) + + err := ctrl.Resume("exec_001") + assert.NoError(t, err) + assert.False(t, exec.IsPaused()) + assert.Nil(t, exec.PausedAt) + }) + + t.Run("resume non-existent returns error", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + + err := ctrl.Resume("non_existent") + assert.Error(t, err) + assert.Contains(t, err.Error(), "not found") + }) + + t.Run("resume not paused returns error", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + ctrl.Track("exec_001", "robot_001", "team_001") + + err := ctrl.Resume("exec_001") + assert.Error(t, err) + assert.Contains(t, err.Error(), "not paused") + }) +} + +// ==================== Stop Tests ==================== + +func TestExecutionControllerStop(t *testing.T) { + t.Run("stop execution", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + err := ctrl.Stop("exec_001") + assert.NoError(t, err) + assert.True(t, exec.IsCancelled()) + assert.Equal(t, types.ExecCancelled, exec.Status) + + // Should be removed from tracking + assert.Nil(t, ctrl.Get("exec_001")) + }) + + t.Run("stop non-existent returns error", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + + err := ctrl.Stop("non_existent") + assert.Error(t, err) + assert.Contains(t, err.Error(), "not found") + }) +} + +// ==================== ControlledExecution Methods Tests ==================== + +func TestControlledExecutionContext(t *testing.T) { + t.Run("context is valid", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + ctx := exec.Context() + assert.NotNil(t, ctx) + + // Context should not be done yet + select { + case <-ctx.Done(): + t.Fatal("context should not be done") + default: + // OK + } + }) + + t.Run("context done after stop", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + ctx := exec.Context() + ctrl.Stop("exec_001") + + select { + case <-ctx.Done(): + // OK + case <-time.After(100 * time.Millisecond): + t.Fatal("context should be done after stop") + } + }) +} + +func TestControlledExecutionCheckCancelled(t *testing.T) { + t.Run("not cancelled returns nil", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + err := exec.CheckCancelled() + assert.NoError(t, err) + }) + + t.Run("cancelled returns error", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + ctrl.Stop("exec_001") + + err := exec.CheckCancelled() + assert.Error(t, err) + assert.Equal(t, types.ErrExecutionCancelled, err) + }) +} + +func TestControlledExecutionUpdatePhase(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + assert.Equal(t, types.PhaseInspiration, exec.Phase) + + exec.UpdatePhase(types.PhaseGoals) + assert.Equal(t, types.PhaseGoals, exec.Phase) + + exec.UpdatePhase(types.PhaseTasks) + assert.Equal(t, types.PhaseTasks, exec.Phase) +} + +func TestControlledExecutionUpdateStatus(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + assert.Equal(t, types.ExecRunning, exec.Status) + + exec.UpdateStatus(types.ExecCompleted) + assert.Equal(t, types.ExecCompleted, exec.Status) + + exec.UpdateStatus(types.ExecFailed) + assert.Equal(t, types.ExecFailed, exec.Status) +} + +// ==================== WaitIfPaused Tests ==================== + +func TestControlledExecutionWaitIfPaused(t *testing.T) { + t.Run("returns immediately if not paused", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + done := make(chan error) + go func() { + done <- exec.WaitIfPaused() + }() + + select { + case err := <-done: + assert.NoError(t, err) + case <-time.After(100 * time.Millisecond): + t.Fatal("WaitIfPaused should return immediately when not paused") + } + }) + + t.Run("blocks when paused, resumes after resume", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + ctrl.Pause("exec_001") + + done := make(chan error) + go func() { + done <- exec.WaitIfPaused() + }() + + // Should be blocked + select { + case <-done: + t.Fatal("WaitIfPaused should block when paused") + case <-time.After(50 * time.Millisecond): + // OK, still blocked + } + + // Resume + ctrl.Resume("exec_001") + + // Should unblock + select { + case err := <-done: + assert.NoError(t, err) + case <-time.After(100 * time.Millisecond): + t.Fatal("WaitIfPaused should unblock after resume") + } + }) + + t.Run("returns error when cancelled while paused", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + exec := ctrl.Track("exec_001", "robot_001", "team_001") + + ctrl.Pause("exec_001") + + done := make(chan error) + go func() { + done <- exec.WaitIfPaused() + }() + + // Should be blocked + select { + case <-done: + t.Fatal("WaitIfPaused should block when paused") + case <-time.After(50 * time.Millisecond): + // OK, still blocked + } + + // Stop instead of resume + ctrl.Stop("exec_001") + + // Should unblock with error + select { + case err := <-done: + assert.Error(t, err) + assert.Equal(t, types.ErrExecutionCancelled, err) + case <-time.After(100 * time.Millisecond): + t.Fatal("WaitIfPaused should unblock after stop") + } + }) +} + +// ==================== Concurrent Access Tests ==================== + +func TestExecutionControllerConcurrency(t *testing.T) { + t.Run("concurrent track and list", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + var wg sync.WaitGroup + + // Concurrent tracking + for i := 0; i < 100; i++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + ctrl.Track( + "exec_"+string(rune('0'+id%10)), + "robot_"+string(rune('0'+id%5)), + "team_001", + ) + }(i) + } + + // Concurrent listing + for i := 0; i < 50; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _ = ctrl.List() + }() + } + + wg.Wait() + // No race conditions or panics + }) + + t.Run("concurrent pause/resume", func(t *testing.T) { + ctrl := trigger.NewExecutionController() + ctrl.Track("exec_001", "robot_001", "team_001") + + var wg sync.WaitGroup + + // Concurrent pause/resume attempts + for i := 0; i < 50; i++ { + wg.Add(2) + go func() { + defer wg.Done() + _ = ctrl.Pause("exec_001") + }() + go func() { + defer wg.Done() + _ = ctrl.Resume("exec_001") + }() + } + + wg.Wait() + // No race conditions or panics + }) +} diff --git a/agent/robot/trigger/trigger.go b/agent/robot/trigger/trigger.go index e27887c7..1986b032 100644 --- a/agent/robot/trigger/trigger.go +++ b/agent/robot/trigger/trigger.go @@ -1,48 +1,142 @@ +// Package trigger provides trigger-related utilities and execution control +// The main trigger logic is in the manager package. +// This package provides: +// - Validation functions for intervention and event requests +// - Builder helpers for TriggerInput +// - ExecutionController for pause/resume/stop +// - ClockMatcher for clock trigger matching (reusable) package trigger -import "github.com/yaoapp/yao/agent/robot/types" +import ( + "fmt" -// Trigger handles all trigger sources -// This is a stub implementation for Phase 2 -type Trigger struct{} + "github.com/yaoapp/yao/agent/robot/types" +) -// New creates a new trigger instance -func New() *Trigger { - return &Trigger{} -} +// ValidateIntervention validates a human intervention request +func ValidateIntervention(req *types.InterveneRequest) error { + if req == nil { + return fmt.Errorf("request is nil") + } + + if req.MemberID == "" { + return fmt.Errorf("member_id is required") + } + + if !isValidAction(req.Action) { + return fmt.Errorf("invalid action: %s", req.Action) + } + + // Validate action-specific requirements + switch req.Action { + case types.ActionTaskAdd, types.ActionGoalAdd, types.ActionInstruct: + // These actions require messages + if len(req.Messages) == 0 { + return fmt.Errorf("messages required for action: %s", req.Action) + } + + case types.ActionPlanAdd: + // Plan add requires plan_time + if req.PlanTime == nil { + return fmt.Errorf("plan_time required for action: plan.add") + } + } -// Clock processes clock trigger -// Stub: returns nil (will be implemented in Phase 3) -func (t *Trigger) Clock(ctx *types.Context, robot *types.Robot) error { return nil } -// Intervene processes human intervention -// Stub: returns empty result (will be implemented in Phase 3) -func (t *Trigger) Intervene(ctx *types.Context, req *types.InterveneRequest) (*types.ExecutionResult, error) { - return &types.ExecutionResult{}, nil -} +// ValidateEvent validates an event trigger request +func ValidateEvent(req *types.EventRequest) error { + if req == nil { + return fmt.Errorf("request is nil") + } -// Event processes event trigger -// Stub: returns empty result (will be implemented in Phase 3) -func (t *Trigger) Event(ctx *types.Context, req *types.EventRequest) (*types.ExecutionResult, error) { - return &types.ExecutionResult{}, nil -} + if req.MemberID == "" { + return fmt.Errorf("member_id is required") + } + + if req.Source == "" { + return fmt.Errorf("source is required") + } + + if req.EventType == "" { + return fmt.Errorf("event_type is required") + } -// Pause pauses a running execution -// Stub: returns nil (will be implemented in Phase 3) -func (t *Trigger) Pause(ctx *types.Context, execID string) error { return nil } -// Resume resumes a paused execution -// Stub: returns nil (will be implemented in Phase 3) -func (t *Trigger) Resume(ctx *types.Context, execID string) error { - return nil +// BuildEventInput creates a TriggerInput from an event request +func BuildEventInput(req *types.EventRequest) *types.TriggerInput { + return &types.TriggerInput{ + Source: types.EventSource(req.Source), + EventType: req.EventType, + Data: req.Data, + } } -// Stop stops a running execution -// Stub: returns nil (will be implemented in Phase 3) -func (t *Trigger) Stop(ctx *types.Context, execID string) error { - return nil +// isValidAction checks if the intervention action is valid +func isValidAction(action types.InterventionAction) bool { + switch action { + case types.ActionTaskAdd, + types.ActionTaskCancel, + types.ActionTaskUpdate, + types.ActionGoalAdjust, + types.ActionGoalAdd, + types.ActionGoalComplete, + types.ActionGoalCancel, + types.ActionPlanAdd, + types.ActionPlanRemove, + types.ActionPlanUpdate, + types.ActionInstruct: + return true + default: + return false + } +} + +// GetActionCategory returns the category of an intervention action +func GetActionCategory(action types.InterventionAction) string { + switch action { + case types.ActionTaskAdd, types.ActionTaskCancel, types.ActionTaskUpdate: + return "task" + case types.ActionGoalAdjust, types.ActionGoalAdd, types.ActionGoalComplete, types.ActionGoalCancel: + return "goal" + case types.ActionPlanAdd, types.ActionPlanRemove, types.ActionPlanUpdate: + return "plan" + case types.ActionInstruct: + return "instruct" + default: + return "unknown" + } +} + +// GetActionDescription returns a human-readable description of an action +func GetActionDescription(action types.InterventionAction) string { + switch action { + case types.ActionTaskAdd: + return "Add a new task" + case types.ActionTaskCancel: + return "Cancel a task" + case types.ActionTaskUpdate: + return "Update task details" + case types.ActionGoalAdjust: + return "Adjust current goal" + case types.ActionGoalAdd: + return "Add a new goal" + case types.ActionGoalComplete: + return "Mark goal as complete" + case types.ActionGoalCancel: + return "Cancel a goal" + case types.ActionPlanAdd: + return "Add to plan queue" + case types.ActionPlanRemove: + return "Remove from plan queue" + case types.ActionPlanUpdate: + return "Update planned item" + case types.ActionInstruct: + return "Direct instruction to robot" + default: + return "Unknown action" + } } diff --git a/agent/robot/trigger/trigger_test.go b/agent/robot/trigger/trigger_test.go new file mode 100644 index 00000000..c00b0dd3 --- /dev/null +++ b/agent/robot/trigger/trigger_test.go @@ -0,0 +1,276 @@ +package trigger_test + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + agentcontext "github.com/yaoapp/yao/agent/context" + "github.com/yaoapp/yao/agent/robot/trigger" + "github.com/yaoapp/yao/agent/robot/types" +) + +// ==================== ValidateIntervention Tests ==================== + +func TestValidateIntervention(t *testing.T) { + t.Run("nil request returns error", func(t *testing.T) { + err := trigger.ValidateIntervention(nil) + assert.Error(t, err) + assert.Contains(t, err.Error(), "request is nil") + }) + + t.Run("empty member_id returns error", func(t *testing.T) { + req := &types.InterveneRequest{ + MemberID: "", + Action: types.ActionTaskAdd, + } + err := trigger.ValidateIntervention(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "member_id is required") + }) + + t.Run("invalid action returns error", func(t *testing.T) { + req := &types.InterveneRequest{ + MemberID: "robot_001", + Action: types.InterventionAction("invalid.action"), + } + err := trigger.ValidateIntervention(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "invalid action") + }) + + t.Run("task.add without messages returns error", func(t *testing.T) { + req := &types.InterveneRequest{ + MemberID: "robot_001", + Action: types.ActionTaskAdd, + Messages: nil, + } + err := trigger.ValidateIntervention(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "messages required") + }) + + t.Run("goal.add without messages returns error", func(t *testing.T) { + req := &types.InterveneRequest{ + MemberID: "robot_001", + Action: types.ActionGoalAdd, + Messages: nil, + } + err := trigger.ValidateIntervention(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "messages required") + }) + + t.Run("instruct without messages returns error", func(t *testing.T) { + req := &types.InterveneRequest{ + MemberID: "robot_001", + Action: types.ActionInstruct, + Messages: nil, + } + err := trigger.ValidateIntervention(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "messages required") + }) + + t.Run("plan.add without plan_time returns error", func(t *testing.T) { + req := &types.InterveneRequest{ + MemberID: "robot_001", + Action: types.ActionPlanAdd, + PlanTime: nil, + } + err := trigger.ValidateIntervention(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "plan_time required") + }) + + t.Run("valid task.add request passes", func(t *testing.T) { + req := &types.InterveneRequest{ + MemberID: "robot_001", + Action: types.ActionTaskAdd, + Messages: []agentcontext.Message{ + {Role: agentcontext.RoleUser, Content: "Add a new task"}, + }, + } + err := trigger.ValidateIntervention(req) + assert.NoError(t, err) + }) + + t.Run("valid plan.add request passes", func(t *testing.T) { + planTime := time.Now().Add(time.Hour) + req := &types.InterveneRequest{ + MemberID: "robot_001", + Action: types.ActionPlanAdd, + PlanTime: &planTime, + } + err := trigger.ValidateIntervention(req) + assert.NoError(t, err) + }) + + t.Run("task.cancel without messages passes", func(t *testing.T) { + req := &types.InterveneRequest{ + MemberID: "robot_001", + Action: types.ActionTaskCancel, + } + err := trigger.ValidateIntervention(req) + assert.NoError(t, err) + }) + + t.Run("goal.adjust without messages passes", func(t *testing.T) { + req := &types.InterveneRequest{ + MemberID: "robot_001", + Action: types.ActionGoalAdjust, + } + err := trigger.ValidateIntervention(req) + assert.NoError(t, err) + }) +} + +// ==================== ValidateEvent Tests ==================== + +func TestValidateEvent(t *testing.T) { + t.Run("nil request returns error", func(t *testing.T) { + err := trigger.ValidateEvent(nil) + assert.Error(t, err) + assert.Contains(t, err.Error(), "request is nil") + }) + + t.Run("empty member_id returns error", func(t *testing.T) { + req := &types.EventRequest{ + MemberID: "", + Source: "webhook", + EventType: "lead.created", + } + err := trigger.ValidateEvent(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "member_id is required") + }) + + t.Run("empty source returns error", func(t *testing.T) { + req := &types.EventRequest{ + MemberID: "robot_001", + Source: "", + EventType: "lead.created", + } + err := trigger.ValidateEvent(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "source is required") + }) + + t.Run("empty event_type returns error", func(t *testing.T) { + req := &types.EventRequest{ + MemberID: "robot_001", + Source: "webhook", + EventType: "", + } + err := trigger.ValidateEvent(req) + assert.Error(t, err) + assert.Contains(t, err.Error(), "event_type is required") + }) + + t.Run("valid request passes", func(t *testing.T) { + req := &types.EventRequest{ + MemberID: "robot_001", + Source: "webhook", + EventType: "lead.created", + Data: map[string]interface{}{"name": "John"}, + } + err := trigger.ValidateEvent(req) + assert.NoError(t, err) + }) +} + +// ==================== BuildEventInput Tests ==================== + +func TestBuildEventInput(t *testing.T) { + t.Run("builds correct TriggerInput", func(t *testing.T) { + req := &types.EventRequest{ + MemberID: "robot_001", + Source: "webhook", + EventType: "lead.created", + Data: map[string]interface{}{"name": "John", "email": "john@example.com"}, + } + + input := trigger.BuildEventInput(req) + + assert.NotNil(t, input) + assert.Equal(t, types.EventSource("webhook"), input.Source) + assert.Equal(t, "lead.created", input.EventType) + assert.Equal(t, "John", input.Data["name"]) + assert.Equal(t, "john@example.com", input.Data["email"]) + }) + + t.Run("handles nil data", func(t *testing.T) { + req := &types.EventRequest{ + MemberID: "robot_001", + Source: "database", + EventType: "order.paid", + Data: nil, + } + + input := trigger.BuildEventInput(req) + + assert.NotNil(t, input) + assert.Equal(t, types.EventSource("database"), input.Source) + assert.Equal(t, "order.paid", input.EventType) + assert.Nil(t, input.Data) + }) +} + +// ==================== GetActionCategory Tests ==================== + +func TestGetActionCategory(t *testing.T) { + tests := []struct { + action types.InterventionAction + expected string + }{ + {types.ActionTaskAdd, "task"}, + {types.ActionTaskCancel, "task"}, + {types.ActionTaskUpdate, "task"}, + {types.ActionGoalAdjust, "goal"}, + {types.ActionGoalAdd, "goal"}, + {types.ActionGoalComplete, "goal"}, + {types.ActionGoalCancel, "goal"}, + {types.ActionPlanAdd, "plan"}, + {types.ActionPlanRemove, "plan"}, + {types.ActionPlanUpdate, "plan"}, + {types.ActionInstruct, "instruct"}, + {types.InterventionAction("unknown"), "unknown"}, + } + + for _, tt := range tests { + t.Run(string(tt.action), func(t *testing.T) { + result := trigger.GetActionCategory(tt.action) + assert.Equal(t, tt.expected, result) + }) + } +} + +// ==================== GetActionDescription Tests ==================== + +func TestGetActionDescription(t *testing.T) { + tests := []struct { + action types.InterventionAction + contains string + }{ + {types.ActionTaskAdd, "Add"}, + {types.ActionTaskCancel, "Cancel"}, + {types.ActionTaskUpdate, "Update"}, + {types.ActionGoalAdjust, "Adjust"}, + {types.ActionGoalAdd, "Add"}, + {types.ActionGoalComplete, "complete"}, + {types.ActionGoalCancel, "Cancel"}, + {types.ActionPlanAdd, "plan"}, + {types.ActionPlanRemove, "Remove"}, + {types.ActionPlanUpdate, "Update"}, + {types.ActionInstruct, "instruction"}, + {types.InterventionAction("unknown"), "Unknown"}, + } + + for _, tt := range tests { + t.Run(string(tt.action), func(t *testing.T) { + result := trigger.GetActionDescription(tt.action) + assert.NotEmpty(t, result) + assert.Contains(t, result, tt.contains) + }) + } +}