evOo-local/packages/adapters/src/schedule-tools.ts

import type { ConnectorTool, JobPublisher } from "@rakazo/adapter-kit";
import { routineJobKey, routineWakeupJob } from "@rakazo/adapter-kit";
import type { RoutineEventBlock } from "@rakazo/contracts";
import {
  cronFromPreset,
  isOneShotRoutineCron,
  isOneShotRoutineCrons,
  nextCronDate,
  ONCE_ROUTINE_CRON,
} from "@rakazo/core";
import type { PrismaClient, ThreadEvents } from "@rakazo/db";
import { markRoutineNoticesRemoved, postRoutineNotice } from "@rakazo/db";
import { getLogger } from "@rakazo/logging";
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<string, unknown>;
  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. Omit to run it yourself.",
        },
      },
    },
  };
}

/** Tools only an autopilot's own runs report through. */
const AUTOPILOT_RUN_TOOLS = new Set(["autopilot_checkpoint", "autopilot_finish"]);

export function filterBuiltinToolsForRun<T extends { name: string }>(
  tools: T[],
  runTrigger: string,
): T[] {
  return tools.filter((tool) => {
    if (tool.name === "schedule_create") return runTrigger !== "routine";
    if (AUTOPILOT_RUN_TOOLS.has(tool.name)) return runTrigger === "autopilot";
    // Only the user starts an autopilot, never a bot, a routine or a webhook.
    if (tool.name === "autopilot_start") return runTrigger === "user" || runTrigger === "autopilot";
    return true;
  });
}

/** Keep cross-bot messaging thread-specific while exposing schedules in DMs and groups. */
export function filterBuiltinToolsForThread<T extends { name: string }>(
  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<ScheduleOwner | { error: string }> {
  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<T extends { id: string; name: string; spaceId: string }>(
  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}.` };
}

/** More active routines than one bot can sensibly keep track of. */
export const MAX_ACTIVE_ROUTINES_PER_BOT = 50;

/** Where the bot that set a routine up writes its timeline line. */
export interface ScheduleNotice {
  threadId: string;
  botId: string;
}

/**
 * Persist and enqueue a bot routine that wakes in the originating thread. A repeating schedule
 * with the name of one of the bot's active repeating routines changes that routine instead of
 * adding a second one. With `notice`, the conversation gets a quiet line about it.
 */
export async function createScheduleFromTool(
  deps: ScheduleToolDeps,
  input: ScheduleOwner & {
    name: string;
    prompt: string;
    timezone?: string;
    schedule: Record<string, unknown>;
    notice?: ScheduleNotice;
  },
) {
  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 };

  if (!resolved.oneShot) {
    const existing = await deps.prisma.routine.findFirst({
      where: {
        botId: input.botId,
        userId: input.userId,
        active: true,
        coachKey: null,
        name: { equals: name, mode: "insensitive" },
        NOT: { crons: { has: ONCE_ROUTINE_CRON } },
      },
    });
    if (existing) {
      return updateScheduleFromTool(deps, input, existing.id, {
        prompt,
        timezone,
        cron: resolved.cron,
        nextRunAt: resolved.nextRunAt,
      });
    }
  }

  const activeCount = await deps.prisma.routine.count({
    where: { botId: input.botId, active: true },
  });
  if (activeCount >= MAX_ACTIVE_ROUTINES_PER_BOT) {
    return {
      error: `This bot already has ${MAX_ACTIVE_ROUTINES_PER_BOT} active schedules. Cancel one before adding another.`,
    };
  }

  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.
  }
  await postNotice(deps, input.notice, input.spaceId, {
    kind: "event",
    event: "routine.created",
    routineId: row.id,
    botId: input.botId,
    name: row.name,
    crons: row.crons,
    timezone: row.timezone,
    nextRunAt: row.nextRunAt?.toISOString() ?? null,
  });

  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 } : {}),
  };
}

async function updateScheduleFromTool(
  deps: ScheduleToolDeps,
  input: ScheduleOwner & { notice?: ScheduleNotice },
  routineId: string,
  change: { prompt: string; timezone: string; cron: string; nextRunAt: Date },
) {
  const row = await deps.prisma.routine.update({
    where: { id: routineId },
    data: {
      prompt: change.prompt,
      timezone: change.timezone,
      crons: [change.cron],
      nextRunAt: change.nextRunAt,
    },
  });
  try {
    await deps.jobs.enqueue(routineWakeupJob(row.id, change.nextRunAt));
  } catch {
    // The job reconciler re-enqueues active routines whose wakeup is missing.
  }
  try {
    await deps.events.append({
      spaceId: row.spaceId,
      threadId: row.threadId ?? input.threadId,
      botId: row.botId,
      type: "routine.updated",
      payload: { name: row.name },
    });
  } catch {
    // The routine is updated even if the thread signal fails.
  }
  await postNotice(deps, input.notice, input.spaceId, {
    kind: "event",
    event: "routine.updated",
    routineId: row.id,
    botId: row.botId,
    name: row.name,
    crons: row.crons,
    timezone: row.timezone,
    nextRunAt: row.nextRunAt?.toISOString() ?? null,
  });
  return {
    ok: true as const,
    updated: true as const,
    routineId: row.id,
    name: row.name,
    cron: row.crons[0],
    nextRunAt: row.nextRunAt?.toISOString() ?? null,
    oneShot: false,
    ...(input.createdByBotId ? { bot: input.botName } : {}),
  };
}

/** The line is a courtesy: a routine that is live stays live when the line cannot be written. */
async function postNotice(
  deps: ScheduleToolDeps,
  notice: ScheduleNotice | undefined,
  spaceId: string,
  block: RoutineEventBlock,
) {
  if (!notice) return;
  try {
    const posted = await postRoutineNotice(deps.prisma, {
      spaceId,
      threadId: notice.threadId,
      botId: notice.botId,
      block,
    });
    await deps.events.notify(notice.threadId, posted.seq);
  } catch (error) {
    getLogger().error("routine notice", error);
  }
}

/** Mark the timeline line of a removed routine; the removal itself has already happened. */
export async function markRemovedRoutineNotice(
  deps: Pick<ScheduleToolDeps, "prisma" | "events">,
  routine: { id: string; botId: string; noticeMessageId: string | null },
) {
  try {
    const marked = await markRoutineNoticesRemoved(deps.prisma, routine);
    if (marked) await deps.events.notify(marked.threadId, marked.seq);
  } catch (error) {
    getLogger().error("routine notice removal", error);
  }
}

/**
 * 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<ScheduleToolDeps, "prisma">,
  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));
  await markRemovedRoutineNotice(deps, existing);
  return { ok: true as const, routineId: existing.id, name: existing.name };
}