Files
clawdbot/src/cli/cron-cli.ts
2026-01-13 04:55:48 +00:00

760 lines
26 KiB
TypeScript

import type { Command } from "commander";
import { parseAbsoluteTimeMs } from "../cron/parse.js";
import type { CronJob, CronSchedule } from "../cron/types.js";
import { danger } from "../globals.js";
import { PROVIDER_IDS } from "../providers/registry.js";
import { normalizeAgentId } from "../routing/session-key.js";
import { defaultRuntime } from "../runtime.js";
import { formatDocsLink } from "../terminal/links.js";
import { colorize, isRich, theme } from "../terminal/theme.js";
import type { GatewayRpcOpts } from "./gateway-rpc.js";
import { addGatewayClientOptions, callGatewayFromCli } from "./gateway-rpc.js";
const CRON_PROVIDER_OPTIONS = ["last", ...PROVIDER_IDS].join("|");
async function warnIfCronSchedulerDisabled(opts: GatewayRpcOpts) {
try {
const res = (await callGatewayFromCli("cron.status", opts, {})) as {
enabled?: boolean;
storePath?: string;
};
if (res?.enabled === true) return;
const store = typeof res?.storePath === "string" ? res.storePath : "";
defaultRuntime.error(
[
"warning: cron scheduler is disabled in the Gateway; jobs are saved but will not run automatically.",
"Re-enable with `cron.enabled: true` (or remove `cron.enabled: false`) and restart the Gateway.",
store ? `store: ${store}` : "",
]
.filter(Boolean)
.join("\n"),
);
} catch {
// Ignore status failures (older gateway, offline, etc.)
}
}
function parseDurationMs(input: string): number | null {
const raw = input.trim();
if (!raw) return null;
const match = raw.match(/^(\d+(?:\.\d+)?)(ms|s|m|h|d)$/i);
if (!match) return null;
const n = Number.parseFloat(match[1] ?? "");
if (!Number.isFinite(n) || n <= 0) return null;
const unit = (match[2] ?? "").toLowerCase();
const factor =
unit === "ms"
? 1
: unit === "s"
? 1000
: unit === "m"
? 60_000
: unit === "h"
? 3_600_000
: 86_400_000;
return Math.floor(n * factor);
}
function parseAtMs(input: string): number | null {
const raw = input.trim();
if (!raw) return null;
const absolute = parseAbsoluteTimeMs(raw);
if (absolute) return absolute;
const dur = parseDurationMs(raw);
if (dur) return Date.now() + dur;
return null;
}
const CRON_ID_PAD = 36;
const CRON_NAME_PAD = 24;
const CRON_SCHEDULE_PAD = 32;
const CRON_NEXT_PAD = 10;
const CRON_LAST_PAD = 10;
const CRON_STATUS_PAD = 9;
const CRON_TARGET_PAD = 9;
const CRON_AGENT_PAD = 10;
const pad = (value: string, width: number) => value.padEnd(width);
const truncate = (value: string, width: number) => {
if (value.length <= width) return value;
if (width <= 3) return value.slice(0, width);
return `${value.slice(0, width - 3)}...`;
};
const formatIsoMinute = (ms: number) => {
const d = new Date(ms);
if (Number.isNaN(d.getTime())) return "-";
const iso = d.toISOString();
return `${iso.slice(0, 10)} ${iso.slice(11, 16)}Z`;
};
const formatDuration = (ms: number) => {
if (ms < 60_000) return `${Math.max(1, Math.round(ms / 1000))}s`;
if (ms < 3_600_000) return `${Math.round(ms / 60_000)}m`;
if (ms < 86_400_000) return `${Math.round(ms / 3_600_000)}h`;
return `${Math.round(ms / 86_400_000)}d`;
};
const formatSpan = (ms: number) => {
if (ms < 60_000) return "<1m";
if (ms < 3_600_000) return `${Math.round(ms / 60_000)}m`;
if (ms < 86_400_000) return `${Math.round(ms / 3_600_000)}h`;
return `${Math.round(ms / 86_400_000)}d`;
};
const formatRelative = (ms: number | null | undefined, nowMs: number) => {
if (!ms) return "-";
const delta = ms - nowMs;
const label = formatSpan(Math.abs(delta));
return delta >= 0 ? `in ${label}` : `${label} ago`;
};
const formatSchedule = (schedule: CronSchedule) => {
if (schedule.kind === "at") return `at ${formatIsoMinute(schedule.atMs)}`;
if (schedule.kind === "every")
return `every ${formatDuration(schedule.everyMs)}`;
return schedule.tz
? `cron ${schedule.expr} @ ${schedule.tz}`
: `cron ${schedule.expr}`;
};
const formatStatus = (job: CronJob) => {
if (!job.enabled) return "disabled";
if (job.state.runningAtMs) return "running";
return job.state.lastStatus ?? "idle";
};
function printCronList(jobs: CronJob[], runtime = defaultRuntime) {
if (jobs.length === 0) {
runtime.log("No cron jobs.");
return;
}
const rich = isRich();
const header = [
pad("ID", CRON_ID_PAD),
pad("Name", CRON_NAME_PAD),
pad("Schedule", CRON_SCHEDULE_PAD),
pad("Next", CRON_NEXT_PAD),
pad("Last", CRON_LAST_PAD),
pad("Status", CRON_STATUS_PAD),
pad("Target", CRON_TARGET_PAD),
pad("Agent", CRON_AGENT_PAD),
].join(" ");
runtime.log(rich ? theme.heading(header) : header);
const now = Date.now();
for (const job of jobs) {
const idLabel = pad(job.id, CRON_ID_PAD);
const nameLabel = pad(truncate(job.name, CRON_NAME_PAD), CRON_NAME_PAD);
const scheduleLabel = pad(
truncate(formatSchedule(job.schedule), CRON_SCHEDULE_PAD),
CRON_SCHEDULE_PAD,
);
const nextLabel = pad(
job.enabled ? formatRelative(job.state.nextRunAtMs, now) : "-",
CRON_NEXT_PAD,
);
const lastLabel = pad(
formatRelative(job.state.lastRunAtMs, now),
CRON_LAST_PAD,
);
const statusRaw = formatStatus(job);
const statusLabel = pad(statusRaw, CRON_STATUS_PAD);
const targetLabel = pad(job.sessionTarget, CRON_TARGET_PAD);
const agentLabel = pad(
truncate(job.agentId ?? "default", CRON_AGENT_PAD),
CRON_AGENT_PAD,
);
const coloredStatus = (() => {
if (statusRaw === "ok") return colorize(rich, theme.success, statusLabel);
if (statusRaw === "error")
return colorize(rich, theme.error, statusLabel);
if (statusRaw === "running")
return colorize(rich, theme.warn, statusLabel);
if (statusRaw === "skipped")
return colorize(rich, theme.muted, statusLabel);
return colorize(rich, theme.muted, statusLabel);
})();
const coloredTarget =
job.sessionTarget === "isolated"
? colorize(rich, theme.accentBright, targetLabel)
: colorize(rich, theme.accent, targetLabel);
const coloredAgent = job.agentId
? colorize(rich, theme.info, agentLabel)
: colorize(rich, theme.muted, agentLabel);
const line = [
colorize(rich, theme.accent, idLabel),
colorize(rich, theme.info, nameLabel),
colorize(rich, theme.info, scheduleLabel),
colorize(rich, theme.muted, nextLabel),
colorize(rich, theme.muted, lastLabel),
coloredStatus,
coloredTarget,
coloredAgent,
].join(" ");
runtime.log(line.trimEnd());
}
}
export function registerCronCli(program: Command) {
addGatewayClientOptions(
program
.command("wake")
.description(
"Enqueue a system event and optionally trigger an immediate heartbeat",
)
.requiredOption("--text <text>", "System event text")
.option(
"--mode <mode>",
"Wake mode (now|next-heartbeat)",
"next-heartbeat",
)
.option("--json", "Output JSON", false),
).action(async (opts) => {
try {
const result = await callGatewayFromCli(
"wake",
opts,
{ mode: opts.mode, text: opts.text },
{ expectFinal: false },
);
if (opts.json) defaultRuntime.log(JSON.stringify(result, null, 2));
else defaultRuntime.log("ok");
} catch (err) {
defaultRuntime.error(danger(String(err)));
defaultRuntime.exit(1);
}
});
const cron = program
.command("cron")
.description("Manage cron jobs (via Gateway)")
.addHelpText(
"after",
() =>
`\n${theme.muted("Docs:")} ${formatDocsLink(
"/cron-jobs",
"docs.clawd.bot/cron-jobs",
)}\n`,
);
addGatewayClientOptions(
cron
.command("status")
.description("Show cron scheduler status")
.option("--json", "Output JSON", false)
.action(async (opts) => {
try {
const res = await callGatewayFromCli("cron.status", opts, {});
defaultRuntime.log(JSON.stringify(res, null, 2));
} catch (err) {
defaultRuntime.error(danger(String(err)));
defaultRuntime.exit(1);
}
}),
);
addGatewayClientOptions(
cron
.command("list")
.description("List cron jobs")
.option("--all", "Include disabled jobs", false)
.option("--json", "Output JSON", false)
.action(async (opts) => {
try {
const res = await callGatewayFromCli("cron.list", opts, {
includeDisabled: Boolean(opts.all),
});
if (opts.json) {
defaultRuntime.log(JSON.stringify(res, null, 2));
return;
}
const jobs = (res as { jobs?: CronJob[] } | null)?.jobs ?? [];
printCronList(jobs, defaultRuntime);
} catch (err) {
defaultRuntime.error(danger(String(err)));
defaultRuntime.exit(1);
}
}),
);
addGatewayClientOptions(
cron
.command("add")
.alias("create")
.description("Add a cron job")
.requiredOption("--name <name>", "Job name")
.option("--description <text>", "Optional description")
.option("--disabled", "Create job disabled", false)
.option("--delete-after-run", "Delete one-shot job after it succeeds", false)
.option("--agent <id>", "Agent id for this job")
.option("--session <target>", "Session target (main|isolated)", "main")
.option(
"--wake <mode>",
"Wake mode (now|next-heartbeat)",
"next-heartbeat",
)
.option("--at <when>", "Run once at time (ISO) or +duration (e.g. 20m)")
.option("--every <duration>", "Run every duration (e.g. 10m, 1h)")
.option("--cron <expr>", "Cron expression (5-field)")
.option("--tz <iana>", "Timezone for cron expressions (IANA)", "")
.option("--system-event <text>", "System event payload (main session)")
.option("--message <text>", "Agent message payload")
.option(
"--thinking <level>",
"Thinking level for agent jobs (off|minimal|low|medium|high)",
)
.option(
"--model <model>",
"Model override for agent jobs (provider/model or alias)",
)
.option("--timeout-seconds <n>", "Timeout seconds for agent jobs")
.option("--deliver", "Deliver agent output", false)
.option(
"--provider <provider>",
`Delivery provider (${CRON_PROVIDER_OPTIONS})`,
"last",
)
.option(
"--to <dest>",
"Delivery destination (E.164, Telegram chatId, or Discord channel/user)",
)
.option(
"--best-effort-deliver",
"Do not fail the job if delivery fails",
false,
)
.option(
"--post-prefix <prefix>",
"Prefix for summary system event",
"Cron",
)
.option("--json", "Output JSON", false)
.action(async (opts) => {
try {
const schedule = (() => {
const at = typeof opts.at === "string" ? opts.at : "";
const every = typeof opts.every === "string" ? opts.every : "";
const cronExpr = typeof opts.cron === "string" ? opts.cron : "";
const chosen = [
Boolean(at),
Boolean(every),
Boolean(cronExpr),
].filter(Boolean).length;
if (chosen !== 1) {
throw new Error(
"Choose exactly one schedule: --at, --every, or --cron",
);
}
if (at) {
const atMs = parseAtMs(at);
if (!atMs)
throw new Error(
"Invalid --at; use ISO time or duration like 20m",
);
return { kind: "at" as const, atMs };
}
if (every) {
const everyMs = parseDurationMs(every);
if (!everyMs)
throw new Error("Invalid --every; use e.g. 10m, 1h, 1d");
return { kind: "every" as const, everyMs };
}
return {
kind: "cron" as const,
expr: cronExpr,
tz:
typeof opts.tz === "string" && opts.tz.trim()
? opts.tz.trim()
: undefined,
};
})();
const sessionTarget = String(opts.session ?? "main");
if (sessionTarget !== "main" && sessionTarget !== "isolated") {
throw new Error("--session must be main or isolated");
}
const wakeMode = String(opts.wake ?? "next-heartbeat");
if (wakeMode !== "now" && wakeMode !== "next-heartbeat") {
throw new Error("--wake must be now or next-heartbeat");
}
const agentId =
typeof opts.agent === "string" && opts.agent.trim()
? normalizeAgentId(opts.agent)
: undefined;
const payload = (() => {
const systemEvent =
typeof opts.systemEvent === "string"
? opts.systemEvent.trim()
: "";
const message =
typeof opts.message === "string" ? opts.message.trim() : "";
const chosen = [Boolean(systemEvent), Boolean(message)].filter(
Boolean,
).length;
if (chosen !== 1) {
throw new Error(
"Choose exactly one payload: --system-event or --message",
);
}
if (systemEvent)
return { kind: "systemEvent" as const, text: systemEvent };
const timeoutSeconds = opts.timeoutSeconds
? Number.parseInt(String(opts.timeoutSeconds), 10)
: undefined;
return {
kind: "agentTurn" as const,
message,
model:
typeof opts.model === "string" && opts.model.trim()
? opts.model.trim()
: undefined,
thinking:
typeof opts.thinking === "string" && opts.thinking.trim()
? opts.thinking.trim()
: undefined,
timeoutSeconds:
timeoutSeconds && Number.isFinite(timeoutSeconds)
? timeoutSeconds
: undefined,
deliver: Boolean(opts.deliver),
provider:
typeof opts.provider === "string" ? opts.provider : "last",
to:
typeof opts.to === "string" && opts.to.trim()
? opts.to.trim()
: undefined,
bestEffortDeliver: Boolean(opts.bestEffortDeliver),
};
})();
if (sessionTarget === "main" && payload.kind !== "systemEvent") {
throw new Error("Main jobs require --system-event (systemEvent).");
}
if (sessionTarget === "isolated" && payload.kind !== "agentTurn") {
throw new Error("Isolated jobs require --message (agentTurn).");
}
const isolation =
sessionTarget === "isolated"
? {
postToMainPrefix:
typeof opts.postPrefix === "string" &&
opts.postPrefix.trim()
? opts.postPrefix.trim()
: "Cron",
}
: undefined;
const name = String(opts.name ?? "").trim();
if (!name) throw new Error("--name is required");
const description =
typeof opts.description === "string" && opts.description.trim()
? opts.description.trim()
: undefined;
const params = {
name,
description,
enabled: !opts.disabled,
deleteAfterRun: Boolean(opts.deleteAfterRun),
agentId,
schedule,
sessionTarget,
wakeMode,
payload,
isolation,
};
const res = await callGatewayFromCli("cron.add", opts, params);
defaultRuntime.log(JSON.stringify(res, null, 2));
await warnIfCronSchedulerDisabled(opts);
} catch (err) {
defaultRuntime.error(danger(String(err)));
defaultRuntime.exit(1);
}
}),
);
addGatewayClientOptions(
cron
.command("rm")
.alias("remove")
.alias("delete")
.description("Remove a cron job")
.argument("<id>", "Job id")
.option("--json", "Output JSON", false)
.action(async (id, opts) => {
try {
const res = await callGatewayFromCli("cron.remove", opts, { id });
defaultRuntime.log(JSON.stringify(res, null, 2));
} catch (err) {
defaultRuntime.error(danger(String(err)));
defaultRuntime.exit(1);
}
}),
);
addGatewayClientOptions(
cron
.command("enable")
.description("Enable a cron job")
.argument("<id>", "Job id")
.action(async (id, opts) => {
try {
const res = await callGatewayFromCli("cron.update", opts, {
id,
patch: { enabled: true },
});
defaultRuntime.log(JSON.stringify(res, null, 2));
await warnIfCronSchedulerDisabled(opts);
} catch (err) {
defaultRuntime.error(danger(String(err)));
defaultRuntime.exit(1);
}
}),
);
addGatewayClientOptions(
cron
.command("disable")
.description("Disable a cron job")
.argument("<id>", "Job id")
.action(async (id, opts) => {
try {
const res = await callGatewayFromCli("cron.update", opts, {
id,
patch: { enabled: false },
});
defaultRuntime.log(JSON.stringify(res, null, 2));
await warnIfCronSchedulerDisabled(opts);
} catch (err) {
defaultRuntime.error(danger(String(err)));
defaultRuntime.exit(1);
}
}),
);
addGatewayClientOptions(
cron
.command("runs")
.description("Show cron run history (JSONL-backed)")
.requiredOption("--id <id>", "Job id")
.option("--limit <n>", "Max entries (default 50)", "50")
.action(async (opts) => {
try {
const limitRaw = Number.parseInt(String(opts.limit ?? "50"), 10);
const limit =
Number.isFinite(limitRaw) && limitRaw > 0 ? limitRaw : 50;
const id = String(opts.id);
const res = await callGatewayFromCli("cron.runs", opts, {
id,
limit,
});
defaultRuntime.log(JSON.stringify(res, null, 2));
} catch (err) {
defaultRuntime.error(danger(String(err)));
defaultRuntime.exit(1);
}
}),
);
addGatewayClientOptions(
cron
.command("edit")
.description("Edit a cron job (patch fields)")
.argument("<id>", "Job id")
.option("--name <name>", "Set name")
.option("--description <text>", "Set description")
.option("--enable", "Enable job", false)
.option("--disable", "Disable job", false)
.option("--delete-after-run", "Delete one-shot job after it succeeds", false)
.option("--keep-after-run", "Keep one-shot job after it succeeds", false)
.option("--session <target>", "Session target (main|isolated)")
.option("--agent <id>", "Set agent id")
.option("--clear-agent", "Unset agent and use default", false)
.option("--wake <mode>", "Wake mode (now|next-heartbeat)")
.option("--at <when>", "Set one-shot time (ISO) or duration like 20m")
.option("--every <duration>", "Set interval duration like 10m")
.option("--cron <expr>", "Set cron expression")
.option("--tz <iana>", "Timezone for cron expressions (IANA)")
.option("--system-event <text>", "Set systemEvent payload")
.option("--message <text>", "Set agentTurn payload message")
.option("--thinking <level>", "Thinking level for agent jobs")
.option("--model <model>", "Model override for agent jobs")
.option("--timeout-seconds <n>", "Timeout seconds for agent jobs")
.option("--deliver", "Deliver agent output", false)
.option(
"--provider <provider>",
`Delivery provider (${CRON_PROVIDER_OPTIONS})`,
)
.option(
"--to <dest>",
"Delivery destination (E.164, Telegram chatId, or Discord channel/user)",
)
.option(
"--best-effort-deliver",
"Do not fail job if delivery fails",
false,
)
.option("--post-prefix <prefix>", "Prefix for summary system event")
.action(async (id, opts) => {
try {
if (opts.session === "main" && opts.message) {
throw new Error(
"Main jobs cannot use --message; use --system-event or --session isolated.",
);
}
if (opts.session === "isolated" && opts.systemEvent) {
throw new Error(
"Isolated jobs cannot use --system-event; use --message or --session main.",
);
}
if (opts.session === "main" && typeof opts.postPrefix === "string") {
throw new Error("--post-prefix only applies to isolated jobs.");
}
const patch: Record<string, unknown> = {};
if (typeof opts.name === "string") patch.name = opts.name;
if (typeof opts.description === "string")
patch.description = opts.description;
if (opts.enable && opts.disable)
throw new Error("Choose --enable or --disable, not both");
if (opts.enable) patch.enabled = true;
if (opts.disable) patch.enabled = false;
if (opts.deleteAfterRun && opts.keepAfterRun) {
throw new Error(
"Choose --delete-after-run or --keep-after-run, not both",
);
}
if (opts.deleteAfterRun) patch.deleteAfterRun = true;
if (opts.keepAfterRun) patch.deleteAfterRun = false;
if (typeof opts.session === "string")
patch.sessionTarget = opts.session;
if (typeof opts.wake === "string") patch.wakeMode = opts.wake;
if (opts.agent && opts.clearAgent) {
throw new Error("Use --agent or --clear-agent, not both");
}
if (typeof opts.agent === "string" && opts.agent.trim()) {
patch.agentId = normalizeAgentId(opts.agent);
}
if (opts.clearAgent) {
patch.agentId = null;
}
const scheduleChosen = [opts.at, opts.every, opts.cron].filter(
Boolean,
).length;
if (scheduleChosen > 1)
throw new Error("Choose at most one schedule change");
if (opts.at) {
const atMs = parseAtMs(String(opts.at));
if (!atMs) throw new Error("Invalid --at");
patch.schedule = { kind: "at", atMs };
} else if (opts.every) {
const everyMs = parseDurationMs(String(opts.every));
if (!everyMs) throw new Error("Invalid --every");
patch.schedule = { kind: "every", everyMs };
} else if (opts.cron) {
patch.schedule = {
kind: "cron",
expr: String(opts.cron),
tz:
typeof opts.tz === "string" && opts.tz.trim()
? opts.tz.trim()
: undefined,
};
}
const payloadChosen = [opts.systemEvent, opts.message].filter(
Boolean,
).length;
if (payloadChosen > 1)
throw new Error("Choose at most one payload change");
if (opts.systemEvent) {
patch.payload = {
kind: "systemEvent",
text: String(opts.systemEvent),
};
} else if (opts.message) {
const model =
typeof opts.model === "string" && opts.model.trim()
? opts.model.trim()
: undefined;
const thinking =
typeof opts.thinking === "string" && opts.thinking.trim()
? opts.thinking.trim()
: undefined;
const timeoutSeconds = opts.timeoutSeconds
? Number.parseInt(String(opts.timeoutSeconds), 10)
: undefined;
patch.payload = {
kind: "agentTurn",
message: String(opts.message),
model,
thinking,
timeoutSeconds:
timeoutSeconds && Number.isFinite(timeoutSeconds)
? timeoutSeconds
: undefined,
deliver: Boolean(opts.deliver),
provider:
typeof opts.provider === "string" ? opts.provider : undefined,
to: typeof opts.to === "string" ? opts.to : undefined,
bestEffortDeliver: Boolean(opts.bestEffortDeliver),
};
}
if (typeof opts.postPrefix === "string") {
patch.isolation = {
postToMainPrefix: opts.postPrefix.trim()
? opts.postPrefix
: "Cron",
};
}
const res = await callGatewayFromCli("cron.update", opts, {
id,
patch,
});
defaultRuntime.log(JSON.stringify(res, null, 2));
await warnIfCronSchedulerDisabled(opts);
} catch (err) {
defaultRuntime.error(danger(String(err)));
defaultRuntime.exit(1);
}
}),
);
addGatewayClientOptions(
cron
.command("run")
.description("Run a cron job now (debug)")
.argument("<id>", "Job id")
.option("--force", "Run even if not due", false)
.action(async (id, opts) => {
try {
const res = await callGatewayFromCli("cron.run", opts, {
id,
mode: opts.force ? "force" : "due",
});
defaultRuntime.log(JSON.stringify(res, null, 2));
} catch (err) {
defaultRuntime.error(danger(String(err)));
defaultRuntime.exit(1);
}
}),
);
}