fix(cron): bound alerts and complete heartbeat cleanup (#114920)

* fix(cron): bound failure notification delivery

* fix(cron): isolate heartbeat monitor cleanup failures

* perf(cron): read one job for scoped run history

* fix(cron): reject alert deadlines with explicit errors

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

src/gateway/server-cron-heartbeat-jobs.test.ts

@@ -146,4 +146,37 @@ describe("reconcileHeartbeatMonitorJobs", () => {
     expect(add).toHaveBeenCalledTimes(2);
     expect(logger.warn).toHaveBeenCalled();
   });
+
+  it("keeps removing stale monitors when one removal fails", async () => {
+    const add = vi.fn(async () => ({}));
+    const remove = vi
+      .fn(async (_jobId: string) => ({ ok: true }))
+      .mockRejectedValueOnce(new Error("store busy"));
+    const list = vi.fn(async () => [
+      monitorJob("stale-first"),
+      monitorJob("stale-second"),
+      monitorJob("stale-third"),
+    ]);
+    const cleanupLogger = { warn: vi.fn() };
+    const cfg = {
+      agents: { defaults: { heartbeat: { every: "30m" } } },
+    } as OpenClawConfig;
+
+    await expect(
+      reconcileHeartbeatMonitorJobs({
+        cron: { add, list, remove } as never,
+        cfg,
+        logger: cleanupLogger,
+      }),
+    ).resolves.toEqual({ ok: false });
+
+    expect(remove).toHaveBeenCalledTimes(3);
+    expect(remove).toHaveBeenNthCalledWith(1, "job-stale-first", { systemOwned: true });
+    expect(remove).toHaveBeenNthCalledWith(2, "job-stale-second", { systemOwned: true });
+    expect(remove).toHaveBeenNthCalledWith(3, "job-stale-third", { systemOwned: true });
+    expect(cleanupLogger.warn).toHaveBeenCalledExactlyOnceWith(
+      { agentId: "stale-first", err: "Error: store busy" },
+      "cron-heartbeat: stale monitor cleanup failed",
+    );
+  });
 });

src/gateway/server-cron-heartbeat-jobs.ts

@@ -51,19 +51,24 @@ export async function reconcileHeartbeatMonitorJobs(params: {
     }
   }
 
