AgentServer/apps/server/src/schedules/service.ts

// 定时任务（docs/design/14-schedules.md）：在对话中由 AI 创建（用户确认），属于一个会话或一个工作组；
// 调度器到点往同一个会话发出指令（开始一轮），或以系统消息发到工作组（按工作组的规则交给机器人）
import { and, desc, eq, inArray, isNotNull, lte } from 'drizzle-orm'
import type { AgentEvent, ScheduleDto, ScheduleOutcome, ScheduleOverviewDto, ScheduleRunDto, ScheduleSpec, ScheduleTarget } from '@agent-server/shared'
import type { AppLogger } from '../app'
import type { DB } from '../db'
import { runs, scheduleRuns, schedules, sessions, teams } from '../db/schema'
import { badRequest, conflict, notFound } from '../lib/errors'
import { newId, nowIso } from '../lib/ids'
import type { GatewayRunContext } from '../mcp-gateway/tokens'
import type { ProfileService } from '../services/catalog'
import type { EventStore } from '../services/event-store'
import type { RunManager } from '../services/runs'
import { isTeamSource, type SessionService } from '../services/sessions'
import type { SettingsService } from '../services/settings'
import type { BotService } from '../teams/bots'
import type { TeamService } from '../teams/service'
import { describeSpec, formatLocal, nextRun, parseSpec, upcoming, type ScheduleInput } from './spec'
import { scheduleTools } from './tools'

type ScheduleRow = typeof schedules.$inferSelect

const TICK_MS = 30_000

export class ScheduleService {
  /** 会话中正在执行的定时任务：runId → 执行记录 */
  private running = new Map<string, { scheduleId: string; recordId: number; timer: NodeJS.Timeout; timedOut: boolean }>()
  private timer: NodeJS.Timeout

  constructor(private deps: {
    db: DB; log: AppLogger; settings: SettingsService; sessions: SessionService; runs: RunManager; events: EventStore
    teams: TeamService; bots: BotService; profiles: ProfileService
  }) {
    // 服务重启时还在执行的记录：Run 已被标记为中断
    deps.db.update(scheduleRuns).set({ outcome: 'failed', note: '服务重启，本次执行中断' }).where(eq(scheduleRuns.outcome, 'running')).run()
    // 工具的说明中写明时区（设置修改后的新 Run 生效）
    deps.runs.gatewayToolProviders.push(() => scheduleTools(deps.settings.get().schedules.timezone, (name, args, ctx) => this.tool(name, args, ctx)))
    deps.events.subscribe('*', e => this.onEvent(e))
    this.timer = setInterval(() => void this.tick().catch(err => deps.log.error({ err }, 'schedule tick failed')), TICK_MS)
    this.timer.unref()
  }

  stop() {
    clearInterval(this.timer)
  }

  // ---- 查询与管理 ----

  list(target: ScheduleTarget): ScheduleDto[] {
    const where = target.kind === 'session' ? eq(schedules.sessionId, target.sessionId) : eq(schedules.teamId, target.teamId)
    return this.deps.db.select().from(schedules).where(where).orderBy(schedules.createdAt).all().map(r => this.toDto(r))
  }

  /** 后台总览 */
  overview(): ScheduleOverviewDto[] {
    const rows = this.deps.db.select().from(schedules).orderBy(desc(schedules.updatedAt)).all()
    const sessionIds = rows.flatMap(r => (r.sessionId ? [r.sessionId] : []))
    const teamIds = rows.flatMap(r => (r.teamId ? [r.teamId] : []))
    const sessionMap = new Map(sessionIds.length ? this.deps.db.select({ id: sessions.id, title: sessions.title, source: sessions.source }).from(sessions).where(inArray(sessions.id, sessionIds)).all().map(s => [s.id, s]) : [])
    const teamMap = new Map(teamIds.length ? this.deps.db.select({ id: teams.id, title: teams.title, source: teams.source }).from(teams).where(inArray(teams.id, teamIds)).all().map(t => [t.id, t]) : [])
    return rows.map(r => {
      const t = r.sessionId ? sessionMap.get(r.sessionId) : teamMap.get(r.teamId!)
      return { ...this.toDto(r), targetTitle: t?.title || '（已删除）', source: t?.source ?? '' }
    })
  }

  get(id: string): ScheduleRow {
    const row = this.deps.db.select().from(schedules).where(eq(schedules.id, id)).get()
    if (!row) throw notFound('定时任务')
    return row
  }

  /** 平台访问：只能操作自己的会话或工作组的定时任务 */
  getFor(id: string, check: (target: ScheduleTarget) => void): ScheduleRow {
    const row = this.get(id)
    check(this.targetOf(row))
    return row
  }

