From 177828aa75bf03a6befd59e2956c5a438e0b71bd Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 19 Aug 2026 17:25:14 -0700 Subject: [PATCH] fix(cron): honor failure alert thresholds (#126483) * fix(cron): honor failure alert thresholds Per-run Gateway announcements bypassed threshold, cooldown, and opt-out policy. Make the scheduler the sole owner of failure notification decisions. * fix(cron): preserve safe failure details Keep trusted failure detail proof on the scheduler-authorized transport and remove the obsolete Gateway event-context handoff after the ownership consolidation. * test(cron): assert alternate failure route isolation --- docs/automation/cron-jobs.md | 19 +- docs/cli/cron.md | 8 +- docs/gateway/configuration-reference.md | 17 +- src/agents/tools/cron-tool-schema.ts | 10 +- src/agents/tools/cron-tool.ts | 2 + src/cron/cron-delivery-outcomes.e2e.test.ts | 257 ++++++++++-- src/cron/delivery-plan.ts | 100 +++-- src/cron/delivery.failure-notify.test.ts | 380 +++--------------- src/cron/delivery.test.ts | 79 ++-- src/cron/delivery.ts | 69 +--- src/cron/service.failure-alert.test.ts | 86 +++- .../service.persists-delivered-status.test.ts | 90 +---- .../failure-alerts.persistence.test.ts | 22 +- src/cron/service/failure-alerts.ts | 306 +++++++++----- src/cron/service/ops.test.ts | 3 + src/cron/service/run-recovery.test.ts | 113 +++++- src/cron/service/startup-run-repair.ts | 58 ++- src/cron/service/state.ts | 1 + src/cron/service/timer-job-runner.ts | 3 +- src/cron/service/timer-outcomes.ts | 61 +-- src/cron/service/timer-trigger.ts | 39 +- src/cron/service/timer.regression.test.ts | 13 +- .../service/timer.timeout-watchdog.test.ts | 1 + ...ron-notifications.delivery-failure.test.ts | 138 ------- src/gateway/server-cron-notifications.test.ts | 333 ++++----------- src/gateway/server-cron-notifications.ts | 196 +-------- src/gateway/server-cron.test.ts | 16 +- src/gateway/server-cron.ts | 4 +- src/gateway/server-methods/cron.ts | 66 +-- .../server-methods/cron.validation.test.ts | 20 +- src/gateway/server.cron.test.ts | 43 +- .../codex-dynamic-tools.telegram-direct.json | 6 +- .../discord-group-codex-message-tool.md | 8 +- .../telegram-direct-codex-message-tool.md | 8 +- .../telegram-heartbeat-codex-tool.md | 8 +- 35 files changed, 1092 insertions(+), 1491 deletions(-) delete mode 100644 src/gateway/server-cron-notifications.delivery-failure.test.ts diff --git a/docs/automation/cron-jobs.md b/docs/automation/cron-jobs.md index ddcd691a883a..52ee6d55529d 100644 --- a/docs/automation/cron-jobs.md +++ b/docs/automation/cron-jobs.md @@ -379,19 +379,28 @@ Implicit announce delivery uses configured channel allowlists to validate and re ### Failure notifications -Failure notifications follow a separate destination path: +Execution failures use one scheduler-owned threshold and cooldown policy. A job with an existing failure route is covered by default after 2 consecutive failures with a 1-hour cooldown. The route can be a resolved failure destination or the job's primary announce target. Jobs with no such route stay quiet unless a per-job or global `failureAlert` object explicitly activates the policy. -- The destination fields on `cron.failureAlert` (`mode`, `channel`, `to`, `accountId`) set a global default for failure notifications. The retired `cron.failureDestination` block is merged into them by `openclaw doctor --fix`. -- `job.delivery.failureDestination` overrides that per job. -- If neither is set and the job already delivers via `announce`, failure notifications fall back to that primary announce target. +Failure notification routes resolve in this order: + +1. Route fields in the job's `failureAlert` object. +2. `job.delivery.failureDestination`, layered over the destination fields in global `cron.failureAlert` (`mode`, `channel`, `to`, `accountId`). The retired `cron.failureDestination` block is merged into the global object by `openclaw doctor --fix`. +3. The job's primary announce target. + +- `job.failureAlert: false` disables execution and required-delivery failure alerts for that job. The auto-disable safety notification remains active. +- Global `cron.failureAlert.enabled: false` disables inherited notifications. A per-job `failureAlert` object explicitly re-enables that job; `enabled: true` explicitly enables the global policy. +- A per-job `failureAlert` object or any global `cron.failureAlert` object activates and tunes the policy even when the job had no existing route. +- `delivery.bestEffort: true` suppresses inherited/default execution-failure alerts. An explicit per-job `failureAlert` remains authoritative. - `delivery.failureDestination` is only supported on `sessionTarget="isolated"` jobs unless the primary delivery mode is `webhook`. - `failureAlert.includeSkipped: true` opts a job or global automation alert policy into repeated skipped-run alerts. Skipped runs keep a separate consecutive-skip counter, so they do not affect execution-error backoff. - `openclaw automations edit` exposes per-job alert tuning: `--failure-alert`/`--no-failure-alert`, `--failure-alert-after `, `--failure-alert-channel`, `--failure-alert-to`, `--failure-alert-cooldown`, `--failure-alert-include-skipped`/`--failure-alert-exclude-skipped`, `--failure-alert-mode`, and `--failure-alert-account-id`. +A required completion-delivery failure is distinct from an execution failure: a run can record `status: "ok"` with `completionStatus: "failed"`. It does not increment the execution-failure streak or backoff. The scheduler may notify immediately only through a resolved alternate failure destination; it never retries the already-failed primary route. + Chat failure notifications include the run start time in the agent's configured user timezone. Webhook message text stays stable; integrations can read the same instant from the structured `runAtMs` field. Chat notifications show normalized failure causes or allowlisted producer facts for known command and script failures. Arbitrary commands, paths, provider bodies, secrets, delivery errors, skip reasons, diagnostics, and stack/error text remain in automation history. Failure webhooks retain the structured raw error for diagnostic integrations. -Failure alerts are opt-in, but the scheduler also provides an unconditional safety backstop. A time-based recurring job is auto-disabled after 10 consecutive execution failures; a successful run resets that streak. Repeated schedule-computation failures auto-disable after 3 errors. The job records `state.autoDisabled.reason` as `consecutive-failures` or `schedule-errors`, and the owning agent receives a notification with a safe cause and recovery command. Raw errors stay in automation history. After fixing the cause, run `openclaw automations enable `; enabling clears the recorded reason and failure streaks. Because disabled jobs are hidden by the default list, use `openclaw automations list --all` to inspect them. +The scheduler also provides an unconditional safety backstop. A time-based recurring job is auto-disabled after 10 consecutive execution failures; a successful run resets that streak. On the terminal failure, the richer auto-disable notification replaces the regular threshold alert. Repeated schedule-computation failures auto-disable after 3 errors. The job records `state.autoDisabled.reason` as `consecutive-failures` or `schedule-errors`, and the owning agent receives a notification with a safe cause and recovery command. Raw errors stay in automation history. After fixing the cause, run `openclaw automations enable `; enabling clears the recorded reason and failure streaks. Because disabled jobs are hidden by the default list, use `openclaw automations list --all` to inspect them. ### Output language diff --git a/docs/cli/cron.md b/docs/cli/cron.md index 3d83f3351562..ef038c8728d6 100644 --- a/docs/cli/cron.md +++ b/docs/cli/cron.md @@ -114,10 +114,12 @@ Reminders created from an active chat preserve the live chat delivery target for Failure notifications resolve in this order: -1. `delivery.failureDestination` on the job. -2. The global destination fields on `cron.failureAlert` (`mode`, `channel`, `to`, `accountId`). The retired `cron.failureDestination` block is merged into them by `openclaw doctor --fix`. +1. Route fields in the job's `failureAlert` object. +2. `delivery.failureDestination` on the job, layered over the global destination fields on `cron.failureAlert` (`mode`, `channel`, `to`, `accountId`). The retired `cron.failureDestination` block is merged into them by `openclaw doctor --fix`. 3. The job's primary announce target (when neither of the above resolves to a concrete destination). +Jobs with one of those routes default to an execution-failure alert after 2 consecutive failures and a 1-hour cooldown. A per-job or global `failureAlert` object explicitly activates/tunes the policy even without an existing route. `failureAlert: false` disables execution and required-delivery failure alerts for the job, but not the auto-disable safety notification. Global `enabled: false` disables inheritance unless the job has its own `failureAlert` object. `delivery.bestEffort: true` suppresses inherited/default execution alerts, but not an explicit per-job policy. + Main-session jobs may only use `delivery.failureDestination` when primary delivery mode is `webhook`. Isolated jobs accept it in all modes. @@ -128,6 +130,8 @@ Isolated automation runs treat run-level agent failures as job errors even when Command jobs do not start an isolated agent turn. A zero exit code records `ok`; non-zero exit, signal, timeout, or no-output timeout records `error` and can trigger the same failure notification path. +Required completion delivery is separate: `status: "ok"` with `completionStatus: "failed"` does not increment the execution streak or backoff. It can notify immediately only through a resolved alternate failure destination, never the primary route that just failed. + If an isolated run times out before the first model request, `openclaw automations show` and `openclaw automations runs` include a phase-specific error such as `setup timed out before runner start` or a stall message naming the last-known startup phase (for example `context-engine`). For CLI-backed providers, the pre-model watchdog stays active until the external CLI turn starts, so session lookup, hook, auth, prompt, and CLI setup stalls are reported as pre-model automation failures. ## Scheduling diff --git a/docs/gateway/configuration-reference.md b/docs/gateway/configuration-reference.md index 0edac8580a7b..075c8fc1affa 100644 --- a/docs/gateway/configuration-reference.md +++ b/docs/gateway/configuration-reference.md @@ -1676,11 +1676,14 @@ when preserving announce delivery. `openclaw doctor --fix` strips a leftover } ``` -`cron.failureAlert` owns both the alert threshold and the default failure -destination for every job. The retired `cron.failureDestination` block is merged -into it by [`openclaw doctor --fix`](/cli/doctor). +`cron.failureAlert` owns the global alert policy and its default destination. Jobs +with an existing failure route are covered by default after 2 consecutive +execution failures with a 1-hour cooldown; a `cron.failureAlert` object explicitly +activates/tunes the policy even when no route existed. The retired +`cron.failureDestination` block is merged into it by +[`openclaw doctor --fix`](/cli/doctor). -- `enabled`: enable failure alerts for automation jobs (default: `false`). +- `enabled`: explicitly enable or disable the global policy. `false` disables inherited notifications unless a job has its own `failureAlert` object; `true` explicitly enables globally. Omitting it preserves route-backed defaults. - `after`: consecutive failures before an alert fires (positive integer, min: `1`; default: `2`). - `cooldownMs`: minimum milliseconds between repeated alerts for the same job (non-negative integer; default: `3600000`). - `includeSkipped`: count consecutive skipped runs toward the alert threshold (default: `false`). Skipped runs are tracked separately and do not affect execution-error backoff. @@ -1688,8 +1691,10 @@ into it by [`openclaw doctor --fix`](/cli/doctor). - `channel`: channel override for announce delivery. `"last"` reuses the last known delivery channel. - `to`: explicit announce target or webhook URL. Required for webhook mode. - `accountId`: optional account or channel id to scope alert delivery. -- Per-job `delivery.failureDestination` overrides these global destination fields. -- When neither global nor per-job failure destination is set, jobs that already deliver via `announce` fall back to that primary announce target on failure. +- Route precedence is per-job `failureAlert` route fields, then per-job `delivery.failureDestination` layered over these global destination fields, then the primary announce target. +- Per-job `failureAlert: false` disables execution and required-delivery failure alerts for that job; the auto-disable safety notification remains active. Any per-job `failureAlert` object explicitly enables and tunes that job. +- `delivery.bestEffort: true` suppresses inherited/default execution alerts; an explicit per-job `failureAlert` remains authoritative. +- Required completion-delivery failure (`status: "ok"`, `completionStatus: "failed"`) does not increment execution backoff and may notify immediately only through a resolved alternate failure destination, not the failed primary route. - `delivery.failureDestination` is only supported for `sessionTarget="isolated"` jobs unless the job's primary `delivery.mode` is `"webhook"`. See [Automations](/automation/cron-jobs). Isolated automation runs are tracked as [background tasks](/automation/tasks). diff --git a/src/agents/tools/cron-tool-schema.ts b/src/agents/tools/cron-tool-schema.ts index 9245fc1ed602..a1389a156fcf 100644 --- a/src/agents/tools/cron-tool-schema.ts +++ b/src/agents/tools/cron-tool-schema.ts @@ -238,7 +238,8 @@ function createCronDeliverySchema(): TSchema { accountId: deliveryStringSchema("Delivery account"), failureDestination: Type.Optional( Type.Union([failureDestinationObject, Type.Null()], { - description: "Failure destination; null clears.", + description: + "Failure-alert route override and alternate for immediate required-delivery failure; null clears.", }), ), completionDestination: Type.Optional( @@ -252,8 +253,8 @@ function createCronDeliverySchema(): TSchema { ); } -// Omitting `failureAlert` means "leave defaults/unchanged"; `false` explicitly disables alerts. -// Runtime handles `failureAlert === false` in cron/service/timer.ts. +// Omitting `failureAlert` means "leave defaults/unchanged"; `false` disables regular alerts. +// Runtime handles `failureAlert === false` in cron/service/failure-alerts.ts. // The schema declares `type: "object"` to stay compatible with providers that // enforce an OpenAPI 3.0 subset (e.g. Gemini via GitHub Copilot). The // description tells the LLM that `false` is also accepted. @@ -271,7 +272,8 @@ function createCronFailureAlertSchema(): TSchema { accountId: Type.Optional(Type.String()), }, additionalProperties: true, - description: "Failure alert; false disables.", + description: + "Failure alert policy/route override. Route-backed jobs default to after=2 and cooldownMs=3600000; false disables execution/delivery alerts but not the auto-disable safety notice.", }), ); } diff --git a/src/agents/tools/cron-tool.ts b/src/agents/tools/cron-tool.ts index 6009a72eb688..12085cc68a7c 100644 --- a/src/agents/tools/cron-tool.ts +++ b/src/agents/tools/cron-tool.ts @@ -206,6 +206,8 @@ ${triggerSection} DELIVERY {mode:"none"|"announce"|"webhook",channel?,to?,threadId?,bestEffort?,completionDestination?}: where detached run output goes. Omitted=announce (current=>this chat; isolated=>last route; set channel/to for a specific chat — no messaging tool inside the run).${silentWatcherCue} webhook posts finished-run event to URL in \`to\`. To keep announce delivery and also POST completion, use mode:"announce" with completionDestination:{mode:"webhook",to:"https://..."}. +FAILURE ALERTS: jobs with a failure route default to alerting after 2 consecutive execution failures with a 1h cooldown. Route order: job failureAlert fields, delivery.failureDestination over global cron.failureAlert destination fields, then primary announce. failureAlert:false disables execution/delivery alerts, not the auto-disable safety notice; a failureAlert object activates/tunes. bestEffort suppresses inherited execution alerts. Required completion-delivery failure uses only an alternate route immediately and does not increment the execution streak. + Job wakeMode (main jobs): "now"(default)|"next-heartbeat". Restricted automation-run sessions: self status/list/get/runs/remove + own next_check only. failureAlert {...}|false disables. jobId canonical (id=compat). contextMessages 0-10 embeds recent chat lines into reminder text.`; } diff --git a/src/cron/cron-delivery-outcomes.e2e.test.ts b/src/cron/cron-delivery-outcomes.e2e.test.ts index e847655c9b5b..3da50362446e 100644 --- a/src/cron/cron-delivery-outcomes.e2e.test.ts +++ b/src/cron/cron-delivery-outcomes.e2e.test.ts @@ -6,6 +6,7 @@ import { sendGatewayCronFailureAlert, sendGatewayCronWebhook, } from "../gateway/server-cron-notifications.js"; +import { getActiveGatewayRootWorkCount } from "../process/gateway-work-admission.js"; import { resetTaskRegistryForTests } from "../tasks/task-runtime.test-helpers.js"; import { withOpenClawTestState } from "../test-utils/openclaw-test-state.js"; import { runCronCommandJob } from "./command-runner.js"; @@ -24,9 +25,11 @@ type WebhookRequest = { async function createWebhookReceiver(): Promise<{ close: () => Promise; + requests: WebhookRequest[]; request: Promise; url: string; }> { + const requests: WebhookRequest[] = []; let resolveRequest!: (request: WebhookRequest) => void; const request = new Promise((resolve) => { resolveRequest = resolve; @@ -38,10 +41,12 @@ async function createWebhookReceiver(): Promise<{ body += chunk; }); incoming.on("end", () => { - resolveRequest({ + const received = { body: JSON.parse(body) as Record, path: incoming.url ?? "", - }); + }; + requests.push(received); + resolveRequest(received); response.writeHead(204, { Connection: "close" }); response.end(); }); @@ -53,6 +58,7 @@ async function createWebhookReceiver(): Promise<{ const address = server.address() as AddressInfo; return { request, + requests, url: `http://127.0.0.1:${address.port}/cron`, close: async () => { server.closeAllConnections(); @@ -154,7 +160,7 @@ describe.sequential("cron delivery outcomes", () => { } }); - it("dispatches a failed run to its real failure webhook and keeps durable error state", async () => { + it("applies threshold and cooldown once before transporting execution failure alerts", async () => { const receiver = await createWebhookReceiver(); try { await withOpenClawTestState( @@ -168,8 +174,18 @@ describe.sequential("cron delivery outcomes", () => { log: createNoopLogger(), enqueueSystemEvent: vi.fn(), requestHeartbeat: vi.fn(), - runCommandJob: commandRunner(), - runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })), + runIsolatedAgentJob: vi.fn(async () => ({ + status: "error" as const, + error: "monitor failed", + })), + sendCronFailureAlert: async (params) => + await sendGatewayCronFailureAlert({ + ...params, + deps: {} as never, + logger: createNoopLogger(), + resolveCronAgent: () => ({ agentId: "main", cfg: {} as never }), + ssrfPolicy: { allowedHostnames: ["127.0.0.1"] }, + }), onEvent: (event) => { if (event.action !== "finished") { return; @@ -187,46 +203,79 @@ describe.sequential("cron delivery outcomes", () => { try { await cron.start(); const job = await cron.add({ - name: "failure destination delivery", + name: "60-second monitor", enabled: true, schedule: { kind: "every", everyMs: 60_000 }, sessionTarget: "isolated", wakeMode: "next-heartbeat", - payload: { - kind: "command", - argv: [ - process.execPath, - "-e", - "process.stderr.write('DELIVERY_FAILURE'); process.exit(2)", - ], - }, + payload: { kind: "agentTurn", message: "check health" }, delivery: { mode: "none", failureDestination: { mode: "webhook", to: receiver.url }, }, }); - await expect(cron.run(job.id, "force")).resolves.toEqual({ ok: true, ran: true }); - expect(await receiver.request).toMatchObject({ + for (let index = 0; index < 21; index += 1) { + await expect(cron.run(job.id, "force")).resolves.toEqual({ ok: true, ran: true }); + } + const disabled = await cron.add({ + name: "disabled failure alert", + enabled: true, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: "check health quietly" }, + delivery: { + mode: "none", + failureDestination: { mode: "webhook", to: receiver.url }, + }, + failureAlert: false, + }); + for (let index = 0; index < 2; index += 1) { + await expect(cron.run(disabled.id, "force")).resolves.toEqual({ + ok: true, + ran: true, + }); + } + + await receiver.request; + await vi.waitFor(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + expect(receiver.requests).toHaveLength(1); + expect(receiver.requests[0]).toMatchObject({ path: "/cron", body: { jobId: job.id, - jobName: "failure destination delivery", - status: "error", + jobName: "60-second monitor", message: - 'Automation "failure destination delivery" failed: command exited with code 2', + 'Automation "60-second monitor" failed 2 times\nLast error: monitor failed', }, }); expect(await persistedJob(storePath, job.id)).toMatchObject({ state: { lastRunStatus: "error", - lastError: "command exited with code 2", + lastError: "monitor failed", + consecutiveErrors: 21, + lastFailureAlertAtMs: expect.any(Number), + lastFailureNotificationDeliveryStatus: "not-requested", }, }); - expect(historyEntry(storePath, job.id)).toMatchObject({ - status: "error", - error: "command exited with code 2", + const history = readCronTaskRunHistoryPage({ + storeKey: cronStoreKey(storePath), + jobId: job.id, + limit: 25, }); + expect(history.total).toBe(21); + expect(history.entries).toHaveLength(21); + expect(history.entries).toEqual( + expect.arrayContaining([ + expect.objectContaining({ status: "error", error: "monitor failed" }), + ]), + ); + expect( + history.entries.filter( + (entry) => entry.failureNotificationDelivery?.status === "unknown", + ), + ).toHaveLength(1); } finally { cron.stop(); resetTaskRegistryForTests({ persist: false }); @@ -238,6 +287,168 @@ describe.sequential("cron delivery outcomes", () => { } }); + it("routes required completion-delivery failure immediately without changing execution streak", async () => { + const receiver = await createWebhookReceiver(); + try { + await withOpenClawTestState( + { layout: "state-only", prefix: "openclaw-cron-completion-failure-" }, + async (state) => { + resetTaskRegistryForTests({ persist: false }); + const storePath = state.path("cron", "jobs.json"); + const cron = new CronService({ + storePath, + cronEnabled: true, + log: createNoopLogger(), + enqueueSystemEvent: vi.fn(), + requestHeartbeat: vi.fn(), + runIsolatedAgentJob: vi.fn(async () => ({ + status: "ok" as const, + delivered: false, + deliveryAttempted: true, + deliveryError: "primary route rejected", + })), + sendCronFailureAlert: async (params) => + await sendGatewayCronFailureAlert({ + ...params, + deps: {} as never, + logger: createNoopLogger(), + resolveCronAgent: () => ({ agentId: "main", cfg: {} as never }), + ssrfPolicy: { allowedHostnames: ["127.0.0.1"] }, + }), + }); + try { + await cron.start(); + const job = await cron.add({ + name: "required completion delivery", + enabled: true, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: "build report" }, + delivery: { + mode: "announce", + bestEffort: false, + channel: "telegram", + to: "123", + failureDestination: { mode: "webhook", to: receiver.url }, + }, + }); + + await expect(cron.run(job.id, "force")).resolves.toEqual({ ok: true, ran: true }); + expect(await receiver.request).toMatchObject({ + path: "/cron", + body: { + jobId: job.id, + message: + 'Automation "required completion delivery" delivery failed\nLast error: primary route rejected', + }, + }); + expect(await persistedJob(storePath, job.id)).toMatchObject({ + state: { + lastRunStatus: "ok", + lastDeliveryStatus: "not-delivered", + lastDeliveryError: "primary route rejected", + consecutiveErrors: 0, + lastFailureNotificationDeliveryStatus: "unknown", + }, + }); + + const disabled = await cron.add({ + name: "disabled completion failure alert", + enabled: true, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: "build report quietly" }, + delivery: { + mode: "announce", + bestEffort: false, + channel: "telegram", + to: "123", + failureDestination: { mode: "webhook", to: receiver.url }, + }, + failureAlert: false, + }); + await cron.run(disabled.id, "force"); + await vi.waitFor(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); + expect(receiver.requests).toHaveLength(1); + } finally { + cron.stop(); + resetTaskRegistryForTests({ persist: false }); + } + }, + ); + } finally { + await receiver.close(); + } + }); + + it("falls back to the exact job owner when Gateway alert transport rejects", async () => { + await withOpenClawTestState( + { layout: "state-only", prefix: "openclaw-cron-alert-fallback-" }, + async (state) => { + const enqueueSystemEvent = vi.fn(); + const requestHeartbeat = vi.fn(); + const cron = new CronService({ + storePath: state.path("cron", "jobs.json"), + cronEnabled: true, + log: createNoopLogger(), + enqueueSystemEvent, + requestHeartbeat, + runIsolatedAgentJob: vi.fn(async () => ({ + status: "error" as const, + error: "provider unavailable", + })), + sendCronFailureAlert: async (params) => + await sendGatewayCronFailureAlert({ + ...params, + deps: {} as never, + logger: createNoopLogger(), + resolveCronAgent: () => ({ agentId: "work", cfg: {} as never }), + }), + }); + try { + await cron.start(); + const sessionKey = "agent:work:cron:failure-fallback"; + const job = await cron.add({ + name: "fallback owner", + enabled: true, + agentId: "work", + sessionKey, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "isolated", + wakeMode: "now", + payload: { kind: "agentTurn", message: "check provider" }, + delivery: { mode: "none" }, + failureAlert: { + after: 1, + mode: "webhook", + to: "http://127.0.0.1:9/failure", + }, + }); + + await cron.run(job.id, "force"); + + await vi.waitFor(() => + expect(enqueueSystemEvent).toHaveBeenCalledWith( + expect.stringContaining('Automation "fallback owner" failed 1 times'), + { agentId: "work", sessionKey }, + ), + ); + expect(requestHeartbeat).toHaveBeenCalledWith({ + source: "cron", + intent: "immediate", + reason: `cron:${job.id}:failure-alert`, + agentId: "work", + sessionKey, + }); + } finally { + cron.stop(); + } + }, + ); + }); + it("sends skipped-run alerts through the real webhook path and persists alert state", async () => { const receiver = await createWebhookReceiver(); try { diff --git a/src/cron/delivery-plan.ts b/src/cron/delivery-plan.ts index 5a075bf64ffd..8dcccec789ff 100644 --- a/src/cron/delivery-plan.ts +++ b/src/cron/delivery-plan.ts @@ -59,7 +59,9 @@ function resolveAnnounceChannel(params: { } /** Resolves primary delivery config into the runtime mode/channel/target plan. */ -export function resolveCronDeliveryPlan(job: CronJob): CronDeliveryPlan { +export function resolveCronDeliveryPlan( + job: Pick & Partial>, +): CronDeliveryPlan { const delivery = job.delivery; const hasDelivery = delivery && typeof delivery === "object"; const rawMode = hasDelivery ? (delivery as { mode?: unknown }).mode : undefined; @@ -107,12 +109,15 @@ export function resolveCronDeliveryPlan(job: CronJob): CronDeliveryPlan { // Isolated/current/session output jobs default to announce delivery so their // result reaches the initiating session unless the job opts out. Keep this // aligned with create-time normalization and direct service callers. - const resolvedMode = shouldDefaultCronDeliveryToAnnounce({ - payloadKind: job.payload.kind, - sessionTarget: job.sessionTarget, - }) - ? "announce" - : "none"; + const resolvedMode = + job.payload && + job.sessionTarget && + shouldDefaultCronDeliveryToAnnounce({ + payloadKind: job.payload.kind, + sessionTarget: job.sessionTarget, + }) + ? "announce" + : "none"; return { mode: resolvedMode, @@ -150,12 +155,12 @@ function normalizeFailureMode(value: unknown): "announce" | "webhook" | undefine /** Resolves job-level failure notification routing layered over global defaults. */ export function resolveFailureDestination( - job: CronJob, + job: Pick, globalConfig?: CronFailureDestinationConfig, + jobAlertRoute?: CronFailureDestinationInput, ): CronFailureDeliveryPlan | null { const delivery = job.delivery; const jobFailureDest = delivery?.failureDestination as CronFailureDestinationInput | undefined; - const hasJobFailureDest = jobFailureDest && typeof jobFailureDest === "object"; let channel: CronMessageChannel | undefined; let to: string | undefined; @@ -169,61 +174,76 @@ export function resolveFailureDestination( mode = normalizeFailureMode(globalConfig.mode); } - if (hasJobFailureDest) { - const jobTo = normalizeOptionalString(jobFailureDest.to); - const explicitJobChannel = normalizeChannel(jobFailureDest.channel); - const jobChannel = - explicitJobChannel ?? - (jobTo ? (resolveTargetPrefixedChannel(jobTo) as CronMessageChannel | undefined) : undefined); - const jobAccountId = normalizeOptionalString(jobFailureDest.accountId); - const jobMode = normalizeFailureMode(jobFailureDest.mode); - const hasJobChannelField = "channel" in jobFailureDest; - const hasJobToField = "to" in jobFailureDest; - const hasJobAccountIdField = "accountId" in jobFailureDest; - const hasJobModeField = "mode" in jobFailureDest; + // Apply the delivery override first, then the job's failureAlert route. This + // is the canonical route layering used by mutation validation and finalization. + for (const routeOverride of [jobFailureDest, jobAlertRoute]) { + if (!routeOverride || typeof routeOverride !== "object") { + continue; + } + const overrideTo = normalizeOptionalString(routeOverride.to); + const explicitOverrideChannel = normalizeChannel(routeOverride.channel); + const overrideChannel = + explicitOverrideChannel ?? + (overrideTo + ? (resolveTargetPrefixedChannel(overrideTo) as CronMessageChannel | undefined) + : undefined); + const overrideAccountId = normalizeOptionalString(routeOverride.accountId); + const overrideMode = normalizeFailureMode(routeOverride.mode); + const hasChannelField = "channel" in routeOverride; + const hasToField = "to" in routeOverride; + const hasAccountIdField = "accountId" in routeOverride; + const hasModeField = "mode" in routeOverride; - const jobToExplicitValue = hasJobToField && jobTo !== undefined; + const hasExplicitTo = hasToField && overrideTo !== undefined; const globalChannel = resolveAnnounceChannel({ channel, to }); - if (hasJobChannelField || (jobChannel && jobTo)) { - channel = jobChannel; - if (jobChannel && jobChannel !== globalChannel) { + if (hasChannelField || (overrideChannel && overrideTo)) { + channel = overrideChannel; + if (overrideChannel && overrideChannel !== globalChannel) { // Targets and accounts belong to the channel that supplied them. - if (!hasJobToField) { + if (!hasToField) { to = undefined; } - if (!hasJobAccountIdField) { + if (!hasAccountIdField) { accountId = undefined; } } } - if (hasJobToField) { - to = jobTo; + if (hasToField) { + to = overrideTo; } - if (hasJobAccountIdField) { - accountId = jobAccountId; + if (hasAccountIdField) { + accountId = overrideAccountId; } // Naming a channel makes this an announce route even when mode is omitted; // inheriting webhook here would reinterpret the chat target as a URL. - const jobImpliesAnnounce = !hasJobModeField && jobChannel !== undefined; - if (hasJobModeField || jobImpliesAnnounce) { - const effectiveJobMode = jobImpliesAnnounce ? "announce" : jobMode; + const overrideImpliesAnnounce = !hasModeField && overrideChannel !== undefined; + if (hasModeField || overrideImpliesAnnounce) { + const effectiveOverrideMode = overrideImpliesAnnounce ? "announce" : overrideMode; const globalMode = mode ?? "announce"; - const resolvedJobMode = effectiveJobMode ?? "announce"; - if (globalMode !== resolvedJobMode) { + const resolvedOverrideMode = effectiveOverrideMode ?? "announce"; + if (globalMode !== resolvedOverrideMode) { // Chat targets and accounts cannot be reused as webhook routing, or vice versa. - if (!jobToExplicitValue) { + if (!hasChannelField) { + channel = undefined; + } + if (!hasExplicitTo) { to = undefined; } - if (!hasJobAccountIdField) { + if (!hasAccountIdField) { accountId = undefined; } } - mode = effectiveJobMode; + mode = effectiveOverrideMode; } } - if (!channel && !to && !accountId && !mode) { + const jobAlertOnlySelectsMode = + jobAlertRoute?.mode !== undefined && + jobAlertRoute.channel === undefined && + jobAlertRoute.to === undefined && + jobAlertRoute.accountId === undefined; + if (!channel && !to && !accountId && (!mode || jobAlertOnlySelectsMode)) { return null; } diff --git a/src/cron/delivery.failure-notify.test.ts b/src/cron/delivery.failure-notify.test.ts index 3ec44d532844..541766fbcb75 100644 --- a/src/cron/delivery.failure-notify.test.ts +++ b/src/cron/delivery.failure-notify.test.ts @@ -1,5 +1,5 @@ -// Delivery failure notification tests cover alerts emitted after delivery failures. -import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +// Strict cron announcement transport tests cover scheduler-authorized alert delivery. +import { beforeEach, describe, expect, it, vi } from "vitest"; const mocks = vi.hoisted(() => ({ resolveDeliveryTarget: vi.fn(), @@ -7,65 +7,28 @@ const mocks = vi.hoisted(() => ({ resolveAgentOutboundIdentity: vi.fn().mockReturnValue({ kind: "identity" }), buildOutboundSessionContext: vi.fn().mockReturnValue({ kind: "session" }), createOutboundSendDeps: vi.fn().mockReturnValue({ kind: "deps" }), - warn: vi.fn(), })); vi.mock("./isolated-agent/delivery-target.js", () => ({ resolveDeliveryTarget: mocks.resolveDeliveryTarget, })); - vi.mock("../infra/outbound/deliver.js", () => ({ deliverOutboundPayloads: mocks.deliverOutboundPayloads, deliverOutboundPayloadsInternal: mocks.deliverOutboundPayloads, })); - vi.mock("../infra/outbound/identity.js", () => ({ resolveAgentOutboundIdentity: mocks.resolveAgentOutboundIdentity, })); - vi.mock("../infra/outbound/session-context.js", () => ({ buildOutboundSessionContext: mocks.buildOutboundSessionContext, })); - vi.mock("../cli/outbound-send-deps.js", () => ({ createOutboundSendDeps: mocks.createOutboundSendDeps, })); -vi.mock("../logging.js", () => ({ - getChildLogger: vi.fn(() => ({ - warn: mocks.warn, - })), -})); +const { sendCronAnnouncePayloadStrict } = await import("./delivery.js"); -const { sendCronAnnouncePayloadStrict, sendFailureNotificationAnnounce } = - await import("./delivery.js"); - -type DeliveryRequest = { - abortSignal?: unknown; - accountId?: string; - bestEffort?: boolean; - cfg?: unknown; - channel?: string; - deps?: unknown; - identity?: unknown; - payloads?: unknown; - session?: unknown; - threadId?: number; - to?: string; -}; - -type WarnMeta = { channel?: string; err?: string; to?: string }; - -function firstDeliveryRequest() { - const [deliveryRequest] = mocks.deliverOutboundPayloads.mock.calls[0] as [DeliveryRequest]; - return deliveryRequest; -} - -function firstWarnCall() { - return mocks.warn.mock.calls[0] as [WarnMeta, string]; -} - -describe("sendFailureNotificationAnnounce", () => { +describe("sendCronAnnouncePayloadStrict", () => { beforeEach(() => { vi.clearAllMocks(); mocks.resolveDeliveryTarget.mockResolvedValue({ @@ -79,145 +42,41 @@ describe("sendFailureNotificationAnnounce", () => { mocks.deliverOutboundPayloads.mockResolvedValue([{ ok: true }]); }); - afterEach(() => { - vi.useRealTimers(); - }); - - it("delivers failure alerts to the resolved explicit target with strict send settings", async () => { - const deps = {} as never; - const cfg = {} as never; - - await sendFailureNotificationAnnounce( - deps, - cfg, - "main", - "job-1", - { channel: "telegram", to: "123", accountId: "bot-a" }, - { - text: "Cron failed", - presentation: { - blocks: [ - { - type: "buttons", - buttons: [ - { - label: "Log in to Codex", - action: { type: "command", command: "/login codex" }, - }, - ], - }, - ], - }, - }, - ); - - expect(mocks.resolveDeliveryTarget).toHaveBeenCalledWith( - cfg, - "main", - { - channel: "telegram", - to: "123", - accountId: "bot-a", - }, - undefined, - ); - expect(mocks.buildOutboundSessionContext).toHaveBeenCalledWith({ - cfg, + it("delivers the payload through the resolved target with strict send settings", async () => { + await sendCronAnnouncePayloadStrict({ + deps: {} as never, + cfg: {} as never, agentId: "main", - sessionKey: "cron:job-1:failure", + jobId: "job-1", + target: { channel: "telegram", to: "123", accountId: "bot-a" }, + payload: { text: "Automation failed" }, + abortSignal: new AbortController().signal, }); - expect(mocks.deliverOutboundPayloads).toHaveBeenCalledTimes(1); - const deliveryRequest = firstDeliveryRequest(); - expect(deliveryRequest.cfg).toBe(cfg); - expect(deliveryRequest.channel).toBe("telegram"); - expect(deliveryRequest.to).toBe("123"); - expect(deliveryRequest.accountId).toBe("bot-a"); - expect(deliveryRequest.threadId).toBe(42); - expect(deliveryRequest.payloads).toEqual([ - { - text: "Cron failed", - presentation: { - blocks: [ - { - type: "buttons", - buttons: [ - { - label: "Log in to Codex", - action: { type: "command", command: "/login codex" }, - }, - ], - }, - ], - }, - }, - ]); - expect(deliveryRequest.session).toEqual({ kind: "session" }); - expect(deliveryRequest.identity).toEqual({ kind: "identity" }); - expect(deliveryRequest.bestEffort).toBe(false); - expect(deliveryRequest.deps).toEqual({ kind: "deps" }); - expect(deliveryRequest.abortSignal).toBeInstanceOf(AbortSignal); - }); - - it("uses sessionKey for delivery-target resolution and outbound context", async () => { - await sendFailureNotificationAnnounce( - {} as never, - {} as never, - "main", - "job-1", - { - channel: "telegram", - sessionKey: "agent:main:telegram:direct:123:thread:99", - }, - { text: "Cron failed" }, - ); expect(mocks.resolveDeliveryTarget).toHaveBeenCalledWith( - {} as never, + {}, "main", - { - channel: "telegram", - to: undefined, - accountId: undefined, - sessionKey: "agent:main:telegram:direct:123:thread:99", - }, + { channel: "telegram", to: "123", accountId: "bot-a" }, undefined, ); expect(mocks.buildOutboundSessionContext).toHaveBeenCalledWith({ cfg: {}, agentId: "main", - sessionKey: "agent:main:telegram:direct:123:thread:99", + sessionKey: "cron:job-1:failure", }); - }); - - it("can suppress session-thread inheritance for explicit failure destinations", async () => { - await sendFailureNotificationAnnounce( - {} as never, - {} as never, - "main", - "job-1", - { + expect(mocks.deliverOutboundPayloads).toHaveBeenCalledWith( + expect.objectContaining({ channel: "telegram", - to: "-1001234567890", - sessionKey: "agent:main:telegram:group:-1001234567890:thread:42", - inheritSessionThread: false, - }, - { text: "Cron failed" }, - ); - - expect(mocks.resolveDeliveryTarget).toHaveBeenCalledWith( - {}, - "main", - { - channel: "telegram", - to: "-1001234567890", - accountId: undefined, - sessionKey: "agent:main:telegram:group:-1001234567890:thread:42", - }, - { inheritSessionThread: false }, + to: "123", + accountId: "bot-a", + threadId: 42, + payloads: [{ text: "Automation failed" }], + bestEffort: false, + }), ); }); - it("does not begin strict delivery when target resolution settles after cancellation", async () => { + it("does not begin delivery when target resolution settles after cancellation", async () => { let resolvePendingTarget: (value: unknown) => void = () => {}; mocks.resolveDeliveryTarget.mockImplementationOnce( () => @@ -226,14 +85,13 @@ describe("sendFailureNotificationAnnounce", () => { }), ); const abortController = new AbortController(); - const delivery = sendCronAnnouncePayloadStrict({ deps: {} as never, cfg: {} as never, agentId: "main", jobId: "job-1", target: { channel: "telegram", to: "123" }, - payload: { text: "Cron failed" }, + payload: { text: "Automation failed" }, abortSignal: abortController.signal, }); @@ -250,172 +108,34 @@ describe("sendFailureNotificationAnnounce", () => { expect(mocks.deliverOutboundPayloads).not.toHaveBeenCalled(); }); - it("does not send when target resolution fails", async () => { - mocks.resolveDeliveryTarget.mockResolvedValue({ - ok: false, - error: new Error("target missing"), - }); - - await sendFailureNotificationAnnounce( - {} as never, - {} as never, - "main", - "job-1", - { channel: "telegram", to: "123" }, - { text: "Cron failed" }, - ); - - expect(mocks.deliverOutboundPayloads).not.toHaveBeenCalled(); - expect(mocks.warn).toHaveBeenCalledWith( - { error: "target missing" }, - "cron: failed to resolve failure destination target", - ); - }); - - 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" }, - { text: "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" }, - { text: "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" }, - { text: "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); + { + name: "target resolution", + arrange: () => + mocks.resolveDeliveryTarget.mockResolvedValueOnce({ + ok: false, + error: new Error("target unavailable"), + }), + error: "target unavailable", }, - ); - - it("swallows outbound delivery errors after logging", async () => { - mocks.deliverOutboundPayloads.mockRejectedValue(new Error("send failed")); + { + name: "channel delivery", + arrange: () => mocks.deliverOutboundPayloads.mockRejectedValueOnce(new Error("send failed")), + error: "send failed", + }, + ])("rejects $name failures", async ({ arrange, error }) => { + arrange(); await expect( - sendFailureNotificationAnnounce( - {} as never, - {} as never, - "main", - "job-1", - { channel: "telegram", to: "123" }, - { text: "Cron failed" }, - ), - ).resolves.toBeUndefined(); - - expect(mocks.warn).toHaveBeenCalledTimes(1); - const [warnMeta, warnMessage] = firstWarnCall(); - expect(warnMeta.err).toBe("send failed"); - expect(warnMeta.channel).toBe("telegram"); - expect(warnMeta.to).toBe("123"); - expect(warnMessage).toBe("cron: failure destination announce failed"); + sendCronAnnouncePayloadStrict({ + deps: {} as never, + cfg: {} as never, + agentId: "main", + jobId: "job-1", + target: { channel: "telegram", to: "123" }, + payload: { text: "Automation failed" }, + abortSignal: new AbortController().signal, + }), + ).rejects.toThrow(error); }); }); diff --git a/src/cron/delivery.test.ts b/src/cron/delivery.test.ts index 2a50217933c4..fb08a3f36018 100644 --- a/src/cron/delivery.test.ts +++ b/src/cron/delivery.test.ts @@ -393,19 +393,42 @@ describe("resolveFailureDestination", () => { }); }); - it("returns null for webhook mode without destination URL", () => { - const plan = resolveFailureDestination( - makeCronJob({ - delivery: { - mode: "announce", - channel: "telegram", - to: "111", - failureDestination: { mode: "webhook" }, - }, - }), - undefined, - ); - expect(plan).toBeNull(); + it.each([ + { + name: "explicit announce mode", + failureDestination: { mode: "announce" as const }, + globalConfig: undefined, + expected: { mode: "announce", channel: "last", to: undefined, accountId: undefined }, + }, + { + name: "webhook mode without a URL", + failureDestination: { mode: "webhook" as const }, + globalConfig: undefined, + expected: null, + }, + { + name: "clear-only override", + failureDestination: { + channel: undefined as never, + to: undefined as never, + accountId: undefined as never, + mode: undefined as never, + }, + globalConfig: { + channel: "signal", + to: "group-abc", + accountId: "global-account", + mode: "announce" as const, + }, + expected: null, + }, + ])("resolves $name", ({ failureDestination, globalConfig, expected }) => { + expect( + resolveFailureDestination( + makeCronJob({ delivery: { mode: "none", failureDestination } }), + globalConfig, + ), + ).toEqual(expected); }); it("returns null when failure destination matches primary delivery target", () => { @@ -514,36 +537,6 @@ describe("resolveFailureDestination", () => { expect(plan).toBeNull(); }); - it("allows job-level failure destination fields to clear inherited global values", () => { - const plan = resolveFailureDestination( - makeCronJob({ - delivery: { - mode: "announce", - channel: "telegram", - to: "111", - failureDestination: { - mode: "announce", - channel: undefined as never, - to: undefined as never, - accountId: undefined as never, - }, - }, - }), - { - channel: "signal", - to: "group-abc", - accountId: "global-account", - mode: "announce", - }, - ); - expect(plan).toEqual({ - mode: "announce", - channel: "last", - to: undefined, - accountId: undefined, - }); - }); - it("keeps inherited announce targets when a job clears only failure destination mode", () => { const plan = resolveFailureDestination( makeCronJob({ diff --git a/src/cron/delivery.ts b/src/cron/delivery.ts index 452102f06dfe..799c81557bdd 100644 --- a/src/cron/delivery.ts +++ b/src/cron/delivery.ts @@ -5,12 +5,9 @@ import { sendDurableMessageBatchCore } from "../channels/message/runtime.js"; 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"; -import { resolveFailureDestination, resolveCronDeliveryPlan } from "./delivery-plan.js"; +import { resolveCronDeliveryPlan } from "./delivery-plan.js"; import { resolveDeliveryTarget, type DeliveryTargetResolution, @@ -18,10 +15,7 @@ import { import { resolveCronNotificationSessionKey } from "./session-target.js"; import type { CronMessageChannel } from "./types.js"; -export { resolveCronDeliveryPlan, resolveFailureDestination }; - -const FAILURE_NOTIFICATION_TIMEOUT_MS = 30_000; -const cronDeliveryLogger = getChildLogger({ subsystem: "cron-delivery" }); +export { resolveCronDeliveryPlan }; /** Channel target metadata used for cron announcements and failure notifications. */ type CronAnnounceTarget = { @@ -144,62 +138,3 @@ export async function sendCronAnnouncePayloadStrict(params: { abortSignal: params.abortSignal, }); } - -/** Sends a best-effort cron failure notification, logging resolution/send failures. */ -export async function sendFailureNotificationAnnounce( - deps: CliDeps, - cfg: OpenClawConfig, - agentId: string, - jobId: string, - target: CronAnnounceTarget, - payload: ReplyPayload, -): Promise { - const abortController = new AbortController(); - let resolvedTarget: SuccessfulDeliveryTarget | undefined; - - try { - // 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, - payload, - 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: resolvedTarget?.channel ?? target.channel, - to: resolvedTarget?.to ?? target.to, - }, - "cron: failure destination announce failed", - ); - } -} diff --git a/src/cron/service.failure-alert.test.ts b/src/cron/service.failure-alert.test.ts index 39d0428f4baf..ee69a31e80e2 100644 --- a/src/cron/service.failure-alert.test.ts +++ b/src/cron/service.failure-alert.test.ts @@ -35,7 +35,7 @@ function createFailureAlertJob( async function withFailureAlertCron( params: { - failureAlert: FailureAlertConfig; + failureAlert?: FailureAlertConfig; runResult?: IsolatedAgentRunResult; useFallback?: boolean; }, @@ -58,7 +58,9 @@ async function withFailureAlertCron( const cron = new CronService({ storePath: store.storePath, cronEnabled: true, - cronConfig: { failureAlert: params.failureAlert }, + ...(params.failureAlert === undefined + ? {} + : { cronConfig: { failureAlert: params.failureAlert } }), log: noopLogger, enqueueSystemEvent, requestHeartbeat, @@ -121,6 +123,37 @@ function expectAlertTextContaining( } describe("CronService failure alerts", () => { + it("defaults route-backed jobs to two failures and a one-hour cooldown", async () => { + await withFailureAlertCron({}, async ({ cron, sendCronFailureAlert, addJob }) => { + const job = await addJob("default routed alert", { delivery: createTelegramDelivery() }); + + await cron.run(job.id, "force"); + expect(sendCronFailureAlert).not.toHaveBeenCalled(); + + await cron.run(job.id, "force"); + expect(sendCronFailureAlert).toHaveBeenCalledOnce(); + expectAlertFields(sendCronFailureAlert, { channel: "telegram", to: "19098680" }); + + vi.advanceTimersByTime(60 * 60_000 - 1); + await cron.run(job.id, "force"); + expect(sendCronFailureAlert).toHaveBeenCalledOnce(); + }); + }); + + it("activates policy when the global failureAlert object omits enabled", async () => { + await withFailureAlertCron( + { failureAlert: { after: 1 } }, + async ({ cron, sendCronFailureAlert, addJob }) => { + const job = await addJob("object-enabled alert", { delivery: { mode: "none" } }); + + await cron.run(job.id, "force"); + + expect(sendCronFailureAlert).toHaveBeenCalledOnce(); + expectAlertFields(sendCronFailureAlert, { channel: "last", mode: "announce" }); + }, + ); + }); + it("keeps fallback events and immediate wakes on the failing job owner", async () => { await withFailureAlertCron( { failureAlert: { enabled: true, after: 1 }, useFallback: true }, @@ -289,6 +322,15 @@ describe("CronService failure alerts", () => { bestEffort: true, }, }); + const explicitBestEffortJob = await addJob("explicit best effort alert job", { + delivery: { + mode: "announce", + channel: "telegram", + to: "19098680", + bestEffort: true, + }, + failureAlert: { after: 1, channel: "telegram", to: "19098680" }, + }); await cron.run(normalJob.id, "force"); expect(sendCronFailureAlert).toHaveBeenCalledTimes(1); @@ -300,6 +342,14 @@ describe("CronService failure alerts", () => { await cron.run(bestEffortJob.id, "force"); expect(sendCronFailureAlert).toHaveBeenCalledTimes(1); + + await cron.run(explicitBestEffortJob.id, "force"); + expect(sendCronFailureAlert).toHaveBeenCalledTimes(2); + expectAlertFields(sendCronFailureAlert, { + mode: "announce", + channel: "telegram", + to: "19098680", + }); }, ); }); @@ -375,7 +425,7 @@ describe("CronService failure alerts", () => { }, }, { - name: "never reuses a global webhook URL as an overridden job chat target", + name: "falls back to the primary announce route instead of a global webhook URL", globalAlert: { enabled: true, after: 1, @@ -434,7 +484,7 @@ describe("CronService failure alerts", () => { }, }, { - name: "never reuses a global webhook channel after a job switches to chat", + name: "falls back to the primary announce route instead of global webhook fields", globalAlert: { enabled: true, after: 1, @@ -467,7 +517,7 @@ describe("CronService failure alerts", () => { }, }, { - name: "preserves a global webhook URL when an unused job channel is set", + name: "uses an explicit job channel instead of the global webhook route", globalAlert: { enabled: true, after: 1, @@ -478,8 +528,9 @@ describe("CronService failure alerts", () => { channel: "telegram", }, expected: { - mode: "webhook", - to: "https://alerts.example.test/global-failures", + mode: "announce", + channel: "telegram", + to: "telegram:19098680", }, }, ])("$name", async ({ globalAlert, jobAlert, expected }) => { @@ -517,16 +568,7 @@ describe("CronService failure alerts", () => { to: "https://alerts.example.test/job-failures", }, }, - { - name: "clear-only failure destination opt-out", - failureDestination: { - channel: undefined, - to: undefined, - accountId: undefined, - mode: undefined, - }, - }, - ])("does not duplicate an explicitly owned $name", async ({ failureDestination }) => { + ])("routes one scheduler alert through the $name", async ({ failureDestination }) => { await withFailureAlertCron( { failureAlert: { @@ -544,7 +586,11 @@ describe("CronService failure alerts", () => { expect(job.delivery?.failureDestination).toBeDefined(); await cron.run(job.id, "force"); - expect(sendCronFailureAlert).not.toHaveBeenCalled(); + expect(sendCronFailureAlert).toHaveBeenCalledOnce(); + expectAlertFields(sendCronFailureAlert, { + ...failureDestination, + inheritSessionThread: false, + }); }, ); }); @@ -615,8 +661,8 @@ describe("CronService failure alerts", () => { expect(sendCronFailureAlert).toHaveBeenCalledOnce(); expectAlertFields(sendCronFailureAlert, { mode: "announce", - channel: "telegram", - to: "telegram:19098680", + channel: "slack", + to: "#alerts", }); expectAlertTextContaining( sendCronFailureAlert, diff --git a/src/cron/service.persists-delivered-status.test.ts b/src/cron/service.persists-delivered-status.test.ts index 7ecfd2fb6b47..ca326fccdf83 100644 --- a/src/cron/service.persists-delivered-status.test.ts +++ b/src/cron/service.persists-delivered-status.test.ts @@ -94,21 +94,6 @@ function buildWebhookIsolatedAgentTurnJob(name: string): CronAddInput { }; } -function buildAnnounceWithFailureDestinationJob(name: string): CronAddInput { - return { - ...buildAnnounceIsolatedAgentTurnJob(name), - delivery: { - mode: "announce", - channel: "forum", - to: "123", - failureDestination: { - mode: "webhook", - to: "https://example.invalid/cron-failure", - }, - }, - }; -} - function buildFailureDestinationOnlyJob(name: string): CronAddInput { return { ...buildIsolatedAgentTurnJob(name), @@ -735,7 +720,7 @@ describe("CronService persists delivered status", () => { expect(updated?.state.lastFailureNotificationDeliveryStatus).toBe("not-requested"); }); - it("keeps failure notification delivery separate from successful result delivery", async () => { + it("does not infer scheduler alert delivery from a failed run result", async () => { let capturedEvent: | { delivered?: boolean; @@ -761,77 +746,32 @@ describe("CronService persists delivered status", () => { expect(updated?.state.lastDelivered).toBe(false); expect(updated?.state.lastDeliveryStatus).toBe("not-delivered"); expect(updated?.state.lastDeliveryError).toBe("Agent couldn't generate a response."); - expect(updated?.state.lastFailureNotificationDelivered).toBe(true); - expect(updated?.state.lastFailureNotificationDeliveryStatus).toBe("delivered"); + expect(updated?.state.lastFailureNotificationDelivered).toBeUndefined(); + expect(updated?.state.lastFailureNotificationDeliveryStatus).toBe("not-requested"); expect(updated?.state.lastFailureNotificationDeliveryError).toBeUndefined(); expect(capturedEvent?.delivered).toBe(false); expect(capturedEvent?.deliveryStatus).toBe("not-delivered"); - expect(capturedEvent?.failureNotificationDelivery).toEqual({ - delivered: true, - status: "delivered", - }); + expect(capturedEvent?.failureNotificationDelivery).toBeUndefined(); }); - it("marks failure-destination-only error notification delivery unknown", async () => { - let capturedEvent: - | { - delivered?: boolean; - deliveryStatus?: string; - failureNotificationDelivery?: { - delivered?: boolean; - status: string; - error?: string; - }; - } - | undefined; + it("persists scheduler-authorized alert intent as delivery unknown", async () => { + let failureNotificationDelivery: { delivered?: boolean; status: string; error?: string }; + const job = buildAnnounceIsolatedAgentTurnJob("authorized-failure-notification"); + job.failureAlert = { after: 1 }; + const updated = await runIsolatedJobAndReadState({ - job: buildFailureDestinationOnlyJob("failure-destination-only"), + job, status: "error", - error: "Agent couldn't generate a response.", - onFinished: (evt) => { - capturedEvent = evt; + error: "provider unavailable", + onFinished: (event) => { + failureNotificationDelivery = event.failureNotificationDelivery!; }, }); - expect(updated?.state.lastRunStatus).toBe("error"); - expect(updated?.state.lastDelivered).toBeUndefined(); - expect(updated?.state.lastDeliveryStatus).toBe("not-requested"); expect(updated?.state.lastFailureNotificationDelivered).toBeUndefined(); expect(updated?.state.lastFailureNotificationDeliveryStatus).toBe("unknown"); - expect(capturedEvent?.delivered).toBeUndefined(); - expect(capturedEvent?.deliveryStatus).toBe("not-requested"); - expect(capturedEvent?.failureNotificationDelivery).toEqual({ status: "unknown" }); - }); - - it("does not treat primary error delivery as alternate failure-destination delivery", async () => { - let capturedEvent: - | { - delivered?: boolean; - deliveryStatus?: string; - failureNotificationDelivery?: { - delivered?: boolean; - status: string; - error?: string; - }; - } - | undefined; - const updated = await runIsolatedJobAndReadState({ - job: buildAnnounceWithFailureDestinationJob("announce-plus-failure-destination"), - status: "error", - delivered: true, - error: "Agent couldn't generate a response.", - onFinished: (evt) => { - capturedEvent = evt; - }, - }); - - expect(updated?.state.lastRunStatus).toBe("error"); - expect(updated?.state.lastDelivered).toBe(false); - expect(updated?.state.lastDeliveryStatus).toBe("not-delivered"); - expect(updated?.state.lastFailureNotificationDelivered).toBeUndefined(); - expect(updated?.state.lastFailureNotificationDeliveryStatus).toBe("unknown"); - expect(capturedEvent?.delivered).toBe(false); - expect(capturedEvent?.failureNotificationDelivery).toEqual({ status: "unknown" }); + expect(updated?.state.lastFailureNotificationDeliveryError).toBeUndefined(); + expect(failureNotificationDelivery!).toEqual({ status: "unknown" }); }); it("keeps best-effort failure destinations suppressed", async () => { diff --git a/src/cron/service/failure-alerts.persistence.test.ts b/src/cron/service/failure-alerts.persistence.test.ts index 934391e924d7..93d9debd81df 100644 --- a/src/cron/service/failure-alerts.persistence.test.ts +++ b/src/cron/service/failure-alerts.persistence.test.ts @@ -98,10 +98,9 @@ describe("cron failure alert persistence", () => { const alertDone = new Promise((resolve) => { resolveAlert = resolve; }); + let persistedStateAtSend: CronJob["state"] | undefined; const sendCronFailureAlert = vi.fn(async () => { - expect((await loadCronStore(store.storePath)).jobs[0]?.state.lastFailureAlertAtMs).toBe( - endedAt, - ); + persistedStateAtSend = (await loadCronStore(store.storePath)).jobs[0]?.state; order.push("persist"); order.push("alert"); resolveAlert?.(); @@ -122,6 +121,10 @@ describe("cron failure alert persistence", () => { await alertDone; expect(order).toEqual(["persist", "alert"]); + expect(persistedStateAtSend).toMatchObject({ + lastFailureAlertAtMs: endedAt, + lastFailureNotificationDeliveryStatus: "unknown", + }); expect(sendCronFailureAlert).toHaveBeenCalledOnce(); }); @@ -193,9 +196,10 @@ describe("cron failure alert persistence", () => { endedAt: firstAlertAt, }); expect(sendCronFailureAlert).toHaveBeenCalledOnce(); - expect((await loadCronStore(store.storePath)).jobs[0]?.state.lastFailureAlertAtMs).toBe( - firstAlertAt, - ); + expect((await loadCronStore(store.storePath)).jobs[0]?.state).toMatchObject({ + lastFailureAlertAtMs: firstAlertAt, + lastFailureNotificationDeliveryStatus: "unknown", + }); now += 30_000; const currentJob = state.store?.jobs[0]; @@ -213,7 +217,11 @@ describe("cron failure alert persistence", () => { expect(sendCronFailureAlert).toHaveBeenCalledOnce(); expect((await loadCronStore(store.storePath)).jobs[0]).toMatchObject({ - state: { consecutiveErrors: 2, lastFailureAlertAtMs: firstAlertAt }, + state: { + consecutiveErrors: 2, + lastFailureAlertAtMs: firstAlertAt, + lastFailureNotificationDeliveryStatus: "not-requested", + }, }); }); }); diff --git a/src/cron/service/failure-alerts.ts b/src/cron/service/failure-alerts.ts index 1850c4055641..cfbd79543920 100644 --- a/src/cron/service/failure-alerts.ts +++ b/src/cron/service/failure-alerts.ts @@ -10,6 +10,7 @@ import type { ReplyPayload } from "../../auto-reply/reply-payload.js"; import { normalizeAnyChannelId } from "../../channels/registry-normalize.js"; import { resolveTargetPrefixedChannel } from "../../infra/outbound/channel-target-prefix.js"; import { normalizeTargetForProvider } from "../../infra/outbound/target-normalization.js"; +import { resolveCronDeliveryPlan, resolveFailureDestination } from "../delivery-plan.js"; import { cronFailureDetailLines } from "../failure-notification-text.js"; import type { CronFailureNotificationDelivery, @@ -32,6 +33,7 @@ type ResolvedFailureAlert = { accountId?: string; threadId?: string | number; includeSkipped: boolean; + alternateRoute: boolean; }; /** Returns the last failure-notification delivery trace persisted on a cron job. */ @@ -98,53 +100,74 @@ export function resolveFailureAlert( if (job.failureAlert === false) { return null; } - if (!jobConfig && globalConfig?.enabled !== true) { + if (!jobConfig && globalConfig?.enabled === false) { return null; } - - const mode = jobConfig?.mode ?? globalConfig?.mode; - const inheritsGlobalMode = - !jobConfig?.mode || jobConfig.mode === (globalConfig?.mode ?? "announce"); - const jobTo = normalizeOptionalString(jobConfig?.to); - const jobChannel = resolveFailureAlertChannel(jobConfig?.channel, jobTo); - const configuredGlobalTo = inheritsGlobalMode - ? normalizeOptionalString(globalConfig?.to) + const hasJobRoute = Boolean( + jobConfig && + (jobConfig.channel !== undefined || + jobConfig.to !== undefined || + jobConfig.accountId !== undefined || + jobConfig.mode !== undefined), + ); + const alternateRoute = resolveFailureDestination( + job, + globalConfig, + hasJobRoute ? jobConfig : undefined, + ); + const primaryRoute = resolveCronDeliveryPlan(job); + const primaryAnnounceRoute = + primaryRoute.mode === "announce" && primaryRoute.requested ? primaryRoute : undefined; + const explicitlyConfigured = jobConfig !== undefined || globalConfig !== undefined; + if (!alternateRoute && !primaryAnnounceRoute && !explicitlyConfigured) { + return null; + } + const configuredMode = + jobConfig?.mode ?? (jobConfig?.channel ? "announce" : undefined) ?? globalConfig?.mode; + const route = + alternateRoute ?? + (configuredMode === "webhook" && explicitlyConfigured + ? { + mode: "webhook", + to: normalizeOptionalString(jobConfig?.to ?? globalConfig?.to), + accountId: normalizeOptionalString(jobConfig?.accountId ?? globalConfig?.accountId), + } + : primaryAnnounceRoute); + const mode = (route?.mode ?? configuredMode) === "webhook" ? "webhook" : "announce"; + const primaryChannel = primaryAnnounceRoute + ? (resolveFailureAlertChannel(primaryAnnounceRoute.channel, primaryAnnounceRoute.to) ?? "last") : undefined; - const globalChannel = inheritsGlobalMode - ? resolveFailureAlertChannel(globalConfig?.channel, configuredGlobalTo) - : undefined; - // Webhook destinations have no chat-channel identity. Announce destinations - // must stay on their original channel when a job overrides its route. - const inheritsGlobalRoute = - inheritsGlobalMode && (mode === "webhook" || !jobChannel || jobChannel === globalChannel); - const globalTo = inheritsGlobalRoute ? configuredGlobalTo : undefined; - const deliveryTo = normalizeOptionalString(job.delivery?.to); - const deliveryChannel = resolveFailureAlertChannel(job.delivery?.channel, deliveryTo); - const channel = jobChannel ?? globalChannel ?? deliveryChannel ?? "last"; - const inheritsDeliveryChannel = - channel === deliveryChannel || (channel === "last" && !deliveryChannel); - const compatibleDeliveryTo = inheritsDeliveryChannel ? deliveryTo : undefined; - const explicitTo = jobTo ?? globalTo; - const inheritsDeliveryRoute = - inheritsDeliveryChannel && - (explicitTo === undefined || - explicitTo === deliveryTo || - (deliveryTo !== undefined && - normalizeFailureAlertRecipient(channel, explicitTo) === - normalizeFailureAlertRecipient(channel, deliveryTo))); - const inheritedDeliveryAccountId = - mode !== "webhook" && inheritsDeliveryRoute ? job.delivery?.accountId : undefined; + const hasAnnounceRouteSelector = + jobConfig?.channel !== undefined || + jobConfig?.to !== undefined || + job.delivery?.failureDestination?.channel !== undefined || + job.delivery?.failureDestination?.to !== undefined || + globalConfig?.channel !== undefined || + globalConfig?.to !== undefined; + const channel = + mode === "announce" && !hasAnnounceRouteSelector && primaryChannel + ? primaryChannel + : (resolveFailureAlertChannel(route?.channel, route?.to) ?? "last"); + const routeUsesPrimaryChannel = + mode === "announce" && primaryAnnounceRoute !== undefined && channel === primaryChannel; + const to = + normalizeOptionalString(route?.to) ?? + (routeUsesPrimaryChannel ? primaryAnnounceRoute?.to : undefined); + const primaryRecipientMatches = + primaryAnnounceRoute !== undefined && + mode === "announce" && + channel === primaryChannel && + (to === primaryAnnounceRoute.to || + (to !== undefined && + primaryAnnounceRoute.to !== undefined && + normalizeFailureAlertRecipient(channel, to) === + normalizeFailureAlertRecipient(channel, primaryAnnounceRoute.to))); const accountId = - jobConfig?.accountId ?? - (inheritsGlobalRoute ? globalConfig?.accountId : undefined) ?? - inheritedDeliveryAccountId; - // A topic belongs to its channel, peer, and account; never attach the - // primary topic to an independently routed failure destination. - const inheritsDeliveryThread = - mode !== "webhook" && inheritsDeliveryRoute && accountId === job.delivery?.accountId; + normalizeOptionalString(route?.accountId) ?? + (primaryRecipientMatches ? primaryAnnounceRoute?.accountId : undefined); + const primaryRouteMatches = + primaryRecipientMatches && accountId === primaryAnnounceRoute?.accountId; - // Announce alerts inherit the job delivery target; webhook alerts require an - // explicit alert target so chat recipients are not reused as URLs. return { after: clampPositiveInt(jobConfig?.after ?? globalConfig?.after, DEFAULT_FAILURE_ALERT_AFTER), cooldownMs: clampNonNegativeInt( @@ -152,14 +175,71 @@ export function resolveFailureAlert( DEFAULT_FAILURE_ALERT_COOLDOWN_MS, ), channel, - to: mode === "webhook" ? explicitTo : (explicitTo ?? compatibleDeliveryTo), + to, mode, accountId, - threadId: inheritsDeliveryThread ? job.delivery?.threadId : undefined, + threadId: primaryRouteMatches ? primaryAnnounceRoute.threadId : undefined, includeSkipped: jobConfig?.includeSkipped ?? globalConfig?.includeSkipped ?? false, + alternateRoute: alternateRoute !== null && !primaryRouteMatches, }; } +function enqueueFailureAlertFallback(state: CronServiceState, job: CronJob, text: string): void { + enqueueCronSystemEvent(state, text, { + agentId: job.agentId, + sessionKey: job.sessionKey, + }); + if (job.wakeMode === "now") { + requestCronHeartbeat(state, { + intent: "immediate", + reason: `cron:${job.id}:failure-alert`, + agentId: job.agentId, + sessionKey: job.sessionKey, + }); + } +} + +function markFailureNotificationRequested(job: CronJob): void { + job.state.lastFailureNotificationDelivered = undefined; + job.state.lastFailureNotificationDeliveryStatus = "unknown"; + job.state.lastFailureNotificationDeliveryError = undefined; +} + +function transportFailureAlert( + state: CronServiceState, + params: { + job: CronJob; + payload: ReplyPayload; + runAtMs?: number; + route: ResolvedFailureAlert; + }, +): void { + const fallback = () => enqueueFailureAlertFallback(state, params.job, params.payload.text ?? ""); + if (!state.deps.sendCronFailureAlert) { + fallback(); + return; + } + void state.deps + .sendCronFailureAlert({ + job: params.job, + payload: params.payload, + runAtMs: params.runAtMs, + channel: params.route.channel, + to: params.route.to, + mode: params.route.mode, + accountId: params.route.accountId, + threadId: params.route.threadId, + ...(params.route.alternateRoute ? { inheritSessionThread: false as const } : {}), + }) + .catch((err: unknown) => { + state.deps.log.warn( + { jobId: params.job.id, err: String(err) }, + "cron: failure alert delivery failed", + ); + fallback(); + }); +} + function emitFailureAlert( state: CronServiceState, params: { @@ -169,11 +249,7 @@ function emitFailureAlert( failureNotificationDetail?: CronFailureNotificationDetail; runAtMs?: number; consecutiveErrors: number; - channel: CronMessageChannel; - to?: string; - mode?: "announce" | "webhook"; - accountId?: string; - threadId?: string | number; + route: ResolvedFailureAlert; status: "error" | "skipped"; }, ) { @@ -184,7 +260,7 @@ function emitFailureAlert( const statusVerb = params.status === "skipped" ? "skipped" : "failed"; const detailLabel = params.status === "skipped" ? "Skip reason" : "Last error"; const detailLines = - params.mode === "webhook" + params.route.mode === "webhook" ? [ ...(errorReason ? [`Cause: ${errorReason}`] : []), `${detailLabel}: ${truncateUtf16Safe(params.error?.trim() || "unknown reason", 200)}`, @@ -218,38 +294,49 @@ function emitFailureAlert( : {}), }; - if (state.deps.sendCronFailureAlert) { - void state.deps - .sendCronFailureAlert({ - job: params.job, - payload, - runAtMs: params.runAtMs, - channel: params.channel, - to: params.to, - mode: params.mode, - accountId: params.accountId, - threadId: params.threadId, - }) - .catch((err: unknown) => { - state.deps.log.warn( - { jobId: params.job.id, err: String(err) }, - "cron: failure alert delivery failed", - ); - }); + transportFailureAlert(state, { + job: params.job, + payload, + runAtMs: params.runAtMs, + route: params.route, + }); +} + +/** Emits a required-completion delivery failure only to an alternate route. */ +function maybeEmitDeliveryFailureAlert( + state: CronServiceState, + params: { + job: CronJob; + alertConfig: ResolvedFailureAlert | null; + error?: string; + runAtMs?: number; + deferredNotifications?: DeferredCronNotifications; + }, +): void { + if (!params.alertConfig?.alternateRoute) { return; } - - enqueueCronSystemEvent(state, payload.text ?? "", { - agentId: params.job.agentId, - sessionKey: params.job.sessionKey, - }); - if (params.job.wakeMode === "now") { - requestCronHeartbeat(state, { - intent: "immediate", - reason: `cron:${params.job.id}:failure-alert`, - agentId: params.job.agentId, - sessionKey: params.job.sessionKey, + markFailureNotificationRequested(params.job); + const job = structuredClone(params.job); + const safeJobName = job.name || job.id; + const detailLines = + params.alertConfig.mode === "webhook" + ? [`Last error: ${truncateUtf16Safe(params.error?.trim() || "unknown reason", 200)}`] + : cronFailureDetailLines(job.state.lastErrorReason); + const payload: ReplyPayload = { + text: [`Automation "${safeJobName}" delivery failed`, ...detailLines].join("\n"), + }; + const notify = () => + transportFailureAlert(state, { + job, + payload, + runAtMs: params.runAtMs, + route: params.alertConfig!, }); + if (params.deferredNotifications) { + params.deferredNotifications.push(notify); + } else { + notify(); } } @@ -274,15 +361,6 @@ export function maybeEmitFailureAlert( if (!alertConfig || params.consecutiveCount < alertConfig.after) { return; } - if ( - params.status === "error" && - !params.job.failureAlert && - params.job.delivery?.failureDestination - ) { - // Completion delivery owns explicit failure routes and clear-only opt-outs. - // Suppress failed-run duplicates without disabling global skipped alerts. - return; - } // Best-effort delivery suppresses inherited alert noise, not an independently // configured job alert that the operator explicitly requested. if (params.job.delivery?.bestEffort === true && !params.job.failureAlert) { @@ -297,6 +375,7 @@ export function maybeEmitFailureAlert( if (inCooldown) { return; } + markFailureNotificationRequested(params.job); params.job.state.lastFailureAlertAtMs = now; if (params.delivery === "record-only") { return; @@ -311,11 +390,7 @@ export function maybeEmitFailureAlert( failureNotificationDetail: params.failureNotificationDetail, runAtMs: params.runAtMs, consecutiveErrors: params.consecutiveCount, - channel: alertConfig.channel, - to: alertConfig.to, - mode: alertConfig.mode, - accountId: alertConfig.accountId, - threadId: alertConfig.threadId, + route: alertConfig, status: params.status, }); if (params.deferredNotifications) { @@ -324,3 +399,48 @@ export function maybeEmitFailureAlert( notify(); } } + +/** Finalizes execution or required-delivery alerts after scheduling policy settles. */ +export function finalizeCronFailureNotifications( + state: CronServiceState, + params: { + job: CronJob; + alertConfig: ResolvedFailureAlert | null; + result: { + status: "ok" | "error" | "skipped"; + error?: string; + deliveryError?: string; + failureNotificationDetail?: CronFailureNotificationDetail; + startedAt: number; + }; + completionFailed: boolean; + autoDisableNotificationOwnsFailure: boolean; + replayFailureAlertAtMs?: number; + deferredNotifications?: DeferredCronNotifications; + }, +): void { + if (params.result.status === "error" && !params.autoDisableNotificationOwnsFailure) { + maybeEmitFailureAlert(state, { + job: params.job, + alertConfig: params.alertConfig, + status: "error", + error: params.result.error, + errorReason: params.job.state.lastErrorReason, + failureNotificationDetail: params.result.failureNotificationDetail, + runAtMs: params.result.startedAt, + consecutiveCount: params.job.state.consecutiveErrors ?? 0, + ...(params.replayFailureAlertAtMs !== undefined + ? { delivery: "record-only" as const, occurredAtMs: params.replayFailureAlertAtMs } + : {}), + deferredNotifications: params.deferredNotifications, + }); + } else if (params.result.status === "ok" && params.completionFailed) { + maybeEmitDeliveryFailureAlert(state, { + job: params.job, + alertConfig: params.alertConfig, + error: params.result.deliveryError, + runAtMs: params.result.startedAt, + deferredNotifications: params.deferredNotifications, + }); + } +} diff --git a/src/cron/service/ops.test.ts b/src/cron/service/ops.test.ts index a87aa4453f6d..ab3770ebc744 100644 --- a/src/cron/service/ops.test.ts +++ b/src/cron/service/ops.test.ts @@ -1501,6 +1501,9 @@ describe("cron service ops seam coverage", () => { const persisted = await loadCronStore(storePath); expect(persisted.jobs[0]?.state.lastFailureAlertAtMs).toBe(endedAt); expect(persisted.jobs[0]?.state.consecutiveErrors).toBe(1); + expect(persisted.jobs[0]?.state.lastFailureNotificationDelivered).toBeUndefined(); + expect(persisted.jobs[0]?.state.lastFailureNotificationDeliveryStatus).toBe("unknown"); + expect(persisted.jobs[0]?.state.lastFailureNotificationDeliveryError).toBeUndefined(); expect(sendCronFailureAlert).not.toHaveBeenCalled(); stop(state); }); diff --git a/src/cron/service/run-recovery.test.ts b/src/cron/service/run-recovery.test.ts index 932c9d9f636d..b59614bc516c 100644 --- a/src/cron/service/run-recovery.test.ts +++ b/src/cron/service/run-recovery.test.ts @@ -14,6 +14,7 @@ import { saveCronJobsStoreWithTransactionHooks } from "../store/transaction-hook import type { CronJob } from "../types.js"; import { proposeCronRunRecovery, recoverCronRunProposal } from "./run-recovery.js"; import { createCronServiceState } from "./state.js"; +import { runPostPersistCronNotifications } from "./store.js"; import { tryCreateCronTaskRun, tryFinishCronTaskRunWithoutHistory } from "./task-runs.js"; const { logger, makeStorePath } = setupCronServiceSuite({ prefix: "cron-run-recovery-" }); @@ -34,7 +35,14 @@ function makeJob(id: string, startedAtMs: number): CronJob { }; } -function makeState(storePath: string, nowMs: number) { +type RecoveryStateOverrides = Partial< + Pick< + Parameters[0], + "cronConfig" | "enqueueSystemEvent" | "requestHeartbeat" | "sendCronFailureAlert" + > +>; + +function makeState(storePath: string, nowMs: number, overrides: RecoveryStateOverrides = {}) { return createCronServiceState({ storePath, cronEnabled: true, @@ -43,6 +51,7 @@ function makeState(storePath: string, nowMs: number) { enqueueSystemEvent: vi.fn(), requestHeartbeat: vi.fn(), runIsolatedAgentJob: vi.fn(async () => ({ status: "ok" as const })), + ...overrides, }); } @@ -132,6 +141,108 @@ describe("atomic cron run recovery", () => { releaseLocalCronRunReceiptOwnership(receipt); }); + it("queues a threshold-crossing interrupted-run alert after persistence", async () => { + const { storePath } = await makeStorePath(); + const startedAtMs = Date.parse("2026-08-13T10:35:00.000Z"); + const nowMs = startedAtMs + 30_000; + const job = makeJob("interrupted-threshold-alert", startedAtMs); + job.delivery = { mode: "announce", channel: "last" }; + job.failureAlert = { after: 2, cooldownMs: 60_000 }; + job.state.consecutiveErrors = 1; + await writeCronStoreSnapshot({ storePath, jobs: [job] }); + const sendCronFailureAlert = vi.fn(async () => undefined); + const state = makeState(storePath, nowMs, { sendCronFailureAlert }); + + const result = recoverCronRunProposal(state, { + jobId: job.id, + runningAtMs: startedAtMs, + }); + + expect(result).toMatchObject({ kind: "repaired" }); + if (result.kind !== "repaired") { + throw new Error("expected repaired interrupted run"); + } + expect(sendCronFailureAlert).not.toHaveBeenCalled(); + expect(result.notifications).toHaveLength(1); + expect((await loadCronStore(storePath)).jobs[0]?.state).toMatchObject({ + consecutiveErrors: 2, + lastFailureAlertAtMs: nowMs, + lastFailureNotificationDeliveryStatus: "unknown", + }); + + runPostPersistCronNotifications(state, result.notifications); + await vi.waitFor(() => expect(sendCronFailureAlert).toHaveBeenCalledOnce()); + expect(sendCronFailureAlert).toHaveBeenCalledWith( + expect.objectContaining({ + runAtMs: startedAtMs, + payload: expect.objectContaining({ + text: expect.stringContaining("failed 2 times"), + }), + }), + ); + }); + + it("keeps interrupted-run alerts disabled by failureAlert false", async () => { + const { storePath } = await makeStorePath(); + const startedAtMs = Date.parse("2026-08-13T10:36:00.000Z"); + const job = makeJob("interrupted-alert-disabled", startedAtMs); + job.delivery = { mode: "announce", channel: "last" }; + job.failureAlert = false; + job.state.consecutiveErrors = 1; + await writeCronStoreSnapshot({ storePath, jobs: [job] }); + const sendCronFailureAlert = vi.fn(async () => undefined); + const state = makeState(storePath, startedAtMs + 30_000, { sendCronFailureAlert }); + + const result = recoverCronRunProposal(state, { + jobId: job.id, + runningAtMs: startedAtMs, + }); + + expect(result).toMatchObject({ kind: "repaired", notifications: [] }); + expect(sendCronFailureAlert).not.toHaveBeenCalled(); + expect((await loadCronStore(storePath)).jobs[0]?.state).toMatchObject({ + consecutiveErrors: 2, + lastFailureNotificationDeliveryStatus: "not-requested", + }); + }); + + it("keeps only auto-disable notification on the tenth interrupted failure", async () => { + const { storePath } = await makeStorePath(); + const startedAtMs = Date.parse("2026-08-13T10:37:00.000Z"); + const nowMs = startedAtMs + 30_000; + const job = makeJob("interrupted-auto-disable", startedAtMs); + job.delivery = { mode: "announce", channel: "last" }; + job.failureAlert = { after: 10, cooldownMs: 0 }; + job.state.consecutiveErrors = 9; + await writeCronStoreSnapshot({ storePath, jobs: [job] }); + const enqueueSystemEvent = vi.fn(); + const sendCronFailureAlert = vi.fn(async () => undefined); + const state = makeState(storePath, nowMs, { enqueueSystemEvent, sendCronFailureAlert }); + + const result = recoverCronRunProposal(state, { + jobId: job.id, + runningAtMs: startedAtMs, + }); + + expect(result).toMatchObject({ kind: "repaired" }); + if (result.kind !== "repaired") { + throw new Error("expected repaired interrupted run"); + } + expect(result.notifications).toHaveLength(1); + expect((await loadCronStore(storePath)).jobs[0]).toMatchObject({ + enabled: false, + state: { + consecutiveErrors: 10, + lastFailureNotificationDeliveryStatus: "not-requested", + autoDisabled: { reason: "consecutive-failures", consecutiveErrors: 10 }, + }, + }); + + runPostPersistCronNotifications(state, result.notifications); + expect(enqueueSystemEvent).toHaveBeenCalledOnce(); + expect(sendCronFailureAlert).not.toHaveBeenCalled(); + }); + it("restores a finalized quiet trigger with a skipped receipt", async () => { const { storePath } = await makeStorePath(); const startedAtMs = Date.parse("2026-08-13T10:45:00.000Z"); diff --git a/src/cron/service/startup-run-repair.ts b/src/cron/service/startup-run-repair.ts index af4d7f84e752..b507dabf5b51 100644 --- a/src/cron/service/startup-run-repair.ts +++ b/src/cron/service/startup-run-repair.ts @@ -1,11 +1,11 @@ /** Repairs interrupted and finalized cron runs while the service starts. */ import { asDateTimestampMs } from "@openclaw/normalization-core/number-coercion"; import { resolveCronCompletionStatus } from "../completion-status.js"; -import { resolveCronDeliveryPlan, resolveFailureDestination } from "../delivery-plan.js"; import { parseAbsoluteTimeMs } from "../parse.js"; import type { CronRunLogEntry } from "../run-log-types.js"; import type { CronJob, CronRunStatus } from "../types.js"; import { maybeAutoDisableCronJobAfterRunFailure } from "./auto-disable.js"; +import { finalizeCronFailureNotifications, resolveFailureAlert } from "./failure-alerts.js"; import type { CronServiceState, DeferredCronNotifications } from "./state.js"; import type { CronTriggerEvalOutcome } from "./timer-execution-timeout.js"; import { @@ -38,20 +38,6 @@ function resolveOneShotReplacementAtMs(job: CronJob, runningAtMs: number): numbe return parseAbsoluteTimeMs(job.schedule.at) === nextRunAtMs ? nextRunAtMs : undefined; } -function resolveInterruptedStartupFailureNotificationStatus(params: { - state: CronServiceState; - job: CronJob; -}) { - if (params.job.delivery?.bestEffort === true) { - return "not-requested"; - } - if (resolveFailureDestination(params.job, params.state.deps.cronConfig?.failureAlert)) { - return "unknown"; - } - const primaryPlan = resolveCronDeliveryPlan(params.job); - return primaryPlan.mode === "announce" && primaryPlan.requested ? "unknown" : "not-requested"; -} - export function markInterruptedStartupRun(params: { state: CronServiceState; job: CronJob; @@ -64,10 +50,6 @@ export function markInterruptedStartupRun(params: { const replacementAtMs = resolveOneShotReplacementAtMs(job, runningAtMs); // A persisted running marker means the gateway stopped mid-run; mark it as a // normal failed run so retries, alerts, and run logs all see one outcome. - const failureNotificationStatus = resolveInterruptedStartupFailureNotificationStatus({ - state: params.state, - job, - }); const previousErrors = typeof job.state.consecutiveErrors === "number" && Number.isFinite(job.state.consecutiveErrors) ? Math.max(0, Math.floor(job.state.consecutiveErrors)) @@ -90,24 +72,36 @@ export function markInterruptedStartupRun(params: { job.state.lastDeliveryStatus = "unknown"; job.state.lastDeliveryError = STARTUP_INTERRUPTED_ERROR; job.state.lastFailureNotificationDelivered = undefined; - job.state.lastFailureNotificationDeliveryStatus = failureNotificationStatus; + job.state.lastFailureNotificationDeliveryStatus = "not-requested"; job.state.lastFailureNotificationDeliveryError = undefined; job.state.nextRunAtMs = replacementAtMs; job.updatedAtMs = nowMs; - if ( - maybeAutoDisableCronJobAfterRunFailure({ - state: params.state, - job, - atMs: nowMs, - deferredNotifications: params.deferredNotifications, - }) - ) { + const alertConfig = resolveFailureAlert(params.state, job); + const autoDisableNotificationOwnsFailure = maybeAutoDisableCronJobAfterRunFailure({ + state: params.state, + job, + atMs: nowMs, + deferredNotifications: params.deferredNotifications, + }); + if (autoDisableNotificationOwnsFailure) { params.state.deps.log.error( { jobId: job.id, name: job.name, consecutiveErrors: job.state.consecutiveErrors }, "cron: auto-disabled interrupted job after consecutive run failures", ); } + finalizeCronFailureNotifications(params.state, { + job, + alertConfig, + result: { + status: "error", + error: STARTUP_INTERRUPTED_ERROR, + startedAt: runningAtMs, + }, + completionFailed: false, + autoDisableNotificationOwnsFailure, + deferredNotifications: params.deferredNotifications, + }); if (job.schedule.kind === "at" && replacementAtMs === undefined) { job.enabled = false; @@ -186,9 +180,11 @@ export function restoreFinalizedStartupRun(params: { job.state.lastDelivered = entry.delivered; job.state.lastDeliveryStatus = entry.deliveryStatus; job.state.lastDeliveryError = entry.deliveryError; - job.state.lastFailureNotificationDelivered = entry.failureNotificationDelivery?.delivered; - job.state.lastFailureNotificationDeliveryStatus = entry.failureNotificationDelivery?.status; - job.state.lastFailureNotificationDeliveryError = entry.failureNotificationDelivery?.error; + if (entry.failureNotificationDelivery) { + job.state.lastFailureNotificationDelivered = entry.failureNotificationDelivery.delivered; + job.state.lastFailureNotificationDeliveryStatus = entry.failureNotificationDelivery.status; + job.state.lastFailureNotificationDeliveryError = entry.failureNotificationDelivery.error; + } const finalizedNextRunAtMs = replacementAtMs ?? entry.nextRunAtMs; job.state.nextRunAtMs = job.state.autoDisabled || finalizedNextRunAtMs === undefined diff --git a/src/cron/service/state.ts b/src/cron/service/state.ts index b0a88eff87d6..11ad33b656d9 100644 --- a/src/cron/service/state.ts +++ b/src/cron/service/state.ts @@ -260,6 +260,7 @@ export type CronServiceDeps = { mode?: "announce" | "webhook"; accountId?: string; threadId?: string | number; + inheritSessionThread?: false; }) => Promise; onEvent?: (evt: CronEvent, context?: CronEventContext) => void; }; diff --git a/src/cron/service/timer-job-runner.ts b/src/cron/service/timer-job-runner.ts index f2ba87f64187..983bfd2d2551 100644 --- a/src/cron/service/timer-job-runner.ts +++ b/src/cron/service/timer-job-runner.ts @@ -444,14 +444,13 @@ export function authorCronRunCompletion< CronJobRunResult, "status" | "error" | "deliveryError" | "delivered" | "deliveryAttempted" >, ->(state: CronServiceState, job: CronJob, result: T) { +>(_state: CronServiceState, job: CronJob, result: T) { const deliveryState = resolveDeliveryState({ job, runStatus: result.status, delivered: result.delivered, deliveryAttempted: result.deliveryAttempted, error: result.deliveryError ?? result.error, - globalFailureDestination: state.deps.cronConfig?.failureAlert, }); return { ...result, diff --git a/src/cron/service/timer-outcomes.ts b/src/cron/service/timer-outcomes.ts index c29f5d6ae776..d6a5304425df 100644 --- a/src/cron/service/timer-outcomes.ts +++ b/src/cron/service/timer-outcomes.ts @@ -8,7 +8,11 @@ import { cronSchedulingInputsEqual } from "../schedule-identity.js"; import { computeNextRunAtMs } from "../schedule.js"; import type { CronJob, CronRunStatus } from "../types.js"; import { maybeAutoDisableCronJobAfterRunFailure } from "./auto-disable.js"; -import { maybeEmitFailureAlert, resolveFailureAlert } from "./failure-alerts.js"; +import { + finalizeCronFailureNotifications, + maybeEmitFailureAlert, + resolveFailureAlert, +} from "./failure-alerts.js"; import { computeJobNextRunAtMs, DEFAULT_ERROR_BACKOFF_SCHEDULE_MS, @@ -129,7 +133,6 @@ export function applyJobResult( delivered: result.delivered, deliveryAttempted: result.deliveryAttempted, error: result.deliveryError ?? result.error, - globalFailureDestination: state.deps.cronConfig?.failureAlert, }); job.state.lastDelivered = deliveryState.delivered; job.state.lastDeliveryStatus = deliveryState.status; @@ -137,9 +140,9 @@ export function applyJobResult( deliveryState.status === "not-delivered" && deliveryState.error ? deliveryState.error : undefined; - job.state.lastFailureNotificationDelivered = deliveryState.failureNotification.delivered; - job.state.lastFailureNotificationDeliveryStatus = deliveryState.failureNotification.status; - job.state.lastFailureNotificationDeliveryError = deliveryState.failureNotification.error; + job.state.lastFailureNotificationDelivered = undefined; + job.state.lastFailureNotificationDeliveryStatus = "not-requested"; + job.state.lastFailureNotificationDeliveryError = undefined; job.updatedAtMs = result.endedAt; // Track consecutive errors for backoff / auto-disable; skipped runs use a @@ -149,20 +152,6 @@ export function applyJobResult( if (result.status === "error") { job.state.consecutiveErrors = (job.state.consecutiveErrors ?? 0) + 1; job.state.consecutiveSkipped = 0; - maybeEmitFailureAlert(state, { - job, - alertConfig, - status: "error", - error: result.error, - errorReason: job.state.lastErrorReason, - failureNotificationDetail: result.failureNotificationDetail, - runAtMs: result.startedAt, - consecutiveCount: job.state.consecutiveErrors, - ...(opts?.replayFailureAlertAtMs !== undefined - ? { delivery: "record-only" as const, occurredAtMs: opts.replayFailureAlertAtMs } - : {}), - deferredNotifications: opts?.deferredNotifications, - }); } else if (result.status === "skipped") { job.state.consecutiveErrors = 0; job.state.consecutiveSkipped = (job.state.consecutiveSkipped ?? 0) + 1; @@ -198,15 +187,28 @@ export function applyJobResult( previousScheduleState.nextRunAtMs > (opts.scheduleOwnershipAtMs ?? result.startedAt); const ownsSchedule = opts?.scheduleOwnership !== "stale"; const isOneShotSchedule = job.schedule.kind === "at" || job.schedule.kind === "on-exit"; + const completionStatus = + result.completionStatus ?? + resolveAdmittedCronCompletionStatus(job, result.status, deliveryState.status); const shouldDelete = ownsSchedule && isOneShotSchedule && !preserveOneShotSchedule && job.deleteAfterRun === true && - (result.completionStatus ?? - resolveAdmittedCronCompletionStatus(job, result.status, deliveryState.status)) === - "succeeded"; - const retryDisabledHeartbeatOneShot = shouldRetryDisabledHeartbeatOneShot(job, result); + completionStatus === "succeeded"; + let autoDisableNotificationOwnsFailure = false; + const finish = () => { + finalizeCronFailureNotifications(state, { + job, + alertConfig, + result, + completionFailed: completionStatus === "failed", + autoDisableNotificationOwnsFailure, + replayFailureAlertAtMs: opts?.replayFailureAlertAtMs, + deferredNotifications: opts?.deferredNotifications, + }); + return shouldDelete; + }; if (!ownsSchedule) { // The completed invocation still owns its outcome, but the latest durable @@ -221,7 +223,7 @@ export function applyJobResult( job.state.pacedNextRunAtMs = previousScheduleState.pacedNextRunAtMs; job.state.forcePreservedNextRunAtMs = previousScheduleState.nextRunAtMs; } else if (job.schedule.kind === "at") { - if (retryDisabledHeartbeatOneShot) { + if (shouldRetryDisabledHeartbeatOneShot(job, result)) { const retryDecision = resolveDisabledHeartbeatOneShotRetryDecision({ cronConfig: state.deps.cronConfig, consecutiveSkipped: job.state.consecutiveSkipped, @@ -331,6 +333,7 @@ export function applyJobResult( deferredNotifications: opts?.deferredNotifications, }) ) { + autoDisableNotificationOwnsFailure = true; // Keep this after the ownership and immediate-preserve gates: those paths // restore schedule state and would otherwise silently undo the disable. state.deps.log.error( @@ -389,7 +392,7 @@ export function applyJobResult( deferredNotifications: opts?.deferredNotifications, }); if (retryNextRunAtMs === undefined) { - return shouldDelete; + return finish(); } if (retryNextRunAtMs < normalNext) { state.deps.log.info( @@ -404,7 +407,7 @@ export function applyJobResult( }, "cron: scheduling recurring retry after transient error", ); - return shouldDelete; + return finish(); } } } @@ -421,7 +424,7 @@ export function applyJobResult( candidate: undefined, deferredNotifications: opts?.deferredNotifications, }); - return shouldDelete; + return finish(); } const backoffNext = assignNextRunAtMs({ state, @@ -430,7 +433,7 @@ export function applyJobResult( deferredNotifications: opts?.deferredNotifications, }); if (backoffNext === undefined) { - return shouldDelete; + return finish(); } // Use whichever is later: the natural next run or the backoff delay. job.state.nextRunAtMs = @@ -536,7 +539,7 @@ export function applyJobResult( } } - return shouldDelete; + return finish(); } /** Commits payload-script state only after the complete cron run succeeds. */ diff --git a/src/cron/service/timer-trigger.ts b/src/cron/service/timer-trigger.ts index dfee9df76989..b67314f03d9d 100644 --- a/src/cron/service/timer-trigger.ts +++ b/src/cron/service/timer-trigger.ts @@ -1,10 +1,9 @@ import { asDateTimestampMs } from "@openclaw/normalization-core/number-coercion"; import type { CronConfig } from "../../config/types.cron.js"; -import { resolveCronDeliveryPlan, resolveFailureDestination } from "../delivery-plan.js"; +import { resolveCronDeliveryPlan } from "../delivery-plan.js"; import { type CronRetryOn, resolveCronExecutionRetryHint } from "../retry-hint.js"; import { createCronStreamSourceIdentity } from "../stream-schedule.js"; import type { - CronFailureNotificationDelivery, CronJob, CronResolvedDeliveryState, CronRunErrorClassification, @@ -295,25 +294,17 @@ export function resolveDeliveryState(params: { delivered?: boolean; deliveryAttempted?: boolean; error?: string; - globalFailureDestination?: CronConfig["failureAlert"]; }): CronResolvedDeliveryState { const primaryDeliveryPlan = resolveCronDeliveryPlan(params.job); const primaryDeliveryRequested = primaryDeliveryPlan.requested; - // Failure destinations can receive alerts even when the primary delivery - // path was disabled or failed before direct delivery produced an ack. - const alternateFailureNotificationRequested = - params.runStatus === "error" && - params.job.delivery?.bestEffort !== true && - resolveFailureDestination(params.job, params.globalFailureDestination) !== null; + const noFailureNotification = { status: "not-requested" as const }; if (!primaryDeliveryRequested) { if (primaryDeliveryPlan.mode === "webhook") { if (params.delivered === true) { return { delivered: true, status: "delivered", - failureNotification: { - status: alternateFailureNotificationRequested ? "unknown" : "not-requested", - }, + failureNotification: noFailureNotification, }; } if (params.deliveryAttempted === true) { @@ -321,30 +312,22 @@ export function resolveDeliveryState(params: { delivered: false, status: "not-delivered", error: params.error, - failureNotification: { - status: alternateFailureNotificationRequested ? "unknown" : "not-requested", - }, + failureNotification: noFailureNotification, }; } } return { status: "not-requested", - failureNotification: { - status: alternateFailureNotificationRequested ? "unknown" : "not-requested", - }, + failureNotification: noFailureNotification, }; } if (params.runStatus === "error") { - const failureNotification: CronFailureNotificationDelivery = - alternateFailureNotificationRequested ? { status: "unknown" } : { status: "delivered" }; if (params.delivered === true) { return { delivered: false, status: "not-delivered", error: params.error, - failureNotification: alternateFailureNotificationRequested - ? failureNotification - : { delivered: true, status: "delivered" }, + failureNotification: noFailureNotification, }; } if (params.delivered === false) { @@ -352,19 +335,13 @@ export function resolveDeliveryState(params: { delivered: false, status: "not-delivered", error: params.error, - failureNotification: alternateFailureNotificationRequested - ? failureNotification - : { - delivered: false, - status: "not-delivered", - ...(params.error ? { error: params.error } : {}), - }, + failureNotification: noFailureNotification, }; } return { status: "unknown", error: params.error, - failureNotification: { status: "unknown" }, + failureNotification: noFailureNotification, }; } if (params.delivered === true) { diff --git a/src/cron/service/timer.regression.test.ts b/src/cron/service/timer.regression.test.ts index 22799e04addc..b900f2995dc9 100644 --- a/src/cron/service/timer.regression.test.ts +++ b/src/cron/service/timer.regression.test.ts @@ -2925,13 +2925,16 @@ describe("cron service timer regressions", () => { it("auto-disables a recurring job on its tenth consecutive run failure", () => { const startedAt = Date.parse("2026-08-01T12:00:00.000Z"); const deferredNotifications: Array<() => void> = []; + const enqueueSystemEvent = vi.fn(); + const sendCronFailureAlert = vi.fn(async () => undefined); const state = createCronServiceState({ cronEnabled: true, storePath: "/tmp/cron-consecutive-failure-threshold.json", log: noopLogger, nowMs: () => startedAt, - enqueueSystemEvent: vi.fn(), + enqueueSystemEvent, requestHeartbeat: vi.fn(), + sendCronFailureAlert, runIsolatedAgentJob: createDefaultIsolatedRunner(), }); const job = createIsolatedRegressionJob({ @@ -2942,6 +2945,7 @@ describe("cron service timer regressions", () => { payload: { kind: "agentTurn", message: "fail" }, state: { consecutiveErrors: 8 }, }); + job.failureAlert = { after: 10, cooldownMs: 0 }; applyJobResult( state, @@ -2973,6 +2977,9 @@ describe("cron service timer regressions", () => { consecutiveErrors: 10, }); expect(deferredNotifications).toHaveLength(1); + deferredNotifications[0]?.(); + expect(enqueueSystemEvent).toHaveBeenCalledOnce(); + expect(sendCronFailureAlert).not.toHaveBeenCalled(); }); it("resets the auto-disable streak after a successful recurring run", () => { @@ -3023,7 +3030,7 @@ describe("cron service timer regressions", () => { it.each([ { name: "stale schedule", opts: { scheduleOwnership: "stale" as const } }, { name: "forced run", opts: { scheduleMode: "preserve" as const } }, - ])("does not auto-disable after a $name failure", ({ opts }) => { + ])("does not auto-disable but still alerts after a $name failure", ({ opts }) => { const startedAt = Date.parse("2026-08-01T14:00:00.000Z"); const deferredNotifications: Array<() => void> = []; const state = createCronServiceState({ @@ -3054,7 +3061,7 @@ describe("cron service timer regressions", () => { expect(job.enabled).toBe(true); expect(job.state.consecutiveErrors).toBe(10); expect(job.state.autoDisabled).toBeUndefined(); - expect(deferredNotifications).toHaveLength(0); + expect(deferredNotifications).toHaveLength(1); }); it("keeps state updates when cron next-run computation throws after a successful run (#30905)", () => { diff --git a/src/cron/service/timer.timeout-watchdog.test.ts b/src/cron/service/timer.timeout-watchdog.test.ts index b69a8e68f55c..e1556b091893 100644 --- a/src/cron/service/timer.timeout-watchdog.test.ts +++ b/src/cron/service/timer.timeout-watchdog.test.ts @@ -807,6 +807,7 @@ describe("cron service timer regressions", () => { mode: "announce", accountId: undefined, threadId: undefined, + inheritSessionThread: false, }); } finally { vi.useRealTimers(); diff --git a/src/gateway/server-cron-notifications.delivery-failure.test.ts b/src/gateway/server-cron-notifications.delivery-failure.test.ts deleted file mode 100644 index 761de7afd8be..000000000000 --- a/src/gateway/server-cron-notifications.delivery-failure.test.ts +++ /dev/null @@ -1,138 +0,0 @@ -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(); - return { ...actual, sendFailureNotificationAnnounce }; -}); - -import { dispatchGatewayCronFinishedNotifications } from "./server-cron-notifications.js"; - -function createThreadedJob(withFailureDestination: boolean, bestEffort?: 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, - ...(bestEffort === undefined ? {} : { bestEffort }), - ...(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, - completionStatus: "failed" 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.", - }); - }); - - it.each([ - { completionStatus: "failed" as const, currentBestEffort: true, expected: 1 }, - { completionStatus: "succeeded" as const, currentBestEffort: false, expected: 0 }, - ])( - "uses event completion $completionStatus after the current policy changes", - ({ completionStatus, currentBestEffort, expected }) => { - dispatchGatewayCronFinishedNotifications({ - evt: { - jobId: "cron-delivery-failure", - action: "finished", - status: "ok", - completionStatus, - deliveryStatus: "not-delivered", - deliveryError: "message thread not found", - }, - job: createThreadedJob(true, currentBestEffort), - deps: {} as CliDeps, - logger: { warn: vi.fn() }, - resolveCronAgent: () => ({ agentId: "main", cfg: {} }), - }); - - expect(sendFailureNotificationAnnounce).toHaveBeenCalledTimes(expected); - }, - ); - - it("keeps configured failure destinations from inheriting the primary delivery thread", () => { - const job = createThreadedJob(true); - job.sessionKey = "agent:main:telegram:group:-1001234567890:thread:42"; - - dispatchGatewayCronFinishedNotifications({ - evt: { jobId: job.id, action: "finished", status: "error", error: "boom" }, - job, - deps: {} as CliDeps, - logger: { warn: vi.fn() }, - resolveCronAgent: () => ({ agentId: "main", cfg: {} }), - }); - - expect(sendFailureNotificationAnnounce).toHaveBeenCalledTimes(1); - expect(sendFailureNotificationAnnounce.mock.calls[0]?.[4]).toEqual({ - channel: "telegram", - to: "-1001234567890", - accountId: undefined, - sessionKey: "agent:main:telegram:group:-1001234567890:thread:42", - inheritSessionThread: false, - }); - }); -}); diff --git a/src/gateway/server-cron-notifications.test.ts b/src/gateway/server-cron-notifications.test.ts index 68d937ba3361..dedbc4e96bb2 100644 --- a/src/gateway/server-cron-notifications.test.ts +++ b/src/gateway/server-cron-notifications.test.ts @@ -3,7 +3,6 @@ import { createRequireRecord } from "openclaw/plugin-sdk/test-fixtures"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { CliDeps } from "../cli/deps.types.js"; -import { makeCronJob } from "../cron/delivery.test-helpers.js"; import type { CronJob } from "../cron/types.js"; import { getActiveGatewayRootWorkCount, @@ -19,7 +18,6 @@ const mocks = vi.hoisted(() => ({ finalUrl: "https://example.invalid/cron", release: vi.fn(async () => {}), })), - sendFailureNotificationAnnounce: vi.fn(), sendCronAnnouncePayloadStrict: vi.fn(), })); @@ -31,7 +29,6 @@ vi.mock("../cron/delivery.js", async (importOriginal) => { const actual = await importOriginal(); return { ...actual, - sendFailureNotificationAnnounce: mocks.sendFailureNotificationAnnounce, sendCronAnnouncePayloadStrict: mocks.sendCronAnnouncePayloadStrict, }; }); @@ -107,7 +104,6 @@ describe("dispatchGatewayCronFinishedNotifications", () => { finalUrl: "https://example.invalid/cron", release: vi.fn(async () => {}), })); - mocks.sendFailureNotificationAnnounce.mockResolvedValue(undefined); mocks.sendCronAnnouncePayloadStrict.mockResolvedValue(undefined); }); @@ -306,7 +302,7 @@ describe("dispatchGatewayCronFinishedNotifications", () => { } }); - it("preserves the primary topic on immediate failure alerts", async () => { + it("preserves the primary topic on scheduler-authorized alerts", async () => { const job = createWebhookJob({ mode: "announce", channel: "telegram", @@ -340,7 +336,7 @@ describe("dispatchGatewayCronFinishedNotifications", () => { ); }); - it("keeps immediate failure webhook messages stable and adds structured runAtMs", async () => { + it("keeps failure webhook messages stable and adds structured runAtMs", async () => { const runAtMs = Date.parse("2026-01-15T15:30:00.000Z"); const job = createCompletionWebhookJob(); @@ -369,6 +365,72 @@ describe("dispatchGatewayCronFinishedNotifications", () => { }); }); + it.each([ + { name: "missing", to: undefined }, + { name: "invalid", to: "ftp://example.invalid/failure" }, + ])("rejects a $name failure-alert webhook target", async ({ to }) => { + await expect( + sendGatewayCronFailureAlert({ + deps: {} as CliDeps, + logger: { warn: vi.fn() }, + resolveCronAgent: () => ({ agentId: "main", cfg: {} }), + job: createCompletionWebhookJob(), + payload: { text: "cron failed" }, + channel: "last", + mode: "webhook", + to, + }), + ).rejects.toThrow(/webhook requires/); + expect(mocks.fetchWithSsrFGuard).not.toHaveBeenCalled(); + }); + + it("rejects failure-alert webhook network errors", async () => { + mocks.fetchWithSsrFGuard.mockRejectedValueOnce(new Error("network unavailable")); + + await expect( + sendGatewayCronFailureAlert({ + deps: {} as CliDeps, + logger: { warn: vi.fn() }, + resolveCronAgent: () => ({ agentId: "main", cfg: {} }), + job: createCompletionWebhookJob(), + payload: { text: "cron failed" }, + channel: "last", + mode: "webhook", + to: "https://example.invalid/failure", + }), + ).rejects.toThrow("network unavailable"); + }); + + it("rejects unavailable failure-alert agents and channels", async () => { + const job = createWebhookJob({ mode: "announce", channel: "telegram", to: "123" }); + await expect( + sendGatewayCronFailureAlert({ + deps: {} as CliDeps, + logger: { warn: vi.fn() }, + resolveCronAgent: () => { + throw new Error("agent unavailable"); + }, + job, + payload: { text: "cron failed" }, + channel: "telegram", + to: "123", + }), + ).rejects.toThrow("agent unavailable"); + + mocks.sendCronAnnouncePayloadStrict.mockRejectedValueOnce(new Error("channel unavailable")); + await expect( + sendGatewayCronFailureAlert({ + deps: {} as CliDeps, + logger: { warn: vi.fn() }, + resolveCronAgent: () => ({ agentId: "main", cfg: {} }), + job, + payload: { text: "cron failed" }, + channel: "telegram", + to: "123", + }), + ).rejects.toThrow("channel unavailable"); + }); + it("delivers a failed cron webhook even when the run produced no summary", async () => { const logger = { warn: vi.fn() }; const job = createCompletionWebhookJob(); @@ -449,7 +511,7 @@ describe("dispatchGatewayCronFinishedNotifications", () => { expect(logger.warn).not.toHaveBeenCalled(); }); - it("independently admits immediate failure alerts", async () => { + it("independently admits scheduler-authorized failure alerts", async () => { const deferred = createVoidDeferred(); mocks.sendCronAnnouncePayloadStrict.mockImplementationOnce(async () => { await deferred.promise; @@ -478,7 +540,7 @@ describe("dispatchGatewayCronFinishedNotifications", () => { { description: "honors cancellation", honorsCancellation: true }, { description: "ignores cancellation", honorsCancellation: false }, ])( - "releases immediate failure alert admission when a stalled sender $description", + "releases failure alert admission when a stalled sender $description", async ({ honorsCancellation }) => { vi.useFakeTimers(); try { @@ -560,178 +622,6 @@ describe("dispatchGatewayCronFinishedNotifications", () => { await waitForFast(() => expect(mocks.fetchWithSsrFGuard).toHaveBeenCalledTimes(1)); }); - it("independently admits failure destination webhook delivery", async () => { - const deferred = createVoidDeferred(); - mocks.fetchWithSsrFGuard.mockImplementationOnce(async () => { - await deferred.promise; - return { - response: new Response(null, { status: 204 }), - finalUrl: "https://example.invalid/failure", - release: vi.fn(async () => {}), - }; - }); - const job = createWebhookJob({ - mode: "announce", - channel: "last", - failureDestination: { - mode: "webhook", - to: "https://example.invalid/failure", - }, - }); - - dispatchGatewayCronFinishedNotifications({ - evt: { jobId: job.id, action: "finished", status: "error", error: "boom" }, - job, - deps: {} as CliDeps, - logger: { warn: vi.fn() }, - resolveCronAgent: () => ({ agentId: "main", cfg: {} }), - ssrfPolicy: webhookSsrfPolicy, - }); - - await waitForFast(() => expect(mocks.fetchWithSsrFGuard).toHaveBeenCalledTimes(1)); - expectWebhookSsrfPolicy(); - expect(getActiveGatewayRootWorkCount()).toBe(1); - deferred.resolve(); - await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); - }); - - it("independently admits failure destination announce delivery", async () => { - const deferred = createVoidDeferred(); - mocks.sendFailureNotificationAnnounce.mockImplementationOnce(() => deferred.promise); - const job = createWebhookJob({ - mode: "announce", - channel: "last", - failureDestination: { - mode: "announce", - channel: "telegram", - to: "-1001234567890", - }, - }); - - dispatchGatewayCronFinishedNotifications({ - evt: { jobId: job.id, action: "finished", status: "error", error: "boom" }, - job, - deps: {} as CliDeps, - logger: { warn: vi.fn() }, - resolveCronAgent: () => ({ agentId: "main", cfg: {} }), - }); - - await waitForFast(() => expect(mocks.sendFailureNotificationAnnounce).toHaveBeenCalledTimes(1)); - expect(getActiveGatewayRootWorkCount()).toBe(1); - deferred.resolve(); - await waitForFast(() => expect(getActiveGatewayRootWorkCount()).toBe(0)); - }); - - it("owns missing-agent failures during detached failure destination delivery", async () => { - const warning = createVoidDeferred(); - const logger = { warn: vi.fn(() => warning.resolve()) }; - const job = createWebhookJob({ - mode: "announce", - channel: "last", - failureDestination: { - mode: "announce", - channel: "telegram", - to: "-1001234567890", - }, - }); - - expect(() => - dispatchGatewayCronFinishedNotifications({ - evt: { jobId: job.id, action: "finished", status: "error", error: "boom" }, - job, - deps: {} as CliDeps, - logger, - resolveCronAgent: () => { - throw new Error("cron job agent is unavailable: removed-agent"); - }, - }), - ).not.toThrow(); - - await warning.promise; - expect(mocks.sendFailureNotificationAnnounce).not.toHaveBeenCalled(); - expect(logger.warn).toHaveBeenCalledWith( - { - jobId: job.id, - err: "cron job agent is unavailable: removed-agent", - }, - "cron: detached notification delivery failed", - ); - expect(getActiveGatewayRootWorkCount()).toBe(0); - }); - - it("uses classified or closed producer facts for chat failure detail", () => { - const runAtMs = Date.parse("2026-01-15T15:30:00.000Z"); - const announceJob = createWebhookJob({ - mode: "announce", - failureDestination: { - mode: "announce", - channel: "telegram", - to: "-1001234567890", - }, - }); - announceJob.state.lastErrorReason = "auth"; - - dispatchGatewayCronFinishedNotifications({ - evt: { - jobId: announceJob.id, - action: "finished", - status: "error", - error: "provider overloaded", - runAtMs, - }, - job: announceJob, - deps: {} as CliDeps, - logger: { warn: vi.fn() }, - resolveCronAgent: () => ({ - agentId: "main", - cfg: { agents: { defaults: { userTimezone: "America/New_York" } } }, - }), - }); - - expect(mocks.sendFailureNotificationAnnounce).toHaveBeenCalledWith( - expect.anything(), - expect.anything(), - "main", - announceJob.id, - expect.anything(), - { - text: - '⚠️ Automation "notification admission" failed\n' + - "Cause: auth\n" + - "Run started: 2026-01-15 10:30 EST", - }, - ); - - vi.clearAllMocks(); - announceJob.state.lastErrorReason = undefined; - const rawError = String.raw`TOKEN=opaque-secret-value | /Users/private/automation.sh | C:\Users\private\automation.ps1 | sh -lc "curl --header Authorization:secret" | https://internal.example.test/run?token=query-secret | {"error":{"message":"provider secret body"}} | Error: exploded at runAutomation (internal.js:42:7)`; - const dispatchRawFailure = (withDetail: boolean) => - dispatchGatewayCronFinishedNotifications({ - evt: { jobId: announceJob.id, action: "finished", status: "error", error: rawError }, - ...(withDetail - ? { failureNotificationDetail: { kind: "command-exit" as const, exitCode: 2 } } - : {}), - job: announceJob, - deps: {} as CliDeps, - logger: { warn: vi.fn() }, - resolveCronAgent: () => ({ agentId: "main", cfg: {} }), - }); - dispatchRawFailure(true); - const closedDetailPayload = mocks.sendFailureNotificationAnnounce.mock.calls[0]?.[5]; - expect(closedDetailPayload).toEqual({ - text: '⚠️ Automation "notification admission" failed\nCause: command exited with code 2', - }); - expect(closedDetailPayload?.text).not.toMatch( - /opaque-secret-value|Users.private|curl --header|internal\.example|provider secret|runAutomation/u, - ); - - vi.clearAllMocks(); - dispatchRawFailure(false); - expect(mocks.sendFailureNotificationAnnounce.mock.calls[0]?.[5]).toEqual({ - text: '⚠️ Automation "notification admission" failed\nCheck automation history for details.', - }); - }); - it("redacts invalid completion webhook targets in warnings", () => { const logger = { warn: vi.fn(), @@ -806,89 +696,6 @@ describe("dispatchGatewayCronFinishedNotifications", () => { expect(mocks.fetchWithSsrFGuard).not.toHaveBeenCalled(); }); - it("preserves the primary topic when a failed run falls back to its delivery route", () => { - const job = createWebhookJob({ - mode: "announce", - channel: "telegram", - to: "-1001234567890", - accountId: "bot-a", - threadId: 42, - }); - - dispatchGatewayCronFinishedNotifications({ - evt: { jobId: job.id, action: "finished", status: "error", error: "boom" }, - job, - deps: {} as CliDeps, - logger: { warn: vi.fn() }, - resolveCronAgent: () => ({ agentId: "main", cfg: {} }), - }); - - expect(mocks.sendFailureNotificationAnnounce).toHaveBeenCalledWith( - expect.anything(), - expect.anything(), - "main", - job.id, - expect.objectContaining({ - channel: "telegram", - to: "-1001234567890", - accountId: "bot-a", - threadId: 42, - }), - expect.objectContaining({ text: expect.any(String) }), - ); - }); - - it("announces channel-shaped failure destinations without mode under a global webhook default (#102235)", () => { - const logger = { warn: vi.fn() }; - const job = makeCronJob({ - id: "cron-channel-fd-no-mode", - name: "channel fd no mode", - delivery: { - mode: "none", - failureDestination: { channel: "slack", to: "#alerts" }, - }, - }); - - dispatchGatewayCronFinishedNotifications({ - evt: { - jobId: job.id, - action: "finished", - status: "error", - error: "boom", - }, - job, - deps: {} as CliDeps, - logger, - resolveCronAgent: () => ({ agentId: "main", cfg: {} }), - globalFailureDestination: { - mode: "webhook", - to: "https://hook.example/cron", - }, - }); - - expect(mocks.sendFailureNotificationAnnounce).toHaveBeenCalledWith( - expect.anything(), - expect.anything(), - "main", - job.id, - { - channel: "slack", - to: "#alerts", - accountId: undefined, - sessionKey: undefined, - inheritSessionThread: false, - }, - { - text: '⚠️ Automation "channel fd no mode" failed\nCheck automation history for details.', - }, - ); - expect(logger.warn).not.toHaveBeenCalledWith( - expect.objectContaining({ jobId: job.id }), - "cron: failure destination webhook URL is invalid, skipping", - ); - expect(mocks.fetchWithSsrFGuard).not.toHaveBeenCalled(); - }); - it("redacts command action-required summaries before webhook completion delivery", async () => { const logger = { warn: vi.fn() }; const sensitiveSummary = diff --git a/src/gateway/server-cron-notifications.ts b/src/gateway/server-cron-notifications.ts index e0a2ed5bd590..8ef078d5274c 100644 --- a/src/gateway/server-cron-notifications.ts +++ b/src/gateway/server-cron-notifications.ts @@ -7,20 +7,13 @@ import { import { resolveUserTimezone } from "../agents/date-time.js"; import type { ReplyPayload } from "../auto-reply/reply-payload.js"; import type { CliDeps } from "../cli/deps.types.js"; -import type { CronFailureDestinationConfig } from "../config/types.cron.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { redactCronCommandSummaryForExternalDelivery } from "../cron/command-output-summary.js"; -import { - resolveCronDeliveryPlan, - resolveFailureDestination, - sendCronAnnouncePayloadStrict, - sendFailureNotificationAnnounce as sendFailureAnnounce, -} from "../cron/delivery.js"; -import { cronFailureDetailLines } from "../cron/failure-notification-text.js"; +import { resolveCronDeliveryPlan, sendCronAnnouncePayloadStrict } from "../cron/delivery.js"; import { retryTransientDirectCronDelivery } from "../cron/isolated-agent/delivery-dispatch-policy.js"; import type { CronEvent } from "../cron/service.js"; import { resolveCronDeliverySessionKey } from "../cron/session-target.js"; -import type { CronFailureNotificationDetail, CronJob, CronMessageChannel } from "../cron/types.js"; +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"; @@ -55,6 +48,7 @@ type CronFailureAlertParams = { mode?: "announce" | "webhook"; accountId?: string; threadId?: string | number; + inheritSessionThread?: false; }; function redactWebhookUrl(url: string): string { @@ -152,20 +146,6 @@ function buildCronWebhookHeaders(webhookToken?: string): Record return headers; } -function buildCronFailureWebhookPayload(params: { evt: CronEvent; job: CronJob }) { - return { - jobId: params.job.id, - jobName: params.job.name, - message: `Automation "${params.job.name}" ${params.evt.status === "error" ? "failed" : "delivery failed"}: ${params.evt.error ?? params.evt.deliveryError ?? "unknown error"}`, - status: params.evt.status, - completionStatus: params.evt.completionStatus, - error: params.evt.error ?? params.evt.deliveryError, - runAtMs: params.evt.runAtMs, - durationMs: params.evt.durationMs, - nextRunAtMs: params.evt.nextRunAtMs, - }; -} - function appendCronRunStarted( message: string, runAtMs: number | undefined, @@ -349,7 +329,7 @@ function dispatchDetachedCronNotification(params: { }); } -/** Sends the immediate failure alert for cron jobs that failed before normal completion delivery. */ +/** Transports a scheduler-authorized cron failure alert. */ export async function sendGatewayCronFailureAlert(params: CronFailureAlertParams): Promise { await runWithGatewayIndependentRootWorkAdmission(async () => { await sendGatewayCronFailureAlertUnderAdmission(params); @@ -363,40 +343,25 @@ async function sendGatewayCronFailureAlertUnderAdmission( const webhookToken = normalizeOptionalString(params.webhookToken); if (params.mode === "webhook" && !params.to) { - params.logger.warn( - { jobId: params.job.id }, - "cron: failure alert webhook mode requires URL, skipping", - ); - return; + throw new Error("cron failure alert webhook requires a URL"); } if (params.mode === "webhook" && params.to) { const webhookUrl = normalizeHttpWebhookUrl(params.to); - if (webhookUrl) { - await postCronWebhook({ - webhookUrl, - webhookToken, - ssrfPolicy: params.ssrfPolicy, - payload: { - jobId: params.job.id, - jobName: params.job.name, - message: params.payload.text ?? "", - runAtMs: params.runAtMs, - }, - logContext: { jobId: params.job.id }, - blockedLog: "cron: failure alert webhook blocked by SSRF guard", - failedLog: "cron: failure alert webhook failed", - logger: params.logger, - }); - } else { - params.logger.warn( - { - jobId: params.job.id, - webhookUrl: redactWebhookUrl(params.to), - }, - "cron: failure alert webhook URL is invalid, skipping", - ); + if (!webhookUrl) { + throw new Error("cron failure alert webhook requires a valid http(s) URL"); } + await postCronWebhookStrict({ + webhookUrl, + webhookToken, + ssrfPolicy: params.ssrfPolicy, + payload: { + jobId: params.job.id, + jobName: params.job.name, + message: params.payload.text ?? "", + runAtMs: params.runAtMs, + }, + }); return; } @@ -415,6 +380,7 @@ async function sendGatewayCronFailureAlertUnderAdmission( accountId: params.accountId, threadId: params.threadId, sessionKey: resolveCronDeliverySessionKey(params.job), + inheritSessionThread: params.inheritSessionThread, }, payload: { ...params.payload, @@ -432,17 +398,15 @@ async function sendGatewayCronFailureAlertUnderAdmission( ); } -/** Dispatches completion and failure-destination notifications after a cron run finishes. */ +/** Fans out completion webhooks after a cron run finishes. */ export function dispatchGatewayCronFinishedNotifications(params: { evt: CronEvent; - failureNotificationDetail?: CronFailureNotificationDetail; job?: CronJob; deps: CliDeps; logger: CronLogger; resolveCronAgent: CronAgentResolver; webhookToken?: unknown; ssrfPolicy?: SsrFPolicy; - globalFailureDestination?: CronFailureDestinationConfig; }): void { const webhookToken = normalizeOptionalString(params.webhookToken); const redactedWebhookEvent = redactCommandCronEventForExternalDelivery(params.evt, params.job); @@ -494,124 +458,4 @@ export function dispatchGatewayCronFinishedNotifications(params: { }), }); } - - dispatchCronFailureDestinationNotifications({ - evt: params.evt, - failureNotificationDetail: params.failureNotificationDetail, - job: params.job, - deps: params.deps, - logger: params.logger, - resolveCronAgent: params.resolveCronAgent, - webhookToken, - ssrfPolicy: params.ssrfPolicy, - globalFailureDestination: params.globalFailureDestination, - }); -} - -function dispatchCronFailureDestinationNotifications(params: { - evt: CronEvent; - failureNotificationDetail?: CronFailureNotificationDetail; - job?: CronJob; - deps: CliDeps; - logger: CronLogger; - resolveCronAgent: CronAgentResolver; - webhookToken?: string; - ssrfPolicy?: SsrFPolicy; - globalFailureDestination?: CronFailureDestinationConfig; -}): void { - if (!params.job) { - return; - } - - const job = params.job; - const executionFailed = params.evt.status === "error"; - const deliveryOnlyFailed = params.evt.status === "ok" && params.evt.completionStatus === "failed"; - if (!executionFailed && !deliveryOnlyFailed) { - return; - } - if (executionFailed && job.delivery?.bestEffort === true) { - return; - } - const failureDest = resolveFailureDestination(job, params.globalFailureDestination); - if (deliveryOnlyFailed && !failureDest) { - return; - } - const deliverySessionKey = resolveCronDeliverySessionKey(job); - const failurePayload = buildCronFailureWebhookPayload({ evt: params.evt, job }); - - if (failureDest) { - if (failureDest.mode === "webhook" && failureDest.to) { - const webhookUrl = normalizeHttpWebhookUrl(failureDest.to); - if (webhookUrl) { - // Failure destinations mirror completion webhooks: notify in the - // background and log failures without rewriting the cron event result. - dispatchDetachedCronNotification({ - jobId: params.evt.jobId, - logger: params.logger, - deliver: () => - postCronWebhook({ - webhookUrl, - webhookToken: params.webhookToken, - ssrfPolicy: params.ssrfPolicy, - payload: failurePayload, - logContext: { jobId: params.evt.jobId }, - blockedLog: "cron: failure destination webhook blocked by SSRF guard", - failedLog: "cron: failure destination webhook failed", - logger: params.logger, - }), - }); - } else { - params.logger.warn( - { - jobId: params.evt.jobId, - webhookUrl: redactWebhookUrl(failureDest.to), - }, - "cron: failure destination webhook URL is invalid, skipping", - ); - } - return; - } - - if (failureDest.mode !== "announce") { - return; - } - } - - const primaryPlan = resolveCronDeliveryPlan(job); - const announceTarget = failureDest - ? { - channel: failureDest.channel, - to: failureDest.to, - accountId: failureDest.accountId, - sessionKey: deliverySessionKey, - // Explicit failure routes escape rejected primary delivery without inheriting its topic. - inheritSessionThread: false, - } - : primaryPlan.mode === "announce" && primaryPlan.requested - ? { - channel: primaryPlan.channel, - to: primaryPlan.to, - accountId: primaryPlan.accountId, - threadId: primaryPlan.threadId, - sessionKey: deliverySessionKey, - } - : undefined; - if (!announceTarget) { - return; - } - - const failureAlertText = [ - `Automation "${job.name}" ${params.evt.status === "error" ? "failed" : "delivery failed"}`, - ...cronFailureDetailLines(job.state.lastErrorReason, params.failureNotificationDetail), - ].join("\n"); - dispatchDetachedCronNotification({ - jobId: job.id, - logger: params.logger, - deliver: () => { - const { agentId, cfg: runtimeConfig } = params.resolveCronAgent(job.agentId); - return sendFailureAnnounce(params.deps, runtimeConfig, agentId, job.id, announceTarget, { - text: appendCronRunStarted(`⚠️ ${failureAlertText}`, params.evt.runAtMs, runtimeConfig), - }); - }, - }); } diff --git a/src/gateway/server-cron.test.ts b/src/gateway/server-cron.test.ts index eeb4bdc73b39..65a7bda1c603 100644 --- a/src/gateway/server-cron.test.ts +++ b/src/gateway/server-cron.test.ts @@ -29,7 +29,6 @@ const { loadConfigMock, fetchWithSsrFGuardMock, sendCronAnnouncePayloadStrictMock, - sendFailureNotificationAnnounceMock, runCronIsolatedAgentTurnMock, getGlobalHookRunnerMock, runCronChangedMock, @@ -51,7 +50,6 @@ const { loadConfigMock: vi.fn(), fetchWithSsrFGuardMock: vi.fn(), sendCronAnnouncePayloadStrictMock: vi.fn(async () => {}), - sendFailureNotificationAnnounceMock: vi.fn(async () => {}), runCronIsolatedAgentTurnMock: vi.fn(async () => ({ status: "ok", summary: "ok", @@ -181,7 +179,6 @@ vi.mock("../cron/delivery.js", async () => { return { ...actual, sendCronAnnouncePayloadStrict: sendCronAnnouncePayloadStrictMock, - sendFailureNotificationAnnounce: sendFailureNotificationAnnounceMock, }; }); @@ -314,7 +311,6 @@ describe("buildGatewayCronService", () => { loadConfigMock.mockClear(); fetchWithSsrFGuardMock.mockClear(); sendCronAnnouncePayloadStrictMock.mockClear(); - sendFailureNotificationAnnounceMock.mockClear(); runCronIsolatedAgentTurnMock.mockClear(); runCronChangedMock.mockClear(); getGlobalHookRunnerMock.mockClear(); @@ -1969,6 +1965,7 @@ describe("buildGatewayCronService", () => { sessionTarget: "isolated", wakeMode: "next-heartbeat", payload: { kind: "script", script: "return invalid" }, + failureAlert: { after: 1 }, delivery: { mode: "announce", channel: "telegram", @@ -1984,15 +1981,16 @@ describe("buildGatewayCronService", () => { runCronChangedMock.mockClear(); await state.cron.run(job.id, "force"); - await vi.waitFor(() => expect(sendFailureNotificationAnnounceMock).toHaveBeenCalledOnce()); + await vi.waitFor(() => expect(sendCronAnnouncePayloadStrictMock).toHaveBeenCalledOnce()); await vi.waitFor(() => expect(fetchWithSsrFGuardMock).toHaveBeenCalledOnce()); - const announce = requireRecord( - callArg(sendFailureNotificationAnnounceMock, 0, 5, "script failure announce"), - "script failure announce", + const announceRequest = requireRecord( + callArg(sendCronAnnouncePayloadStrictMock, 0, 0, "script failure announce request"), + "script failure announce request", ); + const announce = requireRecord(announceRequest.payload, "script failure announce"); expect(announce.text).toContain( - '⚠️ Automation "script failure detail" failed\nCause: automation script failed internally', + 'Automation "script failure detail" failed 1 times\nCause: automation script failed internally', ); expect(announce.text).not.toContain(rawError); diff --git a/src/gateway/server-cron.ts b/src/gateway/server-cron.ts index 06ff9ac1bf4b..b4df94a2c1db 100644 --- a/src/gateway/server-cron.ts +++ b/src/gateway/server-cron.ts @@ -987,7 +987,7 @@ export function buildGatewayCronService(params: { ssrfPolicy: webhookSsrfPolicy, }), log: getChildLogger({ module: "cron", storeKey: storePath }), - onEvent: (evt, context) => { + onEvent: (evt) => { // Any job/store change can alter session automation bindings, including // in-place enable flips during runs; run/schedule events bump too (cheap). bumpSessionAutomationVersion(); @@ -1074,14 +1074,12 @@ export function buildGatewayCronService(params: { const job = evt.job ?? cron.getJob(evt.jobId); dispatchGatewayCronFinishedNotifications({ evt, - failureNotificationDetail: context?.failureNotificationDetail, job, deps: params.deps, logger: cronLogger, resolveCronAgent, webhookToken: params.cfg.cron?.webhookToken, ssrfPolicy: webhookSsrfPolicy, - globalFailureDestination: params.cfg.cron?.failureAlert, }); } }, diff --git a/src/gateway/server-methods/cron.ts b/src/gateway/server-methods/cron.ts index 4a60c8c643e5..b5a67844b719 100644 --- a/src/gateway/server-methods/cron.ts +++ b/src/gateway/server-methods/cron.ts @@ -30,6 +30,7 @@ import { normalizeCronJobCreate, normalizeCronJobPatch } from "../../cron/normal import { toPublicCronJob } from "../../cron/public-job.js"; import type { CronRuntimeAuthority } from "../../cron/runtime-authority.js"; import { CRON_JOB_SCRATCH_MAX_BYTES } from "../../cron/scratch-contract.js"; +import { resolveFailureAlert } from "../../cron/service/failure-alerts.js"; import { applyJobPatch } from "../../cron/service/jobs.js"; import { isInvalidCronSessionTargetIdError, @@ -253,60 +254,31 @@ async function assertValidCronUpdatePatch(params: { delivery: effectiveDelivery, }); } - // failureAlert is a separate field from delivery, so a failureAlert-only patch - // skips the delivery check above. Validate when this edit touches a field that - // can change the announce channel routing: the alert's own channel/target/mode, - // or delivery itself (an alert without its own channel/target inherits the job - // delivery channel, so a delivery change can invalidate it). Editing unrelated - // alert fields (after/cooldown/includeSkipped) must not be blocked by a channel - // stored before this validation existed. The merged value carries the effective - // mode, and the validator no-ops for alerts that only inherit delivery. + // Compare the canonical before/after policy so route-changing edits are + // validated without blocking threshold-only edits on legacy stored channels. const failureAlertPatch = params.patch.failureAlert; const failureAlertRoutingPatched = failureAlertPatch && ("channel" in failureAlertPatch || "to" in failureAlertPatch || "mode" in failureAlertPatch); - // Enabling a previously OFF alert makes it start inheriting the job delivery - // route, so validate even when the enabling patch (`--failure-alert`, - // `--failure-alert-after`) carries no routing key of its own. An alert is - // already ON - so an object-only edit only changes threshold/cooldown - when it - // has per-job config or when global `cron.failureAlert.enabled` is true; - // resolveFailureAlert() treats those as active, so re-validating their inherited - // route would block unrelated edits on a legacy channel that already delivers. - const globalAlertsEnabled = params.cfg.cron?.failureAlert?.enabled === true; - const currentAlertActive = - params.currentJob.failureAlert !== false && - (params.currentJob.failureAlert !== undefined || globalAlertsEnabled); - const nextAlertActive = - nextJob.failureAlert !== false && (nextJob.failureAlert !== undefined || globalAlertsEnabled); - const alertNewlyEnabled = !currentAlertActive && nextAlertActive; - // A delivery change only affects the alert when the alert inherits the changed - // delivery field (its own channel/to is unset). Gating on that avoids blocking - // unrelated delivery edits (bestEffort, failureDestination) on jobs that carry - // a stale explicit alert channel. A delivery `mode` change is included because - // switching to/from webhook clears the inherited channel/target in - // mergeCronDelivery, which can make an inheriting alert ambiguous. - const deliveryPatch = params.patch.delivery; - const mergedAlert = nextJob.failureAlert; - const alertUsesInheritedChannel = !mergedAlert || mergedAlert.channel === undefined; - const alertUsesInheritedTarget = !mergedAlert || mergedAlert.to === undefined; - const deliveryAffectsInheritedAlert = - deliveryPatch && - nextAlertActive && - (("channel" in deliveryPatch && alertUsesInheritedChannel) || - ("to" in deliveryPatch && alertUsesInheritedTarget) || - ("mode" in deliveryPatch && alertUsesInheritedChannel)); - const currentAlertRoutingOverride = - params.currentJob.failureAlert && - (params.currentJob.failureAlert.channel !== undefined || - params.currentJob.failureAlert.to !== undefined || - params.currentJob.failureAlert.mode !== undefined); - const alertResetToGlobal = - failureAlertPatch === null && nextAlertActive && currentAlertRoutingOverride; + const currentAlert = resolveFailureAlert( + { deps: { cronConfig: params.cfg.cron } }, + params.currentJob, + ); + const nextAlert = resolveFailureAlert( + { deps: { cronConfig: params.cfg.cron } }, + { ...nextJob, delivery: effectiveDelivery }, + ); + const alertNewlyEnabled = currentAlert === null && nextAlert !== null; + const alertRouteChanged = + currentAlert?.mode !== nextAlert?.mode || + currentAlert?.channel !== nextAlert?.channel || + currentAlert?.to !== nextAlert?.to || + currentAlert?.accountId !== nextAlert?.accountId || + currentAlert?.threadId !== nextAlert?.threadId; if ( failureAlertRoutingPatched || alertNewlyEnabled || - deliveryAffectsInheritedAlert || - alertResetToGlobal + (alertRouteChanged && (params.patch.delivery !== undefined || failureAlertPatch === null)) ) { await assertValidCronFailureAlert({ cfg: params.cfg, diff --git a/src/gateway/server-methods/cron.validation.test.ts b/src/gateway/server-methods/cron.validation.test.ts index 27c0028d25ee..46065d639900 100644 --- a/src/gateway/server-methods/cron.validation.test.ts +++ b/src/gateway/server-methods/cron.validation.test.ts @@ -2952,17 +2952,17 @@ describe("cron method validation", () => { createCronJob({ failureAlert: { channel: "c0example01", mode: "announce" } }), ); - // Job omits mode and inherits the global webhook mode, so runtime never uses - // the channel; validation must not reject it (matches resolveFailureAlert). - failureAlertUpdateAccepted( - "does not validate an inherited-webhook failureAlert channel on cron.update (global mode)", + // An explicit job channel selects announce routing ahead of the global + // webhook destination, so the canonical resolver must validate it. + failureAlertUpdateRejected( + "validates an explicit failureAlert channel ahead of a global webhook on cron.update", { failureAlert: { channel: "C0EXAMPLE01", to: "https://example.invalid/hook" } }, createCronJob(), globalFailureAlertConfig(telegramSlackConfig(), { enabled: true, mode: "webhook" }), ); - failureAlertAddAccepted( - "does not validate an inherited-webhook failureAlert channel on cron.add (global mode)", + failureAlertAddRejected( + "validates an explicit failureAlert channel ahead of a global webhook on cron.add", agentTurnCronParams({ name: "inherited webhook alert", failureAlert: { channel: "C0EXAMPLE01", to: "https://example.invalid/hook" }, @@ -3031,10 +3031,10 @@ describe("cron method validation", () => { }), ); - // Enabling an alert with no routing key of its own makes it inherit the job - // delivery channel; a legacy-invalid one must be rejected, not persisted. - failureAlertUpdateRejected( - "validates a newly enabled alert (--failure-alert-after) that inherits a legacy-invalid delivery channel", + // Route-backed alerts are already active by default, so a threshold-only edit + // must not revalidate a legacy channel that the patch does not change. + failureAlertUpdateAccepted( + "does not revalidate a route-backed legacy channel for a threshold-only edit", { failureAlert: { after: 3 } }, createRoutedCronJob("c0legacyinvalid", "123"), ); diff --git a/src/gateway/server.cron.test.ts b/src/gateway/server.cron.test.ts index bfca2818b71a..fbff3a6f0962 100644 --- a/src/gateway/server.cron.test.ts +++ b/src/gateway/server.cron.test.ts @@ -34,10 +34,8 @@ const fetchWithSsrFGuardMock = vi.hoisted(() => })), ); -const sendFailureNotificationAnnounceMock = vi.hoisted(() => - vi.fn( - async () => undefined, - ), +const sendCronAnnouncePayloadStrictMock = vi.hoisted(() => + vi.fn(async () => undefined), ); const closeTrackedBrowserTabsForSessionsMock = vi.hoisted(() => vi.fn(async () => 0)); @@ -56,7 +54,7 @@ vi.mock("../cron/delivery.js", async () => { const actual = await vi.importActual("../cron/delivery.js"); return { ...actual, - sendFailureNotificationAnnounce: sendFailureNotificationAnnounceMock, + sendCronAnnouncePayloadStrict: sendCronAnnouncePayloadStrictMock, }; }); @@ -335,6 +333,7 @@ async function addWebhookCronJob(params: { sessionTarget?: "main" | "isolated"; payloadText?: string; delivery: Record; + failureAlert?: Record; }) { const response = await rpcReq(params.ws, "cron.add", { name: params.name, @@ -349,6 +348,7 @@ async function addWebhookCronJob(params: { : { text: params.payloadText ?? "send webhook" }), }, delivery: params.delivery, + ...(params.failureAlert ? { failureAlert: params.failureAlert } : {}), }); return expectCronJobIdFromResponse(response); } @@ -392,22 +392,23 @@ function expectFailureAnnounceCall(params: { message: string; includeRunStarted?: boolean; }) { - expect(sendFailureNotificationAnnounceMock).toHaveBeenCalledTimes(1); - const call = sendFailureNotificationAnnounceMock.mock.calls.at(0); + expect(sendCronAnnouncePayloadStrictMock).toHaveBeenCalledTimes(1); + const call = sendCronAnnouncePayloadStrictMock.mock.calls.at(0); if (!call) { throw new Error("expected failure announcement call"); } - const args = call; - expect(typeof args[2]).toBe("string"); - expect(args[3]).toBe(params.jobId); - expect(args[4]).toEqual({ + const [request] = call; + expect(typeof request.agentId).toBe("string"); + expect(request.jobId).toBe(params.jobId); + expect(request.target).toEqual({ channel: params.channel, to: params.to, accountId: undefined, + threadId: undefined, sessionKey: params.sessionKey, ...(params.inheritSessionThread === false ? { inheritSessionThread: false } : {}), }); - const payload = expectDefined(args[5], "failure reply payload"); + const payload = expectDefined(request.payload, "failure reply payload"); if (params.includeRunStarted) { const lines = expectDefined(payload.text, "failure reply text").split("\n"); expect(lines).toEqual([ @@ -466,7 +467,7 @@ describe("gateway server cron", () => { beforeEach(() => { // Keep polling helpers deterministic even if other tests left fake timers enabled. vi.useRealTimers(); - sendFailureNotificationAnnounceMock.mockClear(); + sendCronAnnouncePayloadStrictMock.mockClear(); closeTrackedBrowserTabsForSessionsMock.mockClear(); }); @@ -1913,6 +1914,7 @@ describe("gateway server cron", () => { await writeCronConfig({ cron: { webhookToken: "cron-webhook-token", + failureAlert: { after: 1 }, }, }); @@ -2033,7 +2035,7 @@ describe("gateway server cron", () => { expect(failureDestCall.url).toBe("https://example.invalid/failure-destination"); const failureDestBody = failureDestCall.body; expect(failureDestBody.message).toBe( - 'Automation "failure destination webhook" failed: unknown error', + 'Automation "failure destination webhook" failed 1 times\nLast error: unknown reason', ); fetchWithSsrFGuardMock.mockClear(); @@ -2203,6 +2205,7 @@ describe("gateway server cron", () => { mode: "announce", channel: "last", }, + failureAlert: { after: 1 }, }); const updateRes = await rpcReq(ws, "cron.update", { @@ -2225,7 +2228,7 @@ describe("gateway server cron", () => { channel: "last", sessionKey: "agent:main:telegram:direct:123:thread:99", message: - '⚠️ Automation "primary delivery fallback" failed\n' + + 'Automation "primary delivery fallback" failed 1 times\n' + "Check automation history for details.", includeRunStarted: true, }); @@ -2255,7 +2258,7 @@ describe("gateway server cron", () => { await connectOk(ws); try { - sendFailureNotificationAnnounceMock.mockClear(); + sendCronAnnouncePayloadStrictMock.mockClear(); fetchWithSsrFGuardMock.mockClear(); cronIsolatedRun.mockResolvedValueOnce({ status: "error", summary: "delivery failed" }); @@ -2270,6 +2273,7 @@ describe("gateway server cron", () => { to: "#alerts", }, }, + failureAlert: { after: 1 }, }); const finished = waitForCronEvent( @@ -2285,7 +2289,9 @@ describe("gateway server cron", () => { to: "#alerts", sessionKey: undefined, inheritSessionThread: false, - message: '⚠️ Automation "channel fd no mode" failed\nCheck automation history for details.', + message: + 'Automation "channel fd no mode" failed 1 times\n' + + "Check automation history for details.", includeRunStarted: true, }); expect(fetchWithSsrFGuardMock).not.toHaveBeenCalled(); @@ -2316,6 +2322,7 @@ describe("gateway server cron", () => { mode: "announce", channel: "last", }, + failureAlert: { after: 1 }, }); const jobId = expectCronJobIdFromResponse(addRes); @@ -2339,7 +2346,7 @@ describe("gateway server cron", () => { channel: "last", sessionKey: "agent:avery:feishu:direct:ou_founder", message: - '⚠️ Automation "session target failure fallback" failed\n' + + 'Automation "session target failure fallback" failed 1 times\n' + "Check automation history for details.", includeRunStarted: true, }); diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json index 21da7c6a81f2..e6c6a9a001ef 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json +++ b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json @@ -279,7 +279,7 @@ "tools": [ { "deferLoading": true, - "description": "Gateway scheduler: reminders, delayed self-wakeups, loops, recurring work, event watchers. Never exec sleep/poll as timer.\n\nACTIONS: status | list [includeDisabled,limit?,offset?] (use nextOffset for the next page) | get jobId | add job | update jobId job (partial: only supplied fields change; null clears) | remove jobId | run jobId (runMode \"force\"=now) | runs jobId = history | next_check in:\"30m\" (own paced run only) | wake text mode?:\"now\"|\"next-heartbeat\"(default) nudges a caller-owned lane (sessionKey/agentId to pick another).\n\nADD: {name?,schedule,payload,sessionTarget?,pacing?,trigger?,delivery?,enabled?}. Required: schedule+payload.\n\nSCHEDULE:\n- {kind:\"at\",at:\"ISO-8601\"} one-shot; no tz=UTC; auto-deletes after run.\n- {kind:\"every\",everyMs}.\n- {kind:\"cron\",expr,tz?:\"IANA\"}: expr is wall time in tz; never pre-convert to UTC; no tz=gateway host local. 18:00 Shanghai => {expr:\"0 18 * * *\",tz:\"Asia/Shanghai\"}.\n- {kind:\"stream\",command:[argv],mode?:\"line\"|\"match\",match?}: fires on supervised process output; disabled only when cron.triggers.enabled=false.\n\nTARGET+PAYLOAD:\n- \"current\" (agentTurn default) = this conversation: run carries this chat's context, result lands here. Self-wakeup/\"continue later\"/loop = at|every + agentTurn + current.\n- \"isolated\" = fresh detached session (shows in `openclaw tasks`); standalone background work.\n- \"main\" = heartbeat lane; payload {kind:\"systemEvent\",text} (systemEvent default target).\n- \"session:\" = named session.\n- agentTurn {kind:\"agentTurn\",message,model?,thinking?,timeoutSeconds?}; timeoutSeconds 0=none.\n- Inherited configured MCP authority includes only model-callable tools; interactive app-view-only capabilities are excluded from headless jobs.\n- script {kind:\"script\",script,timeoutSeconds?,toolBudget?}: main|isolated only; disabled only when cron.triggers.enabled=false.\n\nPACED LOOP: recurring job + pacing{min?,max?} durations (\"15m\",\"4h\"; at least one). Inside its run, job calls next_check in:\"\" to set the next delay (clamped to bounds, measured from run end; failed runs keep normal backoff). Adaptive polling: tighten when active, back off when quiet.\n\nTRIGGER (condition watcher on every/cron): {script,once?}; available unless cron.triggers.enabled=false — if off, say so; never model-poll instead. Quiet headless check, no model; 30s/5 tool calls/16KB state. Read frozen trigger.state, return json({fire,message?,state?}) with NEW state; dedupe via state, never memory. fire:false saves state only. fire:true runs payload; message is that run's entire context — self-contained. Fire on failures/timeouts too; success-only watchers look healthy when broken. Script stays read-only; actions belong in payload. once:true disables after first fire. Code Mode: await tools.call(\"exec\",{command:\"...\"}).\n\nDELIVERY {mode:\"none\"|\"announce\"|\"webhook\",channel?,to?,threadId?,bestEffort?,completionDestination?}: where detached run output goes. Omitted=announce (current=>this chat; isolated=>last route; set channel/to for a specific chat — no messaging tool inside the run). Silent watcher=>mode:\"none\". webhook posts finished-run event to URL in `to`. To keep announce delivery and also POST completion, use mode:\"announce\" with completionDestination:{mode:\"webhook\",to:\"https://...\"}.\n\nJob wakeMode (main jobs): \"now\"(default)|\"next-heartbeat\". Restricted automation-run sessions: self status/list/get/runs/remove + own next_check only. failureAlert {...}|false disables. jobId canonical (id=compat). contextMessages 0-10 embeds recent chat lines into reminder text.", + "description": "Gateway scheduler: reminders, delayed self-wakeups, loops, recurring work, event watchers. Never exec sleep/poll as timer.\n\nACTIONS: status | list [includeDisabled,limit?,offset?] (use nextOffset for the next page) | get jobId | add job | update jobId job (partial: only supplied fields change; null clears) | remove jobId | run jobId (runMode \"force\"=now) | runs jobId = history | next_check in:\"30m\" (own paced run only) | wake text mode?:\"now\"|\"next-heartbeat\"(default) nudges a caller-owned lane (sessionKey/agentId to pick another).\n\nADD: {name?,schedule,payload,sessionTarget?,pacing?,trigger?,delivery?,enabled?}. Required: schedule+payload.\n\nSCHEDULE:\n- {kind:\"at\",at:\"ISO-8601\"} one-shot; no tz=UTC; auto-deletes after run.\n- {kind:\"every\",everyMs}.\n- {kind:\"cron\",expr,tz?:\"IANA\"}: expr is wall time in tz; never pre-convert to UTC; no tz=gateway host local. 18:00 Shanghai => {expr:\"0 18 * * *\",tz:\"Asia/Shanghai\"}.\n- {kind:\"stream\",command:[argv],mode?:\"line\"|\"match\",match?}: fires on supervised process output; disabled only when cron.triggers.enabled=false.\n\nTARGET+PAYLOAD:\n- \"current\" (agentTurn default) = this conversation: run carries this chat's context, result lands here. Self-wakeup/\"continue later\"/loop = at|every + agentTurn + current.\n- \"isolated\" = fresh detached session (shows in `openclaw tasks`); standalone background work.\n- \"main\" = heartbeat lane; payload {kind:\"systemEvent\",text} (systemEvent default target).\n- \"session:\" = named session.\n- agentTurn {kind:\"agentTurn\",message,model?,thinking?,timeoutSeconds?}; timeoutSeconds 0=none.\n- Inherited configured MCP authority includes only model-callable tools; interactive app-view-only capabilities are excluded from headless jobs.\n- script {kind:\"script\",script,timeoutSeconds?,toolBudget?}: main|isolated only; disabled only when cron.triggers.enabled=false.\n\nPACED LOOP: recurring job + pacing{min?,max?} durations (\"15m\",\"4h\"; at least one). Inside its run, job calls next_check in:\"\" to set the next delay (clamped to bounds, measured from run end; failed runs keep normal backoff). Adaptive polling: tighten when active, back off when quiet.\n\nTRIGGER (condition watcher on every/cron): {script,once?}; available unless cron.triggers.enabled=false — if off, say so; never model-poll instead. Quiet headless check, no model; 30s/5 tool calls/16KB state. Read frozen trigger.state, return json({fire,message?,state?}) with NEW state; dedupe via state, never memory. fire:false saves state only. fire:true runs payload; message is that run's entire context — self-contained. Fire on failures/timeouts too; success-only watchers look healthy when broken. Script stays read-only; actions belong in payload. once:true disables after first fire. Code Mode: await tools.call(\"exec\",{command:\"...\"}).\n\nDELIVERY {mode:\"none\"|\"announce\"|\"webhook\",channel?,to?,threadId?,bestEffort?,completionDestination?}: where detached run output goes. Omitted=announce (current=>this chat; isolated=>last route; set channel/to for a specific chat — no messaging tool inside the run). Silent watcher=>mode:\"none\". webhook posts finished-run event to URL in `to`. To keep announce delivery and also POST completion, use mode:\"announce\" with completionDestination:{mode:\"webhook\",to:\"https://...\"}.\n\nFAILURE ALERTS: jobs with a failure route default to alerting after 2 consecutive execution failures with a 1h cooldown. Route order: job failureAlert fields, delivery.failureDestination over global cron.failureAlert destination fields, then primary announce. failureAlert:false disables execution/delivery alerts, not the auto-disable safety notice; a failureAlert object activates/tunes. bestEffort suppresses inherited execution alerts. Required completion-delivery failure uses only an alternate route immediately and does not increment the execution streak.\n\nJob wakeMode (main jobs): \"now\"(default)|\"next-heartbeat\". Restricted automation-run sessions: self status/list/get/runs/remove + own next_check only. failureAlert {...}|false disables. jobId canonical (id=compat). contextMessages 0-10 embeds recent chat lines into reminder text.", "inputSchema": { "additionalProperties": true, "properties": { @@ -460,7 +460,7 @@ "type": "null" } ], - "description": "Failure destination; null clears." + "description": "Failure-alert route override and alternate for immediate required-delivery failure; null clears." }, "mode": { "description": "Delivery mode", @@ -516,7 +516,7 @@ }, "failureAlert": { "additionalProperties": true, - "description": "Failure alert; false disables.", + "description": "Failure alert policy/route override. Route-backed jobs default to after=2 and cooldownMs=3600000; false disables execution/delivery alerts but not the auto-disable safety notice.", "properties": { "accountId": { "type": "string" diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md index 484f6d18c504..ccc658155a8e 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md +++ b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md @@ -231,8 +231,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the "roughTokens": 0 }, "dynamicToolsJson": { - "chars": 53942, - "roughTokens": 13486 + "chars": 54719, + "roughTokens": 13680 }, "openClawDeveloperInstructions": { "chars": 4499, @@ -243,8 +243,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the "roughTokens": 7221 }, "totalWithDynamicToolsJson": { - "chars": 82826, - "roughTokens": 20707 + "chars": 83603, + "roughTokens": 20901 }, "userInputText": { "chars": 1300, diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md index c7c3aed292bc..523a7685bfd6 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md +++ b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md @@ -231,8 +231,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the "roughTokens": 0 }, "dynamicToolsJson": { - "chars": 53634, - "roughTokens": 13409 + "chars": 54411, + "roughTokens": 13603 }, "openClawDeveloperInstructions": { "chars": 3390, @@ -243,8 +243,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the "roughTokens": 6851 }, "totalWithDynamicToolsJson": { - "chars": 81038, - "roughTokens": 20260 + "chars": 81815, + "roughTokens": 20454 }, "userInputText": { "chars": 929, diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md index 18242e987395..43e4b8294279 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md +++ b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md @@ -226,8 +226,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the "roughTokens": 0 }, "dynamicToolsJson": { - "chars": 55191, - "roughTokens": 13798 + "chars": 55968, + "roughTokens": 13992 }, "openClawDeveloperInstructions": { "chars": 3390, @@ -238,8 +238,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the "roughTokens": 6955 }, "totalWithDynamicToolsJson": { - "chars": 83011, - "roughTokens": 20753 + "chars": 83788, + "roughTokens": 20947 }, "userInputText": { "chars": 1271,