ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-23 02:45:38 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(cron): honor failure alert thresholds (#126483)

Browse Source 
* 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
This commit is contained in:
Peter Steinberger
2026-08-19 17:25:14 -07:00
committed by GitHub
parent 035853b921
commit 177828aa75
35 changed files with 1092 additions and 1491 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/automation/cron-jobs.md
+14 -5
View File
@@ -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 <n>`, `--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 <jobId>`; 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 <jobId>`; 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
docs/cli/cron.md
+6 -2
View File
@@ -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.
<Note>
Main-session jobs may only use `delivery.failureDestination` when primary delivery mode is `webhook`. Isolated jobs accept it in all modes.
</Note>
@@ -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
docs/gateway/configuration-reference.md
+11 -6
View File
@@ -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).
src/agents/tools/cron-tool-schema.ts
+6 -4
View File
@@ -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.",
}),
);
}
src/agents/tools/cron-tool.ts
+2
View File
@@ -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.`;
}
src/cron/cron-delivery-outcomes.e2e.test.ts
+234 -23
View File
@@ -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<void>;
requests: WebhookRequest[];
request: Promise<WebhookRequest>;
url: string;
}> {
const requests: WebhookRequest[] = [];
let resolveRequest!: (request: WebhookRequest) => void;
const request = new Promise<WebhookRequest>((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<string, unknown>,
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 {
src/cron/delivery-plan.ts
+60 -40
View File
@@ -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<CronJob, "delivery"> & Partial<Pick<CronJob, "payload" | "sessionTarget">>,
): 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<CronJob, "delivery">,
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;
}
src/cron/delivery.failure-notify.test.ts
+50 -330
View File
@@ -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<void>((_resolve, reject) => {
deliverySignal = abortSignal;
if (honorsCancellation) {
abortSignal.addEventListener(
"abort",
() =>
reject(
abortSignal.reason instanceof Error
? abortSignal.reason
: new Error("failure notification was aborted"),
),
{ once: true },
);
}
}),
);
const notification = sendFailureNotificationAnnounce(
{} as never,
{} as never,
"main",
"job-1",
{ channel: "telegram", to: "123" },
{ 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);
});
});
src/cron/delivery.test.ts
+36 -43
View File
@@ -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({
src/cron/delivery.ts
+2 -67
View File
@@ -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<void> {
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",
);
}
}
src/cron/service.failure-alert.test.ts
+66 -20
View File
@@ -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,
src/cron/service.persists-delivered-status.test.ts
+15 -75
View File
@@ -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 () => {
src/cron/service/failure-alerts.persistence.test.ts
+15 -7
View File
@@ -98,10 +98,9 @@ describe("cron failure alert persistence", () => {
const alertDone = new Promise<void>((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",
},
});
});
});
src/cron/service/failure-alerts.ts
+213 -93
View File
@@ -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,
});
}
}
src/cron/service/ops.test.ts
+3
View File
@@ -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);
});
src/cron/service/run-recovery.test.ts
+112 -1
View File
@@ -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<typeof createCronServiceState>[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");
src/cron/service/startup-run-repair.ts
+27 -31
View File
@@ -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
src/cron/service/state.ts
+1
View File
@@ -260,6 +260,7 @@ export type CronServiceDeps = {
mode?: "announce" | "webhook";
accountId?: string;
threadId?: string | number;
inheritSessionThread?: false;
}) => Promise<void>;
onEvent?: (evt: CronEvent, context?: CronEventContext) => void;
};
src/cron/service/timer-job-runner.ts
+1 -2
View File
@@ -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,
src/cron/service/timer-outcomes.ts
+32 -29
View File
@@ -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. */
src/cron/service/timer-trigger.ts
+8 -31
View File
@@ -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) {
src/cron/service/timer.regression.test.ts
+10 -3
View File
@@ -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)", () => {
src/cron/service/timer.timeout-watchdog.test.ts
+1
View File
@@ -807,6 +807,7 @@ describe("cron service timer regressions", () => {
mode: "announce",
accountId: undefined,
threadId: undefined,
inheritSessionThread: false,
});
} finally {
vi.useRealTimers();
src/gateway/server-cron-notifications.delivery-failure.test.ts
-138
View File
@@ -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<typeof import("../cron/delivery.js")>();
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,
});
});
});
src/gateway/server-cron-notifications.test.ts
+70 -263
View File
@@ -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<typeof import("../cron/delivery.js")>();
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 =
src/gateway/server-cron-notifications.ts
+20 -176
View File
@@ -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<string, string>
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<void> {
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),
});
},
});
}
src/gateway/server-cron.test.ts
+7 -9
View File
@@ -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<RunCronIsolatedAgentTurnMock>(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);
src/gateway/server-cron.ts
+1 -3
View File
@@ -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,
});
}
},
src/gateway/server-methods/cron.ts
+19 -47
View File
@@ -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,
src/gateway/server-methods/cron.validation.test.ts
+10 -10
View File
@@ -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"),
);
src/gateway/server.cron.test.ts
+25 -18
View File
@@ -34,10 +34,8 @@ const fetchWithSsrFGuardMock = vi.hoisted(() =>
})),
);
const sendFailureNotificationAnnounceMock = vi.hoisted(() =>
vi.fn<typeof import("../cron/delivery.js").sendFailureNotificationAnnounce>(
async () => undefined,
),
const sendCronAnnouncePayloadStrictMock = vi.hoisted(() =>
vi.fn<typeof import("../cron/delivery.js").sendCronAnnouncePayloadStrict>(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<typeof import("../cron/delivery.js")>("../cron/delivery.js");
return {
...actual,
sendFailureNotificationAnnounce: sendFailureNotificationAnnounceMock,
sendCronAnnouncePayloadStrict: sendCronAnnouncePayloadStrictMock,
};
});
@@ -335,6 +333,7 @@ async function addWebhookCronJob(params: {
sessionTarget?: "main" | "isolated";
payloadText?: string;
delivery: Record<string, unknown>;
failureAlert?: Record<string, unknown>;
}) {
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,
});
test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json
Vendored
+3 -3
View File
@@ -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:<key>\" = 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:\"<dur>\" 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:<key>\" = 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:\"<dur>\" 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"
test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md
Vendored
+4 -4
View File
@@ -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,
test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md
Vendored
+4 -4
View File
@@ -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,
test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md
Vendored
+4 -4
View File
@@ -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,