  runs(id: string, limit = 50): ScheduleRunDto[] {
    return this.deps.db.select().from(scheduleRuns).where(eq(scheduleRuns.scheduleId, id)).orderBy(desc(scheduleRuns.id)).limit(limit).all()
      .map(r => ({ id: r.id, scheduleId: r.scheduleId, ts: r.ts, outcome: r.outcome, runId: r.runId, costUsd: r.costUsd, note: r.note }))
  }

  /** 暂停或恢复；恢复时从现在起计算下次执行时间 */
  setStatus(id: string, status: 'active' | 'paused'): ScheduleDto {
    const row = this.get(id)
    if (row.status === 'ended') throw conflict('schedule_ended', '定时任务已结束')
    const next = status === 'active' ? nextRun(row.spec, row.timezone, Date.now()) : null
    if (status === 'active' && next === null) throw conflict('schedule_ended', '执行时间已经过去')
    this.patch(id, { status, statusNote: null, ...(status === 'active' ? { nextRunAt: new Date(next!).toISOString() } : {}) })
    return this.toDto(this.get(id))
  }

  delete(id: string) {
    this.get(id)
    this.deps.db.delete(schedules).where(eq(schedules.id, id)).run()
    this.deps.db.delete(scheduleRuns).where(eq(scheduleRuns.scheduleId, id)).run()
  }

  /** 会话或工作组删除时一起删除 */
  deleteFor(target: ScheduleTarget) {
    for (const s of this.list(target)) this.delete(s.id)
  }

  // ---- 调度 ----

  /** 执行到期的定时任务；停机期间错过的只补跑一次 */
  async tick(now = Date.now()) {
    const due = this.deps.db.select().from(schedules)
      .where(and(eq(schedules.status, 'active'), isNotNull(schedules.nextRunAt), lte(schedules.nextRunAt, new Date(now).toISOString()))).all()
    for (const row of due) {
      try {
        this.execute(row, now)
      } catch (err) {
        this.deps.log.error({ err, scheduleId: row.id }, 'schedule execution failed')
      }
    }
  }

  private execute(row: ScheduleRow, now: number) {
    const next = nextRun(row.spec, row.timezone, now)
    const text = `【定时任务「${row.title}」· ${formatLocal(now, row.timezone)}】\n${row.prompt}`
    let outcome: ScheduleOutcome
    let note: string | null = null
    let runId: string | null = null

    if (row.sessionId) {
      const session = this.deps.db.select().from(sessions).where(eq(sessions.id, row.sessionId)).get()
      if (!session) {
        this.delete(row.id)
        return
      }
      try {
        runId = this.deps.runs.start(session, text, []).id
        outcome = 'running'
      } catch (err) {
        const busy = (err as { code?: string }).code === 'session_busy'
        outcome = busy ? 'skipped' : 'failed'
        note = busy ? '会话正在执行，跳过本次' : (err as Error).message
      }
    } else {
      const team = this.deps.db.select().from(teams).where(eq(teams.id, row.teamId!)).get()
      if (!team) {
        this.delete(row.id)
        return
      }
      if (team.status !== 'active') {
        outcome = 'skipped'
        note = team.status === 'paused' ? '工作组已暂停，跳过本次' : '工作组已关闭，跳过本次'
      } else if (this.deps.teams.isBusy(team.id)) {
        outcome = 'skipped'
        note = '工作组还有进行中的任务或机器人在工作，跳过本次'
      } else {
        try {
          this.deps.teams.postScheduled(team.id, text, row.createdByBotId ? [row.createdByBotId] : [])
          outcome = 'sent'
        } catch (err) {
          outcome = 'failed'
          note = (err as Error).message
        }
      }
    }

    const record = this.deps.db.insert(scheduleRuns).values({ scheduleId: row.id, ts: new Date(now).toISOString(), outcome, runId, note }).returning({ id: scheduleRuns.id }).get()
    this.patch(row.id, {
      lastRunAt: new Date(now).toISOString(), lastOutcome: outcome, runCount: row.runCount + 1,
      nextRunAt: next === null ? null : new Date(next).toISOString(),
      ...(next === null ? { status: 'ended' as const } : {}),
    })
    if (runId) {
      const minutes = this.deps.settings.get().schedules.maxMinutesPerRun
      const entry = { scheduleId: row.id, recordId: record.id, timedOut: false, timer: setTimeout(() => {
        entry.timedOut = true
        this.deps.runs.interrupt(runId!, row.sessionId!)
      }, minutes * 60_000) }
      entry.timer.unref()
      this.running.set(runId, entry)
    }
  }

