AgentServer/apps/server/test/schedules.test.ts

// 定时任务（docs/design/14-schedules.md）：时间表计算；在对话中创建（确认卡片）、到点执行、跳过、成本超限暂停；工作组中的定时任务
import pino from 'pino'
import { afterAll, beforeAll, describe, expect, it } from 'vitest'
import type { InteractionRequest, ProviderStatusDto, ScheduleDto, TeamDetailDto, TeamEvent, TeamMessage } from '@agent-server/shared'
import { createApp, type App } from '../src/app'
import { buildServer } from '../src/http/server'
import type { AgentProvider, RunContext } from '../src/providers/types'
import { describeSpec, nextRun, parseSpec } from '../src/schedules/spec'

type Call = (tool: string, args: Record<string, unknown>) => Promise<{ text: string; isError: boolean }>

/** 假引擎：输入中带"【定时任务"时按 onScheduled 执行，否则按 script 执行（可以调用网关工具） */
class ScriptedProvider implements AgentProvider {
  readonly id = 'claude' as const
  script: (input: string, call: Call) => Promise<string> = async () => 'ok'
  inputs: { sessionId: string; text: string }[] = []
  cost = 0.01
  /** 为 true 时一直执行到被中断 */
  hang = false
  async run(ctx: RunContext) {
    ctx.setProviderSessionId(`fake-${ctx.sessionId}`)
    this.inputs.push({ sessionId: ctx.sessionId, text: ctx.input.text })
    const gctx = app.gatewayTokens.get(ctx.gateway.token)!
    const call: Call = async (tool, args) => {
      const t = gctx.extraTools?.find(x => x.name === tool)
      if (!t) return { text: `没有工具 ${tool}`, isError: true }
      const r = await t.handler(t.schema.parse(args) as Record<string, unknown>, gctx)
      return { text: String(r.content[0]?.text ?? ''), isError: !!r.isError }
    }
    if (this.hang) {
      await new Promise((_, reject) => ctx.signal.addEventListener('abort', () => reject(new Error('aborted'))))
    }
    const reply = await this.script(ctx.input.text, call)
    ctx.emit('message.completed', { messageId: `m${this.inputs.length}`, text: reply })
    return { usage: { inputTokens: 10, outputTokens: 5, cacheReadTokens: 0, cacheWriteTokens: 0, costUsd: this.cost } }
  }
  async status(): Promise<ProviderStatusDto> {
    return { id: 'claude', enabled: true, available: true, loggedIn: true, account: 'test', cliPath: '', version: 'x', models: ['m'], defaultModel: 'm', message: null }
  }
}

let app: App
let server: Awaited<ReturnType<typeof buildServer>>
const fake = new ScriptedProvider()
let apiKey: string
let otherKey: string
let botKey: string
const auth = (key = apiKey) => ({ authorization: `Bearer ${key}` })

async function waitFor<T>(fn: () => T | undefined | false | null, ms = 4000): Promise<T> {
  const end = Date.now() + ms
  while (Date.now() < end) {
    const v = fn()
    if (v) return v
    await new Promise(r => setTimeout(r, 20))
  }
  throw new Error('timeout')
}

const DAILY = { kind: 'daily', time: '09:00' }
const later = (hours: number) => Date.now() + hours * 3_600_000

beforeAll(async () => {
  const log = pino({ level: 'silent' })
  app = createApp(log, { providers: { claude: fake } })
  server = await buildServer(app, { logger: log })
  const builtin = app.profiles.getBuiltin().id
  apiKey = app.apiKeys.create('平台A', { allowedProfileIds: [builtin], defaultProfileId: builtin }).key
  otherKey = app.apiKeys.create('平台B', { allowedProfileIds: [builtin], defaultProfileId: builtin }).key
  const bot = app.bots.create({ name: '日报员', avatar: 'kl:bot:blue', kind: 'worker', description: '写日报', profileId: builtin, enabled: true })
  botKey = app.botKeys.create('平台A', [bot.id]).key
})
afterAll(() => {
  app.schedules.stop()
  return server.close()
})

