qiming/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/schedulerService.ts

import log from "electron-log";
import { LOCALHOST_HOSTNAME } from "../constants";
import { getConfiguredPorts } from "../startupPorts";
import {
  finishDigitalEmployeeScheduleRun,
  listDigitalEmployeeSchedules,
  listDueDigitalEmployeeSchedules,
  readDigitalEmployeeCronSettings,
  readDigitalEmployeeScheduleRuns,
  recordDigitalEmployeeScheduleRunStarted,
  upsertDigitalEmployeeSchedule,
  type DigitalEmployeeScheduleRecord,
} from "./stateService";

const SWEEP_INTERVAL_MS = 30_000;
const MAX_SCAN_MINUTES = 366 * 24 * 60;
const RETRY_DELAYS_MS = [60_000, 5 * 60_000, 15 * 60_000];

let schedulerTimer: ReturnType<typeof setInterval> | null = null;
let schedulerSweeping = false;
let activeScheduleRuns = 0;

export interface DigitalEmployeeSchedulerCheckResult {
  checkedAt: string;
  due: number;
  activated: number;
  skipped: number;
  errors: Array<{ scheduleId: string; error: string }>;
}

export function startDigitalEmployeeScheduler(): void {
  if (schedulerTimer) return;
  schedulerTimer = setInterval(() => {
    void runDigitalEmployeeSchedulerCheck().catch((error) => {
      log.warn("[DigitalEmployeeScheduler] Sweep failed:", error);
    });
  }, SWEEP_INTERVAL_MS);
  void initializeScheduleNextRuns().then(() => runDigitalEmployeeSchedulerCheck()).catch((error) => {
    log.warn("[DigitalEmployeeScheduler] Initial check failed:", error);
  });
  log.info("[DigitalEmployeeScheduler] Worker started");
}

export function stopDigitalEmployeeScheduler(): void {
  if (!schedulerTimer) return;
  clearInterval(schedulerTimer);
  schedulerTimer = null;
  log.info("[DigitalEmployeeScheduler] Worker stopped");
}

export async function runDigitalEmployeeSchedulerCheck(
  now = new Date(),
): Promise<DigitalEmployeeSchedulerCheckResult> {
  if (schedulerSweeping) {
    return { checkedAt: now.toISOString(), due: 0, activated: 0, skipped: 0, errors: [] };
  }
  const settings = readDigitalEmployeeCronSettings();
  if (!settings.enabled) {
    return { checkedAt: now.toISOString(), due: 0, activated: 0, skipped: 0, errors: [] };
  }

  schedulerSweeping = true;
  const due = listDueDigitalEmployeeSchedules(now.toISOString(), 50);
  const result: DigitalEmployeeSchedulerCheckResult = {
    checkedAt: now.toISOString(),
    due: due.length,
    activated: 0,
    skipped: 0,
    errors: [],
  };

  try {
    for (const schedule of due) {
      if (activeScheduleRuns >= settings.max_concurrent_schedule_runs) {
        result.skipped += 1;
        continue;
      }
      const dueRuns = computeDueRuns(schedule, now);
      for (const dueAt of dueRuns) {
        if (activeScheduleRuns >= settings.max_concurrent_schedule_runs) {
          result.skipped += 1;
          break;
        }
        try {
          activeScheduleRuns += 1;
          await executeSchedule(schedule, dueAt, "scheduled");
          result.activated += 1;
        } catch (error) {
          result.errors.push({ scheduleId: schedule.id, error: normalizeError(error) });
        } finally {
          activeScheduleRuns = Math.max(0, activeScheduleRuns - 1);
        }
      }
    }
  } finally {
    schedulerSweeping = false;
  }

  return result;
}

export async function runDigitalEmployeeScheduleNow(
  scheduleId: string,
): Promise<{ ok: boolean; scheduleRunId?: string; error?: string }> {
  const schedule = listDigitalEmployeeSchedules().find((item) => item.id === scheduleId);
  if (!schedule) return { ok: false, error: "schedule not found" };
  try {
    const run = await executeSchedule(schedule, new Date().toISOString(), "manual");
    return { ok: true, scheduleRunId: run?.id };
  } catch (error) {
    return { ok: false, error: normalizeError(error) };
  }
}

