diff --git a/agent/autonomous/TECHNICAL.md b/agent/autonomous/TECHNICAL.md index fcf59bff..c7291305 100644 --- a/agent/autonomous/TECHNICAL.md +++ b/agent/autonomous/TECHNICAL.md @@ -1017,420 +1017,7 @@ type Dedup interface { --- -## 4. Key Implementations - -### 4.1 Manager Implementation - -```go -// manager.go -package autonomous - -import ( - "context" - "sync" - "time" - - "github.com/yaoapp/kun/log" - "github.com/yaoapp/yao/agent/autonomous/types" -) - -// Ensure manager implements types.Manager -var _ types.Manager = (*manager)(nil) - -type manager struct { - cache types.Cache - scheduler types.Scheduler - dedup types.Dedup - executor types.Executor - - ticker *time.Ticker - tickerMu sync.Mutex - stopCh chan struct{} - wg sync.WaitGroup -} - -// NewManager creates a new manager -func NewManager(cfg *types.SchedulerConfig) types.Manager { - return &manager{ - cache: newCache(), - scheduler: newScheduler(cfg), - dedup: newDedup(), - executor: newExecutor(), - stopCh: make(chan struct{}), - } -} - -func (m *manager) Start() error { - // Load active robots - if err := m.cache.LoadAll(context.Background()); err != nil { - return err - } - - // Start scheduler - if err := m.scheduler.Start(); err != nil { - return err - } - - // Start ticker (every minute) - m.ticker = time.NewTicker(time.Minute) - m.wg.Add(1) - go m.tickLoop() - - log.Info("Autonomous manager started with %d robots", m.cache.Count()) - return nil -} - -func (m *manager) Stop() error { - close(m.stopCh) - m.ticker.Stop() - m.wg.Wait() - return m.scheduler.Stop() -} - -func (m *manager) tickLoop() { - defer m.wg.Done() - for { - select { - case <-m.stopCh: - return - case t := <-m.ticker.C: - if err := m.Tick(context.Background(), t); err != nil { - log.Error("Tick error: %v", err) - } - } - } -} - -func (m *manager) Tick(ctx context.Context, now time.Time) error { - robots := m.cache.List("") // all teams - for _, robot := range robots { - // Skip if not autonomous or paused - if !robot.AutonomousMode || robot.Status == types.RobotPaused { - continue - } - - // Check if clock trigger is enabled - if !robot.Config.Triggers.IsEnabled(types.TriggerClock) { - continue - } - - // Check if should run now - if !m.shouldRun(robot, now) { - continue - } - - // Check dedup - result, err := m.dedup.CheckExecution(ctx, robot.MemberID, types.TriggerClock) - if err != nil { - log.Warn("Dedup check error for %s: %v", robot.MemberID, err) - continue - } - if result == types.DedupSkip { - continue - } - - // Submit to scheduler - if err := m.scheduler.Submit(ctx, robot, types.TriggerClock, nil); err != nil { - log.Warn("Submit error for %s: %v", robot.MemberID, err) - } - } - return nil -} - -func (m *manager) shouldRun(robot *types.Robot, now time.Time) bool { - cfg := robot.Config.Clock - if cfg == nil { - return false - } - - loc := cfg.GetLocation() - now = now.In(loc) - - switch cfg.Mode { - case types.ClockTimes: - return m.shouldRunTimes(cfg, now) - case types.ClockInterval: - return m.shouldRunInterval(robot, cfg, now) - case types.ClockDaemon: - return robot.CanRun() // always run if can - } - return false -} - -func (m *manager) shouldRunTimes(cfg *types.Clock, now time.Time) bool { - // Check day - if len(cfg.Days) > 0 && cfg.Days[0] != "*" { - dayMatch := false - for _, d := range cfg.Days { - if d == now.Weekday().String()[:3] { - dayMatch = true - break - } - } - if !dayMatch { - return false - } - } - - // Check time (within 1 minute window) - nowTime := now.Format("15:04") - for _, t := range cfg.Times { - if t == nowTime { - return true - } - } - return false -} - -func (m *manager) shouldRunInterval(robot *types.Robot, cfg *types.Clock, now time.Time) bool { - every, err := time.ParseDuration(cfg.Every) - if err != nil { - return false - } - return now.Sub(robot.LastExecution) >= every -} -``` - -### 4.2 Executor Implementation - -```go -// executor.go -package autonomous - -import ( - "context" - "fmt" - "time" - - gonanoid "github.com/matoous/go-nanoid/v2" - "github.com/yaoapp/kun/log" - "github.com/yaoapp/yao/agent/autonomous/types" - "github.com/yaoapp/yao/job" -) - -// Ensure executor implements types.Executor -var _ types.Executor = (*executor)(nil) - -type executor struct{} - -func newExecutor() types.Executor { - return &executor{} -} - -func (e *executor) Execute(ctx context.Context, robot *types.Robot, triggerType types.TriggerType, triggerData interface{}) (*types.Execution, error) { - // Create execution - exec := &types.Execution{ - ID: gonanoid.Must(), - MemberID: robot.MemberID, - TeamID: robot.TeamID, - TriggerType: triggerType, - TriggerData: triggerData, - StartTime: time.Now(), - Status: types.ExecRunning, - } - - // Create context with timeout - timeout := robot.Config.Clock.GetTimeout() - exec.ctx, exec.cancel = context.WithTimeout(ctx, timeout) - defer exec.cancel() - - // Update robot status - robot.IncrRunning() - defer robot.DecrRunning() - - // Determine phases to run - phases := e.getPhasesToRun(triggerType) - - // Run phases - for _, phase := range phases { - exec.Phase = phase - if err := e.RunPhase(exec.ctx, exec, phase); err != nil { - exec.Status = ExecFailed - exec.Error = err.Error() - e.saveExecution(exec) - return exec, err - } - } - - // Mark completed - now := time.Now() - exec.EndTime = &now - exec.Status = types.ExecCompleted - e.saveExecution(exec) - - return exec, nil -} - -func (e *executor) getPhasesToRun(triggerType types.TriggerType) []types.Phase { - if triggerType == types.TriggerClock { - return types.AllPhases // P0 -> P5 - } - // Human/Event: skip P0 - return []types.Phase{types.PhaseGoals, types.PhaseTasks, types.PhaseRun, types.PhaseDelivery, types.PhaseLearning} -} - -func (e *executor) RunPhase(ctx context.Context, exec *types.Execution, phase types.Phase) error { - log.Debug("Running phase %s for %s", phase, exec.MemberID) - - switch phase { - case types.PhaseInspiration: - return e.runInspiration(ctx, exec) - case types.PhaseGoals: - return e.runGoals(ctx, exec) - case types.PhaseTasks: - return e.runTasks(ctx, exec) - case types.PhaseRun: - return e.runExecution(ctx, exec) - case types.PhaseDelivery: - return e.runDelivery(ctx, exec) - case types.PhaseLearning: - return e.runLearning(ctx, exec) - default: - return fmt.Errorf("unknown phase: %s", phase) - } -} - -func (e *executor) saveExecution(exec *types.Execution) { - // Save to job system - jobExec := &job.Execution{ - ExecutionID: exec.ID, - JobID: "robot_" + exec.MemberID, - Status: string(exec.Status), - TriggerCategory: string(exec.TriggerType), - } - if exec.StartTime.IsZero() == false { - jobExec.StartedAt = &exec.StartTime - } - if exec.EndTime != nil { - jobExec.EndedAt = exec.EndTime - } - job.SaveExecution(jobExec) -} -``` - -### 4.3 Yao Process - -```go -// process.go -package autonomous - -import ( - "context" - - "github.com/yaoapp/gou/process" - "github.com/yaoapp/kun/log" - "github.com/yaoapp/yao/agent/autonomous/types" -) - -func init() { - process.Register("autonomous.Execute", processExecute) - process.Register("autonomous.Intervene", processIntervene) - process.Register("autonomous.HandleEvent", processHandleEvent) - process.Register("autonomous.GetStatus", processGetStatus) - process.Register("autonomous.Pause", processPause) - process.Register("autonomous.Resume", processResume) -} - -// processExecute - autonomous.Execute(memberID, triggerType, triggerData) -func processExecute(p *process.Process) interface{} { - memberID := p.ArgsString(0) - triggerType := types.TriggerType(p.ArgsString(1, "clock")) - triggerData := p.Args[2] - - mgr := GetManager() - robot := mgr.GetRobot("", memberID) // teamID not needed for lookup - if robot == nil { - return map[string]interface{}{"error": "robot not found"} - } - - exec, err := GetExecutor().Execute(context.Background(), robot, triggerType, triggerData) - if err != nil { - log.Error("Execute error: %v", err) - return map[string]interface{}{"error": err.Error()} - } - - return map[string]interface{}{ - "execution_id": exec.ID, - "status": exec.Status, - } -} - -// processIntervene - autonomous.Intervene(teamID, memberID, action, description, priority) -func processIntervene(p *process.Process) interface{} { - req := &types.InterveneRequest{ - TeamID: p.ArgsString(0), - MemberID: p.ArgsString(1), - Action: types.InterventionAction(p.ArgsString(2)), - Description: p.ArgsString(3), - Priority: types.Priority(p.ArgsString(4, "normal")), - } - - result, err := GetTrigger().Intervene(context.Background(), req) - if err != nil { - return map[string]interface{}{"error": err.Error()} - } - return result -} - -// processHandleEvent - autonomous.HandleEvent(memberID, source, eventType, data) -func processHandleEvent(p *process.Process) interface{} { - req := &types.EventRequest{ - MemberID: p.ArgsString(0), - Source: p.ArgsString(1), - EventType: p.ArgsString(2), - Data: p.ArgsMap(3), - } - - result, err := GetTrigger().HandleEvent(context.Background(), req) - if err != nil { - return map[string]interface{}{"error": err.Error()} - } - return result -} - -// processGetStatus - autonomous.GetStatus(teamID, memberID) -func processGetStatus(p *process.Process) interface{} { - state, err := GetTrigger().GetStatus( - context.Background(), - p.ArgsString(0), - p.ArgsString(1), - ) - if err != nil { - return map[string]interface{}{"error": err.Error()} - } - return state -} - -// processPause - autonomous.Pause(teamID, memberID) -func processPause(p *process.Process) interface{} { - err := GetTrigger().Pause( - context.Background(), - p.ArgsString(0), - p.ArgsString(1), - ) - if err != nil { - return map[string]interface{}{"error": err.Error()} - } - return map[string]interface{}{"success": true} -} - -// processResume - autonomous.Resume(teamID, memberID) -func processResume(p *process.Process) interface{} { - err := GetTrigger().Resume( - context.Background(), - p.ArgsString(0), - p.ArgsString(1), - ) - if err != nil { - return map[string]interface{}{"error": err.Error()} - } - return map[string]interface{}{"success": true} -} -``` - ---- - -## 5. Errors +## 4. Errors ```go // types/errors.go @@ -1460,191 +1047,3 @@ var ( ErrDeliveryFailed = errors.New("delivery failed") ) ``` - ---- - -## 6. Global Singletons - -```go -// global.go -package autonomous - -import ( - "sync" - - "github.com/yaoapp/yao/agent/autonomous/types" -) - -var ( - globalManager types.Manager - globalTrigger types.Trigger - globalExecutor types.Executor - globalOnce sync.Once -) - -// Init initializes the autonomous system -func Init(cfg *types.SchedulerConfig) error { - var initErr error - globalOnce.Do(func() { - mgr := NewManager(cfg) - if err := mgr.Start(); err != nil { - initErr = err - return - } - globalManager = mgr - globalTrigger = newTrigger(mgr) - globalExecutor = newExecutor() - }) - return initErr -} - -// GetManager returns the global manager -func GetManager() types.Manager { - return globalManager -} - -// GetTrigger returns the global trigger -func GetTrigger() types.Trigger { - return globalTrigger -} - -// GetExecutor returns the global executor -func GetExecutor() types.Executor { - return globalExecutor -} - -// Shutdown stops the autonomous system -func Shutdown() error { - if globalManager != nil { - return globalManager.Stop() - } - return nil -} -``` - ---- - -## 7. Integration Points - -### 7.1 With Job System - -```go -// Job creation on robot create -func createRobotJob(robot *types.Robot) error { - j, err := job.Once(job.GOROUTINE, map[string]interface{}{ - "job_id": "robot_" + robot.MemberID, - "category_id": "autonomous_robot", - "name": robot.DisplayName, - }) - if err != nil { - return err - } - return job.SaveJob(j) -} -``` - -### 7.2 With Assistant - -```go -// Call phase agent -func callPhaseAgent(ctx context.Context, agentID string, prompt string) (string, error) { - ast, err := assistant.Get(agentID) - if err != nil { - return "", err - } - - messages := []chatctx.Message{ - {Role: "user", Content: prompt}, - } - - resp, err := ast.Stream(chatctx.New(ctx), messages) - if err != nil { - return "", err - } - - return resp.Content, nil -} -``` - -### 7.3 With Member Model - -```go -// Load robot from __yao.member -func loadRobotFromMember(memberID string) (*types.Robot, error) { - mod := model.Select("__yao.member") - data, err := mod.Find(memberID, model.QueryParam{}) - if err != nil { - return nil, err - } - - robot := &types.Robot{ - MemberID: data.Get("member_id").(string), - TeamID: data.Get("team_id").(string), - DisplayName: data.Get("display_name").(string), - SystemPrompt: data.Get("system_prompt").(string), - Status: types.RobotStatus(data.Get("robot_status").(string)), - AutonomousMode: data.Get("autonomous_mode").(bool), - } - - // Parse robot_config - if cfgData := data.Get("robot_config"); cfgData != nil { - var cfg types.Config - if err := jsoniter.Unmarshal(cfgData.([]byte), &cfg); err != nil { - return nil, err - } - robot.Config = &cfg - } - - return robot, nil -} -``` - ---- - -## 8. Testing - -```go -// manager_test.go -package autonomous - -import ( - "context" - "testing" - "time" - - "github.com/yaoapp/yao/agent/autonomous/types" -) - -func TestManagerTick(t *testing.T) { - mgr := NewManager(&types.SchedulerConfig{Workers: 2}) - defer mgr.Stop() - - // Add test robot - robot := &types.Robot{ - MemberID: "test_robot", - TeamID: "test_team", - AutonomousMode: true, - Status: types.RobotIdle, - Config: &types.Config{ - Clock: &types.Clock{ - Mode: types.ClockTimes, - Times: []string{"09:00"}, - Days: []string{"*"}, - }, - Identity: &types.Identity{Role: "Test"}, - }, - } - - mgr.(*manager).cache.Add(robot) - - // Tick at 09:00 - now := time.Date(2024, 1, 1, 9, 0, 0, 0, time.Local) - err := mgr.Tick(context.Background(), now) - if err != nil { - t.Fatalf("Tick error: %v", err) - } - - // Check execution was submitted - // ... -} -```