describe('时间表', () => {
  const tz = 'Asia/Shanghai'
  // 2026-10-27 是周二；北京时间 10:00 = UTC 02:00
  const now = Date.parse('2026-10-27T02:00:00Z')
  const at = (ms: number | null) => (ms === null ? null : new Date(ms).toISOString())

  it('按时区计算下次执行时间', () => {
    expect(at(nextRun({ kind: 'daily', time: '09:00' }, tz, now))).toBe('2026-10-28T01:00:00.000Z')
    expect(at(nextRun({ kind: 'daily', time: '11:30' }, tz, now))).toBe('2026-10-27T03:30:00.000Z')
    // 周五之后的工作日是下周一
    expect(at(nextRun({ kind: 'weekdays', time: '09:00' }, tz, Date.parse('2026-10-30T02:00:00Z')))).toBe('2026-11-02T01:00:00.000Z')
    expect(at(nextRun({ kind: 'weekly', days: [1, 5], time: '09:00' }, tz, now))).toBe('2026-10-30T01:00:00.000Z')
    // 每月 31 日：11 月没有 31 日，取 30 日
    expect(at(nextRun({ kind: 'monthly', day: 31, time: '09:00' }, tz, Date.parse('2026-11-01T00:00:00Z')))).toBe('2026-11-30T01:00:00.000Z')
    expect(at(nextRun({ kind: 'interval', hours: 2 }, tz, now))).toBe('2026-10-27T04:00:00.000Z')
    expect(at(nextRun({ kind: 'once', at: '2026-10-27 18:00' }, tz, now))).toBe('2026-10-27T10:00:00.000Z')
    expect(nextRun({ kind: 'once', at: '2026-10-27 08:00' }, tz, now)).toBeNull()
    // 其他时区
    expect(at(nextRun({ kind: 'daily', time: '09:00' }, 'America/New_York', now))).toBe('2026-10-27T13:00:00.000Z')
  })

  it('校验与描述', () => {
    const opts = { minIntervalHours: 1, timezone: tz, now }
    expect(parseSpec({ kind: 'daily', time: '9:05' }, opts)).toEqual({ kind: 'daily', time: '09:05' })
    // 相对时间：北京时间 10:00 的 90 分钟后
    expect(parseSpec({ kind: 'once', after_minutes: 90 }, opts)).toEqual({ kind: 'once', at: '2026-10-27 11:30' })
    expect(() => parseSpec({ kind: 'daily', time: '25:00' }, opts)).toThrow('HH:mm')
    expect(() => parseSpec({ kind: 'interval', hours: 0.5 }, opts)).toThrow('不能小于 1')
    expect(() => parseSpec({ kind: 'once', at: '2026-10-27 08:00' }, opts)).toThrow('已经过去')
    expect(() => parseSpec({ kind: 'once', at: '2026-02-30 08:00' }, { ...opts, now: Date.parse('2026-01-01T00:00:00Z') })).toThrow('不存在')
    expect(() => parseSpec({ kind: 'weekly', days: [], time: '09:00' }, opts)).toThrow('days')
    expect(describeSpec({ kind: 'weekly', days: [1, 3, 5], time: '09:00' })).toBe('每周一、三、五 09:00')
    expect(describeSpec({ kind: 'weekdays', time: '09:00' })).toBe('每个工作日（周一到周五）09:00')
  })
})

