ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-12 21:53:00 -06:00
Code Issues Packages Projects Releases Wiki Activity

feat(cron): add per-job dynamic cadence (#110978)

Browse Source 
Add optional per-job pacing bounds across the cron API, CLI, tool schema, public output, and SQLite job envelope, requiring at least one bound. Allow only the currently running paced job to record a one-shot next_check proposal and carry it through isolated-run completion.

After successful runs, clamp the proposal to the job bounds and persist an exact one-shot slot marker so maintenance preserves only that timestamp. Clear the marker on runs, edits, and schedule normalization; preserve existing no-proposal, skip, timeout, and error scheduling behavior.
This commit is contained in:
Peter Steinberger
2026-07-18 23:43:39 +01:00
committed by GitHub
parent 5114b45927
commit 556a2ee276
40 changed files with 1033 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift
+8
View File
@@ -10737,6 +10737,7 @@ public struct CronJob: Codable, Sendable {
public let updatedatms: Int
public let configrevision: String?
public let schedule: AnyCodable
public let pacing: [String: AnyCodable]?
public let trigger: [String: AnyCodable]?
public let sessiontarget: AnyCodable
public let wakemode: AnyCodable
@@ -10770,6 +10771,7 @@ public struct CronJob: Codable, Sendable {
updatedatms: Int,
configrevision: String? = nil,
schedule: AnyCodable,
pacing: [String: AnyCodable]? = nil,
trigger: [String: AnyCodable]? = nil,
sessiontarget: AnyCodable,
wakemode: AnyCodable,
@@ -10802,6 +10804,7 @@ public struct CronJob: Codable, Sendable {
self.updatedatms = updatedatms
self.configrevision = configrevision
self.schedule = schedule
self.pacing = pacing
self.trigger = trigger
self.sessiontarget = sessiontarget
self.wakemode = wakemode
@@ -10836,6 +10839,7 @@ public struct CronJob: Codable, Sendable {
case updatedatms = "updatedAtMs"
case configrevision = "configRevision"
case schedule
case pacing
case trigger
case sessiontarget = "sessionTarget"
case wakemode = "wakeMode"
@@ -10923,6 +10927,7 @@ public struct CronAddParams: Codable, Sendable {
public let enabled: Bool?
public let deleteafterrun: Bool?
public let schedule: AnyCodable
public let pacing: [String: AnyCodable]?
public let trigger: [String: AnyCodable]?
public let sessiontarget: AnyCodable
public let wakemode: AnyCodable
@@ -10941,6 +10946,7 @@ public struct CronAddParams: Codable, Sendable {
enabled: Bool? = nil,
deleteafterrun: Bool? = nil,
schedule: AnyCodable,
pacing: [String: AnyCodable]? = nil,
trigger: [String: AnyCodable]? = nil,
sessiontarget: AnyCodable,
wakemode: AnyCodable,
@@ -10958,6 +10964,7 @@ public struct CronAddParams: Codable, Sendable {
self.enabled = enabled
self.deleteafterrun = deleteafterrun
self.schedule = schedule
self.pacing = pacing
self.trigger = trigger
self.sessiontarget = sessiontarget
self.wakemode = wakemode
@@ -10977,6 +10984,7 @@ public struct CronAddParams: Codable, Sendable {
case enabled
case deleteafterrun = "deleteAfterRun"
case schedule
case pacing
case trigger
case sessiontarget = "sessionTarget"
case wakemode = "wakeMode"
docs/automation/cron-jobs.md
+8
View File
@@ -76,6 +76,14 @@ Timestamps without a timezone are treated as UTC. Add `--tz America/New_York` to
Recurring top-of-hour expressions (minute `0` with a wildcard hour field) are automatically staggered by up to 5 minutes to reduce load spikes. Use `--exact` to force precise timing, or `--stagger 30s` for an explicit window (cron schedules only).
### Dynamic cadence (pacing)
Recurring jobs can set `pacing.min` and/or `pacing.max` to duration strings such as `15m` or `4h`; at least one bound is required. Use `--pacing-min` and `--pacing-max` with `cron add|edit` (`--clear-pacing` removes both bounds).
During an isolated run, a paced job can call the `cron` tool with `action: "next_check"` and `in: "30m"`. The proposal applies only to that currently running job and is measured from successful run completion. OpenClaw silently clamps it to the configured bounds.
Pacing without a proposal leaves the normal schedule unchanged. Failed, timed-out, and skipped runs discard the proposal, so existing retry and error-backoff behavior takes precedence.
### Day-of-month and day-of-week use OR logic
Cron expressions are parsed by [croner](https://github.com/Hexagon/croner). When both the day-of-month and day-of-week fields are non-wildcard, croner matches when **either** field matches, not both. This is standard Vixie cron behavior.
docs/docs_map.md
+1
View File
@@ -65,6 +65,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`.
- H2: Quick start
- H2: How cron works
- H2: Schedule types
- H3: Dynamic cadence (pacing)
- H3: Day-of-month and day-of-week use OR logic
- H2: Event triggers (condition watchers)
- H2: Payloads
packages/gateway-protocol/src/schema/cron.ts
+15
View File
@@ -224,6 +224,18 @@ export const CronTriggerSchema = closedObject({
once: Type.Optional(Type.Boolean()),
});
/** Optional dynamic-cadence bounds stored with a cron job. */
export const CronPacingSchema = Type.Object(
{
min: Type.Optional(NonBlankString),
max: Type.Optional(NonBlankString),
},
{
additionalProperties: false,
description: "Dynamic-cadence bounds; at least one of min or max is required",
},
);
/** Full cron payload for new jobs. */
export const CronPayloadSchema = Type.Union([
closedObject({
@@ -433,6 +445,7 @@ export const CronJobSchema = closedObject({
/** Opaque Gateway-computed token for the job definition, excluding scheduler state. */
configRevision: Type.Optional(CronConfigRevisionSchema),
schedule: CronScheduleSchema,
pacing: Type.Optional(CronPacingSchema),
trigger: Type.Optional(CronTriggerSchema),
sessionTarget: CronSessionTargetSchema,
wakeMode: CronWakeModeSchema,
@@ -481,6 +494,7 @@ export const CronAddParamsSchema = closedObject({
owner: Type.Optional(CronOwnerSchema),
...CronCommonOptionalFields,
schedule: CronScheduleSchema,
pacing: Type.Optional(CronPacingSchema),
trigger: Type.Optional(CronTriggerSchema),
sessionTarget: CronSessionTargetSchema,
wakeMode: CronWakeModeSchema,
@@ -505,6 +519,7 @@ export const CronJobPatchSchema = closedObject({
displayName: Type.Optional(Type.Union([CronDisplayNameSchema, Type.Null()])),
...CronCommonOptionalFields,
schedule: Type.Optional(CronScheduleSchema),
pacing: Type.Optional(Type.Union([CronPacingSchema, Type.Null()])),
trigger: Type.Optional(Type.Union([CronTriggerSchema, Type.Null()])),
sessionTarget: Type.Optional(CronSessionTargetSchema),
wakeMode: Type.Optional(CronWakeModeSchema),
src/agents/openclaw-tools.ts
+2 -1
View File
@@ -179,7 +179,7 @@ export function createOpenClawTools(
senderIsOwner?: boolean;
/** Server-owned operation-local origin for conversation-read visibility policy. */
conversationReadOrigin?: ConversationReadInvocationOrigin;
/** Restrict the cron tool to self-removing this active cron job. */
/** Restrict cron operations to the active cron job's self-scoped surface. */
cronSelfRemoveOnlyJobId?: string;
/** Require explicit message targets (no implicit last-route sends). */
requireExplicitMessageTarget?: boolean;
@@ -497,6 +497,7 @@ export function createOpenClawTools(
threadId: options?.currentThreadTs ?? options?.agentThreadId,
},
creatorToolAllowlist: options?.cronCreatorToolAllowlist,
runId: options?.runId,
...(options?.cronSelfRemoveOnlyJobId
? { selfRemoveOnlyJobId: options.cronSelfRemoveOnlyJobId }
: {}),
src/agents/tools/cron-tool-canonicalize.ts
+1
View File
@@ -42,6 +42,7 @@ const CRON_RECOVERABLE_OBJECT_KEYS: ReadonlySet<string> = new Set([
"displayName",
"owner",
"schedule",
"pacing",
"trigger",
"sessionTarget",
"wakeMode",
src/agents/tools/cron-tool.pacing.test.ts
+87
View File
@@ -0,0 +1,87 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import {
claimAgentRunContext,
clearAgentRunContext,
consumeCronNextCheckProposal,
} from "../../infra/agent-events.js";
import { createCronTool } from "./cron-tool.js";
const RUN_ID = "paced-run";
const JOB_ID = "paced-job";
afterEach(() => {
clearAgentRunContext(RUN_ID);
});
function createScopedTool() {
return createCronTool(
{ selfRemoveOnlyJobId: JOB_ID, runId: RUN_ID },
{ callGatewayTool: vi.fn() },
);
}
function registerRun(pacingEnabled: boolean) {
claimAgentRunContext(RUN_ID, {
sessionKey: `agent:main:cron:${JOB_ID}`,
cronJobId: JOB_ID,
cronPacingEnabled: pacingEnabled,
});
}
describe("cron next_check action", () => {
it("lets a restricted isolated run record a proposal for its own paced job", async () => {
registerRun(true);
const result = await createScopedTool().execute("call-next-check", {
action: "next_check",
in: "1h30m",
});
expect(result.details).toEqual({ ok: true, delayMs: 90 * 60_000 });
expect(consumeCronNextCheckProposal(RUN_ID, JOB_ID)).toBe(90 * 60_000);
expect(consumeCronNextCheckProposal(RUN_ID, JOB_ID)).toBeUndefined();
});
it("rejects a proposal when the current job has no pacing", async () => {
registerRun(false);
await expect(
createScopedTool().execute("call-next-check", { action: "next_check", in: "15m" }),
).rejects.toThrow("cron next_check requires pacing on the current job");
});
it("rejects arbitrary job targeting", async () => {
registerRun(true);
await expect(
createScopedTool().execute("call-next-check-other", {
action: "next_check",
jobId: "another-job",
in: "15m",
}),
).rejects.toThrow("Cron tool is restricted to the current cron job.");
});
it("rejects next_check outside a current cron run", async () => {
const tool = createCronTool(undefined, { callGatewayTool: vi.fn() });
await expect(
tool.execute("call-next-check-unscoped", { action: "next_check", in: "15m" }),
).rejects.toThrow("cron next_check is only available to the currently running job");
});
it("drops an unconsumed proposal when the run context changes jobs", async () => {
registerRun(true);
await createScopedTool().execute("call-next-check-stale", {
action: "next_check",
in: "15m",
});
claimAgentRunContext(RUN_ID, {
cronJobId: "next-job",
cronPacingEnabled: true,
});
expect(consumeCronNextCheckProposal(RUN_ID, "next-job")).toBeUndefined();
});
});
src/agents/tools/cron-tool.schema.test.ts
+15
View File
@@ -55,6 +55,7 @@ describe("createCronToolSchema", () => {
"failureAlert",
"name",
"owner",
"pacing",
"payload",
"schedule",
"sessionKey",
@@ -85,6 +86,7 @@ describe("createCronToolSchema", () => {
"enabled",
"failureAlert",
"name",
"pacing",
"payload",
"schedule",
"sessionKey",
@@ -95,6 +97,19 @@ describe("createCronToolSchema", () => {
);
});
it("exposes next_check with its relative duration parameter", () => {
expect(Value.Check(schema, { action: "next_check", in: "15m" })).toBe(true);
expect(propertyAt(schemaRecord, "in")?.description).toContain("next_check");
expect(keysAt(schemaRecord, "job.pacing")).toEqual(["max", "min"]);
const patchPacing = propertyAt(schemaRecord, "patch.pacing");
const pacingObject = (patchPacing?.anyOf as Array<Record<string, unknown>> | undefined)?.find(
(entry) => entry.type === "object",
);
expect(
Object.keys((pacingObject?.properties as Record<string, unknown>) ?? {}).toSorted(),
).toEqual(["max", "min"]);
});
it("job.schedule exposes kind, at, everyMs, anchorMs, expr, tz, staggerMs", () => {
expect(keysAt(schemaRecord, "job.schedule")).toEqual(
["anchorMs", "at", "everyMs", "expr", "kind", "staggerMs", "tz"].toSorted(),
src/agents/tools/cron-tool.test.ts
+30 -1
View File
@@ -25,7 +25,7 @@ import { createCronTool } from "./cron-tool.js";
describe("cron tool", () => {
type SchemaLike = {
anyOf?: Array<{ type?: string }>;
anyOf?: Array<SchemaLike>;
description?: string;
properties?: Record<string, SchemaLike>;
type?: string;
@@ -744,6 +744,8 @@ describe("cron tool", () => {
const patch = parameters.properties?.patch;
const payload = patch?.properties?.payload;
const delivery = patch?.properties?.delivery;
const jobPacing = parameters.properties?.job?.properties?.pacing;
const patchPacing = patch?.properties?.pacing?.anyOf?.find((entry) => entry.type === "object");
expect(jobDelivery?.properties?.channel?.anyOf).toBeUndefined();
expect(jobDelivery?.properties?.channel?.type).toBe("string");
@@ -777,6 +779,8 @@ describe("cron tool", () => {
"object",
"null",
]);
expect(jobPacing?.description).toContain("at least one of min or max is required");
expect(patchPacing?.description).toContain("at least one of min or max is required");
});
it.each([
@@ -932,6 +936,31 @@ describe("cron tool", () => {
expect(callGatewayMock).not.toHaveBeenCalled();
});
it.each([
[
"add",
{
action: "add",
job: { ...buildReminderAgentTurnJob(), pacing: {} },
},
],
[
"update",
{
action: "update",
jobId: "paced-job",
patch: { pacing: {} },
},
],
])("rejects empty pacing on cron.%s before calling the gateway", async (_action, args) => {
const tool = createTestCronTool();
await expect(tool.execute("call-empty-pacing", args)).rejects.toThrow(
"cron pacing requires at least one of min or max",
);
expect(callGatewayMock).not.toHaveBeenCalled();
});
it("rejects null agentId on add from the scoped agent cron tool", async () => {
const tool = createTestCronTool({ agentSessionKey: "main" });
await expect(
src/agents/tools/cron-tool.ts
+63 -4
View File
@@ -5,13 +5,16 @@
*/
import { normalizeLowercaseStringOrEmpty } from "@openclaw/normalization-core/string-coerce";
import { Type, type TSchema } from "typebox";
import { parseDurationMs } from "../../cli/parse-duration.js";
import { getRuntimeConfig, type OpenClawConfig } from "../../config/config.js";
import { resolveCronCreationDelivery } from "../../cron/delivery-context.js";
import { assertCronDeliveryInputNonBlankFields } from "../../cron/delivery-target-validation.js";
import { normalizeCronJobCreate, normalizeCronJobPatch } from "../../cron/normalize.js";
import type { CronDelivery } from "../../cron/types.js";
import { parseCronPacingBounds } from "../../cron/pacing.js";
import type { CronDelivery, CronPacing } from "../../cron/types.js";
import { normalizeHttpWebhookUrl } from "../../cron/webhook-url.js";
import { GatewayClientRequestError } from "../../gateway/client.js";
import { recordCronNextCheckProposal } from "../../infra/agent-events.js";
import { normalizeAgentId } from "../../routing/session-key.js";
import { parseAgentSessionKey } from "../../sessions/session-key-utils.js";
import { extractTextFromChatContent } from "../../shared/chat-content.js";
@@ -68,6 +71,7 @@ const CRON_ACTIONS = [
"remove",
"run",
"runs",
"next_check",
"wake",
] as const;
@@ -165,6 +169,30 @@ function createCronScheduleSchema(): TSchema {
);
}
function createCronPacingSchema(params: { nullableClears: boolean }): TSchema {
const pacing = Type.Object(
{
min: Type.Optional(Type.String({ description: "Minimum dynamic delay" })),
max: Type.Optional(Type.String({ description: "Maximum dynamic delay" })),
},
{
additionalProperties: false,
description: "Dynamic-cadence bounds; at least one of min or max is required",
},
);
return Type.Optional(params.nullableClears ? Type.Union([pacing, Type.Null()]) : pacing);
}
function assertCronPacingInput(value: unknown, params: { nullableClears: boolean }): void {
if (value === undefined || (params.nullableClears && value === null)) {
return;
}
if (!isRecord(value)) {
throw new Error("cron pacing must be an object");
}
parseCronPacingBounds(value as CronPacing);
}
function createCronPayloadSchema(): TSchema {
return Type.Optional(
cronPayloadObjectSchema({
@@ -294,6 +322,7 @@ function createCronJobObjectSchema(): TSchema {
),
),
schedule: createCronScheduleSchema(),
pacing: createCronPacingSchema({ nullableClears: false }),
trigger: createCronTriggerSchema({ nullableClears: false }),
sessionTarget: Type.Optional(
Type.String({
@@ -326,6 +355,7 @@ function createCronPatchObjectSchema(): TSchema {
}),
),
schedule: createCronScheduleSchema(),
pacing: createCronPacingSchema({ nullableClears: true }),
trigger: createCronTriggerSchema({ nullableClears: true }),
sessionTarget: Type.Optional(Type.String({ description: "Session target" })),
wakeMode: optionalStringEnum(CRON_WAKE_MODES),
@@ -360,6 +390,11 @@ function createCronToolSchema(): TSchema {
jobId: Type.Optional(Type.String()),
id: Type.Optional(Type.String()),
patch: createCronPatchObjectSchema(),
in: Type.Optional(
Type.String({
description: 'Relative duration for action="next_check" (for example, "15m")',
}),
),
text: Type.Optional(Type.String()),
mode: optionalStringEnum(CRON_WAKE_MODES),
runMode: optionalStringEnum(CRON_RUN_MODES, {
@@ -509,6 +544,9 @@ function assertCronSelfRemoveScope(
if (!selfRemoveOnlyJobId || isCronSelfIntrospectionAction(action)) {
return;
}
if (action === "next_check" && params.jobId === undefined && params.id === undefined) {
return;
}
if (action === "get" || action === "remove" || action === "runs") {
const id = readCronJobIdParam(params);
if (id && id === selfRemoveOnlyJobId) {
@@ -691,11 +729,11 @@ export function createCronTool(opts?: CronToolOptions, deps?: CronToolDeps): Any
ACTIONS:
- status scheduler; list compact summaries (includeDisabled, session agentId auto-filter; get for full); get jobId
- add job; update jobId+patch; remove jobId
- run jobId (due only; runMode="force" now); runs jobId history
- run jobId (due only; runMode="force" now); runs jobId history; next_check in (current paced job only)
- wake text (+ optional mode). Default caller lane; top-level sessionKey/agentId selects another caller-owned lane.
ADD JOB:
{ "name":"...", "schedule":{...}, "trigger":{ "script":"...", "once":false }, "payload":{...}, "delivery":{...}, "sessionTarget":"main|isolated|current|session:<id>", "enabled":true }
{ "name":"...", "schedule":{...}, "pacing":{ "min":"15m", "max":"4h" }, "trigger":{ "script":"...", "once":false }, "payload":{...}, "delivery":{...}, "sessionTarget":"main|isolated|current|session:<id>", "enabled":true }
Required: schedule,payload. enabled default true. trigger only every/cron.
TARGET/PAYLOAD:
@@ -722,7 +760,7 @@ DELIVERY top-level: {mode:"none|announce|webhook",channel?,to?,threadId?,bestEff
- Isolated agentTurn omitted delivery => announce. announce only isolated/current/session; channel/to optional; threadId chat topic. Specific chat: set channel/to; no messaging tool inside run.
- webhook posts finished-run event to URL in to.
Restricted isolated runs may only self status/list, current get/runs, and remove current job. wake mode: next-heartbeat default | now. jobId canonical; id compat. contextMessages 0-10 adds prior messages.`,
Restricted isolated runs may only self status/list, current get/runs/remove, and next_check for their own paced job. wake mode: next-heartbeat default | now. jobId canonical; id compat. contextMessages 0-10 adds prior messages.`,
parameters: createCronToolSchema(),
execute: async (_toolCallId, args) => {
const params = args as Record<string, unknown>;
@@ -829,6 +867,7 @@ Restricted isolated runs may only self status/list, current get/runs, and remove
const canonicalJob = canonicalizeCronToolObject(params.job as Record<string, unknown>);
assertNoCronShellExecution(canonicalJob);
assertCronDeliveryInputNonBlankFields(canonicalJob.delivery);
assertCronPacingInput(canonicalJob.pacing, { nullableClears: false });
if (
typeof canonicalJob.declarationKey === "string" &&
canonicalJob.declarationKey.trim().length === 0
@@ -973,6 +1012,7 @@ Restricted isolated runs may only self status/list, current get/runs, and remove
);
assertNoCronShellExecution(canonicalPatch);
assertCronDeliveryInputNonBlankFields(canonicalPatch.delivery);
assertCronPacingInput(canonicalPatch.pacing, { nullableClears: true });
if (
typeof canonicalPatch.displayName === "string" &&
canonicalPatch.displayName.trim().length === 0
@@ -1035,6 +1075,25 @@ Restricted isolated runs may only self status/list, current get/runs, and remove
}),
);
}
case "next_check": {
const jobId = readCronSelfRemoveOnlyJobId(opts);
const runId = opts?.runId?.trim();
if (!jobId || !runId) {
throw new Error("cron next_check is only available to the currently running job");
}
const rawDuration = readStringParam(params, "in", { required: true });
let delayMs: number;
try {
delayMs = parseDurationMs(rawDuration);
} catch {
throw new Error("cron next_check in must be a positive duration");
}
if (delayMs <= 0) {
throw new Error("cron next_check in must be a positive duration");
}
recordCronNextCheckProposal(runId, jobId, delayMs);
return jsonResult({ ok: true, delayMs });
}
case "wake": {
const text = readStringParam(params, "text", { required: true });
const mode =
src/agents/tools/cron-tool.types.ts
+1
View File
@@ -19,6 +19,7 @@ export type CronToolOptions = {
*/
creatorToolAllowlist?: CronCreatorToolAllowlistEntry[];
selfRemoveOnlyJobId?: string;
runId?: string;
};
export type CronToolCallerScope = {
src/cli/cron-cli/register.cron-add.ts
+18
View File
@@ -102,6 +102,8 @@ export function registerCronAddCommand(cron: Command) {
"Run once at time (ISO with offset, or +duration). Use --tz for offset-less datetimes",
)
.option("--every <duration>", "Run every duration (e.g. 10m, 1h)")
.option("--pacing-min <duration>", "Minimum delay accepted from a dynamic next check")
.option("--pacing-max <duration>", "Maximum delay accepted from a dynamic next check")
.option("--cron <expr>", "Cron expression (5-field or 6-field with seconds)")
.option(
"--on-exit <shell>",
@@ -376,6 +378,14 @@ export function registerCronAddCommand(cron: Command) {
if (typeof opts.displayName === "string" && !displayName) {
throw new Error("--display-name must not be blank");
}
const pacingMin = normalizeOptionalString(opts.pacingMin);
const pacingMax = normalizeOptionalString(opts.pacingMax);
if (typeof opts.pacingMin === "string" && !pacingMin) {
throw new Error("--pacing-min must not be blank");
}
if (typeof opts.pacingMax === "string" && !pacingMax) {
throw new Error("--pacing-max must not be blank");
}
const sessionKey = normalizeOptionalString(opts.sessionKey);
const triggerScriptPath = normalizeOptionalString(opts.triggerScript);
@@ -410,6 +420,14 @@ export function registerCronAddCommand(cron: Command) {
agentId,
sessionKey,
schedule,
...(pacingMin || pacingMax
? {
pacing: {
...(pacingMin ? { min: pacingMin } : {}),
...(pacingMax ? { max: pacingMax } : {}),
},
}
: {}),
trigger,
sessionTarget,
wakeMode,
src/cli/cron-cli/register.cron-edit.test.ts
+17
View File
@@ -37,6 +37,23 @@ describe("cron edit command", () => {
expect(help).toMatch(/also\s+implies --announce when used alone/);
});
it("updates one pacing bound while preserving the other", async () => {
callGatewayFromCli.mockImplementation(async (method: string) => {
if (method === "cron.get") {
return { id: "job-1", pacing: { min: "15m", max: "4h" } };
}
return { ok: true };
});
const program = createCronProgram();
await program.parseAsync(["edit", "job-1", "--pacing-min", "30m"], { from: "user" });
expect(callGatewayFromCli).toHaveBeenCalledWith("cron.update", expect.anything(), {
id: "job-1",
patch: { pacing: { min: "30m", max: "4h" } },
});
});
it("keeps --best-effort-deliver-only edits delivery-only (#83908)", async () => {
const program = createCronProgram();
src/cli/cron-cli/register.cron-edit.ts
+27
View File
@@ -113,6 +113,9 @@ export function registerCronEditCommand(cron: Command) {
.option("--wake <mode>", "Wake mode (now|next-heartbeat)")
.option("--at <when>", "Set one-shot time (ISO, offset-less uses --tz) or duration like 20m")
.option("--every <duration>", "Set interval duration like 10m")
.option("--pacing-min <duration>", "Set minimum delay for a dynamic next check")
.option("--pacing-max <duration>", "Set maximum delay for a dynamic next check")
.option("--clear-pacing", "Remove dynamic-cadence bounds", false)
.option("--cron <expr>", "Set cron expression")
.option(
"--tz <iana>",
@@ -292,6 +295,30 @@ export function registerCronEditCommand(cron: Command) {
patch.sessionKey = null;
}
const pacingMin = normalizeOptionalString(opts.pacingMin);
const pacingMax = normalizeOptionalString(opts.pacingMax);
const hasPacingMin = typeof opts.pacingMin === "string";
const hasPacingMax = typeof opts.pacingMax === "string";
if (hasPacingMin && !pacingMin) {
throw new Error("--pacing-min must not be blank");
}
if (hasPacingMax && !pacingMax) {
throw new Error("--pacing-max must not be blank");
}
if (opts.clearPacing && (hasPacingMin || hasPacingMax)) {
throw new Error("Use --clear-pacing or pacing bounds, not both");
}
if (opts.clearPacing) {
patch.pacing = null;
} else if (hasPacingMin || hasPacingMax) {
const existing = await readCronJobForEdit(opts, String(id));
patch.pacing = {
...existing.pacing,
...(pacingMin ? { min: pacingMin } : {}),
...(pacingMax ? { max: pacingMax } : {}),
};
}
const triggerScriptPath = normalizeOptionalString(opts.triggerScript);
if (opts.clearTrigger && (triggerScriptPath || opts.triggerOnce)) {
throw new Error("Use --clear-trigger or trigger options, not both");
src/cli/cron-cli/trigger-options.test.ts
+43
View File
@@ -79,6 +79,36 @@ describe("cron trigger CLI options", () => {
);
});
it("sends pacing bounds on add", async () => {
const program = new Command().exitOverride();
registerCronAddCommand(program);
await program.parseAsync(
[
"add",
"--name",
"paced",
"--every",
"30m",
"--pacing-min",
"15m",
"--pacing-max",
"4h",
"--system-event",
"check",
"--session",
"main",
],
{ from: "user" },
);
expect(callGatewayFromCli).toHaveBeenCalledWith(
"cron.add",
expect.anything(),
expect.objectContaining({ pacing: { min: "15m", max: "4h" } }),
);
});
it("accepts trigger script files at the byte limit", async () => {
const scriptPath = path.join(fixtureRoot, "at-limit.js");
await fs.writeFile(scriptPath, "x".repeat(65_536), "utf8");
@@ -138,4 +168,17 @@ describe("cron trigger CLI options", () => {
{ id: "job-1", patch: { trigger: null } },
);
});
it("maps --clear-pacing to a nullable edit patch", async () => {
const program = new Command().exitOverride();
registerCronEditCommand(program);
await program.parseAsync(["edit", "job-1", "--clear-pacing"], { from: "user" });
expect(callGatewayFromCli).toHaveBeenCalledWith(
"cron.update",
expect.objectContaining({ clearPacing: true }),
{ id: "job-1", patch: { pacing: null } },
);
});
});
src/cron/cron-protocol-schema.test.ts
+11 -1
View File
@@ -1,8 +1,12 @@
// Cron protocol schema tests cover runtime validation for cron protocol payloads.
import { describe, expect, it } from "vitest";
import { CronJobStateSchema } from "../../packages/gateway-protocol/src/schema.js";
import {
CronJobStateSchema,
CronPacingSchema,
} from "../../packages/gateway-protocol/src/schema.js";
type SchemaLike = {
description?: string;
properties?: Record<string, unknown>;
deprecated?: boolean;
};
@@ -23,4 +27,10 @@ describe("cron protocol schema", () => {
expect(properties.lastFailureNotificationDeliveryStatus).toBeDefined();
expect(properties.lastFailureNotificationDeliveryError).toBeDefined();
});
it("documents that pacing requires at least one bound", () => {
expect((CronPacingSchema as SchemaLike).description).toContain(
"at least one of min or max is required",
);
});
});
src/cron/isolated-agent/run.ts
+8 -1
View File
@@ -22,6 +22,7 @@ import type { OpenClawConfig } from "../../config/types.openclaw.js";
import {
assertAgentRunLifecycleGenerationCurrent,
claimAgentRunContext,
consumeCronNextCheckProposal,
getAgentEventLifecycleGeneration,
getAgentRunContext,
releaseAgentRunContext,
@@ -1773,6 +1774,8 @@ export async function runCronIsolatedAgentTurn(params: {
: existingRunContext.sessionKey,
sessionId: initialSessionId,
lifecycleGeneration: runLifecycleGeneration,
cronJobId: params.job.id,
cronPacingEnabled: params.job.pacing !== undefined,
},
{
trackOwner: true,
@@ -1848,8 +1851,12 @@ export async function runCronIsolatedAgentTurn(params: {
outcome = "error";
outcomeError = finalized.error;
}
return finalized;
const delayMs = consumeCronNextCheckProposal(initialSessionId, params.job.id);
return finalized.status !== "ok" || delayMs === undefined
? finalized
: { ...finalized, nextCheck: { delayMs } };
} catch (err) {
consumeCronNextCheckProposal(initialSessionId, params.job.id);
const isCronLaneTimeout = isAborted() || isCronNestedLaneTaskTimeoutError(err);
const error = isCronLaneTimeout ? abortReason() : String(err);
outcome = "error";
src/cron/isolated-agent/run.types.ts
+7 -1
View File
@@ -1,5 +1,10 @@
/** Result types returned by isolated cron agent runs. */
import type { CronDeliveryTrace, CronRunOutcome, CronRunTelemetry } from "../types.js";
import type {
CronDeliveryTrace,
CronNextCheckProposal,
CronRunOutcome,
CronRunTelemetry,
} from "../types.js";
/** Final isolated cron turn result merged into service state and run logs. */
export type RunCronAgentTurnResult = {
@@ -21,5 +26,6 @@ export type RunCronAgentTurnResult = {
/** Post-run delivery failure on an otherwise successful isolated turn. */
deliveryError?: string;
delivery?: CronDeliveryTrace;
nextCheck?: CronNextCheckProposal;
} & CronRunOutcome &
CronRunTelemetry;
src/cron/pacing.test.ts
+10
View File
@@ -0,0 +1,10 @@
import { describe, expect, it } from "vitest";
import { parseCronPacingBounds } from "./pacing.js";
describe("parseCronPacingBounds", () => {
it("rejects pacing without a minimum or maximum", () => {
expect(() => parseCronPacingBounds({})).toThrow(
"cron pacing requires at least one of min or max",
);
});
});
src/cron/pacing.ts
+50
View File
@@ -0,0 +1,50 @@
import { parseDurationMs } from "../cli/parse-duration.js";
import type { CronPacing } from "./types.js";
/** Parsed positive pacing bounds used for validation and next-run clamping. */
type CronPacingBounds = {
minMs?: number;
maxMs?: number;
};
function parsePositivePacingDuration(value: string, field: "min" | "max"): number {
let durationMs: number;
try {
durationMs = parseDurationMs(value);
} catch {
throw new Error(`cron pacing ${field} must be a positive duration`);
}
if (durationMs <= 0) {
throw new Error(`cron pacing ${field} must be a positive duration`);
}
return durationMs;
}
/** Validates pacing strings and returns their millisecond bounds. */
export function parseCronPacingBounds(pacing: CronPacing): CronPacingBounds {
if (pacing.min === undefined && pacing.max === undefined) {
throw new Error("cron pacing requires at least one of min or max");
}
const minMs =
pacing.min === undefined ? undefined : parsePositivePacingDuration(pacing.min, "min");
const maxMs =
pacing.max === undefined ? undefined : parsePositivePacingDuration(pacing.max, "max");
if (minMs !== undefined && maxMs !== undefined && minMs > maxMs) {
throw new Error("cron pacing min must not exceed max");
}
return { minMs, maxMs };
}
/** Clamps one successful run's proposal against its job-local pacing bounds. */
export function resolvePacedNextRunAtMs(params: {
nowMs: number;
delayMs: number;
pacing: CronPacing;
}): number {
const { minMs, maxMs } = parseCronPacingBounds(params.pacing);
const proposedAtMs = params.nowMs + params.delayMs;
return Math.min(
params.nowMs + (maxMs ?? Number.POSITIVE_INFINITY),
Math.max(params.nowMs + (minMs ?? 0), proposedAtMs),
);
}
src/cron/public-job.test.ts
+19
View File
@@ -0,0 +1,19 @@
import { describe, expect, it } from "vitest";
import { makeCronJob } from "./delivery.test-helpers.js";
import { toPublicCronJob } from "./public-job.js";
describe("toPublicCronJob", () => {
it("strips scheduler-only pacing slots without mutating stored state", () => {
const job = makeCronJob({
state: {
nextRunAtMs: 2_000,
pacedNextRunAtMs: 2_000,
},
});
const publicJob = toPublicCronJob(job);
expect(publicJob.state.pacedNextRunAtMs).toBeUndefined();
expect(job.state.pacedNextRunAtMs).toBe(2_000);
});
});
src/cron/public-job.ts
+1
View File
@@ -5,5 +5,6 @@ export function toPublicCronJob(job: CronJob): CronJob {
const state = { ...job.state };
delete state.queuedAtMs;
delete state.startupCatchupAtMs;
delete state.pacedNextRunAtMs;
return { ...job, state };
}
src/cron/service/jobs.ts
+39 -1
View File
@@ -10,6 +10,7 @@ import type { CronConfig } from "../../config/types.cron.js";
import { normalizeAgentId } from "../../routing/session-key.js";
import { isCronJobActive } from "../active-jobs.js";
import { resolveCronDeliveryPlan } from "../delivery-plan.js";
import { parseCronPacingBounds } from "../pacing.js";
import { parseAbsoluteTimeMs } from "../parse.js";
import {
coerceFiniteScheduleNumber,
@@ -248,7 +249,6 @@ function shouldRepairFutureCronNextRunAtMs(params: {
if (isPendingErrorBackoffSlot({ state, job, nextRunAtMs: nextRun, nowMs })) {
return false;
}
let naturalNext: number | undefined;
try {
naturalNext = computeStaggeredCronNextRunAtMs(job, nowMs);
@@ -613,6 +613,7 @@ function normalizeJobTickState(params: { state: CronServiceState; job: CronJob;
...job.schedule,
anchorMs: normalizedAnchorMs,
};
job.state.pacedNextRunAtMs = undefined;
changed = true;
}
}
@@ -622,6 +623,10 @@ function normalizeJobTickState(params: { state: CronServiceState; job: CronJob;
job.state.startupCatchupAtMs = undefined;
changed = true;
}
if (job.state.pacedNextRunAtMs !== undefined) {
job.state.pacedNextRunAtMs = undefined;
changed = true;
}
if (job.state.nextRunAtMs !== undefined) {
job.state.nextRunAtMs = undefined;
changed = true;
@@ -804,6 +809,7 @@ export function recomputeNextRunsForMaintenance(
let changed = false;
const startupCatchupAtMs = job.state.startupCatchupAtMs;
const pacedNextRunAtMs = job.state.pacedNextRunAtMs;
const nextRunAtMs = job.state.nextRunAtMs;
// The persisted marker owns only its exact future slot. Schedule edits,
// malformed state, or arrival at the slot release normal repair policy.
@@ -816,6 +822,15 @@ export function recomputeNextRunsForMaintenance(
job.state.startupCatchupAtMs = undefined;
changed = true;
}
const hasPendingPacedNextRun =
isFiniteTimestamp(pacedNextRunAtMs) &&
hasScheduledNextRunAtMs(nextRunAtMs) &&
pacedNextRunAtMs === nextRunAtMs &&
now < pacedNextRunAtMs;
if (pacedNextRunAtMs !== undefined && !hasPendingPacedNextRun) {
job.state.pacedNextRunAtMs = undefined;
changed = true;
}
if (!hasScheduledNextRunAtMs(job.state.nextRunAtMs)) {
if (recomputeJob(job, now)) {
@@ -824,6 +839,7 @@ export function recomputeNextRunsForMaintenance(
} else if (
repairFutureCronNextRunAtMs &&
!hasPendingStartupCatchup &&
!hasPendingPacedNextRun &&
shouldRepairFutureCronNextRunAtMs({ state, job, nowMs: now })
) {
if (recomputeJob(job, now)) {
@@ -952,6 +968,7 @@ export function createJob(state: CronServiceState, input: CronJobCreate): CronJo
createdAtMs: now,
updatedAtMs: now,
schedule,
...(input.pacing !== undefined ? { pacing: structuredClone(input.pacing) } : {}),
sessionTarget: input.sessionTarget,
wakeMode: input.wakeMode,
payload: input.payload,
@@ -963,6 +980,9 @@ export function createJob(state: CronServiceState, input: CronJobCreate): CronJo
},
};
assertSupportedJobSpec(job);
if (job.pacing !== undefined) {
parseCronPacingBounds(job.pacing);
}
assertTriggerSupport(job, {
cronConfig: state.deps.cronConfig,
requireEnabled: job.trigger !== undefined,
@@ -1054,6 +1074,13 @@ export function applyJobPatch(
job.trigger = structuredClone(patch.trigger);
}
}
if ("pacing" in patch) {
if (patch.pacing === null || patch.pacing === undefined) {
delete job.pacing;
} else {
job.pacing = structuredClone(patch.pacing);
}
}
if (patch.sessionTarget) {
job.sessionTarget = patch.sessionTarget;
}
@@ -1098,6 +1125,9 @@ export function applyJobPatch(
job.sessionKey = normalizeOptionalString((patch as { sessionKey?: unknown }).sessionKey);
}
assertSupportedJobSpec(job);
if (job.pacing !== undefined) {
parseCronPacingBounds(job.pacing);
}
assertTriggerSupport(job, {
cronConfig: opts?.cronConfig,
requireEnabled: patch.trigger !== null && patch.trigger !== undefined,
@@ -1164,6 +1194,11 @@ export function applyDeclarativeJobSpec(
} else {
job.schedule = structuredClone(input.schedule);
}
if (input.pacing !== undefined) {
job.pacing = structuredClone(input.pacing);
} else {
delete job.pacing;
}
job.payload = structuredClone(input.payload);
if (input.trigger) {
job.trigger = structuredClone(input.trigger);
@@ -1185,6 +1220,9 @@ export function applyDeclarativeJobSpec(
});
assertSupportedJobSpec(job);
if (job.pacing !== undefined) {
parseCronPacingBounds(job.pacing);
}
assertMainSessionAgentId(job, opts.defaultAgentId);
assertDeliverySupport(job);
assertFailureDestinationSupport(job);
src/cron/service/ops.pacing.test.ts
+110
View File
@@ -0,0 +1,110 @@
import path from "node:path";
import { describe, expect, it, vi } from "vitest";
import { withEnvAsync } from "../../test-utils/env.js";
import { setupCronServiceSuite } from "../service.test-harness.js";
import type { CronJobCreate, CronJobPatch, CronPacing } from "../types.js";
import { add, update } from "./ops.js";
import { createCronServiceState } from "./state.js";
const { logger, makeStorePath } = setupCronServiceSuite({ prefix: "cron-pacing-ops" });
function makeInput(pacing: CronPacing): CronJobCreate {
return {
name: "paced job",
enabled: true,
schedule: { kind: "every", everyMs: 60_000 },
pacing,
sessionTarget: "isolated",
wakeMode: "now",
payload: { kind: "agentTurn", message: "check" },
};
}
async function withState(run: (state: ReturnType<typeof createCronServiceState>) => Promise<void>) {
const { storePath } = await makeStorePath();
await withEnvAsync({ OPENCLAW_STATE_DIR: path.dirname(path.dirname(storePath)) }, async () => {
await run(
createCronServiceState({
storePath,
cronEnabled: true,
log: logger,
nowMs: () => Date.parse("2026-07-18T12:00:00.000Z"),
enqueueSystemEvent: vi.fn(),
requestHeartbeat: vi.fn(),
runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })),
}),
);
});
}
describe("cron pacing validation", () => {
it("accepts duration strings on create and update", async () => {
await withState(async (state) => {
const job = await add(state, makeInput({ min: "15m", max: "4h" }));
expect(job.pacing).toEqual({ min: "15m", max: "4h" });
const updated = await update(state, job.id, { pacing: { min: "30m", max: "2h" } });
expect(updated.pacing).toEqual({ min: "30m", max: "2h" });
});
});
it.each([
["no bounds", {}, /pacing requires at least one of min or max/],
["zero minimum", { min: "0s" }, /pacing min must be a positive duration/],
["negative maximum", { max: "-1m" }, /pacing max must be a positive duration/],
["minimum above maximum", { min: "4h", max: "15m" }, /pacing min must not exceed max/],
] as const)("rejects %s on create", async (_label, pacing, error) => {
await withState(async (state) => {
await expect(add(state, makeInput(pacing))).rejects.toThrow(error);
});
});
it("rejects invalid pacing on update without changing the stored job", async () => {
await withState(async (state) => {
const job = await add(state, makeInput({ min: "15m" }));
await expect(update(state, job.id, { pacing: { max: "0m" } })).rejects.toThrow(
"cron pacing max must be a positive duration",
);
expect(state.store?.jobs[0]?.pacing).toEqual({ min: "15m" });
});
});
it("rejects empty pacing on update without changing the stored job", async () => {
await withState(async (state) => {
const job = await add(state, makeInput({ min: "15m" }));
await expect(update(state, job.id, { pacing: {} })).rejects.toThrow(
"cron pacing requires at least one of min or max",
);
expect(state.store?.jobs[0]?.pacing).toEqual({ min: "15m" });
});
});
it("accepts a nullable pacing patch and clears pacing and its pending slot", async () => {
await withState(async (state) => {
const job = await add(state, makeInput({ min: "15m" }));
job.state.pacedNextRunAtMs = job.state.nextRunAtMs;
const patch = { pacing: null } satisfies CronJobPatch;
const updated = await update(state, job.id, patch);
expect(updated.pacing).toBeUndefined();
expect(updated.state.pacedNextRunAtMs).toBeUndefined();
expect(state.store?.jobs[0]?.pacing).toBeUndefined();
expect(state.store?.jobs[0]?.state.pacedNextRunAtMs).toBeUndefined();
});
});
it("clears a pending paced slot on an unrelated edit", async () => {
await withState(async (state) => {
const job = await add(state, makeInput({ max: "4h" }));
job.state.pacedNextRunAtMs = job.state.nextRunAtMs;
const updated = await update(state, job.id, { description: "edited" });
expect(updated.pacing).toEqual({ max: "4h" });
expect(updated.state.pacedNextRunAtMs).toBeUndefined();
});
});
});
src/cron/service/ops.ts
+2
View File
@@ -501,6 +501,7 @@ function finalizeUpdatedJob(params: {
}
nextJob.updatedAtMs = now;
nextJob.state.pacedNextRunAtMs = undefined;
if (schedulingInputsChanged) {
nextJob.state.startupCatchupAtMs = undefined;
if (isJobEnabled(nextJob)) {
@@ -545,6 +546,7 @@ async function persistUpdatedJob(params: {
function declarativeFields(job: CronJob, includeEnabled: boolean) {
return {
schedule: job.schedule,
pacing: job.pacing,
trigger: job.trigger,
payload: job.payload,
delivery: job.delivery,
src/cron/service/state.ts
+2
View File
@@ -13,6 +13,7 @@ import type {
CronDeliveryStatus,
CronDeliveryTrace,
CronJob,
CronNextCheckProposal,
CronJobCreate,
CronJobPatch,
CronRunDiagnostics,
@@ -167,6 +168,7 @@ export type CronServiceDeps = {
*/
deliveryAttempted?: boolean;
delivery?: CronDeliveryTrace;
nextCheck?: CronNextCheckProposal;
} & CronRunOutcome &
CronRunTelemetry
>;
src/cron/service/timer.pacing.test.ts
+159
View File
@@ -0,0 +1,159 @@
import { describe, expect, it, vi } from "vitest";
import { makeCronJob } from "../delivery.test-helpers.js";
import { createNoopLogger } from "../service.test-harness.js";
import type { CronJob, CronPacing } from "../types.js";
import { recomputeNextRunsForMaintenance } from "./jobs.js";
import { createCronServiceState } from "./state.js";
import { applyJobResult } from "./timer.js";
const ENDED_AT = Date.parse("2026-07-18T12:00:00.000Z");
const STARTED_AT = ENDED_AT - 1_000;
function makeState() {
return createCronServiceState({
storePath: "/tmp/cron-pacing-timer/jobs.json",
cronEnabled: true,
log: createNoopLogger(),
nowMs: () => ENDED_AT,
enqueueSystemEvent: vi.fn(),
requestHeartbeat: vi.fn(),
runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })),
});
}
function makePacedJob(pacing: CronPacing, everyMs = 60 * 60_000): CronJob {
return makeCronJob({
pacing,
schedule: { kind: "every", everyMs, anchorMs: STARTED_AT },
state: { nextRunAtMs: STARTED_AT },
});
}
describe("applyJobResult dynamic cadence", () => {
it.each([
["honors an in-range proposal", { min: "15m", max: "4h" }, 60 * 60_000, 60 * 60_000],
["clamps below the minimum", { min: "15m", max: "4h" }, 5 * 60_000, 15 * 60_000],
["clamps above the maximum", { min: "15m", max: "4h" }, 6 * 60 * 60_000, 4 * 60 * 60_000],
["clamps a minimum-only job", { min: "15m" }, 5 * 60_000, 15 * 60_000],
["clamps a maximum-only job", { max: "4h" }, 6 * 60 * 60_000, 4 * 60 * 60_000],
] as const)("%s", (_label, pacing, delayMs, expectedDelayMs) => {
const job = makePacedJob(pacing);
applyJobResult(makeState(), job, {
status: "ok",
startedAt: STARTED_AT,
endedAt: ENDED_AT,
nextCheck: { delayMs },
});
expect(job.state.nextRunAtMs).toBe(ENDED_AT + expectedDelayMs);
expect(job.state.pacedNextRunAtMs).toBe(ENDED_AT + expectedDelayMs);
});
it("keeps existing schedule math when no proposal was recorded", () => {
const job = makePacedJob({ min: "15m", max: "4h" });
job.state.pacedNextRunAtMs = ENDED_AT + 30 * 60_000;
applyJobResult(makeState(), job, {
status: "ok",
startedAt: STARTED_AT,
endedAt: ENDED_AT,
});
expect(job.state.nextRunAtMs).toBe(STARTED_AT + 60 * 60_000);
expect(job.state.pacedNextRunAtMs).toBeUndefined();
});
it("discards proposals on error so normal backoff wins", () => {
const job = makePacedJob({ min: "1h", max: "2h" }, 10_000);
job.state.pacedNextRunAtMs = ENDED_AT + 90 * 60_000;
applyJobResult(makeState(), job, {
status: "error",
error: "temporary failure",
startedAt: STARTED_AT,
endedAt: ENDED_AT,
nextCheck: { delayMs: 90 * 60_000 },
});
expect(job.state.nextRunAtMs).toBe(ENDED_AT + 30_000);
expect(job.state.pacedNextRunAtMs).toBeUndefined();
});
it("preserves a paced cron-expression override during future-slot repair", () => {
const state = makeState();
const job = makeCronJob({
pacing: { min: "15m", max: "4h" },
schedule: { kind: "cron", expr: "* * * * *", tz: "UTC" },
state: { nextRunAtMs: STARTED_AT },
});
state.store = { version: 1, jobs: [job] };
applyJobResult(state, job, {
status: "ok",
startedAt: STARTED_AT,
endedAt: ENDED_AT,
nextCheck: { delayMs: 30 * 60_000 },
});
recomputeNextRunsForMaintenance(state, { nowMs: ENDED_AT + 1_000 });
expect(job.state.nextRunAtMs).toBe(ENDED_AT + 30 * 60_000);
expect(job.state.pacedNextRunAtMs).toBe(ENDED_AT + 30 * 60_000);
});
it("repairs an unmarked future slot even when it falls within pacing bounds", () => {
const state = makeState();
const job = makeCronJob({
pacing: { min: "15m", max: "4h" },
schedule: { kind: "cron", expr: "* * * * *", tz: "UTC" },
state: { nextRunAtMs: STARTED_AT },
});
state.store = { version: 1, jobs: [job] };
applyJobResult(state, job, {
status: "ok",
startedAt: STARTED_AT,
endedAt: ENDED_AT,
});
job.state.nextRunAtMs = ENDED_AT + 30 * 60_000 + 1_234;
recomputeNextRunsForMaintenance(state, { nowMs: ENDED_AT + 1_000 });
expect(job.state.nextRunAtMs).toBe(ENDED_AT + 60_000);
});
it("repairs a future slot whose persisted pacing marker does not match", () => {
const state = makeState();
const job = makeCronJob({
pacing: { min: "15m", max: "4h" },
schedule: { kind: "cron", expr: "* * * * *", tz: "UTC" },
state: {
nextRunAtMs: ENDED_AT + 30 * 60_000 + 1_234,
pacedNextRunAtMs: ENDED_AT + 45 * 60_000,
},
});
state.store = { version: 1, jobs: [job] };
recomputeNextRunsForMaintenance(state, { nowMs: ENDED_AT + 1_000 });
expect(job.state.nextRunAtMs).toBe(ENDED_AT + 60_000);
expect(job.state.pacedNextRunAtMs).toBeUndefined();
});
it("clears a paced marker when maintenance normalizes the schedule", () => {
const state = makeState();
const pacedNextRunAtMs = ENDED_AT + 30 * 60_000;
const job = makeCronJob({
createdAtMs: STARTED_AT,
updatedAtMs: STARTED_AT,
pacing: { min: "15m" },
schedule: { kind: "every", everyMs: 60 * 60_000 },
state: { nextRunAtMs: pacedNextRunAtMs, pacedNextRunAtMs },
});
state.store = { version: 1, jobs: [job] };
recomputeNextRunsForMaintenance(state, { nowMs: ENDED_AT + 1_000 });
expect(job.schedule).toEqual({ kind: "every", everyMs: 60 * 60_000, anchorMs: STARTED_AT });
expect(job.state.pacedNextRunAtMs).toBeUndefined();
});
});
src/cron/service/timer.ts
+25
View File
@@ -28,6 +28,7 @@ import {
type CronActiveJobMarker,
} from "../active-jobs.js";
import { resolveCronDeliveryPlan, resolveFailureDestination } from "../delivery-plan.js";
import { resolvePacedNextRunAtMs } from "../pacing.js";
import { resolveCronExecutionRetryHint } from "../retry-hint.js";
import {
createCronRunDiagnosticsFromError,
@@ -43,6 +44,7 @@ import type {
CronDeliveryTrace,
CronFailureNotificationDelivery,
CronJob,
CronNextCheckProposal,
CronRunOutcome,
CronRunStatus,
CronRunTelemetry,
@@ -144,6 +146,7 @@ type TimedCronRunOutcome = CronRunOutcome &
startedAt: number;
endedAt: number;
triggerEval?: CronTriggerEvalOutcome;
nextCheck?: CronNextCheckProposal;
};
type CronJobRunResult = CronRunOutcome &
@@ -152,6 +155,7 @@ type CronJobRunResult = CronRunOutcome &
delivered?: boolean;
startedAt: number;
endedAt: number;
nextCheck?: CronNextCheckProposal;
};
export type CronTriggerEvalOutcome = {
@@ -762,6 +766,7 @@ export function applyJobResult(
};
job.state.queuedAtMs = undefined;
job.state.runningAtMs = undefined;
job.state.pacedNextRunAtMs = undefined;
job.state.lastRunAtMs = result.startedAt;
job.state.lastRunStatus = result.status;
job.state.lastStatus = result.status;
@@ -1019,6 +1024,21 @@ export function applyJobResult(
},
"cron: applying error backoff",
);
} else if (
isJobEnabled(job) &&
result.status === "ok" &&
job.pacing !== undefined &&
result.nextCheck !== undefined
) {
// Pacing bounds are the explicit per-job cadence contract. Do not apply
// normal schedule floors here; that would change the promised clamp.
const nextRunAtMs = resolvePacedNextRunAtMs({
nowMs: result.endedAt,
delayMs: result.nextCheck.delayMs,
pacing: job.pacing,
});
job.state.nextRunAtMs = nextRunAtMs;
job.state.pacedNextRunAtMs = nextRunAtMs;
} else if (isJobEnabled(job)) {
let naturalNext: number | undefined;
try {
@@ -1191,6 +1211,7 @@ function applyOutcomeToStoredJob(
triggerEval: result.triggerEval,
});
job.state.startupCatchupAtMs = undefined;
job.state.pacedNextRunAtMs = undefined;
return undefined;
}
@@ -2284,6 +2305,7 @@ async function runStartupCatchupCandidate(
diagnostics: result.diagnostics,
delivered: result.delivered,
deliveryError: result.deliveryError,
nextCheck: result.nextCheck,
sessionId: result.sessionId,
sessionKey: result.sessionKey,
model: result.model,
@@ -2434,6 +2456,7 @@ async function executeJobCore(
deliveryAttempted?: boolean;
deliveryError?: string;
delivery?: CronDeliveryTrace;
nextCheck?: CronNextCheckProposal;
triggerEval?: CronTriggerEvalOutcome;
}
> {
@@ -2693,6 +2716,7 @@ async function executeDetachedCronJob(
deliveryAttempted?: boolean;
deliveryError?: string;
delivery?: CronDeliveryTrace;
nextCheck?: CronNextCheckProposal;
}
> {
if (job.payload.kind === "command") {
@@ -2782,6 +2806,7 @@ async function executeDetachedCronJob(
// successful run so the service can persist it as `lastDeliveryError` and
// emit it on the finished event for CLI/UI/API run logs (#95419).
deliveryError: res.deliveryError,
nextCheck: res.nextCheck,
summary: res.summary,
delivered: res.delivered,
deliveryAttempted: res.deliveryAttempted,
src/cron/store.test.ts
+3
View File
@@ -504,6 +504,7 @@ describe("cron store", () => {
job.state = {
nextRunAtMs: job.createdAtMs,
startupCatchupAtMs: job.createdAtMs,
pacedNextRunAtMs: job.createdAtMs,
queuedAtMs: job.createdAtMs + 1,
};
@@ -516,10 +517,12 @@ describe("cron store", () => {
expect(JSON.parse(queuedRow.state_json)).toMatchObject({
queuedAtMs: job.createdAtMs + 1,
startupCatchupAtMs: job.createdAtMs,
pacedNextRunAtMs: job.createdAtMs,
});
expect((await loadCronStore(store.storePath)).jobs[0]?.state).toMatchObject({
queuedAtMs: job.createdAtMs + 1,
startupCatchupAtMs: job.createdAtMs,
pacedNextRunAtMs: job.createdAtMs,
});
job.state.queuedAtMs = undefined;
src/cron/store/row-codec.schedule.test.ts
+8
View File
@@ -11,6 +11,14 @@ function roundTrip(schedule: CronSchedule): CronSchedule | null {
}
describe("schedule column codec round-trip", () => {
it("round-trips pacing through the additive job_json envelope", () => {
const job = projectCronJobThroughStorageCodec(
makeCronJob({ pacing: { min: "15m", max: "4h" } }),
);
expect(job.pacing).toEqual({ min: "15m", max: "4h" });
});
it("round-trips an on-exit schedule with command + cwd", () => {
expect(roundTrip({ kind: "on-exit", command: "make build", cwd: "/repo" })).toEqual({
kind: "on-exit",
src/cron/store/row-codec.ts
+14 -1
View File
@@ -6,7 +6,7 @@ import { normalizeCronJobIdentityFields } from "../normalize-job-identity.js";
import { normalizeCronJobInput } from "../normalize.js";
import { getInvalidPersistedCronJobReason } from "../persisted-shape.js";
import { tryCronScheduleIdentity } from "../schedule-identity.js";
import type { CronJob, CronJobState, CronSchedule, CronStoreFile } from "../types.js";
import type { CronJob, CronJobState, CronPacing, CronSchedule, CronStoreFile } from "../types.js";
import { bindDeliveryColumns, deliveryFromRow } from "./delivery-codec.js";
import { bindFailureAlertColumns, failureAlertFromRow } from "./failure-alert-codec.js";
import { bindPayloadColumns, payloadFromRow } from "./payload-codec.js";
@@ -239,12 +239,24 @@ function scheduleFromRow(row: CronJobRow): CronSchedule | null {
return null;
}
function pacingFromRow(row: CronJobRow): CronPacing | undefined {
const pacing = parseJsonObject<Record<string, unknown>>(row.job_json, {}).pacing;
if (!isRecord(pacing) || Array.isArray(pacing)) {
return undefined;
}
return {
...(typeof pacing.min === "string" ? { min: pacing.min } : {}),
...(typeof pacing.max === "string" ? { max: pacing.max } : {}),
};
}
function rowToCronJob(row: CronJobRow): CronJob | null {
const schedule = scheduleFromRow(row);
const payload = payloadFromRow(row);
const delivery = deliveryFromRow(row);
const failureAlert = failureAlertFromRow(row);
const trigger = triggerFromRow(row);
const pacing = pacingFromRow(row);
if (!schedule || !payload) {
return null;
}
@@ -273,6 +285,7 @@ function rowToCronJob(row: CronJobRow): CronJob | null {
...(row.agent_id ? { agentId: row.agent_id } : {}),
...(row.session_key ? { sessionKey: row.session_key } : {}),
schedule,
...(pacing !== undefined ? { pacing } : {}),
sessionTarget: row.session_target as CronJob["sessionTarget"],
wakeMode: row.wake_mode as CronJob["wakeMode"],
...(trigger ? { trigger } : {}),
src/cron/types-shared.ts
+7
View File
@@ -1,3 +1,9 @@
/** Optional dynamic-cadence bounds for one cron job. */
export type CronPacing = {
min?: string;
max?: string;
};
/** Shared persisted cron job envelope used by runtime and external config shapes. */
export type CronJobBase<TSchedule, TSessionTarget, TWakeMode, TPayload, TDelivery, TFailureAlert> =
{
@@ -11,6 +17,7 @@ export type CronJobBase<TSchedule, TSessionTarget, TWakeMode, TPayload, TDeliver
createdAtMs: number;
updatedAtMs: number;
schedule: TSchedule;
pacing?: CronPacing;
sessionTarget: TSessionTarget;
wakeMode: TWakeMode;
payload: TPayload;
src/cron/types.ts
+12 -1
View File
@@ -3,7 +3,9 @@ import type { FailoverReason } from "../agents/embedded-agent-helpers/types.js";
import type { EmbeddedAgentExecutionPhase } from "../agents/embedded-agent-runner/execution-phase.js";
import type { ChannelId } from "../channels/plugins/types.public.js";
import type { HookExternalContentSource } from "../security/external-content.js";
import type { CronJobBase } from "./types-shared.js";
import type { CronJobBase, CronPacing } from "./types-shared.js";
export type { CronPacing } from "./types-shared.js";
/** Supported schedule forms persisted in cron job specs. */
export type CronSchedule =
@@ -197,6 +199,11 @@ export type CronRunOutcome = {
diagnostics?: CronRunDiagnostics;
};
/** One run's requested delay before the same paced job runs again. */
export type CronNextCheckProposal = {
delayMs: number;
};
/** Embedded-agent execution phase names surfaced to cron watchdog progress. */
export type CronAgentExecutionPhase = EmbeddedAgentExecutionPhase;
@@ -315,6 +322,8 @@ export type CronJobState = {
nextRunAtMs?: number;
/** Exact startup catch-up slot protected from future-slot repair across restarts. */
startupCatchupAtMs?: number;
/** Exact paced completion slot protected from future-slot repair until consumed. */
pacedNextRunAtMs?: number;
/** Durable pre-admission reservation. Cleared on restart without recording a run. */
queuedAtMs?: number;
runningAtMs?: number;
@@ -431,9 +440,11 @@ export type CronJobPatch = Partial<
| "declarationKey"
| "displayName"
| "owner"
| "pacing"
>
> & {
displayName?: string | null;
pacing?: CronPacing | null;
trigger?: CronTrigger | null;
payload?: CronPayloadPatch;
delivery?: CronDeliveryPatch;
src/infra/agent-events.ts
+36
View File
@@ -142,6 +142,10 @@ type AgentRunContext = {
/** Whether control UI clients should receive chat/agent updates for this run. */
isControlUiVisible?: boolean;
projectSessionActive?: boolean;
/** Cron job allowed to record a one-shot dynamic-cadence proposal on this run. */
cronJobId?: string;
cronPacingEnabled?: boolean;
cronNextCheckMs?: number;
/** Timestamp when this context was first registered (for TTL-based cleanup). */
registeredAt?: number;
/** Timestamp of last activity (updated on every emitAgentEvent). */
@@ -294,6 +298,15 @@ export function registerAgentRunContext(runId: string, context: AgentRunContext,
if (context.projectSessionActive !== undefined) {
existing.projectSessionActive = context.projectSessionActive;
}
if (context.cronJobId !== undefined) {
if (existing.cronJobId !== context.cronJobId) {
delete existing.cronNextCheckMs;
}
existing.cronJobId = context.cronJobId;
}
if (context.cronPacingEnabled !== undefined) {
existing.cronPacingEnabled = context.cronPacingEnabled;
}
if (context.isHeartbeat !== undefined && existing.isHeartbeat !== context.isHeartbeat) {
existing.isHeartbeat = context.isHeartbeat;
}
@@ -401,6 +414,29 @@ export function getAgentRunContext(runId: string) {
return getAgentEventState().runContextById.get(runId);
}
/** Records the latest next-check proposal on the matching paced cron run. */
export function recordCronNextCheckProposal(runId: string, jobId: string, delayMs: number): void {
const context = getAgentEventState().runContextById.get(runId);
if (!context || context.cronJobId !== jobId) {
throw new Error("cron next_check is only available to the currently running job");
}
if (context.cronPacingEnabled !== true) {
throw new Error("cron next_check requires pacing on the current job");
}
context.cronNextCheckMs = delayMs;
}
/** Consumes one successful cron run's proposal so it cannot affect a later run. */
export function consumeCronNextCheckProposal(runId: string, jobId: string): number | undefined {
const context = getAgentEventState().runContextById.get(runId);
if (!context || context.cronJobId !== jobId) {
return undefined;
}
const delayMs = context.cronNextCheckMs;
delete context.cronNextCheckMs;
return delayMs;
}
export function getAgentRunContextOwnerStatus(
runId: string,
claimId: string,
test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.discord-group.json
Vendored
+54 -2
View File
@@ -427,12 +427,23 @@
},
{
"deferLoading": true,
"description": "Gateway schedules/wakes: reminders, later checks/follow-ups, recurring work. Never exec sleep/process-poll as timer. Main job => heartbeat system event; isolated => background task in `openclaw tasks`.\n\nACTIONS:\n- status scheduler; list compact summaries (includeDisabled, session agentId auto-filter; get for full); get jobId\n- add job; update jobId+patch; remove jobId\n- run jobId (due only; runMode=\"force\" now); runs jobId history\n- wake text (+ optional mode). Default caller lane; top-level sessionKey/agentId selects another caller-owned lane.\n\nADD JOB:\n{ \"name\":\"...\", \"schedule\":{...}, \"trigger\":{ \"script\":\"...\", \"once\":false }, \"payload\":{...}, \"delivery\":{...}, \"sessionTarget\":\"main|isolated|current|session:<id>\", \"enabled\":true }\nRequired: schedule,payload. enabled default true. trigger only every/cron.\n\nTARGET/PAYLOAD:\n- main => systemEvent {kind:\"systemEvent\",text:\"...\"}; systemEvent defaults main.\n- isolated/current/session:<id> => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session:<id> is persistent. Prefer isolated unless user explicitly wants current binding.\n\nSCHEDULE:\n- at: {kind:\"at\",at:\"ISO-8601\"}; timezone-less = UTC.\n- every: {kind:\"every\",everyMs:<ms>,anchorMs?}.\n- cron: {kind:\"cron\",expr:\"...\",tz?:\"IANA\"}. Expr is requested local wall time; never pre-convert to UTC. Missing tz = Gateway host local, not UTC. Shanghai 18:00: {kind:\"cron\",expr:\"0 18 * * *\",tz:\"Asia/Shanghai\"}.\n\nTRIGGER SCRIPT:\n- Requires cron.triggers.enabled; if off, explain and never model-poll fallback.\n- Headless owner allowlist; quiet check has no model. Prior trigger.state is frozen JSON. Return/json({fire:boolean,message?:string,state?:JSONValue}); create new state, never mutate prior.\n- fire:false saves state only; no payload/history. fire:true runs payload and appends message; fired state saves only after payload success.\n- Fire on every actionable state, including failures/timeouts; success-only watchers go silent when broken, which looks healthy. Dedupe by comparing trigger.state and returning new state, never memory.\n- Keep scripts read-only; actions belong in payload. message must be self-contained: it is the fired run's entire context.\n- Silent watcher: top-level delivery.mode=\"none\". Omitted delivery on isolated agentTurn announces and missing route may fail.\n- once:true disables after first successful fire. Per check: 30s, 5 tool calls, 16KB state.\n- Hidden Code Mode tools: await tools.call(\"exec\", {command:\"...\"}); unknown id => search/describe.\n\nDELIVERY top-level: {mode:\"none|announce|webhook\",channel?,to?,threadId?,bestEffort?}\n- Isolated agentTurn omitted delivery => announce. announce only isolated/current/session; channel/to optional; threadId chat topic. Specific chat: set channel/to; no messaging tool inside run.\n- webhook posts finished-run event to URL in to.\n\nRestricted isolated runs may only self status/list, current get/runs, and remove current job. wake mode: next-heartbeat default | now. jobId canonical; id compat. contextMessages 0-10 adds prior messages.",
"description": "Gateway schedules/wakes: reminders, later checks/follow-ups, recurring work. Never exec sleep/process-poll as timer. Main job => heartbeat system event; isolated => background task in `openclaw tasks`.\n\nACTIONS:\n- status scheduler; list compact summaries (includeDisabled, session agentId auto-filter; get for full); get jobId\n- add job; update jobId+patch; remove jobId\n- run jobId (due only; runMode=\"force\" now); runs jobId history; next_check in (current paced job only)\n- wake text (+ optional mode). Default caller lane; top-level sessionKey/agentId selects another caller-owned lane.\n\nADD JOB:\n{ \"name\":\"...\", \"schedule\":{...}, \"pacing\":{ \"min\":\"15m\", \"max\":\"4h\" }, \"trigger\":{ \"script\":\"...\", \"once\":false }, \"payload\":{...}, \"delivery\":{...}, \"sessionTarget\":\"main|isolated|current|session:<id>\", \"enabled\":true }\nRequired: schedule,payload. enabled default true. trigger only every/cron.\n\nTARGET/PAYLOAD:\n- main => systemEvent {kind:\"systemEvent\",text:\"...\"}; systemEvent defaults main.\n- isolated/current/session:<id> => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session:<id> is persistent. Prefer isolated unless user explicitly wants current binding.\n\nSCHEDULE:\n- at: {kind:\"at\",at:\"ISO-8601\"}; timezone-less = UTC.\n- every: {kind:\"every\",everyMs:<ms>,anchorMs?}.\n- cron: {kind:\"cron\",expr:\"...\",tz?:\"IANA\"}. Expr is requested local wall time; never pre-convert to UTC. Missing tz = Gateway host local, not UTC. Shanghai 18:00: {kind:\"cron\",expr:\"0 18 * * *\",tz:\"Asia/Shanghai\"}.\n\nTRIGGER SCRIPT:\n- Requires cron.triggers.enabled; if off, explain and never model-poll fallback.\n- Headless owner allowlist; quiet check has no model. Prior trigger.state is frozen JSON. Return/json({fire:boolean,message?:string,state?:JSONValue}); create new state, never mutate prior.\n- fire:false saves state only; no payload/history. fire:true runs payload and appends message; fired state saves only after payload success.\n- Fire on every actionable state, including failures/timeouts; success-only watchers go silent when broken, which looks healthy. Dedupe by comparing trigger.state and returning new state, never memory.\n- Keep scripts read-only; actions belong in payload. message must be self-contained: it is the fired run's entire context.\n- Silent watcher: top-level delivery.mode=\"none\". Omitted delivery on isolated agentTurn announces and missing route may fail.\n- once:true disables after first successful fire. Per check: 30s, 5 tool calls, 16KB state.\n- Hidden Code Mode tools: await tools.call(\"exec\", {command:\"...\"}); unknown id => search/describe.\n\nDELIVERY top-level: {mode:\"none|announce|webhook\",channel?,to?,threadId?,bestEffort?}\n- Isolated agentTurn omitted delivery => announce. announce only isolated/current/session; channel/to optional; threadId chat topic. Specific chat: set channel/to; no messaging tool inside run.\n- webhook posts finished-run event to URL in to.\n\nRestricted isolated runs may only self status/list, current get/runs/remove, and next_check for their own paced job. wake mode: next-heartbeat default | now. jobId canonical; id compat. contextMessages 0-10 adds prior messages.",
"inputSchema": {
"additionalProperties": true,
"properties": {
"action": {
"enum": ["status", "list", "get", "add", "update", "remove", "run", "runs", "wake"],
"enum": [
"status",
"list",
"get",
"add",
"update",
"remove",
"run",
"runs",
"next_check",
"wake"
],
"type": "string"
},
"agentId": {
@@ -453,6 +464,10 @@
"id": {
"type": "string"
},
"in": {
"description": "Relative duration for action=\"next_check\" (for example, \"15m\")",
"type": "string"
},
"includeDisabled": {
"type": "boolean"
},
@@ -611,6 +626,21 @@
},
"type": "object"
},
"pacing": {
"additionalProperties": false,
"description": "Dynamic-cadence bounds; at least one of min or max is required",
"properties": {
"max": {
"description": "Maximum dynamic delay",
"type": "string"
},
"min": {
"description": "Minimum dynamic delay",
"type": "string"
}
},
"type": "object"
},
"payload": {
"additionalProperties": true,
"properties": {
@@ -943,6 +973,28 @@
"description": "Job name",
"type": "string"
},
"pacing": {
"anyOf": [
{
"additionalProperties": false,
"description": "Dynamic-cadence bounds; at least one of min or max is required",
"properties": {
"max": {
"description": "Maximum dynamic delay",
"type": "string"
},
"min": {
"description": "Minimum dynamic delay",
"type": "string"
}
},
"type": "object"
},
{
"type": "null"
}
]
},
"payload": {
"additionalProperties": true,
"properties": {
test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.heartbeat-turn.json
Vendored
+54 -2
View File
@@ -423,12 +423,23 @@
},
{
"deferLoading": true,
"description": "Gateway schedules/wakes: reminders, later checks/follow-ups, recurring work. Never exec sleep/process-poll as timer. Main job => heartbeat system event; isolated => background task in `openclaw tasks`.\n\nACTIONS:\n- status scheduler; list compact summaries (includeDisabled, session agentId auto-filter; get for full); get jobId\n- add job; update jobId+patch; remove jobId\n- run jobId (due only; runMode=\"force\" now); runs jobId history\n- wake text (+ optional mode). Default caller lane; top-level sessionKey/agentId selects another caller-owned lane.\n\nADD JOB:\n{ \"name\":\"...\", \"schedule\":{...}, \"trigger\":{ \"script\":\"...\", \"once\":false }, \"payload\":{...}, \"delivery\":{...}, \"sessionTarget\":\"main|isolated|current|session:<id>\", \"enabled\":true }\nRequired: schedule,payload. enabled default true. trigger only every/cron.\n\nTARGET/PAYLOAD:\n- main => systemEvent {kind:\"systemEvent\",text:\"...\"}; systemEvent defaults main.\n- isolated/current/session:<id> => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session:<id> is persistent. Prefer isolated unless user explicitly wants current binding.\n\nSCHEDULE:\n- at: {kind:\"at\",at:\"ISO-8601\"}; timezone-less = UTC.\n- every: {kind:\"every\",everyMs:<ms>,anchorMs?}.\n- cron: {kind:\"cron\",expr:\"...\",tz?:\"IANA\"}. Expr is requested local wall time; never pre-convert to UTC. Missing tz = Gateway host local, not UTC. Shanghai 18:00: {kind:\"cron\",expr:\"0 18 * * *\",tz:\"Asia/Shanghai\"}.\n\nTRIGGER SCRIPT:\n- Requires cron.triggers.enabled; if off, explain and never model-poll fallback.\n- Headless owner allowlist; quiet check has no model. Prior trigger.state is frozen JSON. Return/json({fire:boolean,message?:string,state?:JSONValue}); create new state, never mutate prior.\n- fire:false saves state only; no payload/history. fire:true runs payload and appends message; fired state saves only after payload success.\n- Fire on every actionable state, including failures/timeouts; success-only watchers go silent when broken, which looks healthy. Dedupe by comparing trigger.state and returning new state, never memory.\n- Keep scripts read-only; actions belong in payload. message must be self-contained: it is the fired run's entire context.\n- Silent watcher: top-level delivery.mode=\"none\". Omitted delivery on isolated agentTurn announces and missing route may fail.\n- once:true disables after first successful fire. Per check: 30s, 5 tool calls, 16KB state.\n- Hidden Code Mode tools: await tools.call(\"exec\", {command:\"...\"}); unknown id => search/describe.\n\nDELIVERY top-level: {mode:\"none|announce|webhook\",channel?,to?,threadId?,bestEffort?}\n- Isolated agentTurn omitted delivery => announce. announce only isolated/current/session; channel/to optional; threadId chat topic. Specific chat: set channel/to; no messaging tool inside run.\n- webhook posts finished-run event to URL in to.\n\nRestricted isolated runs may only self status/list, current get/runs, and remove current job. wake mode: next-heartbeat default | now. jobId canonical; id compat. contextMessages 0-10 adds prior messages.",
"description": "Gateway schedules/wakes: reminders, later checks/follow-ups, recurring work. Never exec sleep/process-poll as timer. Main job => heartbeat system event; isolated => background task in `openclaw tasks`.\n\nACTIONS:\n- status scheduler; list compact summaries (includeDisabled, session agentId auto-filter; get for full); get jobId\n- add job; update jobId+patch; remove jobId\n- run jobId (due only; runMode=\"force\" now); runs jobId history; next_check in (current paced job only)\n- wake text (+ optional mode). Default caller lane; top-level sessionKey/agentId selects another caller-owned lane.\n\nADD JOB:\n{ \"name\":\"...\", \"schedule\":{...}, \"pacing\":{ \"min\":\"15m\", \"max\":\"4h\" }, \"trigger\":{ \"script\":\"...\", \"once\":false }, \"payload\":{...}, \"delivery\":{...}, \"sessionTarget\":\"main|isolated|current|session:<id>\", \"enabled\":true }\nRequired: schedule,payload. enabled default true. trigger only every/cron.\n\nTARGET/PAYLOAD:\n- main => systemEvent {kind:\"systemEvent\",text:\"...\"}; systemEvent defaults main.\n- isolated/current/session:<id> => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session:<id> is persistent. Prefer isolated unless user explicitly wants current binding.\n\nSCHEDULE:\n- at: {kind:\"at\",at:\"ISO-8601\"}; timezone-less = UTC.\n- every: {kind:\"every\",everyMs:<ms>,anchorMs?}.\n- cron: {kind:\"cron\",expr:\"...\",tz?:\"IANA\"}. Expr is requested local wall time; never pre-convert to UTC. Missing tz = Gateway host local, not UTC. Shanghai 18:00: {kind:\"cron\",expr:\"0 18 * * *\",tz:\"Asia/Shanghai\"}.\n\nTRIGGER SCRIPT:\n- Requires cron.triggers.enabled; if off, explain and never model-poll fallback.\n- Headless owner allowlist; quiet check has no model. Prior trigger.state is frozen JSON. Return/json({fire:boolean,message?:string,state?:JSONValue}); create new state, never mutate prior.\n- fire:false saves state only; no payload/history. fire:true runs payload and appends message; fired state saves only after payload success.\n- Fire on every actionable state, including failures/timeouts; success-only watchers go silent when broken, which looks healthy. Dedupe by comparing trigger.state and returning new state, never memory.\n- Keep scripts read-only; actions belong in payload. message must be self-contained: it is the fired run's entire context.\n- Silent watcher: top-level delivery.mode=\"none\". Omitted delivery on isolated agentTurn announces and missing route may fail.\n- once:true disables after first successful fire. Per check: 30s, 5 tool calls, 16KB state.\n- Hidden Code Mode tools: await tools.call(\"exec\", {command:\"...\"}); unknown id => search/describe.\n\nDELIVERY top-level: {mode:\"none|announce|webhook\",channel?,to?,threadId?,bestEffort?}\n- Isolated agentTurn omitted delivery => announce. announce only isolated/current/session; channel/to optional; threadId chat topic. Specific chat: set channel/to; no messaging tool inside run.\n- webhook posts finished-run event to URL in to.\n\nRestricted isolated runs may only self status/list, current get/runs/remove, and next_check for their own paced job. wake mode: next-heartbeat default | now. jobId canonical; id compat. contextMessages 0-10 adds prior messages.",
"inputSchema": {
"additionalProperties": true,
"properties": {
"action": {
"enum": ["status", "list", "get", "add", "update", "remove", "run", "runs", "wake"],
"enum": [
"status",
"list",
"get",
"add",
"update",
"remove",
"run",
"runs",
"next_check",
"wake"
],
"type": "string"
},
"agentId": {
@@ -449,6 +460,10 @@
"id": {
"type": "string"
},
"in": {
"description": "Relative duration for action=\"next_check\" (for example, \"15m\")",
"type": "string"
},
"includeDisabled": {
"type": "boolean"
},
@@ -607,6 +622,21 @@
},
"type": "object"
},
"pacing": {
"additionalProperties": false,
"description": "Dynamic-cadence bounds; at least one of min or max is required",
"properties": {
"max": {
"description": "Maximum dynamic delay",
"type": "string"
},
"min": {
"description": "Minimum dynamic delay",
"type": "string"
}
},
"type": "object"
},
"payload": {
"additionalProperties": true,
"properties": {
@@ -939,6 +969,28 @@
"description": "Job name",
"type": "string"
},
"pacing": {
"anyOf": [
{
"additionalProperties": false,
"description": "Dynamic-cadence bounds; at least one of min or max is required",
"properties": {
"max": {
"description": "Maximum dynamic delay",
"type": "string"
},
"min": {
"description": "Minimum dynamic delay",
"type": "string"
}
},
"type": "object"
},
{
"type": "null"
}
]
},
"payload": {
"additionalProperties": true,
"properties": {
test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json
Vendored
+54 -2
View File
@@ -423,12 +423,23 @@
},
{
"deferLoading": true,
"description": "Gateway schedules/wakes: reminders, later checks/follow-ups, recurring work. Never exec sleep/process-poll as timer. Main job => heartbeat system event; isolated => background task in `openclaw tasks`.\n\nACTIONS:\n- status scheduler; list compact summaries (includeDisabled, session agentId auto-filter; get for full); get jobId\n- add job; update jobId+patch; remove jobId\n- run jobId (due only; runMode=\"force\" now); runs jobId history\n- wake text (+ optional mode). Default caller lane; top-level sessionKey/agentId selects another caller-owned lane.\n\nADD JOB:\n{ \"name\":\"...\", \"schedule\":{...}, \"trigger\":{ \"script\":\"...\", \"once\":false }, \"payload\":{...}, \"delivery\":{...}, \"sessionTarget\":\"main|isolated|current|session:<id>\", \"enabled\":true }\nRequired: schedule,payload. enabled default true. trigger only every/cron.\n\nTARGET/PAYLOAD:\n- main => systemEvent {kind:\"systemEvent\",text:\"...\"}; systemEvent defaults main.\n- isolated/current/session:<id> => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session:<id> is persistent. Prefer isolated unless user explicitly wants current binding.\n\nSCHEDULE:\n- at: {kind:\"at\",at:\"ISO-8601\"}; timezone-less = UTC.\n- every: {kind:\"every\",everyMs:<ms>,anchorMs?}.\n- cron: {kind:\"cron\",expr:\"...\",tz?:\"IANA\"}. Expr is requested local wall time; never pre-convert to UTC. Missing tz = Gateway host local, not UTC. Shanghai 18:00: {kind:\"cron\",expr:\"0 18 * * *\",tz:\"Asia/Shanghai\"}.\n\nTRIGGER SCRIPT:\n- Requires cron.triggers.enabled; if off, explain and never model-poll fallback.\n- Headless owner allowlist; quiet check has no model. Prior trigger.state is frozen JSON. Return/json({fire:boolean,message?:string,state?:JSONValue}); create new state, never mutate prior.\n- fire:false saves state only; no payload/history. fire:true runs payload and appends message; fired state saves only after payload success.\n- Fire on every actionable state, including failures/timeouts; success-only watchers go silent when broken, which looks healthy. Dedupe by comparing trigger.state and returning new state, never memory.\n- Keep scripts read-only; actions belong in payload. message must be self-contained: it is the fired run's entire context.\n- Silent watcher: top-level delivery.mode=\"none\". Omitted delivery on isolated agentTurn announces and missing route may fail.\n- once:true disables after first successful fire. Per check: 30s, 5 tool calls, 16KB state.\n- Hidden Code Mode tools: await tools.call(\"exec\", {command:\"...\"}); unknown id => search/describe.\n\nDELIVERY top-level: {mode:\"none|announce|webhook\",channel?,to?,threadId?,bestEffort?}\n- Isolated agentTurn omitted delivery => announce. announce only isolated/current/session; channel/to optional; threadId chat topic. Specific chat: set channel/to; no messaging tool inside run.\n- webhook posts finished-run event to URL in to.\n\nRestricted isolated runs may only self status/list, current get/runs, and remove current job. wake mode: next-heartbeat default | now. jobId canonical; id compat. contextMessages 0-10 adds prior messages.",
"description": "Gateway schedules/wakes: reminders, later checks/follow-ups, recurring work. Never exec sleep/process-poll as timer. Main job => heartbeat system event; isolated => background task in `openclaw tasks`.\n\nACTIONS:\n- status scheduler; list compact summaries (includeDisabled, session agentId auto-filter; get for full); get jobId\n- add job; update jobId+patch; remove jobId\n- run jobId (due only; runMode=\"force\" now); runs jobId history; next_check in (current paced job only)\n- wake text (+ optional mode). Default caller lane; top-level sessionKey/agentId selects another caller-owned lane.\n\nADD JOB:\n{ \"name\":\"...\", \"schedule\":{...}, \"pacing\":{ \"min\":\"15m\", \"max\":\"4h\" }, \"trigger\":{ \"script\":\"...\", \"once\":false }, \"payload\":{...}, \"delivery\":{...}, \"sessionTarget\":\"main|isolated|current|session:<id>\", \"enabled\":true }\nRequired: schedule,payload. enabled default true. trigger only every/cron.\n\nTARGET/PAYLOAD:\n- main => systemEvent {kind:\"systemEvent\",text:\"...\"}; systemEvent defaults main.\n- isolated/current/session:<id> => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session:<id> is persistent. Prefer isolated unless user explicitly wants current binding.\n\nSCHEDULE:\n- at: {kind:\"at\",at:\"ISO-8601\"}; timezone-less = UTC.\n- every: {kind:\"every\",everyMs:<ms>,anchorMs?}.\n- cron: {kind:\"cron\",expr:\"...\",tz?:\"IANA\"}. Expr is requested local wall time; never pre-convert to UTC. Missing tz = Gateway host local, not UTC. Shanghai 18:00: {kind:\"cron\",expr:\"0 18 * * *\",tz:\"Asia/Shanghai\"}.\n\nTRIGGER SCRIPT:\n- Requires cron.triggers.enabled; if off, explain and never model-poll fallback.\n- Headless owner allowlist; quiet check has no model. Prior trigger.state is frozen JSON. Return/json({fire:boolean,message?:string,state?:JSONValue}); create new state, never mutate prior.\n- fire:false saves state only; no payload/history. fire:true runs payload and appends message; fired state saves only after payload success.\n- Fire on every actionable state, including failures/timeouts; success-only watchers go silent when broken, which looks healthy. Dedupe by comparing trigger.state and returning new state, never memory.\n- Keep scripts read-only; actions belong in payload. message must be self-contained: it is the fired run's entire context.\n- Silent watcher: top-level delivery.mode=\"none\". Omitted delivery on isolated agentTurn announces and missing route may fail.\n- once:true disables after first successful fire. Per check: 30s, 5 tool calls, 16KB state.\n- Hidden Code Mode tools: await tools.call(\"exec\", {command:\"...\"}); unknown id => search/describe.\n\nDELIVERY top-level: {mode:\"none|announce|webhook\",channel?,to?,threadId?,bestEffort?}\n- Isolated agentTurn omitted delivery => announce. announce only isolated/current/session; channel/to optional; threadId chat topic. Specific chat: set channel/to; no messaging tool inside run.\n- webhook posts finished-run event to URL in to.\n\nRestricted isolated runs may only self status/list, current get/runs/remove, and next_check for their own paced job. wake mode: next-heartbeat default | now. jobId canonical; id compat. contextMessages 0-10 adds prior messages.",
"inputSchema": {
"additionalProperties": true,
"properties": {
"action": {
"enum": ["status", "list", "get", "add", "update", "remove", "run", "runs", "wake"],
"enum": [
"status",
"list",
"get",
"add",
"update",
"remove",
"run",
"runs",
"next_check",
"wake"
],
"type": "string"
},
"agentId": {
@@ -449,6 +460,10 @@
"id": {
"type": "string"
},
"in": {
"description": "Relative duration for action=\"next_check\" (for example, \"15m\")",
"type": "string"
},
"includeDisabled": {
"type": "boolean"
},
@@ -607,6 +622,21 @@
},
"type": "object"
},
"pacing": {
"additionalProperties": false,
"description": "Dynamic-cadence bounds; at least one of min or max is required",
"properties": {
"max": {
"description": "Maximum dynamic delay",
"type": "string"
},
"min": {
"description": "Minimum dynamic delay",
"type": "string"
}
},
"type": "object"
},
"payload": {
"additionalProperties": true,
"properties": {
@@ -939,6 +969,28 @@
"description": "Job name",
"type": "string"
},
"pacing": {
"anyOf": [
{
"additionalProperties": false,
"description": "Dynamic-cadence bounds; at least one of min or max is required",
"properties": {
"max": {
"description": "Maximum dynamic delay",
"type": "string"
},
"min": {
"description": "Minimum dynamic delay",
"type": "string"
}
},
"type": "object"
},
{
"type": "null"
}
]
},
"payload": {
"additionalProperties": true,
"properties": {
test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md
Vendored
+4 -4
View File
@@ -216,8 +216,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
"roughTokens": 0
},
"dynamicToolsJson": {
"chars": 53155,
"roughTokens": 13289
"chars": 54868,
"roughTokens": 13717
},
"openClawDeveloperInstructions": {
"chars": 3559,
@@ -228,8 +228,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
"roughTokens": 7021
},
"totalWithDynamicToolsJson": {
"chars": 81241,
"roughTokens": 20311
"chars": 82954,
"roughTokens": 20739
},
"userInputText": {
"chars": 1442,
test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md
Vendored
+4 -4
View File
@@ -216,8 +216,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
"roughTokens": 0
},
"dynamicToolsJson": {
"chars": 52882,
"roughTokens": 13221
"chars": 54595,
"roughTokens": 13649
},
"openClawDeveloperInstructions": {
"chars": 2450,
@@ -228,8 +228,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
"roughTokens": 6642
},
"totalWithDynamicToolsJson": {
"chars": 79450,
"roughTokens": 19863
"chars": 81163,
"roughTokens": 20291
},
"userInputText": {
"chars": 1033,
test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md
Vendored
+4 -4
View File
@@ -217,8 +217,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
"roughTokens": 0
},
"dynamicToolsJson": {
"chars": 54172,
"roughTokens": 13543
"chars": 55885,
"roughTokens": 13972
},
"openClawDeveloperInstructions": {
"chars": 2469,
@@ -229,8 +229,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
"roughTokens": 6777
},
"totalWithDynamicToolsJson": {
"chars": 81279,
"roughTokens": 20320
"chars": 82992,
"roughTokens": 20748
},
"userInputText": {
"chars": 1271,