Version: 0.9.60.dev.260430

后端：
1.接入主动调度 worker 与飞书通知链路
- 新增 due job scanner 与 active_schedule.triggered workflow
- 接入 notification.feishu.requested handler、飞书 webhook provider 和用户通知配置接口
- 支持 notification_records 去重、重试、skipped/dead 状态流转
- 完成 api / worker / all 启动模式装配与主动调度验收记录
2.后续要做的就是补全从异常发生到给用户推送消息之间的逻辑缺口

backend/active_scheduler/apply/types.go

@@ -93,6 +93,33 @@ func newApplyError(code ErrorCode, message string, err error) error {
 	return &ApplyError{Code: code, Message: message, Err: err}
 }
 
+// NewApplyError 构造 confirm/apply 链路可分类业务错误。
+//
+// 职责边界：
+// 1. 供 service/API 层把预览归属、幂等冲突、adapter 业务拒绝转换为统一错误语义；
+// 2. 不负责写 preview 状态，也不决定 HTTP 状态码；
+// 3. cause 仅用于保留底层错误，展示给前端的文案应放在 message。
+func NewApplyError(code ErrorCode, message string, cause error) error {
+	return newApplyError(code, message, cause)
+}
+
+// AsApplyError 尝试把 error 还原为 ApplyError。
+//
+// 职责边界：
+// 1. 只做 errors.As 类型判断，方便 API 层统一映射业务错误；
+// 2. 不把未知错误强行归类，避免数据库或系统故障被误判为 4xx；
+// 3. 返回 bool=false 时，调用方应按普通系统错误处理。
+func AsApplyError(err error) (*ApplyError, bool) {
+	if err == nil {
+		return nil, false
+	}
+	var applyErr *ApplyError
+	if errors.As(err, &applyErr) {
+		return applyErr, true
+	}
+	return nil, false
+}
+
 func errorCodeOf(err error) ErrorCode {
 	if err == nil {
 		return ""

backend/active_scheduler/job/scanner.go

@@ -0,0 +1,269 @@
+package job
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/LoveLosita/smartflow/backend/active_scheduler/ports"
+	activesvc "github.com/LoveLosita/smartflow/backend/active_scheduler/service"
+	"github.com/LoveLosita/smartflow/backend/active_scheduler/trigger"
+	"github.com/LoveLosita/smartflow/backend/dao"
+	"github.com/LoveLosita/smartflow/backend/model"
+)
+
+const (
+	defaultScanLimit = 50
+)
+
+// Scanner 扫描到期 active_schedule_jobs 并生成正式 trigger。
+//
+// 职责边界：
+// 1. 只负责 due job -> trigger，不执行 dry-run、不写 preview、不发 notification；
+// 2. 扫描时必须重读 task 与 schedule 真值，避免过期 job 误触发；
+// 3. 对已完成、已排入日程或不再符合条件的 job，只更新 job 状态，不物理删除。
+type Scanner struct {
+	activeDAO      *dao.ActiveScheduleDAO
+	taskReader     ports.TaskReader
+	scheduleReader ports.ScheduleReader
+	triggerService *activesvc.TriggerService
+	clock          func() time.Time
+	limit          int
+	scanEvery      time.Duration
+}
+
+type ScannerOptions struct {
+	Limit     int
+	ScanEvery time.Duration
+	Clock     func() time.Time
+}
+
+type ScanResult struct {
+	Scanned   int
+	Triggered int
+	Skipped   int
+	Failed    int
+}
+
+func NewScanner(activeDAO *dao.ActiveScheduleDAO, readers ports.Readers, triggerService *activesvc.TriggerService, options ScannerOptions) (*Scanner, error) {
+	if activeDAO == nil {
+		return nil, errors.New("active schedule dao 不能为空")
+	}
+	if readers.TaskReader == nil {
+		return nil, errors.New("TaskReader 不能为空")
+	}
+	if readers.ScheduleReader == nil {
+		return nil, errors.New("ScheduleReader 不能为空")
+	}
+	if triggerService == nil {
+		return nil, errors.New("trigger service 不能为空")
+	}
+	limit := options.Limit
+	if limit <= 0 {
+		limit = defaultScanLimit
+	}
+	scanEvery := options.ScanEvery
+	if scanEvery <= 0 {
+		scanEvery = time.Minute
+	}
+	clock := options.Clock
+	if clock == nil {
+		clock = time.Now
+	}
+	return &Scanner{
+		activeDAO:      activeDAO,
+		taskReader:     readers.TaskReader,
+		scheduleReader: readers.ScheduleReader,
+		triggerService: triggerService,
+		clock:          clock,
+		limit:          limit,
+		scanEvery:      scanEvery,
+	}, nil
+}
+
+// Start 启动 due job 周期扫描。
+//
+// 说明：
+// 1. worker/all 模式调用；api 模式不启动，避免 API 进程承担后台职责；
+// 2. 每轮扫描失败只记录日志，下一轮继续；
+// 3. ctx 取消后 goroutine 自然退出。
+func (s *Scanner) Start(ctx context.Context) {
+	if s == nil {
+		return
+	}
+	go func() {
+		ticker := time.NewTicker(s.scanEvery)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				result, err := s.ScanDue(ctx, s.now())
+				if err != nil {
+					log.Printf("主动调度 due job 扫描失败: err=%v", err)
+					continue
+				}
+				if result.Scanned > 0 {
+					log.Printf("主动调度 due job 扫描完成: scanned=%d triggered=%d skipped=%d failed=%d", result.Scanned, result.Triggered, result.Skipped, result.Failed)
+				}
+			}
+		}
+	}()
+}
+
+// ScanDue 扫描并处理一批到期 job。
+func (s *Scanner) ScanDue(ctx context.Context, now time.Time) (ScanResult, error) {
+	if s == nil || s.activeDAO == nil {
+		return ScanResult{}, errors.New("scanner 未初始化")
+	}
+	jobs, err := s.activeDAO.ListDueJobs(ctx, now, s.limit)
+	if err != nil {
+		return ScanResult{}, err
+	}
+	result := ScanResult{Scanned: len(jobs)}
+	for _, item := range jobs {
+		handled, handleErr := s.processJob(ctx, item, now)
+		switch {
+		case handleErr != nil:
+			result.Failed++
+			log.Printf("主动调度 due job 处理失败: job_id=%s err=%v", item.ID, handleErr)
+		case handled == model.ActiveScheduleJobStatusTriggered:
+			result.Triggered++
+		default:
+			result.Skipped++
+		}
+	}
+	return result, nil
+}
+
+func (s *Scanner) processJob(ctx context.Context, item model.ActiveScheduleJob, now time.Time) (string, error) {
+	task, found, err := s.taskReader.GetTaskForActiveSchedule(ctx, ports.TaskRequest{
+		UserID: item.UserID,
+		TaskID: item.TaskID,
+		Now:    now,
+	})
+	if err != nil {
+		_ = s.markJobFailed(ctx, item.ID, "task_read_failed", err, now)
+		return "", err
+	}
+	if !found {
+		return model.ActiveScheduleJobStatusSkipped, s.markJobSkipped(ctx, item.ID, model.ActiveScheduleJobStatusSkipped, "task_not_found", now)
+	}
+	if task.IsCompleted {
+		return model.ActiveScheduleJobStatusCanceled, s.markJobSkipped(ctx, item.ID, model.ActiveScheduleJobStatusCanceled, "task_completed", now)
+	}
+	if task.UrgencyThresholdAt == nil {
+		// 1. 到期扫描必须重读 task 真值。
+		// 2. 若上游已经移除了 urgency_threshold_at，说明这条 due job 已经不再具备触发前提。
+		// 3. 这里直接收敛为 canceled，避免继续错误地产生 trigger。
+		return model.ActiveScheduleJobStatusCanceled, s.markJobSkipped(ctx, item.ID, model.ActiveScheduleJobStatusCanceled, "task_not_schedulable", now)
+	}
+	if task.UrgencyThresholdAt != nil && task.UrgencyThresholdAt.After(now) {
+		return model.ActiveScheduleJobStatusPending, s.activeDAO.UpdateJobFields(ctx, item.ID, map[string]any{
+			"trigger_at":      *task.UrgencyThresholdAt,
+			"last_error_code": "threshold_moved_future",
+			"last_scanned_at": &now,
+		})
+	}
+	if task.Priority != 1 && task.Priority != 2 {
+		return model.ActiveScheduleJobStatusSkipped, s.markJobSkipped(ctx, item.ID, model.ActiveScheduleJobStatusSkipped, "task_not_important", now)
+	}
+	alreadyScheduled, err := s.isTaskAlreadyScheduled(ctx, item.UserID, item.TaskID, now)
+	if err != nil {
+		_ = s.markJobFailed(ctx, item.ID, "schedule_read_failed", err, now)
+		return "", err
+	}
+	if alreadyScheduled {
+		return model.ActiveScheduleJobStatusSkipped, s.markJobSkipped(ctx, item.ID, model.ActiveScheduleJobStatusSkipped, "task_already_scheduled", now)
+	}
+
+	payload := struct {
+		JobID              string    `json:"job_id"`
+		UrgencyThresholdAt time.Time `json:"urgency_threshold_at"`
+	}{
+		JobID:              item.ID,
+		UrgencyThresholdAt: item.TriggerAt,
+	}
+	rawPayload, _ := json.Marshal(payload)
+	jobID := item.ID
+	resp, err := s.triggerService.CreateAndPublish(ctx, activesvc.TriggerRequest{
+		UserID:      item.UserID,
+		TriggerType: trigger.TriggerTypeImportantUrgentTask,
+		Source:      trigger.SourceWorkerDueJob,
+		TargetType:  trigger.TargetTypeTaskPool,
+		TargetID:    item.TaskID,
+		DedupeKey:   item.DedupeKey,
+		RequestedAt: now,
+		Payload:     rawPayload,
+		JobID:       &jobID,
+		TraceID:     firstNonEmpty(item.TraceID, fmt.Sprintf("trace_active_job_%s", item.ID)),
+	})
+	if err != nil {
+		_ = s.markJobFailed(ctx, item.ID, "trigger_publish_failed", err, now)
+		return "", err
+	}
+	return model.ActiveScheduleJobStatusTriggered, s.activeDAO.UpdateJobFields(ctx, item.ID, map[string]any{
+		"status":          model.ActiveScheduleJobStatusTriggered,
+		"last_trigger_id": &resp.TriggerID,
+		"last_error_code": nil,
+		"last_error":      nil,
+		"last_scanned_at": &now,
+	})
+}
+
+func (s *Scanner) isTaskAlreadyScheduled(ctx context.Context, userID int, taskID int, now time.Time) (bool, error) {
+	facts, err := s.scheduleReader.GetScheduleFactsByWindow(ctx, ports.ScheduleWindowRequest{
+		UserID:      userID,
+		TargetType:  string(trigger.TargetTypeTaskPool),
+		TargetID:    taskID,
+		WindowStart: now,
+		WindowEnd:   now.Add(24 * time.Hour),
+		Now:         now,
+	})
+	if err != nil {
+		return false, err
+	}
+	return facts.TargetAlreadyScheduled, nil
+}
+
+func (s *Scanner) markJobSkipped(ctx context.Context, jobID string, status string, code string, now time.Time) error {
+	return s.activeDAO.UpdateJobFields(ctx, jobID, map[string]any{
+		"status":          status,
+		"last_error_code": code,
+		"last_error":      nil,
+		"last_scanned_at": &now,
+	})
+}
+
+func (s *Scanner) markJobFailed(ctx context.Context, jobID string, code string, err error, now time.Time) error {
+	message := ""
+	if err != nil {
+		message = err.Error()
+	}
+	return s.activeDAO.UpdateJobFields(ctx, jobID, map[string]any{
+		"status":          model.ActiveScheduleJobStatusFailed,
+		"last_error_code": code,
+		"last_error":      &message,
+		"last_scanned_at": &now,
+	})
+}
+
+func (s *Scanner) now() time.Time {
+	if s == nil || s.clock == nil {
+		return time.Now()
+	}
+	return s.clock()
+}
+
+func firstNonEmpty(values ...string) string {
+	for _, value := range values {
+		if value != "" {
+			return value
+		}
+	}
+	return ""
+}

