From 3b43ca8d3ecbdd81c31fcc74a2d7fc1987c81bae Mon Sep 17 00:00:00 2001 From: Sparkle <2705263765@qq.com> Date: Tue, 1 Sep 2026 15:02:34 +0800 Subject: [PATCH] fix(cron): bound schedules and due batches --- extensions/cron/index.ts | 95 ++++++++++++++------ extensions/cron/schedule.ts | 6 +- tests/extensions/cron/index.test.ts | 115 ++++++++++++++++++++++++- tests/extensions/cron/schedule.test.ts | 13 +++ 4 files changed, 200 insertions(+), 29 deletions(-) diff --git a/extensions/cron/index.ts b/extensions/cron/index.ts index e0ddd03c..57a792ba 100644 --- a/extensions/cron/index.ts +++ b/extensions/cron/index.ts @@ -17,6 +17,9 @@ import type { } from "@earendil-works/pi-coding-agent"; import { advanceDeliveredJobs, + CRON_DELIVERY_MAX_BYTES, + CRON_DELIVERY_MAX_JOBS, + CRON_MAX_JOBS, type CronJob, dueJobs, formatInterval, @@ -41,6 +44,48 @@ const SYSTEM_RUNTIME: CronRuntime = { }, }; +function deliveryMessage(due: readonly CronJob[]) { + const jobs = due.map((job) => ({ + id: job.id, + prompt: job.prompt, + recurring: job.intervalMs !== undefined, + })); + return due.length === 1 + ? { + customType: "cron-fire", + content: `[cron ${jobs[0]!.id} · ${jobs[0]!.recurring ? "recurring" : "once"}]\n${jobs[0]!.prompt}`, + display: true, + details: jobs[0]!, + } + : { + customType: "cron-fire", + content: `${due.length} scheduled prompts are due:\n\n${jobs + .map( + (job) => + `[cron ${job.id} · ${job.recurring ? "recurring" : "once"}]\n${job.prompt}`, + ) + .join("\n\n")}`, + display: true, + details: { count: jobs.length, jobs }, + }; +} + +function dueDeliveryBatch(due: readonly CronJob[]) { + const selected: CronJob[] = []; + for (const job of due.slice(0, CRON_DELIVERY_MAX_JOBS)) { + const candidate = [...selected, job]; + const message = deliveryMessage(candidate); + if ( + new TextEncoder().encode(message.content).byteLength > + CRON_DELIVERY_MAX_BYTES + ) { + break; + } + selected.push(job); + } + return selected; +} + export default function cron( pi: ExtensionAPI, runtime: CronRuntime = SYSTEM_RUNTIME, @@ -58,30 +103,7 @@ export default function cron( const fire = (due: readonly CronJob[]) => { if (due.length === 0) return true; try { - const jobs = due.map((job) => ({ - id: job.id, - prompt: job.prompt, - recurring: job.intervalMs !== undefined, - })); - const message = - due.length === 1 - ? { - customType: "cron-fire", - content: `[cron ${jobs[0]!.id} · ${jobs[0]!.recurring ? "recurring" : "once"}]\n${jobs[0]!.prompt}`, - display: true, - details: jobs[0], - } - : { - customType: "cron-fire", - content: `${due.length} scheduled prompts are due:\n\n${jobs - .map( - (job) => - `[cron ${job.id} · ${job.recurring ? "recurring" : "once"}]\n${job.prompt}`, - ) - .join("\n\n")}`, - display: true, - details: { count: jobs.length, jobs }, - }; + const message = deliveryMessage(due); pi.sendMessage< | { id: number; prompt: string; recurring: boolean } | { @@ -109,9 +131,11 @@ export default function cron( const now = runtime.now(); const due = dueJobs(jobs, now); if (due.length === 0) return; + const batch = dueDeliveryBatch(due); + if (batch.length === 0) return; const deliveredIds = new Set(); - if (fire(due)) { - for (const job of due) deliveredIds.add(job.id); + if (fire(batch)) { + for (const job of batch) deliveredIds.add(job.id); } jobs = advanceDeliveredJobs(jobs, deliveredIds, runtime.now()); if (jobs.length === 0) stopTicker(); @@ -170,12 +194,29 @@ export default function cron( return; } + if (jobs.length >= CRON_MAX_JOBS) { + ctx.ui.notify( + `A session can have at most ${CRON_MAX_JOBS} scheduled prompts. Remove one before adding another.`, + "warning", + ); + return; + } + const intervalMs = parsed.intervalMs!; + const now = runtime.now(); + const nextRunAt = now + intervalMs; + if (!Number.isSafeInteger(now) || !Number.isSafeInteger(nextRunAt)) { + ctx.ui.notify( + "Scheduled time is too far in the future. Use a shorter duration.", + "warning", + ); + return; + } const job: CronJob = { id: nextId++, prompt: parsed.prompt!, ...(parsed.oneShot ? {} : { intervalMs }), - nextRunAt: runtime.now() + intervalMs, + nextRunAt, }; jobs.push(job); startTicker(); diff --git a/extensions/cron/schedule.ts b/extensions/cron/schedule.ts index 5ad88b0f..47a55ae6 100644 --- a/extensions/cron/schedule.ts +++ b/extensions/cron/schedule.ts @@ -22,6 +22,9 @@ const UNITS: Record = { export const MIN_INTERVAL_MS = 30_000; export const CRON_PROMPT_MAX_CHARS = 2_000; +export const CRON_MAX_JOBS = 64; +export const CRON_DELIVERY_MAX_JOBS = 16; +export const CRON_DELIVERY_MAX_BYTES = 48 * 1024; /** * Parse a duration like `30s`, `5m`, `2h`. Deliberately a small duration @@ -33,7 +36,8 @@ export function parseDuration(text: string): number | undefined { if (!match) return undefined; const value = Number(match[1]); if (!Number.isFinite(value) || value <= 0) return undefined; - return value * UNITS[match[2].toLowerCase()]; + const durationMs = value * UNITS[match[2].toLowerCase()]; + return Number.isSafeInteger(durationMs) ? durationMs : undefined; } export interface ParsedCronCommand { diff --git a/tests/extensions/cron/index.test.ts b/tests/extensions/cron/index.test.ts index d9adfca8..88b970b7 100644 --- a/tests/extensions/cron/index.test.ts +++ b/tests/extensions/cron/index.test.ts @@ -5,7 +5,12 @@ import type { ExtensionContext, } from "@earendil-works/pi-coding-agent"; import cron from "../../../extensions/cron/index.ts"; -import { CRON_PROMPT_MAX_CHARS } from "../../../extensions/cron/schedule.ts"; +import { + CRON_DELIVERY_MAX_BYTES, + CRON_DELIVERY_MAX_JOBS, + CRON_MAX_JOBS, + CRON_PROMPT_MAX_CHARS, +} from "../../../extensions/cron/schedule.ts"; type Handler = (event: unknown, ctx: ExtensionContext) => unknown; type CommandHandler = (args: string, ctx: ExtensionContext) => Promise; @@ -85,6 +90,7 @@ function harness() { assert.ok(tick, "scheduler must be running"); tick(); }, + polling: () => tick !== undefined, stopped: () => stopped, }; } @@ -212,3 +218,110 @@ test("cron does not create a job when the prompt exceeds the limit", async () => assert.equal(h.notifications.at(-1), "No scheduled prompts in this session."); assert.equal(h.messages.length, 0); }); + +test("cron rejects an absolute due time that is not safely representable", async () => { + const h = harness(); + await h.emit("session_start"); + h.setNow(Number.MAX_SAFE_INTEGER - 29_999); + + await h.run("in 30s never fire"); + + assert.match(h.notifications.at(-1) ?? "", /too far in the future/i); + assert.equal(h.polling(), false); + await h.run("list"); + assert.equal(h.notifications.at(-1), "No scheduled prompts in this session."); +}); + +test("cron rejects jobs beyond the per-session limit", async () => { + const h = harness(); + await h.emit("session_start"); + + for (let index = 1; index <= CRON_MAX_JOBS; index++) { + await h.run(`in 30s job ${index}`); + } + await h.run("in 30s one too many"); + + assert.match( + h.notifications.at(-1) ?? "", + new RegExp(`at most ${CRON_MAX_JOBS}`, "i"), + ); + h.setNow(30_000); + while (h.polling()) h.poll(); + const delivered = h.messages.flatMap((entry) => { + const details = (entry.message as { details: unknown }).details; + return "jobs" in (details as object) + ? (details as { jobs: Array<{ id: number }> }).jobs + : [details as { id: number }]; + }); + assert.equal(delivered.length, CRON_MAX_JOBS); + assert.deepEqual( + delivered.map((job) => job.id), + Array.from({ length: CRON_MAX_JOBS }, (_, index) => index + 1), + ); +}); + +test("cron bounds each due batch by job count and keeps the rest pending", async () => { + const h = harness(); + await h.emit("session_start"); + + for (let index = 1; index <= CRON_DELIVERY_MAX_JOBS + 1; index++) { + await h.run(`in 30s job ${index}`); + } + h.setNow(30_000); + h.poll(); + + const firstDetails = ( + h.messages[0]?.message as { + details: { count: number; jobs: Array<{ id: number }> }; + } + ).details; + assert.equal(firstDetails.count, CRON_DELIVERY_MAX_JOBS); + assert.deepEqual( + firstDetails.jobs.map((job) => job.id), + Array.from({ length: CRON_DELIVERY_MAX_JOBS }, (_, index) => index + 1), + ); + await h.run("list"); + assert.match(h.notifications.at(-1) ?? "", /17\. once/); + + h.poll(); + assert.deepEqual( + (h.messages[1]?.message as { details: { id: number } }).details, + { + id: CRON_DELIVERY_MAX_JOBS + 1, + prompt: `job ${CRON_DELIVERY_MAX_JOBS + 1}`, + recurring: false, + }, + ); +}); + +test("cron bounds model-visible due batches by UTF-8 bytes", async () => { + const h = harness(); + await h.emit("session_start"); + const prompt = "界".repeat(CRON_PROMPT_MAX_CHARS); + + for (let index = 0; index < CRON_DELIVERY_MAX_JOBS; index++) { + await h.run(`in 30s ${prompt}`); + } + h.setNow(30_000); + h.poll(); + + const first = h.messages[0]?.message as { + content: string; + details: { count: number }; + }; + assert.ok( + Buffer.byteLength(first.content, "utf8") <= CRON_DELIVERY_MAX_BYTES, + ); + assert.ok(first.details.count < CRON_DELIVERY_MAX_JOBS); + await h.run("list"); + assert.notEqual( + h.notifications.at(-1), + "No scheduled prompts in this session.", + ); + + while (h.polling()) h.poll(); + for (const entry of h.messages) { + const content = (entry.message as { content: string }).content; + assert.ok(Buffer.byteLength(content, "utf8") <= CRON_DELIVERY_MAX_BYTES); + } +}); diff --git a/tests/extensions/cron/schedule.test.ts b/tests/extensions/cron/schedule.test.ts index 3a850458..034db1e2 100644 --- a/tests/extensions/cron/schedule.test.ts +++ b/tests/extensions/cron/schedule.test.ts @@ -21,6 +21,19 @@ test("parses duration units and rejects nonsense", () => { assert.equal(parseDuration("* * * * *"), undefined); }); +test("rejects durations that overflow after unit conversion", () => { + const overflowingHours = `${"1"}${"0".repeat(305)}h`; + + assert.equal(parseDuration(overflowingHours), undefined); + for (const schedule of ["in", "every"]) { + const parsed = parseCronCommand( + `${schedule} ${overflowingHours} never fire`, + ); + assert.equal(parsed.action, "help"); + assert.equal(parsed.intervalMs, undefined); + } +}); + test("parses recurring, one-shot, and management commands", () => { const every = parseCronCommand("every 5m check the deploy"); assert.equal(every.action, "add");