fix(cron): bound schedules and due batches

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<number>();
-    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();

extensions/cron/schedule.ts

@@ -22,6 +22,9 @@ const UNITS: Record<string, number> = {
 
 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 {

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<void>;
@@ -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);
+  }
+});

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");