backend/active_scheduler/service/preview_confirm.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"time"
 
 	activeapply "github.com/LoveLosita/smartflow/backend/active_scheduler/apply"
@@ -12,6 +11,7 @@ import (
 	activepreview "github.com/LoveLosita/smartflow/backend/active_scheduler/preview"
 	"github.com/LoveLosita/smartflow/backend/dao"
 	"github.com/LoveLosita/smartflow/backend/model"
+	"gorm.io/gorm"
 )
 
 // PreviewConfirmService 编排第三阶段的预览生成、查询和确认应用。
@@ -88,16 +88,19 @@ func (s *PreviewConfirmService) ConfirmPreview(ctx context.Context, req activeap
 	}
 	previewRow, err := s.activeDAO.GetPreviewByID(ctx, req.PreviewID)
 	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, activeapply.NewApplyError(activeapply.ErrorCodeTargetNotFound, "预览不存在或已被删除", err)
+		}
 		return nil, err
 	}
 	if previewRow.UserID != req.UserID {
-		return nil, fmt.Errorf("preview 不属于当前用户")
+		return nil, activeapply.NewApplyError(activeapply.ErrorCodeForbidden, "预览不属于当前用户", nil)
 	}
 	if previewRow.ApplyStatus == model.ActiveScheduleApplyStatusApplied {
 		if previewRow.ApplyIdempotencyKey == req.IdempotencyKey {
 			return alreadyAppliedResult(*previewRow), nil
 		}
-		return nil, fmt.Errorf("preview 已应用，不能使用新的幂等键重复确认")
+		return nil, activeapply.NewApplyError(activeapply.ErrorCodeAlreadyApplied, "预览已经应用，不能使用新的幂等键重复确认", nil)
 	}
 
 	applyReq, err := activeapply.ConvertConfirmToApplyRequest(*previewRow, req, now)