describe('会话中的定时任务', () => {
  let sessionId: string
  let created: { text: string; isError: boolean } | undefined

  async function approveNext(decision: 'allow_once' | 'deny', headers = auth()) {
    const itx = await waitFor(() => app.interactions.pendingFor(sessionId)[0]) as InteractionRequest
    expect(itx.kind === 'approval' && itx.tool).toBe('schedule_create')
    expect(itx.kind === 'approval' && itx.detail?.command).toContain('接下来：')
    const r = await server.inject({ method: 'POST', url: `/v1/interactions/${itx.interactionId}`, headers, payload: { decision } })
    expect(r.statusCode).toBe(200)
  }

  it('AI 创建定时任务时显示确认卡片，拒绝时不创建，确认后生效', async () => {
    sessionId = (await server.inject({ method: 'POST', url: '/v1/sessions', headers: auth(), payload: {} })).json().id
    fake.script = async (_input, call) => {
      created = await call('schedule_create', { title: '销售日报', prompt: '汇总前一天的销售数据，写到 outputs/daily.md', schedule: DAILY })
      return '好的'
    }
    await server.inject({ method: 'POST', url: `/v1/sessions/${sessionId}/messages`, headers: auth(), payload: { text: '每天 9 点帮我出日报' } })
    await approveNext('deny')
    await waitFor(() => created)
    expect(created!.text).toContain('没有确认')
    expect(app.schedules.list({ kind: 'session', sessionId })).toHaveLength(0)

    created = undefined
    await waitFor(() => !app.runs.activeForSession(sessionId))
    await server.inject({ method: 'POST', url: `/v1/sessions/${sessionId}/messages`, headers: auth(), payload: { text: '确认，每天 9 点' } })
    await approveNext('allow_once')
    await waitFor(() => created)
    expect(created!.text).toContain('已创建定时任务')
    const list = (await server.inject({ method: 'GET', url: `/v1/sessions/${sessionId}/schedules`, headers: auth() })).json() as ScheduleDto[]
    expect(list).toHaveLength(1)
    expect(list[0]).toMatchObject({ title: '销售日报', description: '每天 09:00', status: 'active', timezone: 'Asia/Shanghai', target: { kind: 'session', sessionId } })
    expect(list[0]!.nextRunAt).toBeTruthy()
    // 其他平台看不到、改不了
    expect((await server.inject({ method: 'GET', url: `/v1/sessions/${sessionId}/schedules`, headers: auth(otherKey) })).statusCode).toBe(404)
    expect((await server.inject({ method: 'DELETE', url: `/v1/schedules/${list[0]!.id}`, headers: auth(otherKey) })).statusCode).toBe(404)
  })

  it('到点时往同一个会话发出指令；会话正在执行时跳过；成本超出上限时自动暂停', async () => {
    await waitFor(() => !app.runs.activeForSession(sessionId))
    const [s] = app.schedules.list({ kind: 'session', sessionId })
    fake.script = async () => '日报已生成'
    const before = fake.inputs.length
    await app.schedules.tick(later(25))
    const input = await waitFor(() => fake.inputs.length > before && fake.inputs.at(-1))
    expect(input.sessionId).toBe(sessionId)
    expect(input.text).toMatch(/^【定时任务「销售日报」· .+】\n汇总前一天的销售数据/)
    await waitFor(() => app.schedules.runs(s!.id)[0]?.outcome === 'completed')
    expect(app.schedules.runs(s!.id)[0]).toMatchObject({ outcome: 'completed', costUsd: 0.01 })
    const after = app.schedules.list({ kind: 'session', sessionId })[0]!
    expect(after).toMatchObject({ runCount: 1, lastOutcome: 'completed', status: 'active' })
    expect(Date.parse(after.nextRunAt!)).toBeGreaterThan(Date.now())

    // 会话正在执行时跳过
    fake.hang = true
    await server.inject({ method: 'POST', url: `/v1/sessions/${sessionId}/messages`, headers: auth(), payload: { text: '做别的' } })
    await waitFor(() => app.runs.activeForSession(sessionId))
    await app.schedules.tick(later(50))
    expect(app.schedules.runs(s!.id)[0]).toMatchObject({ outcome: 'skipped', note: '会话正在执行，跳过本次' })
    app.runs.interrupt(app.runs.activeForSession(sessionId)!.runId, sessionId)
    fake.hang = false
    await waitFor(() => !app.runs.activeForSession(sessionId))

    // 成本超出上限：执行完成后暂停
    app.settings.update({ schedules: { maxCostUsdPerRun: 0.05 } })
    fake.cost = 0.2
    await app.schedules.tick(later(75))
    const paused = await waitFor(() => app.schedules.list({ kind: 'session', sessionId }).find(x => x.status === 'paused'))
    expect(paused.statusNote).toContain('超过每次的上限 $0.05')
    expect(paused.nextRunAt).toBeNull()
    fake.cost = 0.01

    // 平台可以恢复、删除
    const resumed = await server.inject({ method: 'PATCH', url: `/v1/schedules/${s!.id}`, headers: auth(), payload: { status: 'active' } })
    expect(resumed.json()).toMatchObject({ status: 'active', statusNote: null })
    const runsRes = (await server.inject({ method: 'GET', url: `/v1/schedules/${s!.id}/runs`, headers: auth() })).json()
    expect(runsRes.map((r: { outcome: string }) => r.outcome)).toEqual(['completed', 'skipped', 'completed'])
    expect((await server.inject({ method: 'DELETE', url: `/v1/schedules/${s!.id}`, headers: auth() })).statusCode).toBe(204)
    expect(app.schedules.list({ kind: 'session', sessionId })).toHaveLength(0)
  })

  it('AI 可以列出、暂停和删除；删除会话时定时任务一起删除', async () => {
    await waitFor(() => !app.runs.activeForSession(sessionId))
    const results: Record<string, string> = {}
    fake.script = async (input, call) => {
      if (input.includes('建一个')) results.create = (await call('schedule_create', { title: '周报', prompt: '写周报', schedule: { kind: 'weekly', days: [1], time: '10:00' } })).text
      if (input.includes('暂停')) {
        const id = app.schedules.list({ kind: 'session', sessionId })[0]!.id
        results.pause = (await call('schedule_update', { schedule_id: id, status: 'paused' })).text
        results.list = (await call('schedule_list', {})).text
      }
      return 'ok'
    }
    await server.inject({ method: 'POST', url: `/v1/sessions/${sessionId}/messages`, headers: auth(), payload: { text: '建一个周报' } })
    await approveNext('allow_once')
    await waitFor(() => results.create && !app.runs.activeForSession(sessionId))
    await server.inject({ method: 'POST', url: `/v1/sessions/${sessionId}/messages`, headers: auth(), payload: { text: '暂停周报' } })
    await waitFor(() => results.list)
    // 暂停不需要确认
    expect(results.pause).toContain('已暂停')
    expect(results.list).toContain('每周一 10:00')
    await waitFor(() => !app.runs.activeForSession(sessionId))
    expect((await server.inject({ method: 'DELETE', url: `/v1/sessions/${sessionId}`, headers: auth() })).statusCode).toBe(204)
    expect(app.schedules.overview().filter(x => x.target.kind === 'session' && x.target.sessionId === sessionId)).toHaveLength(0)
  })
})

