diff --git a/src/gateway/server-cron-heartbeat-jobs.test.ts b/src/gateway/server-cron-heartbeat-jobs.test.ts index a82ad9f5ac2d..43da3b01316f 100644 --- a/src/gateway/server-cron-heartbeat-jobs.test.ts +++ b/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", + ); + }); }); diff --git a/src/gateway/server-cron-heartbeat-jobs.ts b/src/gateway/server-cron-heartbeat-jobs.ts index ce2209fb03f3..3bd94bbf4fdb 100644 --- a/src/gateway/server-cron-heartbeat-jobs.ts +++ b/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 }; } diff --git a/src/gateway/server-cron-notifications.test.ts b/src/gateway/server-cron-notifications.test.ts index e347c5e57657..a87dbab1a058 100644 --- a/src/gateway/server-cron-notifications.test.ts +++ b/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((_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", diff --git a/src/gateway/server-cron-notifications.ts b/src/gateway/server-cron-notifications.ts index b680a622e2fb..458d00ad8b02 100644 --- a/src/gateway/server-cron-notifications.ts +++ b/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((_resolve, reject) => { + abortController.signal.addEventListener("abort", () => reject(deliveryTimeoutError), { + once: true, + }); + }), + ]); + } finally { + clearTimeout(deliveryTimeout); + } } /** 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 d63e805afb74..ac01588ea39f 100644 --- a/src/gateway/server-methods/cron.ts +++ b/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; diff --git a/src/gateway/server-methods/cron.validation.test.ts b/src/gateway/server-methods/cron.validation.test.ts index 6d0f59cd622e..c137098781dc 100644 --- a/src/gateway/server-methods/cron.validation.test.ts +++ b/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",