import type { ConnectorTool, JobPublisher } from "@rakazo/adapter-kit"; import { routineJobKey, routineWakeupJob } from "@rakazo/adapter-kit"; import { cronFromPreset, isOneShotRoutineCron, isOneShotRoutineCrons, nextCronDate, ONCE_ROUTINE_CRON, } from "@rakazo/core"; import type { PrismaClient, ThreadEvents } from "@rakazo/db"; import { loadUserTimeZone } from "./current-time.js"; export { isOneShotRoutineCron, ONCE_ROUTINE_CRON }; export const SCHEDULE_TOOL_NAMES = new Set(["schedule_create", "schedule_list", "schedule_cancel"]); /** schedule_create as the main bot sees it: it may also schedule for one of the user's bots. */ export function withScheduleTargetBot(tool: ConnectorTool): ConnectorTool { const properties = (tool.inputSchema.properties ?? {}) as Record; return { ...tool, inputSchema: { ...tool.inputSchema, properties: { ...properties, bot: { type: "string", description: "Optional name or id of one of the user's specialists who should run this schedule. Omit to schedule for yourself.", }, }, }, }; } export function filterBuiltinToolsForRun( tools: T[], runTrigger: string, ): T[] { return runTrigger === "routine" ? tools.filter((tool) => tool.name !== "schedule_create") : tools; } /** Keep cross-bot messaging thread-specific while exposing schedules in DMs and groups. */ export function filterBuiltinToolsForThread( tools: T[], groupId: string | null | undefined, ): T[] { const inGroup = Boolean(groupId); return tools.filter((tool) => { if (tool.name === "handoff_to_bot") return inGroup; if (tool.name === "message_bot") return !inGroup; return true; }); } type ScheduleUnit = "minutes" | "hours" | "days" | "seconds"; export type ResolvedSchedule = | { ok: true; cron: string; nextRunAt: Date; oneShot: boolean } | { ok: false; error: string }; export function resolveScheduleTiming( input: { cron?: unknown; every?: unknown; unit?: unknown; runAt?: unknown; delayMinutes?: unknown; delaySeconds?: unknown; }, timezone = "UTC", ): ResolvedSchedule { const hasRepeat = input.cron !== undefined || input.every !== undefined || input.unit !== undefined; const hasOneShot = input.runAt !== undefined || input.delayMinutes !== undefined || input.delaySeconds !== undefined; if (hasRepeat && hasOneShot) { return { ok: false, error: "Provide either a repeating schedule or a one-shot time, not both.", }; } if (!hasRepeat && !hasOneShot) { return { ok: false, error: "Provide cron, every/unit for repeating schedules, or runAt/delayMinutes/delaySeconds for one-shot.", }; } if (hasOneShot) { const oneShotFields = [input.runAt, input.delayMinutes, input.delaySeconds].filter( (value) => value !== undefined, ); if (oneShotFields.length !== 1) { return { ok: false, error: "Provide exactly one of runAt, delayMinutes, or delaySeconds for a one-shot schedule.", }; } let nextRunAt: Date; if (input.delaySeconds !== undefined) { const delaySeconds = Number(input.delaySeconds); if (!Number.isFinite(delaySeconds) || delaySeconds < 0) { return { ok: false, error: "delaySeconds must be a non-negative number." }; } nextRunAt = new Date(Date.now() + delaySeconds * 1_000); } else if (input.delayMinutes !== undefined) { const delayMinutes = Number(input.delayMinutes); if (!Number.isFinite(delayMinutes) || delayMinutes < 0) { return { ok: false, error: "delayMinutes must be a non-negative number." }; } nextRunAt = new Date(Date.now() + delayMinutes * 60_000); } else { const parsed = new Date(String(input.runAt ?? "")); if (!Number.isFinite(parsed.getTime())) { return { ok: false, error: "runAt must be a valid ISO datetime." }; } nextRunAt = parsed; } if (nextRunAt.getTime() <= Date.now()) { return { ok: false, error: "One-shot schedules must run in the future." }; } return { ok: true, cron: ONCE_ROUTINE_CRON, nextRunAt, oneShot: true }; } if (input.cron !== undefined) { const cron = String(input.cron).trim(); if (!cron) return { ok: false, error: "cron must be a non-empty 5-field cron expression." }; if (isOneShotRoutineCron(cron)) { return { ok: false, error: "Use runAt, delayMinutes, or delaySeconds for one-shot schedules.", }; } const intervalError = validateRepeatingCron(cron); if (intervalError) return { ok: false, error: intervalError }; try { const nextRunAt = nextCronDate(cron, new Date(), timezone); return { ok: true, cron, nextRunAt, oneShot: false }; } catch { return { ok: false, error: "cron must be a valid 5-field cron expression." }; } } const every = Number(input.every); const unit = String(input.unit ?? "") as ScheduleUnit; if (!Number.isFinite(every) || every <= 0 || !Number.isInteger(every)) { return { ok: false, error: "every must be a positive whole number." }; } if (unit === "seconds") { return { ok: false, error: "minimum interval is 1 minute" }; } if (!["minutes", "hours", "days"].includes(unit)) { return { ok: false, error: 'unit must be "minutes", "hours", or "days".' }; } if (unit === "minutes" && every < 1) { return { ok: false, error: "minimum interval is 1 minute" }; } const cron = cronFromPreset({ freq: "Interval", n: every, unit }); const intervalError = validateRepeatingCron(cron); if (intervalError) return { ok: false, error: intervalError }; try { const nextRunAt = nextCronDate(cron, new Date(), timezone); return { ok: true, cron, nextRunAt, oneShot: false }; } catch { return { ok: false, error: "every/unit produced an invalid cron expression." }; } } function validateRepeatingCron(cron: string): string | null { const parts = cron.trim().split(/\s+/); if (parts.length !== 5) return "cron must be a 5-field cron expression."; const minuteExpr = parts[0] ?? "*"; const step = /^\*\/(\d+)$/.exec(minuteExpr); if (step) { const n = Number(step[1]); if (!Number.isFinite(n) || n < 1 || !Number.isInteger(n)) { return "minimum interval is 1 minute"; } } return null; } export interface ScheduleToolDeps { prisma: PrismaClient; events: ThreadEvents; jobs: JobPublisher; } /** The bot a schedule belongs to and the thread it wakes in. */ export interface ScheduleOwner { spaceId: string; botId: string; userId: string; threadId: string; /** Set when the main bot schedules for another bot. */ createdByBotId?: string; botName?: string; } /** * Without `botRef` a schedule belongs to the creating bot and wakes in its current thread. The * main bot may name one of the user's other bots in any space the user still belongs to: the * schedule then belongs to that bot in its space and wakes in the main conversation when the bot * is a member there, otherwise in the bot's 1:1 thread (`wakeRoutine` accepts only those). */ export async function resolveScheduleOwner( prisma: PrismaClient, creator: { botId: string; kind: string; spaceId: string; userId: string; threadId: string }, botRef: string | undefined, ): Promise { const own = { spaceId: creator.spaceId, botId: creator.botId, userId: creator.userId, threadId: creator.threadId, }; const ref = botRef?.trim(); if (!ref) return own; if (creator.kind !== "main") return { error: "Only the main agent can schedule for another bot." }; const candidates = await prisma.bot.findMany({ where: { userId: creator.userId, archivedAt: null, space: { deletingAt: null, memberships: { some: { userId: creator.userId } } }, }, select: { id: true, name: true, spaceId: true, thread: { select: { id: true } } }, }); const target = pickScheduleTarget(candidates, ref, creator.spaceId); if ("error" in target) return target; if (target.id === creator.botId) return own; if (!target.thread) return { error: `${target.name} has no chat to run in.` }; const mainConversation = await prisma.chatGroup.findFirst({ where: { userId: creator.userId, kind: "main", archivedAt: null, spaceId: target.spaceId, members: { some: { botId: target.id } }, }, select: { thread: { select: { id: true } } }, }); return { spaceId: target.spaceId, botId: target.id, userId: creator.userId, threadId: mainConversation?.thread?.id ?? target.thread.id, createdByBotId: creator.botId, botName: target.name, }; } /** Id first, then exact name, then a case-insensitive name; ties prefer the creator's space. */ function pickScheduleTarget( candidates: T[], ref: string, preferredSpaceId: string, ): T | { error: string } { const byId = candidates.find((candidate) => candidate.id === ref); if (byId) return byId; for (const matches of [ candidates.filter((candidate) => candidate.name === ref), candidates.filter((candidate) => candidate.name.toLowerCase() === ref.toLowerCase()), ]) { if (matches.length === 1) return matches[0]!; if (matches.length === 0) continue; const local = matches.filter((candidate) => candidate.spaceId === preferredSpaceId); if (local.length === 1) return local[0]!; return { error: `Several bots are named ${ref}. Use the bot id.` }; } return { error: `No bot named ${ref}.` }; } /** Persist and enqueue a bot routine that wakes in the originating thread. */ export async function createScheduleFromTool( deps: ScheduleToolDeps, input: ScheduleOwner & { name: string; prompt: string; timezone?: string; schedule: Record; }, ) { const name = input.name.trim(); const prompt = input.prompt.trim(); if (!name) return { error: "name is required." }; if (!prompt) return { error: "prompt is required." }; // Without an explicit zone, "every day at 9" means 9 in the user's time zone. const timezone = input.timezone ? String(input.timezone) : ((await loadUserTimeZone(deps.prisma, input.userId)) ?? "UTC"); const resolved = resolveScheduleTiming(input.schedule, timezone); if (!resolved.ok) return { error: resolved.error }; const row = await deps.prisma.routine.create({ data: { spaceId: input.spaceId, botId: input.botId, userId: input.userId, threadId: input.threadId, createdByBotId: input.createdByBotId ?? null, name, prompt, crons: [resolved.cron], timezone, notify: true, active: true, nextRunAt: resolved.nextRunAt, }, }); try { await deps.jobs.enqueue(routineWakeupJob(row.id, resolved.nextRunAt)); } catch { try { await deps.prisma.routine.delete({ where: { id: row.id } }); } catch { // Do not soft-return while the row may still be active. await deps.prisma.routine.update({ where: { id: row.id }, data: { active: false, nextRunAt: null }, }); } return { error: "Could not schedule the reminder. Try again." }; } try { await deps.events.append({ spaceId: input.spaceId, threadId: input.threadId, botId: input.botId, type: "routine.created", payload: { name: row.name }, }); } catch { // Match routines.create: the reminder is live even if the thread signal fails. } return { ok: true as const, routineId: row.id, name: row.name, cron: row.crons[0], nextRunAt: row.nextRunAt?.toISOString() ?? null, oneShot: resolved.oneShot, ...(input.createdByBotId ? { bot: input.botName } : {}), }; } /** * Routines a bot manages: its own (in a group only the ones of that thread) and the ones it * scheduled for other bots, wherever they wake. */ function managedRoutines(input: { spaceId: string; botId: string; userId: string; threadId?: string; }) { return { OR: [ { spaceId: input.spaceId, botId: input.botId, userId: input.userId, ...(input.threadId ? { threadId: input.threadId } : {}), }, { userId: input.userId, createdByBotId: input.botId }, ], }; } /** List the routines a bot manages; ones it scheduled for another bot name that bot. */ export async function listSchedulesFromTool( deps: Pick, input: { spaceId: string; botId: string; userId: string; threadId?: string }, ) { const rows = await deps.prisma.routine.findMany({ where: managedRoutines(input), orderBy: { createdAt: "desc" }, include: { bot: { select: { name: true } } }, }); return { routines: rows.map((row) => ({ routineId: row.id, name: row.name, prompt: row.prompt, crons: row.crons, active: row.active, nextRunAt: row.nextRunAt?.toISOString() ?? null, oneShot: isOneShotRoutineCrons(row.crons), ...(row.botId === input.botId ? {} : { bot: row.bot.name }), })), }; } /** Cancel a routine the bot manages by id or by a name that matches exactly one of them. */ export async function cancelScheduleFromTool( deps: ScheduleToolDeps, input: { spaceId: string; botId: string; userId: string; threadId?: string; routineId?: string; name?: string; }, ) { const routineId = input.routineId?.trim(); const name = input.name?.trim(); if (!routineId && !name) { return { error: "Provide routineId or the exact schedule name to cancel." }; } const matches = await deps.prisma.routine.findMany({ where: { ...managedRoutines(input), ...(routineId ? { id: routineId } : { name: name! }) }, take: 2, }); if (matches.length > 1) return { error: "Several schedules have that name. Cancel by routineId." }; const existing = matches[0]; if (!existing) return { error: "Schedule not found." }; await deps.prisma.routine.delete({ where: { id: existing.id } }); await deps.jobs.cancel(routineJobKey(existing.id)); return { ok: true as const, routineId: existing.id, name: existing.name }; }