describe('工作组中的定时任务', () => {
  it('组员创建的定时任务到点时 @ 它；工作组忙时跳过；后台可以查看全部', async () => {
    const bot = app.bots.list()[0]!
    const team = (await server.inject({ method: 'POST', url: '/bot/v1/teams', headers: auth(botKey), payload: { human: { id: 'u1', name: '张三' }, workerBotIds: [bot.id] } })).json() as TeamDetailDto
    let created: { text: string; isError: boolean } | undefined
    fake.script = async (input, call) => {
      if (input.includes('每天')) created = await call('schedule_create', { title: '日报', prompt: '写日报', schedule: DAILY })
      return input.includes('【定时任务') ? '日报写好了' : '好的'
    }
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, headers: auth(botKey), payload: { text: '每天 9 点写日报' } })
    const pending = await waitFor(() => app.teams.detail(app.teams.getRow(team.id)).pendingInteractions[0])
    expect(pending.request.kind === 'approval' && pending.request.tool).toBe('schedule_create')
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/interactions/${pending.request.interactionId}`, headers: auth(botKey), payload: { decision: 'allow_once' } })
    await waitFor(() => created)
    const list = (await server.inject({ method: 'GET', url: `/bot/v1/teams/${team.id}/schedules`, headers: auth(botKey) })).json() as ScheduleDto[]
    expect(list[0]).toMatchObject({ title: '日报', target: { kind: 'team', teamId: team.id }, createdByBotId: bot.id })
    await waitFor(() => app.teams.toDto(app.teams.getRow(team.id)).bots.every(b => b.status === 'idle'))

    const messages = () => app.teams.events.list(team.id, 0, 5000, ['message.created']).map(e => (e as TeamEvent & { type: 'message.created' }).payload as TeamMessage)
    await app.schedules.tick(later(25))
    const sent = messages().find(m => m.text.startsWith('【定时任务「日报」'))!
    expect(sent).toMatchObject({ sender: { kind: 'system' }, mentions: [bot.id], targets: [bot.id] })
    await waitFor(() => messages().some(m => m.text === '日报写好了'))
    expect(app.schedules.runs(list[0]!.id)[0]!.outcome).toBe('sent')

    // 机器人在工作时跳过
    fake.hang = true
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/messages`, headers: auth(botKey), payload: { text: '先做别的' } })
    await waitFor(() => app.teams.isBusy(team.id))
    await app.schedules.tick(later(50))
    expect(app.schedules.runs(list[0]!.id)[0]!.outcome).toBe('skipped')
    await server.inject({ method: 'POST', url: `/bot/v1/teams/${team.id}/bots/${bot.id}/interrupt`, headers: auth(botKey) })
    fake.hang = false

    const overview = (await server.inject({ method: 'GET', url: '/admin/v1/schedules', cookies: await adminCookie() })).json()
    expect(overview.find((x: { id: string }) => x.id === list[0]!.id)).toMatchObject({ targetTitle: team.title, source: expect.any(String) })
    expect((await server.inject({ method: 'DELETE', url: `/bot/v1/teams/${team.id}/schedules/${list[0]!.id}`, headers: auth(botKey) })).statusCode).toBe(204)
  })
})

async function adminCookie() {
  const login = await server.inject({ method: 'POST', url: '/admin/v1/auth/login', payload: { password: 'test-password' } })
  const c = login.cookies[0]!
  return { [c.name]: c.value }
}