diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift index 3dc123ce9b9c..737829cc37f4 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift @@ -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" diff --git a/docs/automation/cron-jobs.md b/docs/automation/cron-jobs.md index 0771d22be5c3..ba1a50c404a2 100644 --- a/docs/automation/cron-jobs.md +++ b/docs/automation/cron-jobs.md @@ -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. diff --git a/docs/docs_map.md b/docs/docs_map.md index 703fdf4e06de..4662b89a8d4f 100644 --- a/docs/docs_map.md +++ b/docs/docs_map.md @@ -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 diff --git a/packages/gateway-protocol/src/schema/cron.ts b/packages/gateway-protocol/src/schema/cron.ts index 0ab9c0d110f9..9316aa065d29 100644 --- a/packages/gateway-protocol/src/schema/cron.ts +++ b/packages/gateway-protocol/src/schema/cron.ts @@ -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), diff --git a/src/agents/openclaw-tools.ts b/src/agents/openclaw-tools.ts index 6a2341fc8d51..14a72c3c24f3 100644 --- a/src/agents/openclaw-tools.ts +++ b/src/agents/openclaw-tools.ts @@ -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 } : {}), diff --git a/src/agents/tools/cron-tool-canonicalize.ts b/src/agents/tools/cron-tool-canonicalize.ts index 8690ba9e253f..e5c3e751010b 100644 --- a/src/agents/tools/cron-tool-canonicalize.ts +++ b/src/agents/tools/cron-tool-canonicalize.ts @@ -42,6 +42,7 @@ const CRON_RECOVERABLE_OBJECT_KEYS: ReadonlySet = new Set([ "displayName", "owner", "schedule", + "pacing", "trigger", "sessionTarget", "wakeMode", diff --git a/src/agents/tools/cron-tool.pacing.test.ts b/src/agents/tools/cron-tool.pacing.test.ts new file mode 100644 index 000000000000..5d3824a33548 --- /dev/null +++ b/src/agents/tools/cron-tool.pacing.test.ts @@ -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(); + }); +}); diff --git a/src/agents/tools/cron-tool.schema.test.ts b/src/agents/tools/cron-tool.schema.test.ts index 1ad9dac69e10..f523fc6b156a 100644 --- a/src/agents/tools/cron-tool.schema.test.ts +++ b/src/agents/tools/cron-tool.schema.test.ts @@ -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> | undefined)?.find( + (entry) => entry.type === "object", + ); + expect( + Object.keys((pacingObject?.properties as Record) ?? {}).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(), diff --git a/src/agents/tools/cron-tool.test.ts b/src/agents/tools/cron-tool.test.ts index 3f23dcdc3705..89a7b120988c 100644 --- a/src/agents/tools/cron-tool.test.ts +++ b/src/agents/tools/cron-tool.test.ts @@ -25,7 +25,7 @@ import { createCronTool } from "./cron-tool.js"; describe("cron tool", () => { type SchemaLike = { - anyOf?: Array<{ type?: string }>; + anyOf?: Array; description?: string; properties?: Record; 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( diff --git a/src/agents/tools/cron-tool.ts b/src/agents/tools/cron-tool.ts index 7d9854882d7f..d814e240a943 100644 --- a/src/agents/tools/cron-tool.ts +++ b/src/agents/tools/cron-tool.ts @@ -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:", "enabled":true } +{ "name":"...", "schedule":{...}, "pacing":{ "min":"15m", "max":"4h" }, "trigger":{ "script":"...", "once":false }, "payload":{...}, "delivery":{...}, "sessionTarget":"main|isolated|current|session:", "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; @@ -829,6 +867,7 @@ Restricted isolated runs may only self status/list, current get/runs, and remove const canonicalJob = canonicalizeCronToolObject(params.job as Record); 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 = diff --git a/src/agents/tools/cron-tool.types.ts b/src/agents/tools/cron-tool.types.ts index b90a5d252406..dc7caf95fbc8 100644 --- a/src/agents/tools/cron-tool.types.ts +++ b/src/agents/tools/cron-tool.types.ts @@ -19,6 +19,7 @@ export type CronToolOptions = { */ creatorToolAllowlist?: CronCreatorToolAllowlistEntry[]; selfRemoveOnlyJobId?: string; + runId?: string; }; export type CronToolCallerScope = { diff --git a/src/cli/cron-cli/register.cron-add.ts b/src/cli/cron-cli/register.cron-add.ts index e11d25649291..9fd3ebd51ead 100644 --- a/src/cli/cron-cli/register.cron-add.ts +++ b/src/cli/cron-cli/register.cron-add.ts @@ -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 ", "Run every duration (e.g. 10m, 1h)") + .option("--pacing-min ", "Minimum delay accepted from a dynamic next check") + .option("--pacing-max ", "Maximum delay accepted from a dynamic next check") .option("--cron ", "Cron expression (5-field or 6-field with seconds)") .option( "--on-exit ", @@ -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, diff --git a/src/cli/cron-cli/register.cron-edit.test.ts b/src/cli/cron-cli/register.cron-edit.test.ts index 02fb689701b9..152a6e8b58d0 100644 --- a/src/cli/cron-cli/register.cron-edit.test.ts +++ b/src/cli/cron-cli/register.cron-edit.test.ts @@ -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(); diff --git a/src/cli/cron-cli/register.cron-edit.ts b/src/cli/cron-cli/register.cron-edit.ts index 9b43d9f5361f..d85acd368d2d 100644 --- a/src/cli/cron-cli/register.cron-edit.ts +++ b/src/cli/cron-cli/register.cron-edit.ts @@ -113,6 +113,9 @@ export function registerCronEditCommand(cron: Command) { .option("--wake ", "Wake mode (now|next-heartbeat)") .option("--at ", "Set one-shot time (ISO, offset-less uses --tz) or duration like 20m") .option("--every ", "Set interval duration like 10m") + .option("--pacing-min ", "Set minimum delay for a dynamic next check") + .option("--pacing-max ", "Set maximum delay for a dynamic next check") + .option("--clear-pacing", "Remove dynamic-cadence bounds", false) .option("--cron ", "Set cron expression") .option( "--tz ", @@ -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"); diff --git a/src/cli/cron-cli/trigger-options.test.ts b/src/cli/cron-cli/trigger-options.test.ts index dab103d36f0d..c43ad1782d6f 100644 --- a/src/cli/cron-cli/trigger-options.test.ts +++ b/src/cli/cron-cli/trigger-options.test.ts @@ -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 } }, + ); + }); }); diff --git a/src/cron/cron-protocol-schema.test.ts b/src/cron/cron-protocol-schema.test.ts index 547f09e960e2..93947de6ac82 100644 --- a/src/cron/cron-protocol-schema.test.ts +++ b/src/cron/cron-protocol-schema.test.ts @@ -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; 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", + ); + }); }); diff --git a/src/cron/isolated-agent/run.ts b/src/cron/isolated-agent/run.ts index 3fe63e915405..faef4f50853c 100644 --- a/src/cron/isolated-agent/run.ts +++ b/src/cron/isolated-agent/run.ts @@ -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"; diff --git a/src/cron/isolated-agent/run.types.ts b/src/cron/isolated-agent/run.types.ts index be3a9e48054b..b13af5535eed 100644 --- a/src/cron/isolated-agent/run.types.ts +++ b/src/cron/isolated-agent/run.types.ts @@ -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; diff --git a/src/cron/pacing.test.ts b/src/cron/pacing.test.ts new file mode 100644 index 000000000000..787732dc58c1 --- /dev/null +++ b/src/cron/pacing.test.ts @@ -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", + ); + }); +}); diff --git a/src/cron/pacing.ts b/src/cron/pacing.ts new file mode 100644 index 000000000000..a7bdac0d70ba --- /dev/null +++ b/src/cron/pacing.ts @@ -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), + ); +} diff --git a/src/cron/public-job.test.ts b/src/cron/public-job.test.ts new file mode 100644 index 000000000000..69e95b8832ba --- /dev/null +++ b/src/cron/public-job.test.ts @@ -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); + }); +}); diff --git a/src/cron/public-job.ts b/src/cron/public-job.ts index d4630f3d7b56..3040cdc0bae2 100644 --- a/src/cron/public-job.ts +++ b/src/cron/public-job.ts @@ -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 }; } diff --git a/src/cron/service/jobs.ts b/src/cron/service/jobs.ts index 25e4fdc5a583..68fa935f44e2 100644 --- a/src/cron/service/jobs.ts +++ b/src/cron/service/jobs.ts @@ -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); diff --git a/src/cron/service/ops.pacing.test.ts b/src/cron/service/ops.pacing.test.ts new file mode 100644 index 000000000000..c052570a11f2 --- /dev/null +++ b/src/cron/service/ops.pacing.test.ts @@ -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) => Promise) { + 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(); + }); + }); +}); diff --git a/src/cron/service/ops.ts b/src/cron/service/ops.ts index 60555a813886..ac1809d02ba3 100644 --- a/src/cron/service/ops.ts +++ b/src/cron/service/ops.ts @@ -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, diff --git a/src/cron/service/state.ts b/src/cron/service/state.ts index 26cc296e5fc0..888b572b91b3 100644 --- a/src/cron/service/state.ts +++ b/src/cron/service/state.ts @@ -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 >; diff --git a/src/cron/service/timer.pacing.test.ts b/src/cron/service/timer.pacing.test.ts new file mode 100644 index 000000000000..e12a341cda97 --- /dev/null +++ b/src/cron/service/timer.pacing.test.ts @@ -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(); + }); +}); diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts index d9b1f5e14243..c935dd41e8ea 100644 --- a/src/cron/service/timer.ts +++ b/src/cron/service/timer.ts @@ -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, diff --git a/src/cron/store.test.ts b/src/cron/store.test.ts index 3c309c7894ea..e39c28475a28 100644 --- a/src/cron/store.test.ts +++ b/src/cron/store.test.ts @@ -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; diff --git a/src/cron/store/row-codec.schedule.test.ts b/src/cron/store/row-codec.schedule.test.ts index 88da7a0b4dd9..b0fb56c9cb8a 100644 --- a/src/cron/store/row-codec.schedule.test.ts +++ b/src/cron/store/row-codec.schedule.test.ts @@ -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", diff --git a/src/cron/store/row-codec.ts b/src/cron/store/row-codec.ts index 8ff15e95facf..41058b8f3335 100644 --- a/src/cron/store/row-codec.ts +++ b/src/cron/store/row-codec.ts @@ -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>(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 } : {}), diff --git a/src/cron/types-shared.ts b/src/cron/types-shared.ts index 133cc37e3be6..fa05034595ff 100644 --- a/src/cron/types-shared.ts +++ b/src/cron/types-shared.ts @@ -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 = { @@ -11,6 +17,7 @@ export type CronJobBase > & { displayName?: string | null; + pacing?: CronPacing | null; trigger?: CronTrigger | null; payload?: CronPayloadPatch; delivery?: CronDeliveryPatch; diff --git a/src/infra/agent-events.ts b/src/infra/agent-events.ts index e558273764d8..80fc8966bfcd 100644 --- a/src/infra/agent-events.ts +++ b/src/infra/agent-events.ts @@ -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, diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.discord-group.json b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.discord-group.json index 87d0c936477a..c7241a209bc8 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.discord-group.json +++ b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.discord-group.json @@ -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:\", \"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: => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session: 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:,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:\", \"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: => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session: 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:,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": { diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.heartbeat-turn.json b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.heartbeat-turn.json index f1ec3d89076c..39abff94962f 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.heartbeat-turn.json +++ b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.heartbeat-turn.json @@ -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:\", \"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: => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session: 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:,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:\", \"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: => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session: 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:,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": { diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json index eee2076e5081..94567a4ebf69 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json +++ b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json @@ -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:\", \"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: => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session: 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:,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:\", \"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: => agentTurn {kind:\"agentTurn\",message:\"...\",model?,thinking?,timeoutSeconds?}; agentTurn defaults isolated. timeoutSeconds=0 means none.\n- current binds caller session at creation. session: 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:,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": { diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md index d949d9a1b5a4..d7a95ed3b1cb 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md +++ b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md @@ -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, diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md index a7bae92a23d1..3340b9b3f0b9 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md +++ b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md @@ -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, diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md index 2680b02d4088..4c5949b0639e 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md +++ b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md @@ -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,