fix(cron): alert when topic delivery fails (#123237)

Notify an explicit cron failure destination when primary delivery fails, even if the run itself succeeded. Preserve fail-closed behavior when no alternate destination is configured.

Co-authored-by: Ayaan Zaidi <hi@obviy.us>

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

@@ -0,0 +1,89 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type { CliDeps } from "../cli/deps.types.js";
+import type { CronJob } from "../cron/types.js";
+import { resetGatewayWorkAdmission } from "../process/gateway-work-admission.js";
+
+const sendFailureNotificationAnnounce = vi.hoisted(() => vi.fn());
+
+vi.mock("../cron/delivery.js", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("../cron/delivery.js")>();
+  return { ...actual, sendFailureNotificationAnnounce };
+});
+
+import { dispatchGatewayCronFinishedNotifications } from "./server-cron-notifications.js";
+
+function createThreadedJob(withFailureDestination: boolean): CronJob {
+  return {
+    id: "cron-delivery-failure",
+    name: "threaded report",
+    enabled: true,
+    createdAtMs: 1,
+    updatedAtMs: 1,
+    schedule: { kind: "every", everyMs: 60_000 },
+    sessionTarget: "isolated",
+    wakeMode: "next-heartbeat",
+    payload: { kind: "agentTurn", message: "report" },
+    delivery: {
+      mode: "announce",
+      channel: "telegram",
+      to: "-1001234567890",
+      threadId: 42,
+      ...(withFailureDestination
+        ? {
+            failureDestination: {
+              mode: "announce" as const,
+              channel: "telegram" as const,
+              to: "-1001234567890",
+            },
+          }
+        : {}),
+    },
+    state: {},
+  };
+}
+
+describe("cron primary delivery failure notifications", () => {
+  beforeEach(() => {
+    resetGatewayWorkAdmission();
+    sendFailureNotificationAnnounce.mockReset();
+    sendFailureNotificationAnnounce.mockResolvedValue(undefined);
+  });
+
+  afterEach(() => resetGatewayWorkAdmission());
+
+  it("uses only an explicit failure destination", () => {
+    const evt = {
+      jobId: "cron-delivery-failure",
+      action: "finished" as const,
+      status: "ok" as const,
+      deliveryStatus: "not-delivered" as const,
+      deliveryError: "message thread not found",
+    };
+    const dispatch = (job: CronJob) =>
+      dispatchGatewayCronFinishedNotifications({
+        evt,
+        job,
+        deps: {} as CliDeps,
+        logger: { warn: vi.fn() },
+        resolveCronAgent: () => ({ agentId: "main", cfg: {} }),
+      });
+
+    dispatch(createThreadedJob(false));
+    expect(sendFailureNotificationAnnounce).not.toHaveBeenCalled();
+
+    dispatch(createThreadedJob(true));
+    expect(sendFailureNotificationAnnounce).toHaveBeenCalledOnce();
+    expect(sendFailureNotificationAnnounce.mock.calls[0]?.[4]).toEqual({
+      channel: "telegram",
+      to: "-1001234567890",
+      accountId: undefined,
+      sessionKey: undefined,
+      inheritSessionThread: false,
+    });
+    expect(sendFailureNotificationAnnounce.mock.calls[0]?.[5]).toEqual({
+      text:
+        '⚠️ Automation "threaded report" delivery failed\n' +
+        "Check automation history for details.",
+    });
+  });
+});

src/gateway/server-cron-notifications.ts

@@ -153,13 +153,12 @@ function buildCronWebhookHeaders(webhookToken?: string): Record<string, string>
 }
 
 function buildCronFailureWebhookPayload(params: { evt: CronEvent; job: CronJob }) {
-  const failureMessage = `Automation "${params.job.name}" failed: ${params.evt.error ?? "unknown error"}`;
   return {
     jobId: params.job.id,
     jobName: params.job.name,
-    message: failureMessage,
+    message: `Automation "${params.job.name}" ${params.evt.status === "error" ? "failed" : "delivery failed"}: ${params.evt.error ?? params.evt.deliveryError ?? "unknown error"}`,
     status: params.evt.status,
-    error: params.evt.error,
+    error: params.evt.error ?? params.evt.deliveryError,
     runAtMs: params.evt.runAtMs,
     durationMs: params.evt.durationMs,
     nextRunAtMs: params.evt.nextRunAtMs,
@@ -516,12 +515,16 @@ function dispatchCronFailureDestinationNotifications(params: {
   ssrfPolicy?: SsrFPolicy;
   globalFailureDestination?: CronFailureDestinationConfig;
 }): void {
-  if (params.evt.status !== "error" || !params.job || params.job.delivery?.bestEffort === true) {
+  if (!params.job || params.job.delivery?.bestEffort === true) {
     return;
   }
 
   const job = params.job;
   const failureDest = resolveFailureDestination(job, params.globalFailureDestination);
+  const deliveryFailed = params.evt.deliveryStatus === "not-delivered";
+  if (params.evt.status !== "error" && (!deliveryFailed || !failureDest)) {
+    return;
+  }
   const deliverySessionKey = resolveCronDeliverySessionKey(job);
   const failurePayload = buildCronFailureWebhookPayload({ evt: params.evt, job });
 
@@ -570,7 +573,7 @@ function dispatchCronFailureDestinationNotifications(params: {
         to: failureDest.to,
         accountId: failureDest.accountId,
         sessionKey: deliverySessionKey,
-        // Explicit failure routes keep run context without inheriting the primary topic.
+        // Explicit failure routes escape rejected primary delivery without inheriting its topic.
         inheritSessionThread: false,
       }
     : primaryPlan.mode === "announce" && primaryPlan.requested
@@ -588,7 +591,7 @@ function dispatchCronFailureDestinationNotifications(params: {
 
   const { agentId, cfg: runtimeConfig } = params.resolveCronAgent(job.agentId);
   const failureAlertText = [
-    `Automation "${job.name}" failed`,
+    `Automation "${job.name}" ${params.evt.status === "error" ? "failed" : "delivery failed"}`,
     ...cronFailureDetailLines(job.state.lastErrorReason),
   ].join("\n");
   dispatchDetachedCronNotification({