fix(cron): preserve lazy ownership and bounded notification lifetimes (#117018)

* fix(cron): preserve lazy ownership and bound notification lifetimes

* fix(cron): respect fs-safe policy boundary

---------

Co-authored-by: Peter Steinberger <steipete@macos.shared>

src/cli/system-cli.test.ts

@@ -69,6 +69,15 @@ describe("system-cli", () => {
     expect(runtimeLogs).toEqual([JSON.stringify({ id: "wake-1" }, null, 2)]);
   });
 
+  it("reports a rejected system event instead of claiming it was enqueued", async () => {
+    callGatewayFromCli.mockResolvedValueOnce({ ok: false, reason: "unwakeable-session-key" });
+
+    await runCli(["system", "event", "--text", "hello"]);
+
+    expect(runtimeLogs).toEqual([]);
+    expect(runtimeErrors[0]).toContain("unwakeable-session-key");
+  });
+
   it("handles invalid wake mode as runtime error", async () => {
     await runCli(["system", "event", "--text", "hello", "--mode", "later"]);
 

src/cli/system-cli.ts

@@ -83,12 +83,20 @@ export function registerSystemCli(program: Command) {
         }
         const mode = normalizeWakeMode(opts.mode);
         const sessionKey = normalizeOptionalString(opts.sessionKey);
-        return await callGatewayFromCli(
+        const result = await callGatewayFromCli(
           "wake",
           opts,
           sessionKey ? { mode, text, sessionKey } : { mode, text },
           { expectFinal: false },
         );
+        if (typeof result === "object" && result !== null && "ok" in result && !result.ok) {
+          const reason =
+            "reason" in result && typeof result.reason === "string"
+              ? result.reason
+              : "Gateway did not accept the system event";
+          throw new Error(reason);
+        }
+        return result;
       },
       "ok",
     );

src/cron/delivery.failure-notify.test.ts

@@ -37,7 +37,8 @@ vi.mock("../logging.js", () => ({
   })),
 }));
 
-const { sendFailureNotificationAnnounce } = await import("./delivery.js");
+const { sendCronAnnouncePayloadStrict, sendFailureNotificationAnnounce } =
+  await import("./delivery.js");
 
 type DeliveryRequest = {
   abortSignal?: unknown;
@@ -184,6 +185,39 @@ describe("sendFailureNotificationAnnounce", () => {
     );
   });
 
