// 定时任务(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() 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, ctx: GatewayRunContext): Promise { 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) { 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, } } }