openclaw / src /cli /cron-cli /register.cron-simple.ts
SaylorTwift's picture
SaylorTwift HF Staff
Add files using upload-large-folder tool
f778c12 verified
Raw History Blame Contribute Delete
11.9 kB
// Cron simple command registration: remove, toggle, show, runs, and run-now.
import {
resolvePositiveTimerTimeoutMs,
resolveTimerTimeoutMs,
} from "@openclaw/normalization-core/number-coercion";
import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
import { Option, type Command } from "commander";
import { resolveCronCompletionStatus } from "../../cron/completion-status.js";
import type { CronRunLogEntry } from "../../cron/run-log-types.js";
import { defaultRuntime } from "../../runtime.js";
import { sleep } from "../../utils/sleep.js";
import type { GatewayRpcOpts } from "../gateway-rpc.js";
import { addGatewayClientOptions, callGatewayFromCli } from "../gateway-rpc.js";
import { exitCliAfterOutput } from "../one-shot-exit.js";
import { parseDurationMs } from "../parse-duration.js";
import { parseTimeoutMs } from "../parse-timeout.js";
import { CronCliError } from "./cron-cli-error.js";
import { findCronJobByIdOrName } from "./list-jobs.js";
import { createCronOutputCommand } from "./output-mode.js";
import {
enrichCronJsonWithStatus,
formatCronLookupMiss,
handleCronCliError,
parseCronIntegerOption,
printCronJson,
printCronShow,
requireCronJobId,
warnIfCronSchedulerDisabled,
} from "./shared.js";
const CRON_RUN_WAIT_TIMEOUT_DEFAULT = "10m";
const CRON_RUN_WAIT_POLL_INTERVAL_DEFAULT = "2s";
type CronRunCommandResult = {
ok?: boolean;
ran?: boolean;
enqueued?: boolean;
runId?: string;
};
function parseCronRunWaitDuration(raw: unknown): number {
const input =
typeof raw === "string" || typeof raw === "number" || typeof raw === "bigint"
? String(raw)
: "";
let durationMs: number;
try {
durationMs = parseDurationMs(input, { defaultUnit: "ms" });
} catch (error) {
if (error instanceof Error) {
throw new CronCliError(error);
}
throw error;
}
return resolveTimerTimeoutMs(durationMs, 0, 0);
}
function parseCronRunPollInterval(raw: unknown): number {
const durationMs = parseCronRunWaitDuration(raw);
if (durationMs <= 0) {
throw new CronCliError("invalid --poll-interval");
}
return resolvePositiveTimerTimeoutMs(durationMs, 2_000);
}
async function waitForCronRunCompletion(params: {
opts: GatewayRpcOpts;
jobId: string;
runId: string;
timeoutMs: number;
pollIntervalMs: number;
}): Promise<CronRunLogEntry> {
// Poll the task ledger rather than cron.run because completion state is written asynchronously.
const startedAt = performance.now();
let hasPolled = false;
for (;;) {
const elapsedBeforePollMs = Math.floor(performance.now() - startedAt);
if (hasPolled && elapsedBeforePollMs >= params.timeoutMs) {
throw new CronCliError(`timed out waiting for cron run ${params.runId}`);
}
const remainingMs = Math.max(1, params.timeoutMs - elapsedBeforePollMs);
const configuredTimeoutMs = parseTimeoutMs(params.opts.timeout);
const pollTimeoutMs =
configuredTimeoutMs === undefined ? remainingMs : Math.min(configuredTimeoutMs, remainingMs);
hasPolled = true;
// History reads share the wait deadline, but enqueue keeps its own RPC
// timeout and a zero-duration wait still gets one immediate ledger poll.
const pollOpts = { ...params.opts, timeout: String(pollTimeoutMs) };
const page = (await callGatewayFromCli("cron.runs", pollOpts, {
id: params.jobId,
runId: params.runId,
limit: 1,
})) as { entries?: CronRunLogEntry[] };
const entry = page.entries?.[0];
if (entry?.status === "ok" || entry?.status === "error" || entry?.status === "skipped") {
return entry;
}
const elapsedMs = Math.floor(performance.now() - startedAt);
if (elapsedMs >= params.timeoutMs) {
throw new CronCliError(`timed out waiting for cron run ${params.runId}`);
}
await sleep(Math.min(params.pollIntervalMs, params.timeoutMs - elapsedMs));
}
}
function registerCronToggleCommand(params: {
cron: Command;
name: "enable" | "disable";
description: string;
enabled: boolean;
}) {
addGatewayClientOptions(
createCronOutputCommand(params.cron, params.name)
.description(params.description)
.argument("<id>", "Job id")
.action(async (id, opts) => {
try {
const res = await callGatewayFromCli("cron.update", opts, {
id: requireCronJobId(id),
patch: { enabled: params.enabled },
});
printCronJson(res);
if (!params.enabled && process.stderr.isTTY) {
process.stderr.write(
`Note: 'openclaw cron list' hides disabled jobs by default. Use 'openclaw cron list --all' to see this job, or 'openclaw cron enable <id>' to re-enable it.\n`,
);
}
await warnIfCronSchedulerDisabled(opts);
} catch (err) {
handleCronCliError(err);
}
}),
);
}
export function registerCronSimpleCommands(cron: Command) {
addGatewayClientOptions(
createCronOutputCommand(cron, "rm")
.description("Remove an automation")
.argument("<id>", "Job id")
.action(async (id, opts) => {
try {
const res = await callGatewayFromCli("cron.remove", opts, { id: requireCronJobId(id) });
printCronJson(res);
} catch (err) {
handleCronCliError(err);
}
}),
);
registerCronToggleCommand({
cron,
name: "enable",
description: "Enable an automation",
enabled: true,
});
registerCronToggleCommand({
cron,
name: "disable",
description: "Disable an automation",
enabled: false,
});
addGatewayClientOptions(
createCronOutputCommand(cron, "get")
.description("Get an automation as JSON")
.argument("<id>", "Job id")
.action(async (id, opts) => {
try {
const res = await callGatewayFromCli("cron.get", opts, { id: requireCronJobId(id) });
printCronJson(res);
} catch (err) {
handleCronCliError(err);
}
}),
);
addGatewayClientOptions(
cron
.command("show")
.description("Show an automation")
.argument("<id>", "Job id or exact name")
.option("--json", "Output JSON", false)
.action(async (idArg, opts) => {
try {
const id = requireCronJobId(idArg);
const { job, deliveryPreview } = await findCronJobByIdOrName(opts, id, {
includeDeliveryPreview: !opts.json,
});
if (!job) {
throw new CronCliError(formatCronLookupMiss(id));
}
if (opts.json) {
printCronJson(enrichCronJsonWithStatus(job));
return;
}
printCronShow(job, defaultRuntime, { deliveryPreview });
} catch (err) {
handleCronCliError(err);
}
}),
);
addGatewayClientOptions(
createCronOutputCommand(cron, "runs")
.description("Show automation run history")
.argument("[id]", "Job id")
.option("--id <id>", "Job id (alternative to positional argument)")
.option("--run-id <runId>", "Filter by cron run id")
.addOption(
new Option("--status <status>", "Filter by run status").choices([
"all",
"ok",
"error",
"skipped",
]),
)
.addOption(
new Option("--delivery-status <status>", "Filter by delivery status").choices([
"delivered",
"not-delivered",
"unknown",
"not-requested",
]),
)
.option("--query <text>", "Filter by run summary or error text")
.option("--offset <n>", "Entries to skip before returning results")
.addOption(new Option("--sort <direction>", "Sort by run time").choices(["asc", "desc"]))
.option("--limit <n>", "Max entries (default 50)", "50")
.action(async (idArg, opts) => {
try {
const argId = normalizeOptionalString(idArg);
const flagId = normalizeOptionalString(opts.id);
if (argId && flagId && argId !== flagId) {
throw new CronCliError(
`Conflicting job ids: positional "${argId}" and --id "${flagId}".`,
);
}
const id = requireCronJobId(argId ?? flagId, "Pass it positionally or with --id.");
const limit = parseCronIntegerOption(opts.limit ?? "50", "--limit");
const offset = parseCronIntegerOption(opts.offset, "--offset", "non-negative");
if (typeof opts.runId === "string" && !opts.runId.trim()) {
throw new CronCliError("--run-id must not be blank");
}
const res = await callGatewayFromCli("cron.runs", opts, {
id,
...(typeof opts.runId === "string" && opts.runId.trim() ? { runId: opts.runId } : {}),
...(typeof opts.status === "string" ? { status: opts.status } : {}),
...(typeof opts.deliveryStatus === "string"
? { deliveryStatus: opts.deliveryStatus }
: {}),
...(typeof opts.query === "string" ? { query: opts.query } : {}),
...(offset !== undefined ? { offset } : {}),
...(typeof opts.sort === "string" ? { sortDir: opts.sort } : {}),
limit,
});
printCronJson(res);
} catch (err) {
handleCronCliError(err);
}
}),
);
addGatewayClientOptions(
createCronOutputCommand(cron, "run")
.description("Run an automation now (debug)")
.argument("<id>", "Job id")
.option("--due", "Run only when due (default behavior in older versions)", false)
.option("--wait", "Wait for the queued run to finish", false)
.option(
"--wait-timeout <duration>",
"Maximum time to wait for --wait",
CRON_RUN_WAIT_TIMEOUT_DEFAULT,
)
.option(
"--poll-interval <duration>",
"Polling interval for --wait",
CRON_RUN_WAIT_POLL_INTERVAL_DEFAULT,
)
.action(async (idArg, opts, command) => {
try {
const id = requireCronJobId(idArg);
let waitTimeoutMs = 0;
let pollIntervalMs = 0;
if (opts.wait) {
waitTimeoutMs = parseCronRunWaitDuration(opts.waitTimeout);
pollIntervalMs = parseCronRunPollInterval(opts.pollInterval);
}
if (command.getOptionValueSource("timeout") === "default") {
opts.timeout = "600000";
}
const res = await callGatewayFromCli("cron.run", opts, {
id,
mode: opts.due ? "due" : "force",
});
const result = res as CronRunCommandResult | undefined;
if (opts.wait && result?.ok && result.enqueued) {
if (!result.runId) {
throw new Error("cron run did not return a runId to wait for");
}
const run = await waitForCronRunCompletion({
opts,
jobId: id,
runId: result.runId,
timeoutMs: waitTimeoutMs,
pollIntervalMs,
});
const completionStatus =
run.completionStatus ??
resolveCronCompletionStatus({
status: run.status,
delivered: run.delivered,
deliveryStatus: run.deliveryStatus,
});
const completedRun = { ...run, completionStatus };
printCronJson({
...res,
completed: true,
status: run.status,
completionStatus,
run: completedRun,
});
exitCliAfterOutput(defaultRuntime, completionStatus === "succeeded" ? 0 : 1);
}
printCronJson(res);
exitCliAfterOutput(defaultRuntime, result?.ok && (result.ran || result.enqueued) ? 0 : 1);
} catch (err) {
handleCronCliError(err);
}
}),
);
}