From 383363e3623374db34c2b855f52f244caa688998 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 31 Jul 2026 14:38:21 -0700 Subject: [PATCH] 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 --- src/cli/system-cli.test.ts | 9 + src/cli/system-cli.ts | 10 +- src/cron/delivery.failure-notify.test.ts | 161 +++++++++++++++++- src/cron/delivery.ts | 68 +++++--- src/gateway/server-cron-contract.ts | 2 + src/gateway/server-cron-lazy.test.ts | 24 +++ src/gateway/server-cron-lazy.ts | 7 +- src/gateway/server-cron-notifications.test.ts | 42 +++++ src/gateway/server-cron-notifications.ts | 70 ++++---- src/gateway/server-methods/cron.ts | 5 +- .../server-methods/cron.validation.test.ts | 6 + 11 files changed, 339 insertions(+), 65 deletions(-) diff --git a/src/cli/system-cli.test.ts b/src/cli/system-cli.test.ts index 7d682e2d8424..b492dd7400a6 100644 --- a/src/cli/system-cli.test.ts +++ b/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"]); diff --git a/src/cli/system-cli.ts b/src/cli/system-cli.ts index a0b6b5c0c367..f6b9fa1497c1 100644 --- a/src/cli/system-cli.ts +++ b/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", ); diff --git a/src/cron/delivery.failure-notify.test.ts b/src/cron/delivery.failure-notify.test.ts index 69fb88156527..d2404cfe49fa 100644 --- a/src/cron/delivery.failure-notify.test.ts +++ b/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((_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")); diff --git a/src/cron/delivery.ts b/src/cron/delivery.ts index e8d6ede2e17f..29392768eff6 100644 --- a/src/cron/delivery.ts +++ b/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 { - 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); } } diff --git a/src/gateway/server-cron-contract.ts b/src/gateway/server-cron-contract.ts index 23a57298f286..e02203c22e95 100644 --- a/src/gateway/server-cron-contract.ts +++ b/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; /** Stop cron and await scheduler-owned child process teardown. */ stopAndDrain?(): Promise; }; diff --git a/src/gateway/server-cron-lazy.test.ts b/src/gateway/server-cron-lazy.test.ts index f0a26019321c..bf76d3ffa4a6 100644 --- a/src/gateway/server-cron-lazy.test.ts +++ b/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)); diff --git a/src/gateway/server-cron-lazy.ts b/src/gateway/server-cron-lazy.ts index c968ece54e7b..b947f84fd122 100644 --- a/src/gateway/server-cron-lazy.ts +++ b/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 diff --git a/src/gateway/server-cron-notifications.test.ts b/src/gateway/server-cron-notifications.test.ts index d22bc3a27637..bf4f8adfe9d2 100644 --- a/src/gateway/server-cron-notifications.test.ts +++ b/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(() => {}) }), + ); + 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", diff --git a/src/gateway/server-cron-notifications.ts b/src/gateway/server-cron-notifications.ts index b631c1e31f95..72915e43d3a2 100644 --- a/src/gateway/server-cron-notifications.ts +++ b/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 { 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((_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. */ diff --git a/src/gateway/server-methods/cron.ts b/src/gateway/server-methods/cron.ts index a2eaf381fb30..2a8bdfa2a8c9 100644 --- a/src/gateway/server-methods/cron.ts +++ b/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, diff --git a/src/gateway/server-methods/cron.validation.test.ts b/src/gateway/server-methods/cron.validation.test.ts index 636d6a57c9b9..e18401f50fbb 100644 --- a/src/gateway/server-methods/cron.validation.test.ts +++ b/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" }); });