  /** 会话中的执行结束：记录结果和成本；成本超出上限时暂停 */
  private onEvent(e: AgentEvent) {
    if (e.type !== 'run.completed' && e.type !== 'run.failed' && e.type !== 'run.interrupted') return
    const entry = e.runId ? this.running.get(e.runId) : undefined
    if (!entry) return
    this.running.delete(e.runId!)
    clearTimeout(entry.timer)
    const cost = this.deps.db.select({ costUsd: runs.costUsd }).from(runs).where(eq(runs.id, e.runId!)).get()?.costUsd ?? null
    const outcome: ScheduleOutcome = e.type === 'run.completed' ? 'completed' : 'failed'
    const minutes = this.deps.settings.get().schedules.maxMinutesPerRun
    const note = entry.timedOut ? `超过 ${minutes} 分钟，已中断` : e.type === 'run.failed' ? e.payload.message : e.type === 'run.interrupted' ? '执行被中断' : null
    this.deps.db.update(scheduleRuns).set({ outcome, costUsd: cost, note }).where(eq(scheduleRuns.id, entry.recordId)).run()
    const row = this.deps.db.select().from(schedules).where(eq(schedules.id, entry.scheduleId)).get()
    if (!row) return
    const max = this.deps.settings.get().schedules.maxCostUsdPerRun
    const over = max !== null && cost !== null && cost > max
    this.patch(row.id, {
      lastOutcome: outcome,
      ...(over && row.status === 'active' ? { status: 'paused' as const, statusNote: `本次执行花费 $${cost!.toFixed(2)}，超过每次的上限 $${max}，已自动暂停` } : {}),
    })
  }

  // ---- 工具（AI 在对话中调用） ----

  private async tool(name: string, args: Record<string, unknown>, ctx: GatewayRunContext): Promise<string> {
    const target = this.targetOfSession(ctx.sessionId)
    const createdBy = target.kind === 'team' ? this.deps.teams.teamOfSession(ctx.sessionId)?.botId ?? null : null
    const s = this.deps.settings.get().schedules
    switch (name) {
      case 'schedule_list': {
        const list = this.list(target)
        const now = `当前时间：${formatLocal(Date.now(), s.timezone)}（${s.timezone}）`
        return `${now}\n\n${list.length ? list.map(x => this.summary(x)).join('\n\n') : '当前没有定时任务。'}`
      }
      case 'schedule_create': {
        if (this.list(target).filter(x => x.status !== 'ended').length >= s.maxPerTarget) throw new Error(`最多 ${s.maxPerTarget} 个定时任务，请先删除不需要的`)
        const title = String(args.title ?? '').trim()
        const prompt = String(args.prompt ?? '').trim()
        const spec = parseSpec(args.schedule as ScheduleInput, { minIntervalHours: s.minIntervalHours, timezone: s.timezone })
        const ok = await this.confirm(ctx, target, 'schedule_create', `创建定时任务「${title}」`, title, prompt, spec, s.timezone)
        if (!ok.allowed) return `用户没有确认，定时任务没有创建${ok.reason ? `：${ok.reason}` : ''}。`
        const next = nextRun(spec, s.timezone, Date.now())
        const id = newId('sch')
        this.deps.db.insert(schedules).values({
          id, title, prompt, spec, timezone: s.timezone, nextRunAt: next === null ? null : new Date(next).toISOString(),
          createdByBotId: createdBy,
          ...(target.kind === 'session' ? { sessionId: target.sessionId } : { teamId: target.teamId }),
        }).run()
        return `已创建定时任务（id: ${id}）：${describeSpec(spec)}，下次执行 ${next === null ? '—' : formatLocal(next, s.timezone)}。`
      }
      case 'schedule_update': {
        const row = this.ownRow(String(args.schedule_id), target)
        const title = typeof args.title === 'string' && args.title.trim() ? args.title.trim() : row.title
        const prompt = typeof args.prompt === 'string' && args.prompt.trim() ? args.prompt.trim() : row.prompt
        const spec = args.schedule ? parseSpec(args.schedule as ScheduleInput, { minIntervalHours: s.minIntervalHours, timezone: row.timezone }) : row.spec
        const changed = title !== row.title || prompt !== row.prompt || JSON.stringify(spec) !== JSON.stringify(row.spec)
        if (changed) {
          if (row.status === 'ended') throw new Error('定时任务已结束，请重新创建')
          const ok = await this.confirm(ctx, target, 'schedule_update', `修改定时任务「${row.title}」`, title, prompt, spec, row.timezone)
          if (!ok.allowed) return `用户没有确认，定时任务没有修改${ok.reason ? `：${ok.reason}` : ''}。`
          const next = nextRun(spec, row.timezone, Date.now())
          this.patch(row.id, { title, prompt, spec, nextRunAt: next === null ? null : new Date(next).toISOString() })
        }
        if (args.status === 'active' || args.status === 'paused') this.setStatus(row.id, args.status)
        return `已更新：\n${this.summary(this.toDto(this.get(row.id)))}`
      }
      case 'schedule_delete': {
        const row = this.ownRow(String(args.schedule_id), target)
        this.delete(row.id)
        return `已删除定时任务「${row.title}」。`
      }
    }
    throw new Error(`未知的工具：${name}`)
  }