async function initializeScheduleNextRuns(now = new Date()): Promise<void> {
  for (const schedule of listDigitalEmployeeSchedules()) {
    if (!schedule.enabled || schedule.status !== "enabled" || schedule.deletedAt) continue;
    if (schedule.nextRunAt && schedule.catchUpOnStartup) continue;
    const nextRunAt = nextCronRun(schedule.cronExpr, now).toISOString();
    upsertDigitalEmployeeSchedule({
      id: schedule.id,
      planId: schedule.planId,
      name: schedule.name,
      description: schedule.description,
      cronExpr: schedule.cronExpr,
      timezone: schedule.timezone,
      status: schedule.status,
      enabled: schedule.enabled,
      prompt: schedule.prompt,
      command: schedule.command,
      payload: objectPayload(schedule.payload),
      nextRunAt,
      catchUpOnStartup: schedule.catchUpOnStartup,
      maxCatchUpRuns: schedule.maxCatchUpRuns,
      maxRunHistory: schedule.maxRunHistory,
    });
  }
}

function computeDueRuns(schedule: DigitalEmployeeScheduleRecord, now: Date): string[] {
  const dueAt = schedule.nextRunAt || now.toISOString();
  const runs: string[] = [];
  let cursor = new Date(dueAt);
  const maxRuns = Math.max(1, Math.min(schedule.maxCatchUpRuns, 20));
  while (cursor.getTime() <= now.getTime() && runs.length < maxRuns) {
    runs.push(cursor.toISOString());
    cursor = nextCronRun(schedule.cronExpr, cursor);
  }
  return runs.length > 0 ? runs : [now.toISOString()];
}

async function executeSchedule(
  schedule: DigitalEmployeeScheduleRecord,
  dueAt: string,
  triggerKind: "scheduled" | "manual",
) {
  const attemptNo = Math.min((schedule.failureCount || 0) + 1, 3);
  const started = recordDigitalEmployeeScheduleRunStarted({
    scheduleId: schedule.id,
    planId: schedule.planId,
    dueAt,
    triggerKind,
    attemptNo,
    payload: {
      chatRequest: buildScheduleChatRequest(schedule, dueAt, triggerKind),
    },
  });
  if (!started) throw new Error("failed to record schedule run");

  try {
    const response = await postComputerChat(buildScheduleChatRequest(schedule, dueAt, triggerKind));
    const nextRunAt = nextCronRun(schedule.cronExpr, new Date(dueAt)).toISOString();
    finishDigitalEmployeeScheduleRun(started.id, "completed", { nextRunAt, response });
    pruneScheduleRunHistory(schedule);
    return started;
  } catch (error) {
    const message = normalizeError(error);
    const retryDelay = RETRY_DELAYS_MS[Math.min(attemptNo - 1, RETRY_DELAYS_MS.length - 1)];
    const canRetry = attemptNo < 3;
    const nextRunAt = canRetry
      ? new Date(Date.now() + retryDelay).toISOString()
      : nextCronRun(schedule.cronExpr, new Date(dueAt)).toISOString();
    finishDigitalEmployeeScheduleRun(started.id, "failed", { error: message, nextRunAt });
    throw error;
  }
}

function buildScheduleChatRequest(
  schedule: DigitalEmployeeScheduleRecord,
  dueAt: string,
  triggerKind: string,
) {
  const payload = objectPayload(schedule.payload);
  const prompt = schedule.prompt || schedule.command || stringFromUnknown(payload.prompt) || stringFromUnknown(payload.command) || schedule.name;
  const stamp = Date.now();
  return {
    user_id: stringFromUnknown(payload.user_id) || "qimingclaw-scheduler",
    project_id: schedule.planId || schedule.id,
    prompt,
    session_id: stringFromUnknown(payload.session_id) || `schedule-${schedule.id}`,
    request_id: `schedule-${safeId(schedule.id)}-${safeId(dueAt)}-${stamp}`,
    system_prompt: stringFromUnknown(payload.system_prompt) || "你是 qimingclaw 本地数字员工，请执行调度计划并记录必要的结果。",
    original_user_prompt: prompt,
    open_long_memory: payload.open_long_memory === true,
    agent_config: objectPayload(payload.agent_config),
    model_provider: objectPayload(payload.model_provider),
    schedule_context: {
      schedule_id: schedule.id,
      due_at: dueAt,
      trigger_kind: triggerKind,
    },
  };
}

