package scheduler import ( "context" "errors" "fmt" "log" "strings" "sync" "time" "github.com/agentmail/gateway/internal/db" "github.com/agentmail/gateway/internal/models" "github.com/agentmail/gateway/internal/notify" "github.com/agentmail/gateway/internal/repo" "github.com/google/uuid" ) // CalendarScheduler 定时扫描日历事件,触发到期提醒。 // // 设计参照 Outlook 的 Exchange 提醒器: // - 每 30 秒扫描一次(精度到分钟够用,不需要秒级) // - 到期事件 → 生成一封提醒邮件 → 投递 // - 重复事件自动推进到下一次 // - 幂等:last_fired_at 保证同一分钟不触发两次 // // 为什么用进程内 goroutine 而不是 cron:调度器与 Gateway 同生命周期, // 不需要外部依赖,也不需要第二套「谁在跑」的信任模型。 var ( schedMu sync.Mutex schedStop chan struct{} schedWg sync.WaitGroup ) // Start 启动日历调度器。重复调用安全(先停旧的)。 func Start() { Stop() schedMu.Lock() defer schedMu.Unlock() schedStop = make(chan struct{}) stop := schedStop schedWg.Add(1) go func() { defer schedWg.Done() runLoop(stop) }() log.Printf("[scheduler] 日历调度器已启动(30s 周期)") } // Stop 停止调度器,等待当前扫描完成。 func Stop() { schedMu.Lock() defer schedMu.Unlock() if schedStop != nil { close(schedStop) schedWg.Wait() schedStop = nil log.Printf("[scheduler] 日历调度器已停止") } } func runLoop(stop chan struct{}) { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() // 启动时先扫一次:进程重启期间到期的事件不该被跳过 scanAndFire() for { select { case <-stop: return case <-ticker.C: scanAndFire() } } } // scanAndFire 扫一轮到期事件。 // // **全程不得 panic 出去**:它跑在后台 goroutine 里,而 Go 的 goroutine panic // 会直接结束整个进程 —— 一条写坏的提醒把邮件网关整个带倒是荒谬的代价。 func scanAndFire() { defer func() { if rec := recover(); rec != nil { log.Printf("[scheduler] 扫描崩溃(已捕获,下一轮重试): %v", rec) } }() // DB 未就绪就什么都不做。 // // 调度器的启动时机比它看起来脆弱:main 里它排在 migrate 之后, // 但 Start() 是个导出函数,测试与将来的调用方都可能在建库前调到它。 // 而 nil *sql.DB 上调 QueryContext 是空指针解引用,不是一个 error。 if db.DB == nil { return } ctx := context.Background() events, err := repo.DueEvents(ctx) if err != nil { log.Printf("[scheduler] 查询到期事件失败: %v", err) return } for _, e := range events { fireEvent(ctx, e) } } // RenderReminder 把提醒模板里的变量替换成事件的实际内容。 // // 单独提出来是为了可测:模板渲染错了不会立刻报错, // 只会让 Agent 收到一封写着 `{title}` 的邮件。 // // **{time} 必须转本地时区再格式化。** // DSN 带 `_timezone=UTC`,从库里读回的 EventTime 是 UTC;直接 Format // 会把人在 +0800 输入的 14:30 写成 06:30,而前端预览用的是本地时间 // —— 于是预览显示 14:30、Agent 收到 06:30,两边差 8 小时且两边都不报错。 // 日历是给人看的,人说「下午两点半」指的就是自己时区的那个时刻。 func RenderReminder(tmpl string, e models.CalendarEvent) string { body := tmpl body = strings.ReplaceAll(body, "{title}", e.Title) body = strings.ReplaceAll(body, "{time}", e.EventTime.Local().Format("2006-01-02 15:04")) body = strings.ReplaceAll(body, "{description}", e.Description) return body } // fireEvent 触发一条事件:渲染提醒文本、投递邮件、推进重复。 // // 顺序很重要:**先标记已触发再发信**。 // 反过来的话,发信成功但标记失败会让下一轮再发一遍 —— // 提醒邮件重复比漏发更糟(Agent 会把同一件事做两次)。 func fireEvent(ctx context.Context, e models.CalendarEvent) { // 单条事件崩溃不得让同一轮里剩下的提醒全部落空。 defer func() { if rec := recover(); rec != nil { log.Printf("[scheduler] 事件 %s 触发崩溃(已捕获): %v", short(e.EventID), rec) } }() // 收件人:Recipients 优先,为空时退回 to_address / agent_name(旧数据) recipients := e.EffectiveRecipients() if len(recipients) == 0 { log.Printf("[scheduler] 事件 %s 没有收件人,跳过(title=%q)", short(e.EventID), e.Title) // 没有收件人的事件永远发不出去,标记已触发免得每 30 秒重试一次 _ = repo.MarkEventFired(ctx, e.EventID) return } if err := repo.MarkEventFired(ctx, e.EventID); err != nil { log.Printf("[scheduler] 标记事件 %s 已触发失败,本轮不发信: %v", short(e.EventID), err) return } body := RenderReminder(e.ReminderText, e) subject := "日程提醒:" + e.Title switch e.EffectiveDeliveryMode() { case models.DeliverTogether: // 一起发:首个是主收件人,其余进 cc_list —— 所有人共享同一条线索, // 能看到彼此的回复。适合「pi 主办、dsh 知情」这种有主次的协作。 primary, cc := recipients[0], recipients[1:] if err := SendCalendarMail(ctx, e.EventID, primary, subject, body, e.CreatedBy, e.PermissionMode, cc...); err != nil { log.Printf("[scheduler] 投递提醒失败(%s → %s +%d抄送): %v", e.Title, primary, len(cc), err) } else { log.Printf("[scheduler] 已触发日历提醒: %s → %s(抄送 %d 人,同一线索)", e.Title, primary, len(cc)) } default: // 各发一封:每人落在自己的会话里,互相看不到。 // 适合「让三个 Agent 各自独立汇报」—— 用 together 会让他们互相 // 看到回复而趋同,那种上下文污染事后无法分离。 // // 一个失败不影响其余:三个 Agent 里有一个离线时, // 另外两个仍该收到提醒。 ok, failed := 0, 0 for _, addr := range recipients { if err := SendCalendarMail(ctx, e.EventID, addr, subject, body, e.CreatedBy, e.PermissionMode); err != nil { failed++ log.Printf("[scheduler] 投递提醒失败(%s → %s): %v", e.Title, addr, err) continue } ok++ } if failed == 0 { log.Printf("[scheduler] 已触发日历提醒: %s → %d 个收件人(各自独立会话)", e.Title, ok) } else { log.Printf("[scheduler] 日历提醒 %s:%d 成功 / %d 失败", e.Title, ok, failed) } } // 重复事件推进到下一次;一次性事件到此结束 if e.Recurrence != models.RecurNone { advanced, err := repo.AdvanceRecurrence(ctx, e.EventID) if err != nil { log.Printf("[scheduler] 推进重复事件 %s 失败: %v", short(e.EventID), err) } else if !advanced { log.Printf("[scheduler] 重复事件 %s 已过 recurrence_end 或无法推进,已置为 cancelled", short(e.EventID)) } } } func short(id string) string { if len(id) > 8 { return id[:8] } return id } // SendCalendarMail 把一条提醒投成邮件。 // // 直接走 repo + sse 而不是自己发一个 HTTP 请求到 /mail/send: // 调度器是进程内 goroutine,绕出去再进来只是多一次鉴权与序列化。 // // **日历提醒不扣会话预算**:预算的语义是「这件事值得模型自主发多少封信」, // 而提醒是人预先设定的定时任务,不是模型的自主行为。让它扣预算会出现 // 「每天 9 点的日报提醒把当天的预算吃掉一格」这种反直觉结果。 // // 收件地址支持完整三维寻址: // - `agent` → 该 Agent 的默认会话(长期提醒应该用这个, // 所有提醒落在同一条线索上,模型看得到历史) // - `agent@/path` → 指定工作目录的默认会话 // - `agent@/path.alias` → 指定已存在的会话(不存在则报错,不静默新建) // - `agent@/path.new` → 每次提醒开一条新会话(适合互不相关的一次性任务) func SendCalendarMail(ctx context.Context, eventID, toAddr, subject, body, createdBy, permMode string, ccAddrs ...string) error { addr, err := models.ParseAddress(toAddr) if err != nil { return fmt.Errorf("收件地址 %q 无法解析: %w", toAddr, err) } // 抄送方逐个解析。单个解析失败只跳过它,不让整封信发不出去 —— // 主收件人能收到提醒比「抄送名单必须完整」重要。 ccList := make([]models.Address, 0, len(ccAddrs)) for _, raw := range ccAddrs { raw = strings.TrimSpace(raw) if raw == "" { continue } ca, cErr := models.ParseAddress(raw) if cErr != nil { log.Printf("[scheduler] 抄送地址 %q 无法解析,已跳过: %v", raw, cErr) continue } // 抄送方的 session 位不参与会话定位(那是主收件人的事), // 但必须保留在地址里:cc_list 的 name/path 决定谁能看到这条线索。 ccList = append(ccList, ca) } sessionID, err := resolveCalendarSession(ctx, addr, subject, permMode) if err != nil { return err } // 人建的日程 → 把他设为会话 owner。 // // 为什么必须设:权限询问的决策人解析是「会话 owner → 线索里最近的人类 → 409」。 // 日历提醒的发件人是 `calendar`(不是人也不是 Agent),所以一旦 Agent 在 // 这条会话里要跑需要授权的命令,线索上根本找不到人类 —— 而那个日程 // 就是人自己在界面上设的,他当然是合理的决策人。 // // 不设的后果(删掉管理员兜底之后暴露):人建的提醒触发后,Agent 的权限询问 // 直接得 409「这条链上没有人类」。 // // created_by 是 Agent(Agent 自己建的日程)时 owner 保持为空 —— // 那条链上确实没有人类,409 是对的。 if u, uErr := repo.GetUserByName(ctx, createdBy); uErr == nil && u != nil { _ = repo.SetSessionOwner(ctx, sessionID, u.ID) } // 发件人固定为 calendar:它不是任何一个 Agent,也不是人。 // 用创建者的名字会让 Agent 以为人在实时找它,而人此刻可能在睡觉 —— // 模型据此判断「要不要马上追问」,来源写错会让它问一个不在线的人。 mailID, err := repo.CreateMail(ctx, sessionID, nil, calendarSender, "", addr.Name, addr.Path, subject, body, ccList) if err != nil { return err } // 事件附件复制成邮件附件。失败不阻断投递:提醒本身(正文)比附件重要得多, // 少一个附件也比整条提醒发不出去好 —— 后者会让人以为定时任务坏了。 if eventID != "" { if n, aErr := repo.AttachCalendarFilesToMail(ctx, eventID, mailID, calendarSender); aErr != nil { log.Printf("[scheduler] 事件 %s 的附件挂载失败(提醒仍已发出): %v", short(eventID), aErr) } else if n > 0 { log.Printf("[scheduler] 事件 %s 随提醒带了 %d 个附件", short(eventID), n) } } // ─── 推送 SSE ─── // // 这是整个系统里**唯一**的推送实现(handler 那条路径也是 notify.Recipients)。 // 原来这里自己拼了一整份 payload(handler 里是另一份), // 加 `platform_session_id` 时只改了那边 → 日历提醒投进接管会话时 // 插件不知道是接管、另开了一条新 pi 会话 → 命名同步把接管会话的别名冲掉 // → 人在补全里选的「项目定位」变成了「日程提醒:…」→ 选哪条都落进同一条。 // 根因只是「同一件事写了两遍」。 // // ReplyToName = addr.Name:日历提醒的回信要落回那条线索,不是回给 calendar // Origin = "calendar":让插件与 UI 能区分「这封是定时提醒」 notify.Recipients(ctx, notify.Mail{ SessionID: sessionID, MailID: mailID, From: calendarSender, To: addr, CC: ccList, Subject: subject, MailType: "normal", Origin: "calendar", ReplyToName: addr.Name, }) // 创建者不是收件方,不在 Recipients 的参与方去重里 —— 他不会收到 // new_mail(他不该被「有人在找你」打扰),但他应该看到会话活跃起来: // 「我设的提醒到底触发了没有」不应该只能去翻 journalctl。 notify.SessionActive(createdBy, sessionID) return nil } // calendarSender 是提醒邮件的发件人名。 // // 刻意不是人类用户名也不是 Agent 名:这样收件方一眼能看出这封信来自定时任务, // 而 `from_name` 又不会与命名空间里任何真实账号冲突。 const calendarSender = "calendar" // resolveCalendarSession 按地址的 session 位定位会话。 // // 与 handler.resolveTarget 同一套三态语义,但**不受新建会话速率限制**: // 那条限制是防 Agent 暴开线索的,而日历事件的数量由人在界面上决定。 // // 档位规则(见 PLAN 7.11 P1): // - 新建会话 → 用事件档位定死(permMode,已规范化) // - 复用已有会话 → ModeAtMost(会话现档, 事件档),取更严, // 不允许因复用而提权(plan 档 Agent 建的日程触发时拿 workspace 就绕开了 plan) func resolveCalendarSession(ctx context.Context, addr models.Address, subject, permMode string) (uuid.UUID, error) { // permMode 由调用方已规范化过,这里再保一次(直接调用本函数的路径上该一样) eventMode := models.NormalizePermissionMode(permMode) apply := func(sessionID uuid.UUID, created bool) { if created { _, _ = repo.SetSessionPermissionMode(ctx, sessionID, eventMode) _ = repo.SetSessionEnforcement(ctx, sessionID, repo.AgentModeEnforcement(ctx, addr.Name)) return } // 复用已有会话:取更严。ModeAtMost 已判过会话现档与事件档,取严的那个。 cur := repo.SessionPermissionMode(ctx, sessionID) merged := models.ModeAtMost(cur, eventMode) if merged != cur { _, _ = repo.SetSessionPermissionMode(ctx, sessionID, merged) } } switch addr.Mode() { case models.SessionNew: id, err := repo.CreateSession(ctx, nil, calendarSender, subject, addr.Path) if err != nil { return uuid.Nil, err } // 与发信路径一致:`.new` 建完必须立刻有别名,否则这条会话 // 除了回复那一封之外再也无法寻址(未命名会话查不到也补全不出来)。 _, _ = repo.EnsureSessionAlias(ctx, id, repo.AutoAliasFor(addr.Name, subject)) apply(id, true) return id, nil case models.SessionNamed: id, err := repo.FindNamedSessionFor(ctx, addr.Name, addr.Path, addr.Session) if errors.Is(err, repo.ErrSessionNotFound) { // 不静默新建:别名指向不存在的会话时报错,与人发信时的语义一致。 // 静默新建会让「提醒发到哪去了」变成一个查不清的问题。 return uuid.Nil, fmt.Errorf("会话 %q 不存在于 %s@%s", addr.Session, addr.Name, addr.Path) } if err != nil { return uuid.Nil, err } repo.TouchSession(ctx, id) apply(id, false) return id, nil default: // SessionDefault id, created, err := repo.FindOrCreateDefaultSessionCreated(ctx, addr.Name, addr.Path, calendarSender, subject) if err != nil { return uuid.Nil, err } _, _ = repo.EnsureSessionAlias(ctx, id, repo.AutoAliasFor(addr.Name, subject)) apply(id, created) return id, nil } }