  /** 确认卡片：时间表、接下来的执行时间、指令；无人值守时会遇到审批的给出提示 */
  private async confirm(ctx: GatewayRunContext, target: ScheduleTarget, tool: string, heading: string, title: string, prompt: string, spec: ScheduleSpec, timezone: string) {
    if (!title) throw new Error('请填写 title')
    if (!prompt) throw new Error('请填写 prompt：到点时要执行的完整指令')
    const next = upcoming(spec, timezone, Date.now())
    const body = [
      `时间：${describeSpec(spec)}（${timezone}）`,
      `接下来：${next.length ? next.map(t => formatLocal(t, timezone)).join('、') : '—'}`,
      '',
      `到点时发出的指令：\n${prompt}`,
    ].join('\n')
    const outcome = await ctx.interact({
      kind: 'approval', tool, title: `${heading}：${describeSpec(spec)}`, input: { title, prompt, schedule: spec },
      detail: { command: body, ...(this.needsApproval(target) ? { reason: `无人值守时，需要审批的操作会在 ${Math.round(this.deps.settings.get().limits.interactionTimeoutSec / 60)} 分钟后被拒绝` } : {}) },
      options: ['allow_once', 'deny'],
    })
    return { allowed: outcome.resolution === 'allowed', reason: outcome.resolution === 'denied' ? outcome.reason : undefined }
  }

  /** 执行时可能要人审批：会话的 Profile（或工作组中任一机器人的 Profile）为"需审批" */
  private needsApproval(target: ScheduleTarget): boolean {
    if (target.kind === 'session') {
      const session = this.deps.sessions.getRow(target.sessionId)
      return this.deps.sessions.effectiveProfile(session).permissionMode === 'ask'
    }
    return this.deps.teams.toDto(this.deps.teams.getRow(target.teamId)).bots.some(b => {
      try {
        return this.deps.profiles.get(this.deps.bots.get(b.id).profileId).permissionMode === 'ask'
      } catch {
        return false
      }
    })
  }

  private ownRow(id: string, target: ScheduleTarget): ScheduleRow {
    const row = this.deps.db.select().from(schedules).where(eq(schedules.id, id)).get()
    const t = row && this.targetOf(row)
    const same = t && (t.kind === 'session' ? target.kind === 'session' && t.sessionId === target.sessionId : target.kind === 'team' && t.teamId === target.teamId)
    if (!row || !same) throw new Error('定时任务不存在，请先调用 schedule_list 查看')
    return row
  }

  private summary(x: ScheduleDto): string {
    const status = { active: '进行中', paused: `已暂停${x.statusNote ? `（${x.statusNote}）` : ''}`, ended: '已结束' }[x.status]
    const next = x.nextRunAt && x.status === 'active' ? `，下次 ${formatLocal(Date.parse(x.nextRunAt), x.timezone)}` : ''
    return `- ${x.title}（id: ${x.id}）：${x.description} · ${status}${next} · 已执行 ${x.runCount} 次\n  指令：${x.prompt}`
  }

  // ---- 内部 ----

  private targetOfSession(sessionId: string): ScheduleTarget {
    const session = this.deps.sessions.getRow(sessionId)
    if (isTeamSource(session.source)) {
      const ref = this.deps.teams.teamOfSession(sessionId)
      if (!ref) throw badRequest('这个会话不属于任何工作组')
      return { kind: 'team', teamId: ref.teamId }
    }
    return { kind: 'session', sessionId }
  }

  private targetOf(r: ScheduleRow): ScheduleTarget {
    return r.sessionId ? { kind: 'session', sessionId: r.sessionId } : { kind: 'team', teamId: r.teamId! }
  }

  private patch(id: string, set: Partial<ScheduleRow>) {
    this.deps.db.update(schedules).set({ ...set, updatedAt: nowIso() }).where(eq(schedules.id, id)).run()
  }

  private toDto(r: ScheduleRow): ScheduleDto {
    return {
      id: r.id, target: this.targetOf(r), title: r.title, prompt: r.prompt, spec: r.spec, timezone: r.timezone, description: describeSpec(r.spec),
      status: r.status, statusNote: r.statusNote, nextRunAt: r.status === 'active' ? r.nextRunAt : null, lastRunAt: r.lastRunAt, lastOutcome: r.lastOutcome,
      runCount: r.runCount, createdByBotId: r.createdByBotId, createdAt: r.createdAt, updatedAt: r.updatedAt,
    }
  }
}