@@ -106,7 +109,7 @@ func (s *PreviewConfirmService) ConfirmPreview(ctx context.Context, req activeap
 		return nil, err
 	}
 	if len(applyReq.Commands) == 0 {
-		return nil, fmt.Errorf("当前候选没有可正式应用的日程变更")
+		return s.markNoopApplied(ctx, *applyReq)
 	}
 	if err = s.markApplying(ctx, *applyReq); err != nil {
 		return nil, err
@@ -115,8 +118,9 @@ func (s *PreviewConfirmService) ConfirmPreview(ctx context.Context, req activeap
 	adapterReq := toAdapterRequest(*applyReq)
 	adapterResult, err := s.applyAdapter.ApplyActiveScheduleChanges(ctx, adapterReq)
 	if err != nil {
-		_ = s.markApplyFailed(ctx, previewRow.ID, applyReq.ApplyID, err)
-		return nil, err
+		classifiedErr := classifyAdapterApplyError(err)
+		_ = s.markApplyFailed(ctx, previewRow.ID, applyReq.ApplyID, classifiedErr)
+		return nil, classifiedErr
 	}
 
 	result := activeapply.ApplyActiveScheduleResult{
@@ -155,13 +159,48 @@ func (s *PreviewConfirmService) markApplying(ctx context.Context, req activeappl
 	})
 }
 
+// markNoopApplied 处理 notify_only / ask_user / close 这类“确认成功但不写正式日程”的候选。
+//
+// 职责边界：
+// 1. 只把 preview 标记为已处理，并保留幂等字段，便于同 key 重试直接命中历史结果；
+// 2. 不调用 apply adapter，因为这些 change 在转换阶段已经被归类为 skipped_changes；
+// 3. 失败时直接返回数据库错误，调用方应按系统错误处理，避免前端误以为确认成功。
+func (s *PreviewConfirmService) markNoopApplied(ctx context.Context, req activeapply.ApplyActiveScheduleRequest) (*activeapply.ConfirmResult, error) {
+	result := activeapply.ApplyActiveScheduleResult{
+		ApplyID:              req.ApplyID,
+		ApplyStatus:          activeapply.ApplyStatusApplied,
+		AppliedChanges:       []activeapply.ApplyChange{},
+		SkippedChanges:       req.SkippedChanges,
+		RequestHash:          req.RequestHash,
+		NormalizedChangeHash: req.NormalizedChangesHash,
+	}
+	if err := s.markApplied(ctx, req, result); err != nil {
+		return nil, err
+	}
+	return &activeapply.ConfirmResult{
+		PreviewID:       req.PreviewID,
+		ApplyID:         req.ApplyID,
+		ApplyStatus:     activeapply.ApplyStatusApplied,
+		CandidateID:     req.CandidateID,
+		RequestHash:     req.RequestHash,
+		RequestBodyHash: req.RequestBodyHash,
+		ApplyRequest:    &req,
+		ApplyResult:     &result,
+		SkippedChanges:  req.SkippedChanges,
+	}, nil
+}
+
 func (s *PreviewConfirmService) markApplied(ctx context.Context, req activeapply.ApplyActiveScheduleRequest, result activeapply.ApplyActiveScheduleResult) error {
 	now := s.now()
 	appliedChangesJSON := mustJSON(result.AppliedChanges)
 	appliedEventIDsJSON := mustJSON(result.AppliedEventIDs)
 	return s.activeDAO.UpdatePreviewFields(ctx, req.PreviewID, map[string]any{
 		"status":                 model.ActiveSchedulePreviewStatusApplied,
+		"apply_id":               req.ApplyID,
 		"apply_status":           model.ActiveScheduleApplyStatusApplied,
+		"apply_candidate_id":     req.CandidateID,
+		"apply_idempotency_key":  req.IdempotencyKey,
+		"apply_request_hash":     req.RequestHash,
 		"applied_changes_json":   &appliedChangesJSON,
 		"applied_event_ids_json": &appliedEventIDsJSON,
 		"apply_error":            nil,
@@ -177,8 +216,19 @@ func (s *PreviewConfirmService) markApplyFailed(ctx context.Context, previewID s
 	if err != nil {
 		message = err.Error()
 	}
+	status := model.ActiveScheduleApplyStatusFailed
+	if applyErr, ok := activeapply.AsApplyError(err); ok {
+		switch applyErr.Code {
+		case activeapply.ErrorCodeExpired:
+			status = model.ActiveScheduleApplyStatusExpired
+		case activeapply.ErrorCodeDBError:
+			status = model.ActiveScheduleApplyStatusFailed
+		default:
+			status = model.ActiveScheduleApplyStatusRejected
+		}
+	}
 	updates := map[string]any{
-		"apply_status": model.ActiveScheduleApplyStatusFailed,
+		"apply_status": status,
 		"apply_error":  &message,
 	}
 	if applyID != "" {
@@ -187,6 +237,40 @@ func (s *PreviewConfirmService) markApplyFailed(ctx context.Context, previewID s
 	return s.activeDAO.UpdatePreviewFields(ctx, previewID, updates)
 }
 
+// classifyAdapterApplyError 把正式写库 adapter 的错误转换为 confirm 层统一错误码。
+//
+// 职责边界：
+// 1. 只处理 applyadapter 已声明的业务错误码，保持 API 层只理解 active_scheduler/apply 包；
+// 2. 未知错误统一归为 db_error，避免把真实系统故障错误映射为用户可修正的 4xx；
+// 3. 原始错误作为 cause 保留，日志和 apply_error 仍能追到 adapter 返回的完整信息。
+func classifyAdapterApplyError(err error) error {
+	if err == nil {
+		return nil
+	}
+	var adapterErr *applyadapter.ApplyError
+	if !errors.As(err, &adapterErr) {
+		return activeapply.NewApplyError(activeapply.ErrorCodeDBError, "主动调度正式写库失败", err)
+	}
+	switch adapterErr.Code {
+	case applyadapter.ErrorCodeInvalidRequest:
+		return activeapply.NewApplyError(activeapply.ErrorCodeInvalidRequest, adapterErr.Message, err)
+	case applyadapter.ErrorCodeUnsupportedChangeType:
+		return activeapply.NewApplyError(activeapply.ErrorCodeUnsupportedChangeType, adapterErr.Message, err)
+	case applyadapter.ErrorCodeTargetNotFound:
+		return activeapply.NewApplyError(activeapply.ErrorCodeTargetNotFound, adapterErr.Message, err)
+	case applyadapter.ErrorCodeTargetCompleted:
+		return activeapply.NewApplyError(activeapply.ErrorCodeTargetCompleted, adapterErr.Message, err)
+	case applyadapter.ErrorCodeTargetAlreadyScheduled:
+		return activeapply.NewApplyError(activeapply.ErrorCodeTargetAlreadySchedule, adapterErr.Message, err)
+	case applyadapter.ErrorCodeSlotConflict:
+		return activeapply.NewApplyError(activeapply.ErrorCodeSlotConflict, adapterErr.Message, err)
+	case applyadapter.ErrorCodeInvalidEditedChanges:
+		return activeapply.NewApplyError(activeapply.ErrorCodeInvalidEditedChanges, adapterErr.Message, err)
+	default:
+		return activeapply.NewApplyError(activeapply.ErrorCodeDBError, adapterErr.Message, err)
+	}
+}
+
 func (s *PreviewConfirmService) now() time.Time {
 	if s == nil || s.clock == nil {
 		return time.Now()

backend/active_scheduler/service/trigger.go

@@ -0,0 +1,270 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/LoveLosita/smartflow/backend/active_scheduler/trigger"
+	"github.com/LoveLosita/smartflow/backend/dao"
+	outboxinfra "github.com/LoveLosita/smartflow/backend/infra/outbox"
+	"github.com/LoveLosita/smartflow/backend/model"
+	sharedevents "github.com/LoveLosita/smartflow/backend/shared/events"
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+)
+
+const triggerDedupeWindow = 30 * time.Minute
+
+// TriggerRequest 是正式主动调度触发入口的请求 DTO。
+//
+// 职责边界：
+// 1. 负责承载 API trigger、worker due job、用户反馈归一后的触发事实；
+// 2. 不承载 dry-run 结果、preview 快照或 notification provider 参数；
+// 3. Payload 只保存触发来源补充信息，不能塞任意业务写库参数。
+type TriggerRequest struct {
+	UserID         int
+	TriggerType    trigger.TriggerType
+	Source         trigger.Source
+	TargetType     trigger.TargetType
+	TargetID       int
+	FeedbackID     string
+	IdempotencyKey string
+	DedupeKey      string
+	MockNow        *time.Time
+	IsMockTime     bool
+	RequestedAt    time.Time
+	Payload        json.RawMessage
+	JobID          *string
+	TraceID        string
+}
+
+// TriggerResponse 是正式触发写入后的结果。
+type TriggerResponse struct {
+	TriggerID string  `json:"trigger_id"`
+	Status    string  `json:"status"`
+	PreviewID *string `json:"preview_id,omitempty"`
+	DedupeHit bool    `json:"dedupe_hit"`
+	TraceID   string  `json:"trace_id,omitempty"`
+}
+
+// TriggerService 负责写入正式 trigger 并发布 active_schedule.triggered 事件。
+//
+// 职责边界：
+// 1. 只负责触发信号持久化、去重和事件发布；
+// 2. 不执行 dry-run、不写 preview、不发飞书；
+// 3. outbox 未启用时返回明确错误，避免调用方误以为正式链路已启动。
+type TriggerService struct {
+	activeDAO *dao.ActiveScheduleDAO
+	publisher outboxinfra.EventPublisher
+	clock     func() time.Time
+}
+
+func NewTriggerService(activeDAO *dao.ActiveScheduleDAO, publisher outboxinfra.EventPublisher) (*TriggerService, error) {
+	if activeDAO == nil {
+		return nil, errors.New("active schedule dao 不能为空")
+	}
+	return &TriggerService{
+		activeDAO: activeDAO,
+		publisher: publisher,
+		clock:     time.Now,
+	}, nil
+}
+
+func (s *TriggerService) SetClock(clock func() time.Time) {
+	if s != nil && clock != nil {
+		s.clock = clock
+	}
+}
+
+// CreateAndPublish 创建正式 trigger 并发布 outbox 事件。
+//
+// 步骤化说明：
+// 1. 先按主动调度 trigger DTO 做入口校验，确保 mock_now 不会从 worker 入口混入；
+// 2. 再用 idempotency_key / dedupe_key 查询已有 trigger，命中则直接返回旧状态；
+// 3. 新 trigger 先落库，再发布 outbox；发布失败会把 trigger 标记 failed，便于排障；
+// 4. 返回 nil error 只表示事件已入 outbox，不表示 worker 已经生成 preview。
+func (s *TriggerService) CreateAndPublish(ctx context.Context, req TriggerRequest) (*TriggerResponse, error) {
+	if s == nil || s.activeDAO == nil {
+		return nil, errors.New("trigger service 未初始化")
+	}
+	if s.publisher == nil {
+		return nil, errors.New("outbox event bus 未启用，无法执行正式主动调度 trigger")
+	}
+
+	now := s.now()
+	if req.RequestedAt.IsZero() {
+		req.RequestedAt = now
+	}
+	if req.IsMockTime && req.MockNow == nil {
+		return nil, errors.New("is_mock_time=true 时 mock_now 不能为空")
+	}
+	trig := trigger.ActiveScheduleTrigger{
+		UserID:         req.UserID,
+		TriggerType:    req.TriggerType,
+		Source:         req.Source,
+		TargetType:     req.TargetType,
+		TargetID:       req.TargetID,
+		FeedbackID:     req.FeedbackID,
+		IdempotencyKey: req.IdempotencyKey,
+		MockNow:        req.MockNow,
+		IsMockTime:     req.IsMockTime,
+		RequestedAt:    req.RequestedAt,
+		TraceID:        firstNonEmpty(req.TraceID, fmt.Sprintf("trace_active_trigger_%d", now.UnixNano())),
+	}
+	if err := trig.Validate(); err != nil {
+		return nil, err
+	}
+	if trig.Source == trigger.SourceAPIDryRun {
+		return nil, errors.New("api_dry_run 不允许创建正式 trigger")
+	}
+
+	dedupeKey := strings.TrimSpace(req.DedupeKey)
+	if dedupeKey == "" {
+		dedupeKey = BuildTriggerDedupeKey(req.UserID, req.TriggerType, req.TargetType, req.TargetID, req.FeedbackID, req.IdempotencyKey, trig.EffectiveNow(req.RequestedAt))
+	}
+	if existing, ok, err := s.findExistingTrigger(ctx, req.UserID, string(req.TriggerType), req.IdempotencyKey, dedupeKey); err != nil {
+		return nil, err
+	} else if ok {
+		return triggerResponseFromModel(existing, true), nil
+	}
+
+	payloadJSON := string(req.Payload)
+	if strings.TrimSpace(payloadJSON) == "" {
+		payloadJSON = "{}"
+	}
+	triggerID := "ast_" + uuid.NewString()
+	row := &model.ActiveScheduleTrigger{
+		ID:             triggerID,
+		UserID:         req.UserID,
+		TriggerType:    string(req.TriggerType),
+		Source:         string(req.Source),
+		TargetType:     string(req.TargetType),
+		TargetID:       req.TargetID,
+		FeedbackID:     strings.TrimSpace(req.FeedbackID),
+		JobID:          req.JobID,
+		IdempotencyKey: strings.TrimSpace(req.IdempotencyKey),
+		DedupeKey:      dedupeKey,
+		Status:         model.ActiveScheduleTriggerStatusPending,
+		MockNow:        req.MockNow,
+		IsMockTime:     req.IsMockTime,
+		RequestedAt:    req.RequestedAt,
+		PayloadJSON:    &payloadJSON,
+		TraceID:        trig.TraceID,
+	}
+	if err := s.activeDAO.CreateTrigger(ctx, row); err != nil {
+		return nil, err
+	}
+
+	eventPayload := sharedevents.ActiveScheduleTriggeredPayload{
+		TriggerID:      row.ID,
+		UserID:         row.UserID,
+		TriggerType:    row.TriggerType,
+		Source:         row.Source,
+		TargetType:     row.TargetType,
+		TargetID:       row.TargetID,
+		FeedbackID:     row.FeedbackID,
+		IdempotencyKey: row.IdempotencyKey,
+		DedupeKey:      row.DedupeKey,
+		MockNow:        row.MockNow,
+		IsMockTime:     row.IsMockTime,
+		RequestedAt:    row.RequestedAt,
+		Payload:        json.RawMessage(payloadJSON),
+		TraceID:        row.TraceID,
+	}
+	if err := eventPayload.Validate(); err != nil {
+		_ = s.markTriggerFailed(ctx, row.ID, "payload_invalid", err)
+		return nil, err
+	}
+	if err := s.publisher.Publish(ctx, outboxinfra.PublishRequest{
+		EventType:    sharedevents.ActiveScheduleTriggeredEventType,
+		EventVersion: sharedevents.ActiveScheduleTriggeredEventVersion,
+		MessageKey:   eventPayload.MessageKey(),
+		AggregateID:  eventPayload.AggregateID(),
+		Payload:      eventPayload,
+	}); err != nil {
+		_ = s.markTriggerFailed(ctx, row.ID, "outbox_publish_failed", err)
+		return nil, err
+	}
+
+	return triggerResponseFromModel(row, false), nil
+}
+
+func (s *TriggerService) findExistingTrigger(ctx context.Context, userID int, triggerType string, idempotencyKey string, dedupeKey string) (*model.ActiveScheduleTrigger, bool, error) {
+	if strings.TrimSpace(idempotencyKey) != "" {
+		existing, err := s.activeDAO.FindTriggerByIdempotencyKey(ctx, userID, triggerType, idempotencyKey)
+		if err == nil {
+			return existing, true, nil
+		}
+		if !errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, false, err
+		}
+	}
+	statuses := []string{
+		model.ActiveScheduleTriggerStatusPending,
+		model.ActiveScheduleTriggerStatusProcessing,
+		model.ActiveScheduleTriggerStatusPreviewGenerated,
+	}
+	existing, err := s.activeDAO.FindTriggerByDedupeKey(ctx, dedupeKey, statuses)
+	if err == nil {
+		return existing, true, nil
+	}
+	if !errors.Is(err, gorm.ErrRecordNotFound) {
+		return nil, false, err
+	}
+	return nil, false, nil
+}
+
+func (s *TriggerService) markTriggerFailed(ctx context.Context, triggerID string, code string, err error) error {
+	message := ""
+	if err != nil {
+		message = err.Error()
+	}
+	now := s.now()
+	return s.activeDAO.UpdateTriggerFields(ctx, triggerID, map[string]any{
+		"status":          model.ActiveScheduleTriggerStatusFailed,
+		"last_error_code": code,
+		"last_error":      &message,
+		"completed_at":    &now,
+	})
+}
+
+func (s *TriggerService) now() time.Time {
+	if s == nil || s.clock == nil {
+		return time.Now()
+	}
+	return s.clock()
+}
+
+// BuildTriggerDedupeKey 生成正式触发去重键。
+//
+// 说明：
+// 1. important_urgent_task 按 30 分钟窗口聚合，避免同一任务反复生成预览；
+// 2. unfinished_feedback 优先使用 feedback_id/idempotency_key，不做固定时间窗强去重；
+// 3. 参数非法时仍返回可读字符串，调用方会在 trigger.Validate 阶段拒绝非法输入。
+func BuildTriggerDedupeKey(userID int, triggerType trigger.TriggerType, targetType trigger.TargetType, targetID int, feedbackID string, idempotencyKey string, at time.Time) string {
+	if triggerType == trigger.TriggerTypeUnfinishedFeedback {
+		return fmt.Sprintf("%d:%s:%s", userID, triggerType, firstNonEmpty(feedbackID, idempotencyKey, fmt.Sprintf("%s:%d", targetType, targetID)))
+	}
+	if at.IsZero() {
+		at = time.Now()
+	}
+	windowStart := at.Truncate(triggerDedupeWindow)
+	return fmt.Sprintf("%d:%s:%s:%d:%s", userID, triggerType, targetType, targetID, windowStart.Format(time.RFC3339))
+}
+
+func triggerResponseFromModel(row *model.ActiveScheduleTrigger, dedupeHit bool) *TriggerResponse {
+	if row == nil {
+		return &TriggerResponse{DedupeHit: dedupeHit}
+	}
+	return &TriggerResponse{
+		TriggerID: row.ID,
+		Status:    row.Status,
+		PreviewID: row.PreviewID,
+		DedupeHit: dedupeHit,
+		TraceID:   row.TraceID,
+	}
+}

backend/active_scheduler/service/trigger_outbox.go

@@ -0,0 +1,219 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	kafkabus "github.com/LoveLosita/smartflow/backend/infra/kafka"
+	outboxinfra "github.com/LoveLosita/smartflow/backend/infra/outbox"
+	"github.com/LoveLosita/smartflow/backend/model"
+	sharedevents "github.com/LoveLosita/smartflow/backend/shared/events"
+)
+
+const requestedNotificationDedupeWindow = 30 * time.Minute
+
+// EnqueueActiveScheduleTriggeredInTx 在事务内写入 active_schedule.triggered outbox 消息。
+//
+// 职责边界：
+// 1. 只负责把已经校验好的事件契约写入 outbox；
+// 2. 不负责创建 trigger 记录，trigger 真值应由调用方先落库；
+// 3. 失败时返回 error，让上层决定是否整体回滚与重试。
+func EnqueueActiveScheduleTriggeredInTx(
+	ctx context.Context,
+	outboxRepo *outboxinfra.Repository,
+	kafkaCfg kafkabus.Config,
+	payload sharedevents.ActiveScheduleTriggeredPayload,
+) error {
+	return enqueueContractEventInTx(
+		ctx,
+		outboxRepo,
+		kafkaCfg,
+		sharedevents.ActiveScheduleTriggeredEventType,
+		sharedevents.ActiveScheduleTriggeredEventVersion,
+		payload.MessageKey(),
+		payload.AggregateID(),
+		payload.AggregateID(),
+		payload,
+		payload.Validate,
+	)
+}
+
+// EnqueueNotificationFeishuRequestedInTx 在事务内写入 notification.feishu.requested outbox 消息。
+//
+// 职责边界：
+// 1. 只做事件契约序列化和 outbox 入队；
+// 2. 不负责 notification_records 幂等与 provider 调用；
+// 3. 失败时直接返回，让 trigger -> preview -> notification 保持同事务回滚。
+func EnqueueNotificationFeishuRequestedInTx(
+	ctx context.Context,
+	outboxRepo *outboxinfra.Repository,
+	kafkaCfg kafkabus.Config,
+	payload sharedevents.FeishuNotificationRequestedPayload,
+) error {
+	return enqueueContractEventInTx(
+		ctx,
+		outboxRepo,
+		kafkaCfg,
+		sharedevents.NotificationFeishuRequestedEventType,
+		sharedevents.NotificationFeishuRequestedEventVersion,
+		payload.MessageKey(),
+		payload.AggregateID(),
+		payload.AggregateID(),
+		payload,
+		payload.Validate,
+	)
+}
+
+// BuildTriggeredPayloadFromModel 把持久化 trigger 还原成事件载荷。
+//
+// 职责边界：
+// 1. 只做 model -> contract DTO 映射；
+// 2. 不校验 trigger 是否应该被处理，业务真值判断由 scanner / worker 完成；
+// 3. 若 payload_json 不是合法 JSON，返回 error，让调用方回滚本次触发。
+func BuildTriggeredPayloadFromModel(row model.ActiveScheduleTrigger) (sharedevents.ActiveScheduleTriggeredPayload, error) {
+	var rawPayload json.RawMessage
+	if row.PayloadJSON != nil && strings.TrimSpace(*row.PayloadJSON) != "" {
+		rawPayload = json.RawMessage(strings.TrimSpace(*row.PayloadJSON))
+		if !json.Valid(rawPayload) {
+			return sharedevents.ActiveScheduleTriggeredPayload{}, errors.New("trigger payload_json 不是合法 JSON")
+		}
+	}
+
+	payload := sharedevents.ActiveScheduleTriggeredPayload{
+		TriggerID:      row.ID,
+		UserID:         row.UserID,
+		TriggerType:    row.TriggerType,
+		Source:         row.Source,
+		TargetType:     row.TargetType,
+		TargetID:       row.TargetID,
+		FeedbackID:     row.FeedbackID,
+		IdempotencyKey: row.IdempotencyKey,
+		DedupeKey:      row.DedupeKey,
+		MockNow:        row.MockNow,
+		IsMockTime:     row.IsMockTime,
+		RequestedAt:    row.RequestedAt,
+		Payload:        rawPayload,
+		TraceID:        row.TraceID,
+	}
+	if err := payload.Validate(); err != nil {
+		return sharedevents.ActiveScheduleTriggeredPayload{}, err
+	}
+	return payload, nil
+}
+
+// BuildFeishuRequestedPayload 生成通知事件载荷。
+//
+// 职责边界：
+// 1. 只做 trigger/preview 快照到通知契约的拼装；
+// 2. 不判断是否真的要发通知，上层应先根据 decision.ShouldNotify 决定是否调用；
+// 3. fallback 文案只做兜底，不替代后续 notification handler 的 provider 级策略。
+func BuildFeishuRequestedPayload(
+	triggerRow model.ActiveScheduleTrigger,
+	previewID string,
+	notificationSummary string,
+	requestedAt time.Time,
+) sharedevents.FeishuNotificationRequestedPayload {
+	summary := strings.TrimSpace(notificationSummary)
+	return sharedevents.FeishuNotificationRequestedPayload{
+		UserID:       triggerRow.UserID,
+		TriggerID:    triggerRow.ID,
+		PreviewID:    strings.TrimSpace(previewID),
+		TriggerType:  triggerRow.TriggerType,
+		TargetType:   triggerRow.TargetType,
+		TargetID:     triggerRow.TargetID,
+		DedupeKey:    BuildNotificationDedupeKey(triggerRow.UserID, triggerRow.TriggerType, triggerRow.RequestedAt),
+		TargetURL:    fmt.Sprintf("/schedule-adjust/%s", strings.TrimSpace(previewID)),
+		SummaryText:  summary,
+		FallbackText: buildNotificationFallbackText(summary, strings.TrimSpace(previewID)),
+		TraceID:      triggerRow.TraceID,
+		RequestedAt:  requestedAt,
+	}
+}
+
+// BuildNotificationDedupeKey 生成通知 30 分钟窗口去重键。
+//
+// 说明：
+// 1. 第一版按 user_id + trigger_type + time_window 聚合；
+// 2. 当 requested_at 缺失时回退到当前时间，避免空值直接写出脏 dedupe_key；
+// 3. 不拼 preview_id，保证同一窗口内多次重试只会落到同一组通知记录。
+func BuildNotificationDedupeKey(userID int, triggerType string, requestedAt time.Time) string {
+	if requestedAt.IsZero() {
+		requestedAt = time.Now()
+	}
+	windowStart := requestedAt.Truncate(requestedNotificationDedupeWindow)
+	return fmt.Sprintf("%d:%s:%s",
+		userID,
+		strings.TrimSpace(triggerType),
+		windowStart.Format(time.RFC3339),
+	)
+}
+
+func enqueueContractEventInTx(
+	ctx context.Context,
+	outboxRepo *outboxinfra.Repository,
+	kafkaCfg kafkabus.Config,
+	eventType string,
+	eventVersion string,
+	messageKey string,
+	aggregateID string,
+	eventID string,
+	payload any,
+	validate func() error,
+) error {
+	if outboxRepo == nil {
+		return errors.New("outbox repository 不能为空")
+	}
+	if validate == nil {
+		return errors.New("事件校验函数不能为空")
+	}
+	if err := validate(); err != nil {
+		return err
+	}
+
+	payloadJSON, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+
+	cfg := normalizeKafkaConfig(kafkaCfg)
+	wrapped := outboxinfra.OutboxEventPayload{
+		EventID:      strings.TrimSpace(eventID),
+		EventType:    eventType,
+		EventVersion: strings.TrimSpace(eventVersion),
+		AggregateID:  strings.TrimSpace(aggregateID),
+		Payload:      payloadJSON,
+	}
+	_, err = outboxRepo.CreateMessage(ctx, eventType, cfg.Topic, strings.TrimSpace(messageKey), wrapped, cfg.MaxRetry)
+	return err
+}
+
+func normalizeKafkaConfig(cfg kafkabus.Config) kafkabus.Config {
+	if strings.TrimSpace(cfg.Topic) == "" {
+		cfg.Topic = kafkabus.DefaultTopic
+	}
+	if cfg.MaxRetry <= 0 {
+		cfg.MaxRetry = 20
+	}
+	return cfg
+}
+
+func buildNotificationFallbackText(summary string, previewID string) string {
+	link := fmt.Sprintf("/schedule-adjust/%s", previewID)
+	if summary == "" {
+		return "你有一条新的日程调整建议，请查看：" + link
+	}
+	return summary + "，请查看：" + link
+}
+
+func firstNonEmpty(values ...string) string {
+	for _, value := range values {
+		if strings.TrimSpace(value) != "" {
+			return strings.TrimSpace(value)
+		}
+	}
+	return ""
+}

backend/active_scheduler/service/trigger_pipeline.go

@@ -0,0 +1,338 @@
+package service
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	activepreview "github.com/LoveLosita/smartflow/backend/active_scheduler/preview"
+	"github.com/LoveLosita/smartflow/backend/active_scheduler/trigger"
+	"github.com/LoveLosita/smartflow/backend/dao"
+	kafkabus "github.com/LoveLosita/smartflow/backend/infra/kafka"
+	outboxinfra "github.com/LoveLosita/smartflow/backend/infra/outbox"
+	"github.com/LoveLosita/smartflow/backend/model"
+	sharedevents "github.com/LoveLosita/smartflow/backend/shared/events"
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+const (
+	triggerErrorCodePayloadMismatch = "payload_mismatch"
+	triggerErrorCodeWorkerFailed    = "worker_failed"
+)
+
+// TriggerWorkflowService 负责第四阶段的 trigger -> dry-run -> preview -> notification 编排。
+//
+// 职责边界：
+// 1. 只推进主动调度 trigger 的后台状态机，不负责启动 outbox worker；
+// 2. dry-run 与 preview 复用现有 service，不再单独实现第二套候选生成逻辑；
+// 3. notification 只发布 requested 事件，不直接接真实飞书 provider。
+type TriggerWorkflowService struct {
+	activeDAO *dao.ActiveScheduleDAO
+	dryRun    *DryRunService
+	outbox    *outboxinfra.Repository
+	kafkaCfg  kafkabus.Config
+	clock     func() time.Time
+}
+
+func NewTriggerWorkflowService(
+	activeDAO *dao.ActiveScheduleDAO,
+	dryRun *DryRunService,
+	outboxRepo *outboxinfra.Repository,
+	kafkaCfg kafkabus.Config,
+) (*TriggerWorkflowService, error) {
+	if activeDAO == nil {
+		return nil, errors.New("active schedule dao 不能为空")
+	}
+	if dryRun == nil {
+		return nil, errors.New("dry-run service 不能为空")
+	}
+	if outboxRepo == nil {
+		return nil, errors.New("outbox repository 不能为空")
+	}
+	return &TriggerWorkflowService{
+		activeDAO: activeDAO,
+		dryRun:    dryRun,
+		outbox:    outboxRepo,
+		kafkaCfg:  kafkaCfg,
+		clock:     time.Now,
+	}, nil
+}
+
+func (s *TriggerWorkflowService) SetClock(clock func() time.Time) {
+	if s != nil && clock != nil {
+		s.clock = clock
+	}
+}
+
+// ProcessTriggeredInTx 在 outbox 消费事务内推进 trigger 主链路。
+//
+// 步骤化说明：
+// 1. 先锁 trigger 行，确保同一 trigger 在并发 worker 下只能由一个事务推进；
+// 2. 再把状态切到 processing，避免排障时看不出消息已经被消费；
+// 3. 复用 dry-run + preview service 生成预览；若发现已有 preview，则直接复用，避免重复写库；
+// 4. preview 成功后回写 trigger 状态，并在同一事务里补发 notification.requested outbox；
+// 5. 任一步失败都返回 error，由外层 handler 负责记录 failed 状态并触发 outbox retry。
+func (s *TriggerWorkflowService) ProcessTriggeredInTx(
+	ctx context.Context,
+	tx *gorm.DB,
+	payload sharedevents.ActiveScheduleTriggeredPayload,
+) error {
+	if s == nil || s.activeDAO == nil || s.dryRun == nil || s.outbox == nil {
+		return errors.New("trigger workflow service 未初始化")
+	}
+	if tx == nil {
+		return errors.New("gorm tx 不能为空")
+	}
+	if err := payload.Validate(); err != nil {
+		return err
+	}
+
+	now := s.now()
+	triggerRow, err := s.lockTrigger(ctx, tx, payload.TriggerID)
+	if err != nil {
+		return err
+	}
+
+	txDAO := s.activeDAO.WithTx(tx)
+	if completed, err := s.tryFinishByTerminalStatus(ctx, txDAO, *triggerRow); err != nil || completed {
+		return err
+	}
+	if handled, err := s.tryRejectMismatchedPayload(ctx, txDAO, *triggerRow, payload, now); err != nil || handled {
+		return err
+	}
+
+	if err := txDAO.UpdateTriggerFields(ctx, triggerRow.ID, map[string]any{
+		"status":          model.ActiveScheduleTriggerStatusProcessing,
+		"processed_at":    &now,
+		"last_error_code": nil,
+		"last_error":      nil,
+	}); err != nil {
+		return err
+	}
+
+	existingPreview, err := txDAO.GetPreviewByTriggerID(ctx, triggerRow.ID)
+	switch {
+	case err == nil:
+		return s.finishWithExistingPreview(ctx, txDAO, *triggerRow, *existingPreview, now)
+	case errors.Is(err, gorm.ErrRecordNotFound):
+		// 继续创建新 preview。
+	default:
+		return err
+	}
+
+	domainTrigger := buildDomainTriggerFromModel(*triggerRow, payload)
+	dryRunResult, err := s.dryRun.DryRun(ctx, domainTrigger)
+	if err != nil {
+		return err
+	}
+	if len(dryRunResult.Candidates) == 0 {
+		return s.markClosedWithoutPreview(ctx, txDAO, triggerRow.ID, now)
+	}
+	if !dryRunResult.Observation.Decision.ShouldNotify && !dryRunResult.Observation.Decision.ShouldWritePreview {
+		return s.markClosedWithoutPreview(ctx, txDAO, triggerRow.ID, now)
+	}
+
+	previewService, err := activepreview.NewService(txDAO)
+	if err != nil {
+		return err
+	}
+	previewResp, err := previewService.CreatePreview(ctx, activepreview.CreatePreviewRequest{
+		ActiveContext: dryRunResult.Context,
+		Observation:   dryRunResult.Observation,
+		Candidates:    dryRunResult.Candidates,
+		TriggerID:     triggerRow.ID,
+		GeneratedAt:   now,
+	})
+	if err != nil {
+		return err
+	}
+
+	previewID := previewResp.Detail.PreviewID
+	if err = txDAO.UpdateTriggerFields(ctx, triggerRow.ID, map[string]any{
+		"status":          model.ActiveScheduleTriggerStatusPreviewGenerated,
+		"preview_id":      &previewID,
+		"completed_at":    &now,
+		"last_error_code": nil,
+		"last_error":      nil,
+	}); err != nil {
+		return err
+	}
+
+	if !dryRunResult.Observation.Decision.ShouldNotify {
+		return nil
+	}
+
+	notificationPayload := BuildFeishuRequestedPayload(
+		*triggerRow,
+		previewID,
+		previewResp.Detail.Notification,
+		now,
+	)
+	return EnqueueNotificationFeishuRequestedInTx(ctx, s.outbox.WithTx(tx), s.kafkaCfg, notificationPayload)
+}
+
+// MarkTriggerFailedBestEffort 在事务外补记 trigger failed 状态，供 outbox retry 前排障。
+//
+// 职责边界：
+// 1. 只做 best-effort 状态回写，不能影响外层对原始错误的返回；
+// 2. 不负责错误分类，当前统一记为 worker_failed；
+// 3. 失败时静默返回，让真正的重试仍由 outbox 状态机负责。
+func (s *TriggerWorkflowService) MarkTriggerFailedBestEffort(ctx context.Context, triggerID string, err error) {
+	if s == nil || s.activeDAO == nil || strings.TrimSpace(triggerID) == "" {
+		return
+	}
+	message := ""
+	if err != nil {
+		message = err.Error()
+	}
+	_ = s.activeDAO.UpdateTriggerFields(ctx, triggerID, map[string]any{
+		"status":          model.ActiveScheduleTriggerStatusFailed,
+		"last_error_code": triggerErrorCodeWorkerFailed,
+		"last_error":      &message,
+	})
+}
+
+func (s *TriggerWorkflowService) lockTrigger(ctx context.Context, tx *gorm.DB, triggerID string) (*model.ActiveScheduleTrigger, error) {
+	var row model.ActiveScheduleTrigger
+	err := tx.WithContext(ctx).
+		Clauses(clause.Locking{Strength: "UPDATE"}).
+		Where("id = ?", triggerID).
+		First(&row).Error
+	if err != nil {
+		return nil, err
+	}
+	return &row, nil
+}
+
+func (s *TriggerWorkflowService) tryFinishByTerminalStatus(
+	ctx context.Context,
+	txDAO *dao.ActiveScheduleDAO,
+	row model.ActiveScheduleTrigger,
+) (bool, error) {
+	switch row.Status {
+	case model.ActiveScheduleTriggerStatusPreviewGenerated,
+		model.ActiveScheduleTriggerStatusClosed,
+		model.ActiveScheduleTriggerStatusSkipped,
+		model.ActiveScheduleTriggerStatusRejected:
+		return true, nil
+	case model.ActiveScheduleTriggerStatusPending,
+		model.ActiveScheduleTriggerStatusProcessing,
+		model.ActiveScheduleTriggerStatusFailed:
+		return false, nil
+	default:
+		// 1. 遇到未知状态时，不直接报错中断，而是继续按 processing 流程推进。
+		// 2. 这样可以兼容迁移期历史脏数据，避免单条异常阻塞整批消费。
+		// 3. 真实状态最终会被下面的 UpdateTriggerFields 覆盖为 processing。
+		return false, nil
+	}
+}
+
+func (s *TriggerWorkflowService) tryRejectMismatchedPayload(
+	ctx context.Context,
+	txDAO *dao.ActiveScheduleDAO,
+	row model.ActiveScheduleTrigger,
+	payload sharedevents.ActiveScheduleTriggeredPayload,
+	now time.Time,
+) (bool, error) {
+	mismatchReason := buildPayloadMismatchReason(row, payload)
+	if mismatchReason == "" {
+		return false, nil
+	}
+	if err := txDAO.UpdateTriggerFields(ctx, row.ID, map[string]any{
+		"status":          model.ActiveScheduleTriggerStatusRejected,
+		"last_error_code": triggerErrorCodePayloadMismatch,
+		"last_error":      &mismatchReason,
+		"completed_at":    &now,
+	}); err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+func (s *TriggerWorkflowService) finishWithExistingPreview(
+	ctx context.Context,
+	txDAO *dao.ActiveScheduleDAO,
+	triggerRow model.ActiveScheduleTrigger,
+	previewRow model.ActiveSchedulePreview,
+	now time.Time,
+) error {
+	previewID := previewRow.ID
+	return txDAO.UpdateTriggerFields(ctx, triggerRow.ID, map[string]any{
+		"status":          model.ActiveScheduleTriggerStatusPreviewGenerated,
+		"preview_id":      &previewID,
+		"completed_at":    &now,
+		"last_error_code": nil,
+		"last_error":      nil,
+	})
+}
+
+func (s *TriggerWorkflowService) markClosedWithoutPreview(
+	ctx context.Context,
+	txDAO *dao.ActiveScheduleDAO,
+	triggerID string,
+	now time.Time,
+) error {
+	return txDAO.UpdateTriggerFields(ctx, triggerID, map[string]any{
+		"status":          model.ActiveScheduleTriggerStatusClosed,
+		"completed_at":    &now,
+		"last_error_code": nil,
+		"last_error":      nil,
+	})
+}
+
+func (s *TriggerWorkflowService) now() time.Time {
+	if s == nil || s.clock == nil {
+		return time.Now()
+	}
+	return s.clock()
+}
+
+func buildDomainTriggerFromModel(
+	row model.ActiveScheduleTrigger,
+	payload sharedevents.ActiveScheduleTriggeredPayload,
+) trigger.ActiveScheduleTrigger {
+	mockNow := row.MockNow
+	if mockNow == nil && payload.MockNow != nil {
+		mockNow = payload.MockNow
+	}
+	traceID := strings.TrimSpace(row.TraceID)
+	if traceID == "" {
+		traceID = strings.TrimSpace(payload.TraceID)
+	}
+	if traceID == "" {
+		traceID = "trace_active_trigger_" + uuid.NewString()
+	}
+	return trigger.ActiveScheduleTrigger{
+		TriggerID:      row.ID,
+		UserID:         row.UserID,
+		TriggerType:    trigger.TriggerType(row.TriggerType),
+		Source:         trigger.Source(row.Source),
+		TargetType:     trigger.TargetType(row.TargetType),
+		TargetID:       row.TargetID,
+		FeedbackID:     row.FeedbackID,
+		IdempotencyKey: row.IdempotencyKey,
+		MockNow:        mockNow,
+		IsMockTime:     row.IsMockTime || payload.IsMockTime,
+		RequestedAt:    row.RequestedAt,
+		TraceID:        traceID,
+	}
+}
+
+func buildPayloadMismatchReason(row model.ActiveScheduleTrigger, payload sharedevents.ActiveScheduleTriggeredPayload) string {
+	switch {
+	case row.UserID != payload.UserID:
+		return fmt.Sprintf("trigger 事件 user_id 不一致: row=%d payload=%d", row.UserID, payload.UserID)
+	case row.TriggerType != payload.TriggerType:
+		return fmt.Sprintf("trigger 事件 trigger_type 不一致: row=%s payload=%s", row.TriggerType, payload.TriggerType)
+	case row.TargetType != payload.TargetType:
+		return fmt.Sprintf("trigger 事件 target_type 不一致: row=%s payload=%s", row.TargetType, payload.TargetType)
+	case row.TargetID != payload.TargetID:
+		return fmt.Sprintf("trigger 事件 target_id 不一致: row=%d payload=%d", row.TargetID, payload.TargetID)
+	default:
+		return ""
+	}
+}