Method NotifyStep

pkg/core/scheduler.go:156–206

NotifyStep 通知步骤变化 (由 Agent 调用)

(stepCount int)

// NotifyStep 通知步骤变化 (由 Agent 调用)
func (s *Scheduler) NotifyStep(stepCount int) {
	s.mu.RLock()

	// 复制监听器和任务列表,避免长时间持锁
	listeners := make([]StepCallback, len(s.stepListeners))
	copy(listeners, s.stepListeners)

	tasks := make([]*StepTask, 0, len(s.stepTasks))
	for _, task := range s.stepTasks {
		tasks = append(tasks, task)
	}

	s.mu.RUnlock()

	// 通知监听器
	for _, listener := range listeners {
		if listener == nil {
			continue // 跳过已取消的监听器
		}
		go func(cb StepCallback) {
			if err := cb(s.ctx, stepCount); err != nil {
				schedulerLog.Warn(s.ctx, "step callback error", map[string]any{"step": stepCount, "error": err})
			}
		}(listener)
	}

	// 检查并触发任务
	for _, task := range tasks {
		shouldTrigger := stepCount-task.LastTriggered >= task.Every
		if !shouldTrigger {
			continue
		}

		// 更新触发时间
		s.mu.Lock()
		task.LastTriggered = stepCount
		s.mu.Unlock()

		// 异步执行回调
		go func(t *StepTask) {
			if err := t.Callback(s.ctx, stepCount); err != nil {
				schedulerLog.Warn(s.ctx, "step task callback error", map[string]any{"task_id": t.ID, "error": err})
			}

			// 通知触发
			if s.opts.OnTrigger != nil {
				s.opts.OnTrigger(t.ID, fmt.Sprintf("step:%d", t.Every), TriggerKindStep)
			}
		}(task)
	}
}

// EveryInterval 每隔一段时间执行
func (s *Scheduler) EveryInterval(interval time.Duration, callback TaskCallback) (string, error) {