-  try {
-    for (const job of jobs) {
-      const agentId = heartbeatMonitorAgentId(job);
-      // Disabled heartbeats retain their stable monitor row (and scratch). Only
-      // agents no longer enrolled in heartbeat are pruned.
-      if (!agentId || desired.has(agentId)) {
-        continue;
-      }
-      await params.cron.remove(job.id, { systemOwned: true });
+  for (const job of jobs) {
+    const agentId = heartbeatMonitorAgentId(job);
+    // Disabled heartbeats retain their stable monitor row (and scratch). Only
+    // agents no longer enrolled in heartbeat are pruned.
+    if (!agentId || desired.has(agentId)) {
+      continue;
+    }
+    // Keep cleanup isolated per agent; a failed removal must not strand later
+    // stale monitors or suppress the existing reconciliation retry.
+    try {
+      await params.cron.remove(job.id, { systemOwned: true });
+    } catch (error) {
+      ok = false;
+      params.logger.warn(
+        { agentId, err: String(error) },
+        "cron-heartbeat: stale monitor cleanup failed",
+      );
     }
-  } catch (error) {
-    ok = false;
-    params.logger.warn({ err: String(error) }, "cron-heartbeat: stale monitor cleanup failed");
   }
   return { ok };
 }

src/gateway/server-cron-notifications.test.ts

@@ -335,6 +335,71 @@ describe("dispatchGatewayCronFinishedNotifications", () => {
     expect(getActiveGatewayRootWorkCount()).toBe(0);
   });
 
+  it.each([
+    { description: "honors cancellation", honorsCancellation: true },
+    { description: "ignores cancellation", honorsCancellation: false },
+  ])(
+    "releases immediate failure alert admission when a stalled sender $description",
+    async ({ honorsCancellation }) => {
+      vi.useFakeTimers();
+      try {
+        let deliverySignal: AbortSignal | undefined;
+        mocks.sendCronAnnouncePayloadStrict.mockImplementationOnce(
+          ({ abortSignal }: { abortSignal: AbortSignal }) =>
+            new Promise<void>((_resolve, reject) => {
+              deliverySignal = abortSignal;
+              if (honorsCancellation) {
+                abortSignal.addEventListener(
+                  "abort",
+                  () =>
+                    reject(
+                      abortSignal.reason instanceof Error
+                        ? abortSignal.reason
+                        : new Error("cron: failure alert announcement timed out"),
+                    ),
+                  { once: true },
+                );
+              }
+            }),
+        );
+        const job = createWebhookJob({ mode: "announce", channel: "discord", to: "channel:ops" });
+
+        const delivery = sendGatewayCronFailureAlert({
+          deps: {} as CliDeps,
+          logger: { warn: vi.fn() },
+          resolveCronAgent: () => ({ agentId: "main", cfg: {} }),
+          job,
+          text: "cron failed",
+          channel: "discord",
+          to: "channel:ops",
+          mode: "announce",
+        });
+        const deliveryOutcome = delivery.then(
+          () => undefined,
+          (error: unknown) => error,
+        );
+
+        expect(mocks.sendCronAnnouncePayloadStrict).toHaveBeenCalledOnce();
+        expect(getActiveGatewayRootWorkCount()).toBe(1);
+        expect(deliverySignal?.aborted).toBe(false);
+
+        await vi.advanceTimersByTimeAsync(9_999);
+        expect(getActiveGatewayRootWorkCount()).toBe(1);
+        expect(deliverySignal?.aborted).toBe(false);
+
+        await vi.advanceTimersByTimeAsync(1);
+        expect(deliverySignal?.aborted).toBe(true);
+        await expect(deliveryOutcome).resolves.toEqual(
+          expect.objectContaining({ message: "cron: failure alert announcement timed out" }),
+        );
+        expect(getActiveGatewayRootWorkCount()).toBe(0);
+        expect(vi.getTimerCount()).toBe(0);
+      } finally {
+        vi.useRealTimers();
+      }
+    },
+  );
+
   it("defers detached completion delivery while suspension is prepared", async () => {
     const job = createWebhookJob({
       mode: "webhook",

src/gateway/server-cron-notifications.ts

@@ -313,20 +313,37 @@ async function sendGatewayCronFailureAlertUnderAdmission(
   }
 
   const abortController = new AbortController();
-  await sendCronAnnouncePayloadStrict({
-    deps: params.deps,
-    cfg: runtimeConfig,
-    agentId,
-    jobId: params.job.id,
-    target: {
-      channel: params.channel,
-      to: params.to,
-      accountId: params.accountId,
-      sessionKey: resolveCronDeliverySessionKey(params.job),
-    },
-    message: params.text,
-    abortSignal: abortController.signal,
-  });
+  const deliveryTimeoutError = new Error("cron: failure alert announcement timed out");
+  const deliveryTimeout = setTimeout(() => {
+    abortController.abort(deliveryTimeoutError);
+  }, CRON_WEBHOOK_TIMEOUT_MS);
+
+  try {
+    // Release Gateway admission on deadline even when a transport ignores abort.
+    await Promise.race([
+      sendCronAnnouncePayloadStrict({
+        deps: params.deps,
+        cfg: runtimeConfig,
+        agentId,
+        jobId: params.job.id,
+        target: {
+          channel: params.channel,
+          to: params.to,
+          accountId: params.accountId,
+          sessionKey: resolveCronDeliverySessionKey(params.job),
+        },
+        message: params.text,
+        abortSignal: abortController.signal,
+      }),
+      new Promise<never>((_resolve, reject) => {
+        abortController.signal.addEventListener("abort", () => reject(deliveryTimeoutError), {
+          once: true,
+        });
+      }),
+    ]);
+  } finally {
+    clearTimeout(deliveryTimeout);
+  }
 }
 
 /** Dispatches completion and failure-destination notifications after a cron run finishes. */

src/gateway/server-methods/cron.ts

@@ -1069,21 +1069,20 @@ export const cronHandlers: GatewayRequestHandlers = {
       return;
     }
     try {
-      const jobs = filterCronRunLogJobsByAgent(
-        await context.cron.list({ includeDisabled: true }),
-        p.agentId,
-        context.cron.getDefaultAgentId(),
-      );
-      const matchedJob = jobs.find(
-        (job) =>
-          job.id === jobId &&
-          cronJobMatchesCallerScope({
-            job,
-            callerScope,
-            defaultAgentId: context.cron.getDefaultAgentId(),
-            allowCurrentJob: true,
-          }),
-      );
+      const job = await context.cron.readJob(jobId as string);
+      const defaultAgentId = context.cron.getDefaultAgentId();
+      const matchedJob =
+        job &&
+        filterCronRunLogJobsByAgent([job], p.agentId, defaultAgentId).length > 0 &&
+        cronJobMatchesCallerScope({
+          job,
+          callerScope,
+          defaultAgentId,
+          allowCurrentJob: true,
+        })
+          ? job
+          : undefined;
+      // Operator history survives job deletion; scoped reads still need a live, matching owner.
       if ((callerScope || p.agentId) && !matchedJob) {
         respondInvalidCronParams(respond, "cron.runs", "id not found");
         return;

src/gateway/server-methods/cron.validation.test.ts

@@ -3293,12 +3293,96 @@ describe("cron method validation", () => {
     });
   });
 
+  it.each([
+    { selector: "id", params: { id: "cron-1" } },
+    { selector: "jobId", params: { jobId: "cron-1" } },
+  ])("reads only the $selector-selected cron job for run history", async ({ params }) => {
+    const context = createCronContext([
+      createCronJob({ id: "cron-1", agentId: "ops" }),
+      createCronJob({ id: "cron-2", agentId: "worker" }),
+    ]);
+
+    const { respond } = await invokeCron("cron.runs", params, { context });
+
+    expect(context.cron.readJob).toHaveBeenCalledExactlyOnceWith("cron-1");
+    expect(context.cron.list).not.toHaveBeenCalled();
+    expect(respond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({ entries: expect.any(Array) }),
+      undefined,
+    );
+  });
+
+  it("preserves deleted-job history without listing unrelated cron jobs", async () => {
+    const context = createCronContext();
+
+    const { respond } = await invokeCron("cron.runs", { id: "deleted-cron" }, { context });
+
+    expect(context.cron.readJob).toHaveBeenCalledExactlyOnceWith("deleted-cron");
+    expect(context.cron.list).not.toHaveBeenCalled();
+    expect(respond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({ entries: expect.any(Array) }),
+      undefined,
+    );
+  });
+
+  it("preserves explicit agent ownership for directly read cron history", async () => {
+    const context = createCronContext(createCronJob({ id: "cron-1", agentId: "ops" }));
+
+    const { respond } = await invokeCron(
+      "cron.runs",
+      { id: "cron-1", agentId: "worker" },
+      { context },
+    );
+
+    expect(context.cron.readJob).toHaveBeenCalledExactlyOnceWith("cron-1");
+    expect(context.cron.list).not.toHaveBeenCalled();
+    expectResponseError(respond, {
+      code: "INVALID_REQUEST",
+      messageIncludes: "invalid cron.runs params: id not found",
+    });
+  });
+
+  it("preserves normalized default-agent ownership for directly read cron history", async () => {
+    const context = createCronContext(createCronJob({ id: "cron-1", agentId: undefined }));
+
+    const { respond } = await invokeCron(
+      "cron.runs",
+      { id: "cron-1", agentId: "MAIN" },
+      { context },
+    );
+
+    expect(context.cron.readJob).toHaveBeenCalledExactlyOnceWith("cron-1");
+    expect(context.cron.list).not.toHaveBeenCalled();
+    expect(respond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({ entries: expect.any(Array) }),
+      undefined,
+    );
+  });
+
+  it("retains full cron job discovery for all-scope history", async () => {
+    const context = createCronContext(createCronJob({ id: "cron-1" }));
+
+    const { respond } = await invokeCron("cron.runs", { scope: "all" }, { context });
+
+    expect(context.cron.list).toHaveBeenCalledExactlyOnceWith({ includeDisabled: true });
+    expect(context.cron.readJob).not.toHaveBeenCalled();
+    expect(respond).toHaveBeenCalledWith(
+      true,
+      expect.objectContaining({ entries: expect.any(Array) }),
+      undefined,
+    );
+  });
+
   it("does not widen a whitespace-only cron.runs selector to all history", async () => {
     const context = createCronContext();
 
     const { respond } = await invokeCron("cron.runs", { id: "   " }, { context });
 
     expect(context.cron.list).not.toHaveBeenCalled();
+    expect(context.cron.readJob).not.toHaveBeenCalled();
     expectResponseError(respond, {
       code: "INVALID_REQUEST",
       messageIncludes: "invalid cron.runs params: missing id",
@@ -3314,6 +3398,8 @@ describe("cron method validation", () => {
       { context, client: callerClient("ops") },
     );
 
+    expect(context.cron.readJob).toHaveBeenCalledExactlyOnceWith("cron-1");
+    expect(context.cron.list).not.toHaveBeenCalled();
     expectResponseError(respond, {
       code: "INVALID_REQUEST",
       messageIncludes: "invalid cron.runs params: id not found",
@@ -3339,6 +3425,8 @@ describe("cron method validation", () => {
       { context, client: callerClient("ops") },
     );
 
+    expect(context.cron.readJob).toHaveBeenCalledExactlyOnceWith("cron-1");
+    expect(context.cron.list).not.toHaveBeenCalled();
     expectResponseError(respond, {
       code: "INVALID_REQUEST",
       messageIncludes: "invalid cron.runs params: id not found",