+  it("does not begin strict delivery when target resolution settles after cancellation", async () => {
+    let resolvePendingTarget: (value: unknown) => void = () => {};
+    mocks.resolveDeliveryTarget.mockImplementationOnce(
+      () =>
+        new Promise((resolve) => {
+          resolvePendingTarget = resolve;
+        }),
+    );
+    const abortController = new AbortController();
+
+    const delivery = sendCronAnnouncePayloadStrict({
+      deps: {} as never,
+      cfg: {} as never,
+      agentId: "main",
+      jobId: "job-1",
+      target: { channel: "telegram", to: "123" },
+      message: "Cron failed",
+      abortSignal: abortController.signal,
+    });
+
+    abortController.abort(new Error("delivery deadline exceeded"));
+    resolvePendingTarget({
+      ok: true,
+      channel: "telegram",
+      to: "123",
+      accountId: "bot-a",
+      mode: "explicit",
+    });
+
+    await expect(delivery).rejects.toThrow("delivery deadline exceeded");
+    expect(mocks.deliverOutboundPayloads).not.toHaveBeenCalled();
+  });
+
   it("does not send when target resolution fails", async () => {
     mocks.resolveDeliveryTarget.mockResolvedValue({
       ok: false,
@@ -206,6 +240,131 @@ describe("sendFailureNotificationAnnounce", () => {
     );
   });
 
+  it("logs thrown target-resolution failures without masking the failed cron run", async () => {
+    mocks.resolveDeliveryTarget.mockRejectedValueOnce(new Error("target lookup failed"));
+
+    await expect(
+      sendFailureNotificationAnnounce(
+        {} as never,
+        {} as never,
+        "main",
+        "job-1",
+        { channel: "telegram", to: "123" },
+        "Cron failed",
+      ),
+    ).resolves.toBeUndefined();
+
+    expect(mocks.deliverOutboundPayloads).not.toHaveBeenCalled();
+    expect(mocks.warn).toHaveBeenCalledWith(
+      { err: "target lookup failed", channel: "telegram", to: "123" },
+      "cron: failure destination announce failed",
+    );
+  });
+
+  it("bounds stalled target resolution without starting a late channel send", async () => {
+    vi.useFakeTimers();
+    let resolvePendingTarget: (value: unknown) => void = () => {};
+    mocks.resolveDeliveryTarget.mockImplementationOnce(
+      () =>
+        new Promise((resolve) => {
+          resolvePendingTarget = resolve;
+        }),
+    );
+
+    const notification = sendFailureNotificationAnnounce(
+      {} as never,
+      {} as never,
+      "main",
+      "job-1",
+      { channel: "telegram", to: "123" },
+      "Cron failed",
+    );
+
+    await vi.advanceTimersByTimeAsync(29_999);
+    expect(mocks.warn).not.toHaveBeenCalled();
+    expect(mocks.deliverOutboundPayloads).not.toHaveBeenCalled();
+
+    await vi.advanceTimersByTimeAsync(1);
+    await expect(notification).resolves.toBeUndefined();
+    expect(mocks.warn).toHaveBeenCalledWith(
+      {
+        err: "cron: failure destination announcement timed out",
+        channel: "telegram",
+        to: "123",
+      },
+      "cron: failure destination announce failed",
+    );
+
+    resolvePendingTarget({
+      ok: true,
+      channel: "telegram",
+      to: "123",
+      accountId: "bot-a",
+      mode: "explicit",
+    });
+    await vi.advanceTimersByTimeAsync(0);
+    expect(mocks.deliverOutboundPayloads).not.toHaveBeenCalled();
+    expect(vi.getTimerCount()).toBe(0);
+  });
+
+  it.each([
+    { description: "honors cancellation", honorsCancellation: true },
+    { description: "ignores cancellation", honorsCancellation: false },
+  ])(
+    "bounds a stalled failure notification when its channel $description",
+    async ({ honorsCancellation }) => {
+      vi.useFakeTimers();
+      let deliverySignal: AbortSignal | undefined;
+      mocks.deliverOutboundPayloads.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("failure notification was aborted"),
+                  ),
+                { once: true },
+              );
+            }
+          }),
+      );
+
+      const notification = sendFailureNotificationAnnounce(
+        {} as never,
+        {} as never,
+        "main",
+        "job-1",
+        { channel: "telegram", to: "123" },
+        "Cron failed",
+      );
+
+      await vi.advanceTimersByTimeAsync(0);
+      expect(mocks.deliverOutboundPayloads).toHaveBeenCalledOnce();
+      expect(deliverySignal?.aborted).toBe(false);
+
+      await vi.advanceTimersByTimeAsync(29_999);
+      expect(deliverySignal?.aborted).toBe(false);
+
+      await vi.advanceTimersByTimeAsync(1);
+      await expect(notification).resolves.toBeUndefined();
+      expect(deliverySignal?.aborted).toBe(true);
+      expect(mocks.warn).toHaveBeenCalledWith(
+        {
+          err: "cron: failure destination announcement timed out",
+          channel: "telegram",
+          to: "123",
+        },
+        "cron: failure destination announce failed",
+      );
+      expect(vi.getTimerCount()).toBe(0);
+    },
+  );
+
   it("swallows outbound delivery errors after logging", async () => {
     mocks.deliverOutboundPayloads.mockRejectedValue(new Error("send failed"));
 

src/cron/delivery.ts

@@ -4,6 +4,7 @@ import type { CliDeps } from "../cli/deps.types.js";
 import { createOutboundSendDeps } from "../cli/outbound-send-deps.js";
 import type { OpenClawConfig } from "../config/types.js";
 import { formatErrorMessage } from "../infra/errors.js";
+import { withTimeout } from "../infra/fs-safe.js";
 import { resolveAgentOutboundIdentity } from "../infra/outbound/identity.js";
 import { buildOutboundSessionContext } from "../infra/outbound/session-context.js";
 import { getChildLogger } from "../logging.js";
@@ -130,6 +131,9 @@ export async function sendCronAnnouncePayloadStrict(params: {
   if (!delivery.ok) {
     throw delivery.error;
   }
+  // Resolution can settle after its caller's deadline; never start plugin
+  // delivery once the Gateway has released ownership of the timed-out work.
+  params.abortSignal.throwIfAborted();
   await deliverCronAnnouncePayload({
     deps: params.deps,
     cfg: params.cfg,
@@ -148,42 +152,52 @@ export async function sendFailureNotificationAnnounce(
   target: CronAnnounceTarget,
   message: string,
 ): Promise<void> {
-  const delivery = await resolveCronAnnounceDelivery({ cfg, agentId, jobId, target });
-
-  if (!delivery.ok) {
-    // Failure alerts must not mask the original cron run failure.
-    cronDeliveryLogger.warn(
-      { error: delivery.error.message },
-      "cron: failed to resolve failure destination target",
-    );
-    return;
-  }
-
   const abortController = new AbortController();
-  const timeout = setTimeout(() => {
-    // Failure notifications are secondary; timeout prevents a stuck channel send
-    // from extending an already-failed cron run.
-    abortController.abort();
-  }, FAILURE_NOTIFICATION_TIMEOUT_MS);
+  let resolvedTarget: SuccessfulDeliveryTarget | undefined;
 
   try {
-    await deliverCronAnnouncePayload({
-      deps,
-      cfg,
-      delivery,
-      message,
-      abortSignal: abortController.signal,
-    });
+    // Bound resolution and transport together; either owner can stall while
+    // retaining the detached Gateway work admission.
+    await withTimeout(
+      (async () => {
+        const delivery = await resolveCronAnnounceDelivery({ cfg, agentId, jobId, target });
+        if (!delivery.ok) {
+          // Failure alerts must not mask the original cron run failure.
+          cronDeliveryLogger.warn(
+            { error: delivery.error.message },
+            "cron: failed to resolve failure destination target",
+          );
+          return;
+        }
+        resolvedTarget = delivery.resolvedTarget;
+        // A resolver can settle after its deadline; never start a late send
+        // after detached work ownership has already been released.
+        abortController.signal.throwIfAborted();
+        await deliverCronAnnouncePayload({
+          deps,
+          cfg,
+          delivery,
+          message,
+          abortSignal: abortController.signal,
+        });
+      })(),
+      FAILURE_NOTIFICATION_TIMEOUT_MS,
+      {
+        createError: () => {
+          const error = new Error("cron: failure destination announcement timed out");
+          abortController.abort(error);
+          return error;
+        },
+      },
+    );
   } catch (err) {
     cronDeliveryLogger.warn(
       {
         err: formatErrorMessage(err),
-        channel: delivery.resolvedTarget.channel,
-        to: delivery.resolvedTarget.to,
+        channel: resolvedTarget?.channel ?? target.channel,
+        to: resolvedTarget?.to ?? target.to,
       },
       "cron: failure destination announce failed",
     );
-  } finally {
-    clearTimeout(timeout);
   }
 }

src/gateway/server-cron-contract.ts

@@ -22,6 +22,8 @@ export type GatewayCronServiceContract = CronServiceContract & {
   resumeScheduling(): void;
   /** Scheduler-owned work not represented by active cron run markers. */
   getSuspensionBlockerCount?(): number;
+  /** Materialize lazy cron dependencies before a synchronous operator wake. */
+  prepareWake?(): Promise<void>;
   /** Stop cron and await scheduler-owned child process teardown. */
   stopAndDrain?(): Promise<void>;
 };

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

@@ -99,6 +99,30 @@ describe("createLazyGatewayCronState", () => {
     expect(cron["run"]).toHaveBeenCalledWith("demo", "force", { payload });
   });
 
+  it("preserves system-owned removal authority across lazy cron loading", async () => {
+    const cron = createCronService();
+    hoisted.setState(createCronState(cron));
+
+    const lazy = createLazyGatewayCronState(createParams());
+    await lazy.cron.remove("heartbeat-monitor", { systemOwned: true });
+
+    expect(cron["remove"]).toHaveBeenCalledExactlyOnceWith("heartbeat-monitor", {
+      systemOwned: true,
+    });
+  });
+
+  it("prepares a lazy scheduler before an operator wake without starting it", async () => {
+    const cron = createCronService();
+    hoisted.setState(createCronState(cron));
+
+    const lazy = createLazyGatewayCronState(createParams());
+    await lazy.cron.prepareWake?.();
+
+    expect(lazy.cron.wake({ mode: "now", text: "ping" })).toEqual({ ok: true });
+    expect(cron["start"]).not.toHaveBeenCalled();
+    expect(cron["wake"]).toHaveBeenCalledExactlyOnceWith({ mode: "now", text: "ping" });
+  });
+
   it("starts the loaded cron service once", async () => {
     const cron = createCronService();
     hoisted.setState(createCronState(cron));

src/gateway/server-cron-lazy.ts

@@ -259,8 +259,8 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew
     async updateWithPrecondition(id, patch, precondition) {
       return await (await load()).state.cron.updateWithPrecondition(id, patch, precondition);
     },
-    async remove(id) {
-      return await (await load()).state.cron.remove(id);
+    async remove(id, opts) {
+      return await (await load()).state.cron.remove(id, opts);
     },
     async removeStaleJobFamily(family) {
       return await (await load()).state.cron.removeStaleJobFamily(family);
@@ -295,6 +295,9 @@ export function createLazyGatewayCronState(params: LazyGatewayCronParams): Gatew
       }
       return loaded.state.cron.getDefaultAgentId();
     },
+    async prepareWake() {
+      await load();
+    },
     wake(opts) {
       if (!loaded) {
         // A wake should kick off lazy loading but cannot claim success before

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

@@ -261,6 +261,48 @@ describe("dispatchGatewayCronFinishedNotifications", () => {
     expect(cleanupOrder).toEqual(["cancel", "release"]);
   });
 
+  it("releases Gateway admission when webhook response cancellation never settles", async () => {
+    vi.useFakeTimers();
+    try {
+      const release = vi.fn(async () => {});
+      const response = new Response(
+        new ReadableStream({ cancel: () => new Promise<void>(() => {}) }),
+      );
+      mocks.fetchWithSsrFGuard.mockResolvedValueOnce({
+        response,
+        finalUrl: "https://example.invalid/cron",
+        release,
+      });
+
+      const delivery = sendGatewayCronFailureAlert({
+        deps: {} as CliDeps,
+        logger: { warn: vi.fn() },
+        resolveCronAgent: () => ({ agentId: "main", cfg: {} }),
+        job: createWebhookJob({
+          mode: "webhook",
+          to: "https://example.invalid/cron",
+        }),
+        text: "cron failed",
+        channel: "last",
+        mode: "webhook",
+        to: "https://example.invalid/cron",
+      });
+
+      await vi.advanceTimersByTimeAsync(0);
+      expect(getActiveGatewayRootWorkCount()).toBe(1);
+      await vi.advanceTimersByTimeAsync(9_999);
+      expect(release).not.toHaveBeenCalled();
+
+      await vi.advanceTimersByTimeAsync(1);
+      await expect(delivery).resolves.toBeUndefined();
+      expect(release).toHaveBeenCalledOnce();
+      expect(getActiveGatewayRootWorkCount()).toBe(0);
+      expect(vi.getTimerCount()).toBe(0);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("adds the run start time to immediate chat alerts in the agent timezone", async () => {
     const job = createWebhookJob({
       mode: "announce",

src/gateway/server-cron-notifications.ts

@@ -21,6 +21,7 @@ import type { CronJob, CronMessageChannel } from "../cron/types.js";
 import { normalizeHttpWebhookUrl } from "../cron/webhook-url.js";
 import { formatErrorMessage } from "../infra/errors.js";
 import { formatZonedTimestamp } from "../infra/format-time/format-datetime.js";
+import { withTimeout } from "../infra/fs-safe.js";
 import { fetchWithSsrFGuard } from "../infra/net/fetch-guard.js";
 import { SsrFBlockedError } from "../infra/net/ssrf.js";
 import { runWithGatewayIndependentRootWorkAdmission } from "../process/gateway-work-admission.js";
@@ -225,6 +226,7 @@ async function postCronWebhook(params: {
   logger: CronLogger;
 }): Promise<void> {
   const abortController = new AbortController();
+  const deadlineAtMs = Date.now() + CRON_WEBHOOK_TIMEOUT_MS;
   try {
     assertSecretOwnerAvailable("capability", "cron-webhook");
     const result = await fetchWithSsrFGuard({
@@ -243,9 +245,17 @@ async function postCronWebhook(params: {
       }
     } finally {
       // Guard release closes the dispatcher, not an unread response stream.
-      // Settle the terminal body first so streaming webhooks cannot retain the socket.
+      // Keep response cleanup inside the request deadline; a non-settling
+      // stream cancellation must not retain the dispatcher or Gateway root.
       if (!result.response.bodyUsed) {
-        await result.response.body?.cancel().catch(() => undefined);
+        const cancellation = result.response.body?.cancel();
+        if (cancellation) {
+          await withTimeout(
+            cancellation,
+            Math.max(1, deadlineAtMs - Date.now()),
+            "cron webhook response cleanup",
+          ).catch(() => undefined);
+        }
       }
       await result.release();
     }
@@ -338,37 +348,31 @@ async function sendGatewayCronFailureAlertUnderAdmission(
 
   const abortController = new AbortController();
   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,
-          threadId: params.threadId,
-          sessionKey: resolveCronDeliverySessionKey(params.job),
-        },
-        message: appendCronRunStarted(params.text, params.runAtMs, runtimeConfig),
-        abortSignal: abortController.signal,
-      }),
-      new Promise<never>((_resolve, reject) => {
-        abortController.signal.addEventListener("abort", () => reject(deliveryTimeoutError), {
-          once: true,
-        });
-      }),
-    ]);
-  } finally {
-    clearTimeout(deliveryTimeout);
-  }
+  // Release Gateway admission on deadline even when a transport ignores abort.
+  await withTimeout(
+    sendCronAnnouncePayloadStrict({
+      deps: params.deps,
+      cfg: runtimeConfig,
+      agentId,
+      jobId: params.job.id,
+      target: {
+        channel: params.channel,
+        to: params.to,
+        accountId: params.accountId,
+        threadId: params.threadId,
+        sessionKey: resolveCronDeliverySessionKey(params.job),
+      },
+      message: appendCronRunStarted(params.text, params.runAtMs, runtimeConfig),
+      abortSignal: abortController.signal,
+    }),
+    CRON_WEBHOOK_TIMEOUT_MS,
+    {
+      createError: () => {
+        abortController.abort(deliveryTimeoutError);
+        return deliveryTimeoutError;
+      },
+    },
+  );
 }
 
 /** Dispatches completion and failure-destination notifications after a cron run finishes. */

src/gateway/server-methods/cron.ts

@@ -337,7 +337,7 @@ function respondMissingCronJobId(respond: RespondFn, method: string): void {
 
 /** Gateway request handlers for cron jobs and cron run-log access. */
 export const cronHandlers: GatewayRequestHandlers = {
-  wake: ({ params, respond, context, client }) => {
+  wake: async ({ params, respond, context, client }) => {
     if (!assertValidParams(params, validateWakeParams, "wake", respond)) {
       return;
     }
@@ -414,6 +414,9 @@ export const cronHandlers: GatewayRequestHandlers = {
       );
       return;
     }
+    // Gateway becomes request-ready before scheduled services start; load the
+    // wake owner first so an early operator event cannot disappear on cold start.
+    await context.cron.prepareWake?.();
     const result = context.cron.wake({
       mode: p.mode,
       text: p.text,

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

@@ -137,6 +137,7 @@ function createCronContext(currentJobs?: CronJob | CronJob[]) {
       enqueueRun: vi.fn(async () => ({ ok: true, enqueued: true, runId: "run-1" })),
       getDefaultAgentId: vi.fn(() => "main"),
       getJob: vi.fn((id: string) => jobs.find((job) => job.id === id)),
+      prepareWake: vi.fn(async () => undefined),
       wake: vi.fn(() => ({ ok: true }) as const),
       readJob: vi.fn(async (id: string) => jobs.find((job) => job.id === id)),
       list: vi.fn(async () => jobs),
@@ -3480,6 +3481,10 @@ describe("cron method validation", () => {
         text: "ping",
         sessionKey: "agent:main:telegram:dm:42",
       });
+      expect(context.cron.prepareWake).toHaveBeenCalledOnce();
+      expect(context.cron.prepareWake.mock.invocationCallOrder[0]).toBeLessThan(
+        context.cron.wake.mock.invocationCallOrder[0]!,
+      );
       expect(respond).toHaveBeenCalledWith(true, { ok: true }, undefined);
     });
 
@@ -3509,6 +3514,7 @@ describe("cron method validation", () => {
         sessionKey,
       });
       expect(context.cron.wake).not.toHaveBeenCalled();
+      expect(context.cron.prepareWake).not.toHaveBeenCalled();
       expectResponseError(respond, { code: "INVALID_REQUEST", messageIncludes: "sessionKey" });
     });