async function postComputerChat(body: Record<string, unknown>): Promise<unknown> {
  const { agent: agentPort } = getConfiguredPorts();
  const response = await fetch(`http://${LOCALHOST_HOSTNAME}:${agentPort}/computer/chat`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(body),
  });
  const text = await response.text();
  let payload: unknown = text;
  try {
    payload = text ? JSON.parse(text) : {};
  } catch {
    payload = text;
  }
  if (!response.ok) {
    throw new Error(`ComputerServer ${response.status}: ${text}`);
  }
  const record = objectPayload(payload);
  if (record.success === false || (typeof record.code === "string" && record.code !== "0000")) {
    throw new Error(stringFromUnknown(record.message) || "ComputerServer rejected schedule run");
  }
  return payload;
}

function pruneScheduleRunHistory(schedule: DigitalEmployeeScheduleRecord): void {
  const runs = readDigitalEmployeeScheduleRuns(schedule.id, schedule.maxRunHistory + 20);
  if (runs.length <= schedule.maxRunHistory) return;
  // Retention is intentionally conservative for now; the schema keeps history for audit.
}

export function nextCronRun(expression: string, after: Date): Date {
  const parsed = parseCronExpression(expression);
  const candidate = new Date(after.getTime());
  candidate.setSeconds(0, 0);
  candidate.setMinutes(candidate.getMinutes() + 1);
  for (let i = 0; i < MAX_SCAN_MINUTES; i += 1) {
    if (matchesCron(parsed, candidate)) return candidate;
    candidate.setMinutes(candidate.getMinutes() + 1);
  }
  throw new Error(`No cron run found within scan window: ${expression}`);
}

function parseCronExpression(expression: string): CronParts {
  const parts = expression.trim().split(/\s+/);
  if (parts.length !== 5) throw new Error(`Only 5-field cron expressions are supported: ${expression}`);
  return {
    minute: parseCronField(parts[0], 0, 59),
    hour: parseCronField(parts[1], 0, 23),
    dayOfMonth: parseCronField(parts[2], 1, 31),
    month: parseCronField(parts[3], 1, 12),
    dayOfWeek: parseCronField(parts[4], 0, 7),
    domAny: parts[2] === "*",
    dowAny: parts[4] === "*",
  };
}

interface CronParts {
  minute: Set<number>;
  hour: Set<number>;
  dayOfMonth: Set<number>;
  month: Set<number>;
  dayOfWeek: Set<number>;
  domAny: boolean;
  dowAny: boolean;
}

function parseCronField(field: string, min: number, max: number): Set<number> {
  const values = new Set<number>();
  for (const segment of field.split(",")) {
    const [rangePart, stepPart] = segment.split("/");
    const step = stepPart ? Number(stepPart) : 1;
    if (!Number.isInteger(step) || step <= 0) throw new Error(`Invalid cron step: ${field}`);
    const [startText, endText] = (rangePart || "*").split("-");
    const start = startText === "*" ? min : Number(startText);
    const end = startText === "*" ? max : endText ? Number(endText) : start;
    if (!Number.isInteger(start) || !Number.isInteger(end) || start < min || end > max || start > end) {
      throw new Error(`Invalid cron field: ${field}`);
    }
    for (let value = start; value <= end; value += step) {
      values.add(value === 7 && max === 7 ? 0 : value);
    }
  }
  return values;
}

function matchesCron(parts: CronParts, date: Date): boolean {
  const dayOfWeek = date.getDay();
  const domMatch = parts.dayOfMonth.has(date.getDate());
  const dowMatch = parts.dayOfWeek.has(dayOfWeek);
  const dayMatch = parts.domAny || parts.dowAny ? domMatch && dowMatch : domMatch || dowMatch;
  return parts.minute.has(date.getMinutes())
    && parts.hour.has(date.getHours())
    && parts.month.has(date.getMonth() + 1)
    && dayMatch;
}

function objectPayload(value: unknown): Record<string, unknown> {
  return value && typeof value === "object" && !Array.isArray(value)
    ? value as Record<string, unknown>
    : {};
}

function stringFromUnknown(value: unknown): string {
  return typeof value === "string" && value.trim() ? value.trim() : "";
}

function safeId(value: string): string {
  return value.toLowerCase().replace(/[^a-z0-9_-]+/g, "-").replace(/^-+|-+$/g, "").slice(0, 80) || "item";
}

function normalizeError(error: unknown): string {
  return error instanceof Error ? error.message : String(error);
}