From 2bcc06cc2270db529b1b6a5fd301b9fa401e65ba Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 18 Aug 2026 22:44:33 -0700 Subject: [PATCH] fix(cron): required delivery failures no longer report success (#126164) * fix(cron): preserve required delivery completion Record durable completion independently from payload execution so required delivery failure cannot delete one-shots or report successful waits.\n\nCloses #126163 * fix(cron): keep completion contracts acyclic * fix(cron): keep delivery predicate private --- .../OpenClawProtocol/GatewayModels.swift | 4 + docs/automation/cron-jobs.md | 6 +- docs/cli/cron.md | 4 +- .../src/cron-validators.test.ts | 27 +++++ packages/gateway-protocol/src/schema/cron.ts | 6 ++ src/cli/cron-cli.test.ts | 38 +++++-- src/cli/cron-cli/register.cron-simple.ts | 27 ++++- src/cron/completion-status.ts | 63 +++++++++++ src/cron/run-log-types.ts | 2 + .../service.persists-delivered-status.test.ts | 92 ++++++++++++++++ ...runs-one-shot-main-job-disables-it.test.ts | 101 ++++++++++++++++++ .../failure-alerts.persistence.test.ts | 7 +- src/cron/service/ops-run-preparation.ts | 12 ++- src/cron/service/ops-run.ts | 7 +- src/cron/service/ops.test.ts | 3 + src/cron/service/run-admission.ts | 12 ++- .../startup-run-repair.auto-disable.test.ts | 70 ++++++++++++ src/cron/service/startup-run-repair.ts | 8 ++ src/cron/service/state.ts | 2 + src/cron/service/task-runs.test.ts | 2 + src/cron/service/task-runs.ts | 14 ++- src/cron/service/timer-execution-timeout.ts | 6 ++ src/cron/service/timer-job-runner.ts | 36 ++++++- ...imer-outcome-finalization.receipts.test.ts | 25 +++-- .../service/timer-outcome-finalization.ts | 3 +- src/cron/service/timer-outcomes.ts | 28 ++--- src/cron/service/timer-trigger.ts | 9 +- .../service/timer.batch-finalization.test.ts | 44 +++++--- src/cron/service/timer.pacing.test.ts | 20 ++-- src/cron/service/timer.ts | 2 +- src/cron/task-run-detail.ts | 20 +++- src/cron/task-run-event-codec.test.ts | 3 +- src/cron/task-run-event-codec.ts | 1 + src/cron/task-run-history.test.ts | 67 ++++++++++++ src/cron/types.ts | 9 ++ ...ron-notifications.delivery-failure.test.ts | 29 ++++- src/gateway/server-cron-notifications.ts | 14 ++- src/gateway/server-cron.test.ts | 97 +++++++++++++++-- src/gateway/server-cron.ts | 22 +--- src/plugins/hook-types.ts | 1 + 40 files changed, 827 insertions(+), 116 deletions(-) create mode 100644 src/cron/completion-status.ts diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift index eea0066deed0..5006694ba94a 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift @@ -16541,6 +16541,7 @@ public struct CronRunLogEntry: Codable, Sendable { public let jobid: String public let action: String public let status: AnyCodable? + public let completionstatus: AnyCodable? public let error: String? public let errorreason: AnyCodable? public let summary: String? @@ -16567,6 +16568,7 @@ public struct CronRunLogEntry: Codable, Sendable { jobid: String, action: String, status: AnyCodable? = nil, + completionstatus: AnyCodable? = nil, error: String? = nil, errorreason: AnyCodable? = nil, summary: String? = nil, @@ -16592,6 +16594,7 @@ public struct CronRunLogEntry: Codable, Sendable { self.jobid = jobid self.action = action self.status = status + self.completionstatus = completionstatus self.error = error self.errorreason = errorreason self.summary = summary @@ -16619,6 +16622,7 @@ public struct CronRunLogEntry: Codable, Sendable { case jobid = "jobId" case action case status + case completionstatus = "completionStatus" case error case errorreason = "errorReason" case summary diff --git a/docs/automation/cron-jobs.md b/docs/automation/cron-jobs.md index d6c60d9f6bdf..9cc6300b0244 100644 --- a/docs/automation/cron-jobs.md +++ b/docs/automation/cron-jobs.md @@ -44,7 +44,7 @@ Manage automations with the `openclaw automations` CLI; `openclaw cron` remains - Automations run **inside the Gateway process**, not inside the model. The Gateway must be running for schedules to fire. - Job definitions, runtime state, and run history persist in OpenClaw's shared SQLite state database, so restarts do not lose schedules. - Every automation run creates a [background task](/automation/tasks) record. -- One-shot jobs (`--at`) auto-delete after success by default; pass `--keep-after-run` to keep them. +- One-shot jobs (`--at`) auto-delete only when run `completionStatus` is `succeeded`; pass `--keep-after-run` to keep successful jobs. A required-delivery failure or unknown completion keeps the job disabled for inspection and restart recovery without replaying the payload. - Per-run wall-clock budget: `--timeout-seconds` when set. Otherwise, isolated/detached agent-turn jobs are bounded by the scheduler's own 60-minute watchdog before the underlying agent-turn timeout (`agents.defaults.timeoutSeconds`, default 48 hours) would ever apply; command jobs default to 10 minutes, and script payloads default to 5 minutes. - On Gateway startup, overdue isolated agent-turn jobs are rescheduled instead of replayed immediately, keeping model/tool bootstrap work out of the channel-connect window. - If you drive `openclaw agent` from system cron or another external scheduler, wrap it with a hard-kill escalation even though the CLI already handles `SIGTERM`/`SIGINT`. Gateway-backed runs ask the Gateway to abort accepted runs; `--local` runs get the same abort signal. For GNU `timeout`, prefer `timeout -k 60 600 openclaw agent ...` over plain `timeout 600 ...` — the `-k` value is the backstop if the process cannot drain in time. For systemd units, use a `SIGTERM` stop signal with a grace window (`TimeoutStopSec`) before the final kill. Reusing a `--run-id` while the original Gateway run is still active reports the duplicate as in-flight instead of starting a second run. @@ -510,7 +510,9 @@ openclaw automations edit --clear-agent Archiving a session (Control UI, or `sessions.patch { key, archived: true, expectedSessionId }` using the durable ID from `sessions.list`) disables every enabled automation job bound to that session: its isolated `cron:` session, a `session:` target, or a delivery/wake `sessionKey` lane. Restoring the session requires the same observed identity and does not re-enable those jobs; use `openclaw automations enable `. Sessions with an enabled bound job show a clock badge in the Control UI sidebar. -`openclaw automations run ` returns after enqueueing the manual run. Use `--wait` for shutdown hooks, maintenance scripts, or other automation that must block until the queued run finishes; it polls the returned `runId` (default timeout `10m`, poll interval `2s`) and exits `0` for status `ok`, non-zero for `error`, `skipped`, or a wait timeout. +`openclaw automations run ` returns after enqueueing the manual run. Use `--wait` for shutdown hooks, maintenance scripts, or other automation that must block until the queued run finishes; it polls the returned `runId` (default timeout `10m`, poll interval `2s`) and exits `0` only for `completionStatus: "succeeded"`. Failed or unknown completion and wait timeouts exit non-zero. + +Run history keeps payload execution in `status` (`ok`, `error`, or `skipped`) and whole-run completion in `completionStatus` (`succeeded`, `failed`, or `unknown`). Delivery is required only when the admitted job explicitly sets `delivery.bestEffort: false`; delivery-only failure leaves execution `status: "ok"`, does not increment execution error counters or enter retry backoff, and records `completionStatus: "failed"`. Direct Gateway event sources can use `cron.run` with `mode: "if-enabled"` to run immediately without overriding an operator-disabled or auto-disabled job. Explicit operator run-now commands continue to use `force`. diff --git a/docs/cli/cron.md b/docs/cli/cron.md index e8b9e83526c4..3d83f3351562 100644 --- a/docs/cli/cron.md +++ b/docs/cli/cron.md @@ -137,7 +137,7 @@ If an isolated run times out before the first model request, `openclaw automatio `--at ` schedules a one-shot run. Offset-less datetimes are treated as UTC unless you also pass `--tz `, which interprets the wall-clock time in the given timezone. -One-shot jobs delete after success by default. Use `--keep-after-run` to preserve them. +One-shot jobs delete only after `completionStatus: "succeeded"`. Required-delivery failure or unknown completion keeps the job disabled, with no next run, so restarts do not replay payload side effects. Use `--keep-after-run` to preserve successful jobs too. ### Recurring jobs @@ -165,7 +165,7 @@ Add `--wait` when a script should block until that exact queued run records a te openclaw automations run --wait --wait-timeout 10m --poll-interval 2s ``` -With `--wait`, the CLI still calls `cron.run` first, then polls `cron.runs` for the returned `runId`. The command exits `0` only when the run finishes with status `ok`. It exits non-zero when the run finishes with `error` or `skipped`, when the Gateway response does not include a `runId`, or when `--wait-timeout` expires (default `10m`, polled every `2s` by default). `--poll-interval` must be greater than zero. +With `--wait`, the CLI calls `cron.run` first, then polls the durable `cron.runs` row for the returned `runId`; it does not reread mutable job delivery settings. JSON reports payload execution as `status` and whole-run completion as `completionStatus`. The command exits `0` only for `completionStatus: "succeeded"`; `failed`, `unknown`, execution errors/skips, a missing `runId`, and timeout expiry exit non-zero (default `10m`, polled every `2s` by default). `--poll-interval` must be greater than zero. Use `--due` when you want the manual command to run only if the job is currently due. If `--due --wait` does not enqueue a run, the command returns the normal non-run response instead of polling. diff --git a/packages/gateway-protocol/src/cron-validators.test.ts b/packages/gateway-protocol/src/cron-validators.test.ts index b9644c7e6d86..b63cf5c35b1b 100644 --- a/packages/gateway-protocol/src/cron-validators.test.ts +++ b/packages/gateway-protocol/src/cron-validators.test.ts @@ -102,6 +102,33 @@ describe("cron protocol validators", () => { ).toBe(false); }); + it.each(["succeeded", "failed", "unknown"] as const)( + "accepts additive cron completion status %s", + (completionStatus) => { + expect( + Value.Check(CronRunLogEntrySchema, { + ts: 1, + jobId: "job-1", + action: "finished", + status: "ok", + completionStatus, + }), + ).toBe(true); + }, + ); + + it("rejects unknown cron completion status values", () => { + expect( + Value.Check(CronRunLogEntrySchema, { + ts: 1, + jobId: "job-1", + action: "finished", + status: "ok", + completionStatus: "partial", + }), + ).toBe(false); + }); + it("rejects client-authored scheduled authority provenance", () => { const scheduledToolPolicy = { version: 1, mode: "trusted" } as const; expectCases(validateCronAddParams, false, [add({ scheduledToolPolicy })]); diff --git a/packages/gateway-protocol/src/schema/cron.ts b/packages/gateway-protocol/src/schema/cron.ts index f4ce9f9824cf..706ed9dacb4a 100644 --- a/packages/gateway-protocol/src/schema/cron.ts +++ b/packages/gateway-protocol/src/schema/cron.ts @@ -97,6 +97,11 @@ function cronRunStatusSchema(options: Record = {}) { } const CronRunStatusSchema = cronRunStatusSchema(); +const CronCompletionStatusSchema = Type.Union([ + Type.Literal("succeeded"), + Type.Literal("failed"), + Type.Literal("unknown"), +]); const CronConfigRevisionSchema = Type.String({ minLength: 1, maxLength: 128 }); const DeprecatedCronRunStatusSchema = cronRunStatusSchema({ deprecated: true, @@ -753,6 +758,7 @@ export const CronRunLogEntrySchema = closedObject({ jobId: NonEmptyString, action: Type.Literal("finished"), status: Type.Optional(CronRunStatusSchema), + completionStatus: Type.Optional(CronCompletionStatusSchema), error: Type.Optional(Type.String()), errorReason: Type.Optional(FailoverReasonSchema), summary: Type.Optional(Type.String()), diff --git a/src/cli/cron-cli.test.ts b/src/cli/cron-cli.test.ts index 7c43d4d579fe..f4d0dccf7354 100644 --- a/src/cli/cron-cli.test.ts +++ b/src/cli/cron-cli.test.ts @@ -314,6 +314,7 @@ async function runCronRunAndCaptureExit(params: { runId?: string; runStatus?: "ok" | "error" | "skipped"; runStatuses?: Array<"ok" | "error" | "skipped" | undefined>; + completionStatus?: "succeeded" | "failed" | "unknown"; args?: string[]; }) { resetGatewayMock(); @@ -336,7 +337,17 @@ async function runCronRunAndCaptureExit(params: { const runStatus = params.runStatuses?.[runPollCount] ?? params.runStatus; runPollCount += 1; return { - entries: runStatus ? [{ status: runStatus }] : [], + entries: runStatus + ? [ + { + status: runStatus, + completionStatus: params.completionStatus, + ...(params.completionStatus === undefined && runStatus === "ok" + ? { deliveryStatus: "not-requested" } + : {}), + }, + ] + : [], }; } return { ok: true, params: callParams }; @@ -378,6 +389,18 @@ describe("cron cli", () => { }, ); + it("uses legacy stored delivery for --wait completion", async () => { + const { calls, exitSpy } = await runCronRunAndCaptureExit({ + enqueued: true, + runId: "manual:legacy:123:0", + runStatus: "ok", + args: ["cron", "run", "job-1", "--wait", "--wait-timeout", "1s", "--poll-interval", "1ms"], + }); + + expect(exitSpy).toHaveBeenCalledWith(0); + expect(calls.some((call) => call[0] === "cron.get")).toBe(false); + }); + it.each(CRON_GATEWAY_COMMANDS)( "accepts leaf Gateway options for cron $name", async ({ name, args }) => { @@ -454,16 +477,17 @@ describe("cron cli", () => { }); it.each([ - { status: "ok" as const, expectedExitCode: 0 }, - { status: "error" as const, expectedExitCode: 1 }, - { status: "skipped" as const, expectedExitCode: 1 }, + { status: "ok" as const, completionStatus: "succeeded" as const, expectedExitCode: 0 }, + { status: "ok" as const, completionStatus: "failed" as const, expectedExitCode: 1 }, + { status: "ok" as const, completionStatus: "unknown" as const, expectedExitCode: 1 }, ])( - "waits for queued cron run completion with status $status", - async ({ status, expectedExitCode }) => { + "waits for execution $status with completion $completionStatus", + async ({ status, completionStatus, expectedExitCode }) => { const { calls, exitSpy } = await runCronRunAndCaptureExit({ enqueued: true, runId: "manual:job-1:123:0", runStatus: status, + completionStatus, args: ["cron", "run", "job-1", "--wait", "--wait-timeout", "1s", "--poll-interval", "1ms"], }); @@ -476,6 +500,8 @@ describe("cron cli", () => { }); expect(stdoutText()).toContain('"completed": true'); expect(stdoutText()).toContain(`"status": "${status}"`); + expect(stdoutText()).toContain(`"completionStatus": "${completionStatus}"`); + expect(calls.some((call) => call[0] === "cron.get")).toBe(false); }, ); diff --git a/src/cli/cron-cli/register.cron-simple.ts b/src/cli/cron-cli/register.cron-simple.ts index 6a74781df1d2..b3be83fd2e8d 100644 --- a/src/cli/cron-cli/register.cron-simple.ts +++ b/src/cli/cron-cli/register.cron-simple.ts @@ -5,6 +5,8 @@ import { resolveTimerTimeoutMs, } from "@openclaw/normalization-core/number-coercion"; import type { Command } from "commander"; +import { resolveCronCompletionStatus } from "../../cron/completion-status.js"; +import type { CronRunLogEntry } from "../../cron/run-log-types.js"; import { defaultRuntime } from "../../runtime.js"; import { sleep } from "../../utils/sleep.js"; import type { GatewayRpcOpts } from "../gateway-rpc.js"; @@ -32,9 +34,10 @@ type CronRunCommandResult = { runId?: string; }; -type CronRunLogEntryResult = { - status?: "ok" | "error" | "skipped"; -}; +type CronRunLogEntryResult = Pick< + CronRunLogEntry, + "status" | "completionStatus" | "delivered" | "deliveryStatus" +>; function parseCronRunWaitDuration(raw: unknown, label: string): number { const input = @@ -264,8 +267,22 @@ export function registerCronSimpleCommands(cron: Command) { timeoutMs: waitTimeoutMs, pollIntervalMs, }); - printCronJson({ ...res, completed: true, status: run.status, run }); - defaultRuntime.exit(run.status === "ok" ? 0 : 1); + const completionStatus = + run.completionStatus ?? + resolveCronCompletionStatus({ + status: run.status, + delivered: run.delivered, + deliveryStatus: run.deliveryStatus, + }); + const completedRun = { ...run, completionStatus }; + printCronJson({ + ...res, + completed: true, + status: run.status, + completionStatus, + run: completedRun, + }); + defaultRuntime.exit(completionStatus === "succeeded" ? 0 : 1); return; } printCronJson(res); diff --git a/src/cron/completion-status.ts b/src/cron/completion-status.ts new file mode 100644 index 000000000000..ffcb2aaa4ef6 --- /dev/null +++ b/src/cron/completion-status.ts @@ -0,0 +1,63 @@ +type CronCompletionRunStatus = "ok" | "error" | "skipped"; + +type CronCompletionDeliveryStatus = "delivered" | "not-delivered" | "unknown" | "not-requested"; + +type CronCompletionJob = { + delivery?: { + mode?: "none" | "announce" | "webhook"; + bestEffort?: boolean; + }; +}; + +/** Whole-run completion after execution and any explicitly required delivery settle. */ +export type CronCompletionStatus = "succeeded" | "failed" | "unknown"; + +/** Required delivery is an explicit admitted policy, never an inferred default. */ +function isCronDeliveryRequired(job: CronCompletionJob): boolean { + return ( + job.delivery?.bestEffort === false && + (job.delivery.mode === "announce" || job.delivery.mode === "webhook") + ); +} + +/** Resolves authored completion from an admitted job, or legacy completion from stored facts. */ +export function resolveCronCompletionStatus(params: { + status?: CronCompletionRunStatus; + delivered?: boolean; + deliveryStatus?: CronCompletionDeliveryStatus; + requiredDelivery?: boolean; +}): CronCompletionStatus { + if (params.status === "error" || params.status === "skipped") { + return "failed"; + } + if (params.status !== "ok") { + return "unknown"; + } + if (params.requiredDelivery === undefined) { + return params.delivered === true || + params.deliveryStatus === "delivered" || + params.deliveryStatus === "not-requested" + ? "succeeded" + : "unknown"; + } + if (!params.requiredDelivery) { + return "succeeded"; + } + if (params.deliveryStatus === "delivered") { + return "succeeded"; + } + return params.deliveryStatus === "not-delivered" ? "failed" : "unknown"; +} + +/** Resolves completion from the immutable delivery contract admitted for this run. */ +export function resolveAdmittedCronCompletionStatus( + job: CronCompletionJob, + status: CronCompletionRunStatus, + deliveryStatus: CronCompletionDeliveryStatus, +): CronCompletionStatus { + return resolveCronCompletionStatus({ + status, + deliveryStatus, + requiredDelivery: isCronDeliveryRequired(job), + }); +} diff --git a/src/cron/run-log-types.ts b/src/cron/run-log-types.ts index e144c634e980..de97a95b8169 100644 --- a/src/cron/run-log-types.ts +++ b/src/cron/run-log-types.ts @@ -1,6 +1,7 @@ /** Stable cron run-history wire shape and legacy JSONL migration input. */ import type { FailoverReason } from "../agents/failover/signal.js"; import type { + CronCompletionStatus, CronDeliveryStatus, CronDeliveryTrace, CronFailureNotificationDelivery, @@ -15,6 +16,7 @@ export type CronRunLogEntry = { jobId: string; action: "finished"; status?: CronRunStatus; + completionStatus?: CronCompletionStatus; error?: string; errorReason?: FailoverReason; summary?: string; diff --git a/src/cron/service.persists-delivered-status.test.ts b/src/cron/service.persists-delivered-status.test.ts index f3ff3cd398b9..7ecfd2fb6b47 100644 --- a/src/cron/service.persists-delivered-status.test.ts +++ b/src/cron/service.persists-delivered-status.test.ts @@ -2,6 +2,7 @@ import { createServer } from "node:http"; import type { AddressInfo } from "node:net"; import { describe, expect, it, vi } from "vitest"; +import { createDeferred } from "../../test/helpers/promise.js"; const mocks = vi.hoisted(() => ({ fetchWithSsrFGuard: vi.fn(), })); @@ -159,6 +160,7 @@ function createIsolatedCronWithFinishedBarrier(params: { delivered?: boolean; deliveryStatus?: string; deliveryError?: string; + completionStatus?: string; failureNotificationDelivery?: { delivered?: boolean; status: string; @@ -189,6 +191,7 @@ function createIsolatedCronWithFinishedBarrier(params: { delivered: evt.delivered, deliveryStatus: evt.deliveryStatus, deliveryError: evt.deliveryError, + completionStatus: evt.completionStatus, failureNotificationDelivery: evt.failureNotificationDelivery, }); } @@ -976,4 +979,93 @@ describe("CronService persists delivered status", () => { expect(capturedEvent?.deliveryStatus).toBe("not-delivered"); expect(capturedEvent?.deliveryError).toBe("Message delivery failed"); }); + + it.each([ + { + name: "required to best-effort", + admittedBestEffort: false, + edits: [true], + expectedCompletionStatus: "failed", + }, + { + name: "default to required", + admittedBestEffort: undefined, + edits: [false], + expectedCompletionStatus: "succeeded", + }, + { + name: "best-effort to required", + admittedBestEffort: true, + edits: [false], + expectedCompletionStatus: "succeeded", + }, + { + name: "required A to B to A", + admittedBestEffort: false, + edits: [true, false], + expectedCompletionStatus: "failed", + }, + ])( + "authors completion from the admitted delivery policy: $name", + async ({ admittedBestEffort, edits, expectedCompletionStatus }) => { + const store = await makeStorePath(); + const started = createDeferred(); + const finish = createDeferred<{ + status: "ok"; + delivered: false; + deliveryError: string; + }>(); + let finishedEvent: CronEvent | undefined; + const cron = new CronService({ + storePath: store.storePath, + cronEnabled: true, + log: noopLogger, + enqueueSystemEvent: vi.fn(), + requestHeartbeat: vi.fn(), + runIsolatedAgentJob: vi.fn(async () => { + started.resolve(); + return await finish.promise; + }), + onEvent: (event) => { + if (event.action === "finished") { + finishedEvent = event; + } + }, + }); + await cron.start(); + const job = await cron.add({ + ...buildAnnounceIsolatedAgentTurnJob(`admitted-policy-${admittedBestEffort}`), + delivery: { + mode: "announce", + channel: "forum", + to: "123", + ...(admittedBestEffort === undefined ? {} : { bestEffort: admittedBestEffort }), + }, + }); + + const run = cron.run(job.id, "force"); + await started.promise; + for (const bestEffort of edits) { + await cron.update(job.id, { delivery: { bestEffort } }); + } + finish.resolve({ + status: "ok", + delivered: false, + deliveryError: "delivery rejected", + }); + await run; + + expect(finishedEvent).toMatchObject({ + status: "ok", + deliveryStatus: "not-delivered", + completionStatus: expectedCompletionStatus, + }); + expect(cron.getJob(job.id)?.state).toMatchObject({ + lastRunStatus: "ok", + consecutiveErrors: 0, + }); + cron.stop(); + await store.cleanup(); + }, + ); }); diff --git a/src/cron/service.runs-one-shot-main-job-disables-it.test.ts b/src/cron/service.runs-one-shot-main-job-disables-it.test.ts index ecdd5edf198c..541eacf1b3cc 100644 --- a/src/cron/service.runs-one-shot-main-job-disables-it.test.ts +++ b/src/cron/service.runs-one-shot-main-job-disables-it.test.ts @@ -362,6 +362,107 @@ describe("CronService", () => { await stopCronAndCleanup(cron, store); }); + it("retains a required-delivery-failed one-shot and never replays it after restart", async () => { + const runIsolatedAgentJob = vi.fn(async () => ({ + status: "ok" as const, + summary: "payload completed", + delivered: false, + deliveryError: "delivery rejected", + })); + const { store, cron, events } = await createIsolatedAnnounceHarness(runIsolatedAgentJob); + const runAt = new Date("2025-12-13T00:00:03.000Z"); + const job = await cron.add({ + name: "required one-shot", + enabled: true, + schedule: { kind: "at", at: runAt.toISOString() }, + sessionTarget: "isolated", + wakeMode: "now", + payload: { kind: "agentTurn", message: "do it once" }, + delivery: { mode: "announce", bestEffort: false }, + }); + + vi.setSystemTime(runAt); + await vi.runOnlyPendingTimersAsync(); + const event = await events.waitFor( + (candidate) => candidate.jobId === job.id && candidate.action === "finished", + ); + const retained = cron.getJob(job.id); + + expect(event).toMatchObject({ + status: "ok", + completionStatus: "failed", + deliveryStatus: "not-delivered", + }); + expect(retained).toMatchObject({ + enabled: false, + state: { + lastRunStatus: "ok", + consecutiveErrors: 0, + }, + }); + expect(retained?.state.nextRunAtMs).toBeUndefined(); + expect(runIsolatedAgentJob).toHaveBeenCalledOnce(); + + cron.stop(); + const restartedRun = vi.fn(async () => ({ status: "ok" as const })); + const restarted = createStartedCronService(store.storePath, restartedRun); + await restarted.start(); + await vi.runOnlyPendingTimersAsync(); + expect(restartedRun).not.toHaveBeenCalled(); + expect(restarted.getJob(job.id)).toMatchObject({ enabled: false }); + + await stopCronAndCleanup(restarted, store); + }); + + it("counts the next execution error from one after a delivery-only failure", async () => { + const runIsolatedAgentJob = vi + .fn() + .mockResolvedValueOnce({ + status: "ok" as const, + delivered: false, + deliveryError: "delivery rejected", + }) + .mockResolvedValueOnce({ + status: "error" as const, + error: "provider overloaded", + }); + const { store, cron, events } = await createIsolatedAnnounceHarness(runIsolatedAgentJob); + const job = await cron.add({ + name: "delivery then execution error", + enabled: true, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "isolated", + wakeMode: "now", + payload: { kind: "agentTurn", message: "run" }, + delivery: { mode: "announce", bestEffort: false }, + }); + + const firstAt = job.state.nextRunAtMs!; + vi.setSystemTime(firstAt); + await vi.runOnlyPendingTimersAsync(); + await events.waitFor( + (candidate) => candidate.jobId === job.id && candidate.action === "finished", + ); + const secondAt = cron.getJob(job.id)?.state.nextRunAtMs; + expect(secondAt).toBeTypeOf("number"); + expect(cron.getJob(job.id)?.state.consecutiveErrors).toBe(0); + + vi.setSystemTime(secondAt!); + await vi.runOnlyPendingTimersAsync(); + await vi.waitFor(() => expect(runIsolatedAgentJob).toHaveBeenCalledTimes(2)); + const updated = cron.getJob(job.id); + expect(updated).toMatchObject({ + enabled: true, + state: { + lastRunStatus: "error", + consecutiveErrors: 1, + }, + }); + expect(updated?.state.nextRunAtMs).toBeGreaterThan(secondAt!); + + await stopCronAndCleanup(cron, store); + }); + it("deletes a recurring job converted to at when retention is omitted", async () => { const { store, cron, events } = await createMainOneShotHarness(); const job = await cron.add({ diff --git a/src/cron/service/failure-alerts.persistence.test.ts b/src/cron/service/failure-alerts.persistence.test.ts index fd333d721b47..934391e924d7 100644 --- a/src/cron/service/failure-alerts.persistence.test.ts +++ b/src/cron/service/failure-alerts.persistence.test.ts @@ -12,6 +12,7 @@ import { cronStoreKey } from "../store/key.js"; import type { CronJob, CronRunStatus } from "../types.js"; import { createCronServiceState } from "./state.js"; import { finalizeCompletedCronRunOutcomes } from "./timer-outcome-finalization.js"; +import { authorCronRunCompletion } from "./timer.js"; const fixtures = setupCronRegressionFixtures({ prefix: "cron-failure-alert-persistence-", @@ -67,8 +68,10 @@ async function finalizeAlertOutcome(params: { jobId: params.job.id, job: structuredClone(params.job), activeJobMarker: markCronJobActive(params.job.id), - status: params.status, - error: params.error, + ...authorCronRunCompletion(params.state, params.job, { + status: params.status, + error: params.error, + }), startedAt: params.startedAt, endedAt: params.endedAt, }, diff --git a/src/cron/service/ops-run-preparation.ts b/src/cron/service/ops-run-preparation.ts index 1af0887ff5d4..5f3960e58d93 100644 --- a/src/cron/service/ops-run-preparation.ts +++ b/src/cron/service/ops-run-preparation.ts @@ -1,5 +1,6 @@ import type { CommandLaneTaskMarker } from "../../process/command-queue.js"; import type { CronActiveJobMarker } from "../active-jobs.js"; +import { resolveCronCompletionStatus } from "../completion-status.js"; import { resolveCronJobConfigRevision } from "../config-revision.js"; import { createCronRunDiagnosticsFromError } from "../run-diagnostics.js"; import { @@ -98,15 +99,21 @@ export function emitCronRunFinished( errorClassification?: CronRunErrorClassification; }, ): void { + const event = { + ...evt, + completionStatus: + evt.completionStatus ?? + resolveCronCompletionStatus({ status: evt.status, deliveryStatus: evt.deliveryStatus }), + }; tryFinishCronTaskRun(state, { taskRunId, job: evt.job, - event: evt, + event, errorClassification: details?.errorClassification, ...(details?.scriptResult ? { scriptResult: details.scriptResult } : {}), ...(details?.triggerEval ? { triggerEval: details.triggerEval } : {}), }); - emit(state, evt); + emit(state, event); if (tracker) { tracker.emitted = true; } @@ -162,6 +169,7 @@ async function skipInvalidPersistedManualRun(params: { params.job, { status: "skipped", + completionStatus: "failed", error: errorText, diagnostics, startedAt: endedAt, diff --git a/src/cron/service/ops-run.ts b/src/cron/service/ops-run.ts index debeaec7e74a..b91767b196cc 100644 --- a/src/cron/service/ops-run.ts +++ b/src/cron/service/ops-run.ts @@ -55,6 +55,7 @@ import { applyTriggerNoFireResult, applyTriggerRunResult, armTimer, + authorCronRunCompletion, executeJobCoreWithTimeout, } from "./timer.js"; import { wake } from "./wake.js"; @@ -165,11 +166,11 @@ async function finishPreparedManualRun( if (err instanceof CronRunReceiptRevisionError && err.reason === "owner-unavailable") { receiptSettlementDisposition = "owner-unavailable"; } - coreResult = { + coreResult = authorCronRunCompletion(state, executionJob, { status: "error", error: err instanceof CronRunReceiptRevisionError ? err.message : normalizeCronRunErrorText(err), - }; + }); } if (prepared.onTriggerDisposition) { const disposition = coreResult.triggerEval?.busy @@ -205,6 +206,7 @@ async function finishPreparedManualRun( action: "finished", job, status: triggerSkipped ? "skipped" : coreResult.status, + completionStatus: triggerSkipped ? "failed" : coreResult.completionStatus, error: triggerSkipped ? "queued manual run skipped: trigger condition not met" : coreResult.error, @@ -361,6 +363,7 @@ async function finishPreparedManualRun( action: "finished", job: committed.job, status: coreResult.status, + completionStatus: coreResult.completionStatus, error: coreResult.error, summary: coreResult.summary, diagnostics: coreResult.diagnostics, diff --git a/src/cron/service/ops.test.ts b/src/cron/service/ops.test.ts index 5c770ef0b4c6..a87aa4453f6d 100644 --- a/src/cron/service/ops.test.ts +++ b/src/cron/service/ops.test.ts @@ -1217,6 +1217,7 @@ describe("cron service ops seam coverage", () => { action: "finished", job, status: "ok", + completionStatus: "succeeded", runAtMs: startedAt, durationMs: 1_000, }, @@ -1285,6 +1286,7 @@ describe("cron service ops seam coverage", () => { action: "finished", job, status: "ok", + completionStatus: "succeeded", summary: "completed before restart", runAtMs: startedAt, durationMs: endedAt - startedAt, @@ -1425,6 +1427,7 @@ describe("cron service ops seam coverage", () => { action: "finished", job: original, status, + completionStatus: status === "ok" ? "succeeded" : "failed", ...(status === "error" ? { error: "original failed before restart" } : {}), summary: "original completed before restart", runAtMs: startedAt, diff --git a/src/cron/service/run-admission.ts b/src/cron/service/run-admission.ts index 16997ee79d96..7582e0003080 100644 --- a/src/cron/service/run-admission.ts +++ b/src/cron/service/run-admission.ts @@ -32,7 +32,7 @@ import { runsDetachedFromMainSession, type TimedCronRunOutcome, } from "./timer-execution-timeout.js"; -import { executeJobCoreWithTimeout } from "./timer-job-runner.js"; +import { authorCronRunCompletion, executeJobCoreWithTimeout } from "./timer-job-runner.js"; import { isRunnableJob } from "./timer-runnable.js"; export function resolveRunConcurrency(): number { @@ -665,10 +665,12 @@ export async function executeQueuedCronRun(params: { params.onSetupError?.(executionJob, errorText); outcome = { ...base, - status: "error", - error: errorText, - diagnostics: createCronRunDiagnosticsFromError("cron-setup", errorText, { - nowMs: state.deps.nowMs, + ...authorCronRunCompletion(state, executionJob, { + status: "error", + error: errorText, + diagnostics: createCronRunDiagnosticsFromError("cron-setup", errorText, { + nowMs: state.deps.nowMs, + }), }), ...(receiptSettlementDisposition ? { receiptSettlementDisposition } : {}), endedAt: state.deps.nowMs(), diff --git a/src/cron/service/startup-run-repair.auto-disable.test.ts b/src/cron/service/startup-run-repair.auto-disable.test.ts index 6a930d4daef8..0bd477837a4f 100644 --- a/src/cron/service/startup-run-repair.auto-disable.test.ts +++ b/src/cron/service/startup-run-repair.auto-disable.test.ts @@ -138,6 +138,76 @@ describe("startup run repair auto-disable", () => { expect(state.deps.requestHeartbeat).toHaveBeenCalledOnce(); }); + it.each([ + { + name: "required delivery failed", + completionStatus: "failed" as const, + deliveryStatus: "not-delivered" as const, + }, + { + name: "completion evidence unknown", + completionStatus: "unknown" as const, + deliveryStatus: "unknown" as const, + }, + { + name: "legacy row missing completion evidence", + completionStatus: undefined, + deliveryStatus: "not-delivered" as const, + }, + ])("retains finalized one-shot after $name", ({ completionStatus, deliveryStatus }) => { + const runningAtMs = Date.parse("2026-08-01T17:00:00.000Z"); + const state = createCronServiceState({ + storePath: "/tmp/startup-run-repair-completion.json", + cronEnabled: true, + log: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + nowMs: () => runningAtMs + 1_000, + enqueueSystemEvent: vi.fn(), + requestHeartbeat: vi.fn(), + runIsolatedAgentJob: vi.fn(), + }); + const job: CronJob = { + id: "finalized-required-delivery", + name: "finalized required delivery", + enabled: true, + deleteAfterRun: true, + createdAtMs: runningAtMs - 60_000, + updatedAtMs: runningAtMs, + schedule: { kind: "at", at: new Date(runningAtMs).toISOString() }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "agentTurn", message: "do not replay" }, + // Current policy is intentionally mutable and must not decide replay. + delivery: { mode: "announce", bestEffort: true }, + state: { runningAtMs }, + }; + + const restored = restoreFinalizedStartupRun({ + state, + job, + runningAtMs, + entry: { + ts: runningAtMs + 1_000, + jobId: job.id, + action: "finished", + status: "ok", + ...(completionStatus === undefined ? {} : { completionStatus }), + deliveryStatus, + runAtMs: runningAtMs, + durationMs: 1_000, + }, + }); + + expect(restored?.shouldDelete).toBe(false); + expect(job).toMatchObject({ + enabled: false, + state: { + lastRunStatus: "ok", + consecutiveErrors: 0, + }, + }); + expect(job.state.nextRunAtMs).toBeUndefined(); + }); + it("buffers quiet-trigger repair notifications until the recovery commit", () => { const runningAtMs = Date.parse("2026-08-01T16:30:00.000Z"); const enqueueSystemEvent = vi.fn(); diff --git a/src/cron/service/startup-run-repair.ts b/src/cron/service/startup-run-repair.ts index ace3f0fa18b0..af4d7f84e752 100644 --- a/src/cron/service/startup-run-repair.ts +++ b/src/cron/service/startup-run-repair.ts @@ -1,5 +1,6 @@ /** 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"; @@ -162,6 +163,13 @@ export function restoreFinalizedStartupRun(params: { job, { ...entry, + completionStatus: + entry.completionStatus ?? + resolveCronCompletionStatus({ + status: entry.status, + delivered: entry.delivered, + deliveryStatus: entry.deliveryStatus, + }), startedAt, endedAt, }, diff --git a/src/cron/service/state.ts b/src/cron/service/state.ts index bf85bbbdb882..f00834cad6fe 100644 --- a/src/cron/service/state.ts +++ b/src/cron/service/state.ts @@ -12,6 +12,7 @@ import type { CronScheduledToolPolicy } from "../scheduled-tool-policy.js"; import type { QuarantinedCronConfigJob } from "../store.js"; import type { CronRunReceiptHandle } from "../store/run-receipt-store.js"; import type { + CronCompletionStatus, CronTriggerEvaluationResult, CronAgentExecutionPhaseUpdate, CronAgentExecutionStarted, @@ -41,6 +42,7 @@ export type CronEvent = { runAtMs?: number; durationMs?: number; status?: CronRunStatus; + completionStatus?: CronCompletionStatus; error?: string; summary?: string; diagnostics?: CronRunDiagnostics; diff --git a/src/cron/service/task-runs.test.ts b/src/cron/service/task-runs.test.ts index f86605c206ab..0527fc5fae57 100644 --- a/src/cron/service/task-runs.test.ts +++ b/src/cron/service/task-runs.test.ts @@ -196,6 +196,7 @@ describe("cron task run terminal records", () => { action: "finished", job, status: "ok", + completionStatus: "succeeded", runAtMs: 1_000, durationMs: 100, }, @@ -536,6 +537,7 @@ describe("cron task run terminal records", () => { action: "finished", job, status: "ok", + completionStatus: "succeeded", summary: "done", sessionKey: "agent:main:cron:retry-job:run:actual", runAtMs: startedAt, diff --git a/src/cron/service/task-runs.ts b/src/cron/service/task-runs.ts index ca6f53d26ce6..9ff0a996c60f 100644 --- a/src/cron/service/task-runs.ts +++ b/src/cron/service/task-runs.ts @@ -31,7 +31,12 @@ import { resolveCronTaskRecordTimestamp, } from "../task-run-detail.js"; import { cronRunLogEntryFromEvent } from "../task-run-event-codec.js"; -import type { CronJob, CronRunErrorClassification, CronRunStatus } from "../types.js"; +import type { + CronCompletionStatus, + CronJob, + CronRunErrorClassification, + CronRunStatus, +} from "../types.js"; import { normalizeCronRunErrorText } from "./execution-errors.js"; import type { CronEvent, CronServiceState } from "./state.js"; import { CRON_TASK_RUNNING_PROGRESS_SUMMARY } from "./task-ledger.js"; @@ -263,6 +268,7 @@ export function tryFinishCronTaskRunWithoutHistory( result: { taskRunId?: string; status: "ok" | "error" | "skipped"; + completionStatus?: CronCompletionStatus; error?: unknown; endedAt: number; summary?: string; @@ -286,7 +292,11 @@ export function tryFinishCronTaskRunWithoutHistory( finalizeTaskRunByRunIdCore({ runId: result.taskRunId, runtime: "cron", - status: cronRunStatusToTaskStatus({ status: result.status, error }), + status: cronRunStatusToTaskStatus({ + status: result.status, + completionStatus: quietTriggerEval ? "succeeded" : result.completionStatus, + error, + }), endedAt: result.endedAt, lastEventAt: result.endedAt, error, diff --git a/src/cron/service/timer-execution-timeout.ts b/src/cron/service/timer-execution-timeout.ts index b0d0799da58c..cea870104a01 100644 --- a/src/cron/service/timer-execution-timeout.ts +++ b/src/cron/service/timer-execution-timeout.ts @@ -8,9 +8,11 @@ import type { CronRunReceiptHandle } from "../store/run-receipt-store.js"; import type { CronAgentExecutionPhaseUpdate, CronAgentExecutionStarted, + CronCompletionStatus, CronDeliveryTrace, CronJob, CronNextCheckProposal, + CronResolvedDeliveryState, CronRunOutcome, CronRunStatus, CronRunTelemetry, @@ -42,6 +44,8 @@ export type TimedCronRunOutcome = CronRunOutcome & jobId: string; job: CronJob; taskRunId?: string; + completionStatus: CronCompletionStatus; + deliveryState: CronResolvedDeliveryState; delivered?: boolean; deliveryAttempted?: boolean; deliveryError?: string; @@ -61,6 +65,8 @@ export type TimedCronRunOutcome = CronRunOutcome & export type CronJobRunResult = CronRunOutcome & Pick & { + completionStatus?: CronCompletionStatus; + deliveryState?: CronResolvedDeliveryState; deliveryError?: string; delivered?: boolean; deliveryAttempted?: boolean; diff --git a/src/cron/service/timer-job-runner.ts b/src/cron/service/timer-job-runner.ts index 72d264589e97..f2ba87f64187 100644 --- a/src/cron/service/timer-job-runner.ts +++ b/src/cron/service/timer-job-runner.ts @@ -1,6 +1,7 @@ import { formatErrorMessage } from "../../infra/errors.js"; import type { CommandLaneTaskMarker } from "../../process/command-queue.js"; import { type CronActiveJobMarker, isCronActiveJobMarkerCurrent } from "../active-jobs.js"; +import { resolveAdmittedCronCompletionStatus } from "../completion-status.js"; import { resolveCronDeliveryPlan } from "../delivery-plan.js"; import { createCronRunDiagnosticsFromError } from "../run-diagnostics.js"; import type { CronAgentExecutionStarted, CronJob } from "../types.js"; @@ -26,6 +27,7 @@ import type { CronServiceState } from "./state.js"; import { tryUpdateCronTaskRunSession, withCronTaskRunId } from "./task-runs.js"; import { resolveCronJobTimeoutMs } from "./timeout-policy.js"; import { + type CronJobRunResult, type IsolatedAgentSetupTimeoutSignal, runsDetachedFromMainSession, } from "./timer-execution-timeout.js"; @@ -36,6 +38,7 @@ import { withPrimaryWebhookTrace, type CronRunProgress, } from "./timer-job-runner.interruption.js"; +import { resolveDeliveryState } from "./timer-trigger.js"; type CronCoreRunOutcome = Awaited> & { isolatedAgentSetupTimeout?: IsolatedAgentSetupTimeoutSignal; @@ -185,7 +188,7 @@ function cronRunAttributionFromExecution(execution?: CronAgentExecutionStarted): } /** Executes cron job core logic with the configured wall-clock timeout and watchdog cleanup. */ -export async function executeJobCoreWithTimeout( +async function executeJobCoreWithTimeoutUnfinalized( state: CronServiceState, job: CronJob, opts?: CronCoreRunOptions, @@ -435,3 +438,34 @@ export async function executeJobCoreWithTimeout( releaseCronTaskRun?.(); } } + +export function authorCronRunCompletion< + T extends Pick< + CronJobRunResult, + "status" | "error" | "deliveryError" | "delivered" | "deliveryAttempted" + >, +>(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, + deliveryState, + completionStatus: resolveAdmittedCronCompletionStatus(job, result.status, deliveryState.status), + }; +} + +/** Authors completion after execution and primary delivery have both settled. */ +export async function executeJobCoreWithTimeout( + state: CronServiceState, + job: CronJob, + opts?: CronCoreRunOptions, +) { + const result = await executeJobCoreWithTimeoutUnfinalized(state, job, opts); + return authorCronRunCompletion(state, job, result); +} diff --git a/src/cron/service/timer-outcome-finalization.receipts.test.ts b/src/cron/service/timer-outcome-finalization.receipts.test.ts index b5efb698c4b6..fb17465b2b6e 100644 --- a/src/cron/service/timer-outcome-finalization.receipts.test.ts +++ b/src/cron/service/timer-outcome-finalization.receipts.test.ts @@ -18,7 +18,9 @@ import { } from "../store/run-receipt-store.js"; import type { CronJob } from "../types.js"; import { createCronServiceState } from "./state.js"; +import type { TimedCronRunOutcome } from "./timer-execution-timeout.js"; import { finalizeCompletedCronRunOutcomes } from "./timer-outcome-finalization.js"; +import { authorCronRunCompletion } from "./timer.js"; import { onTimer } from "./timer.test-support.js"; const fixtures = setupCronRegressionFixtures({ prefix: "cron-finalization-receipts-" }); @@ -39,6 +41,13 @@ function claimReceipt(storePath: string, job: CronJob, startedAtMs: number) { ); } +function authorOutcome( + state: ReturnType, + outcome: Omit, +) { + return authorCronRunCompletion(state, outcome.job, outcome); +} + describe("cron outcome receipt finalization", () => { it("emits only committed authoritative outcomes after a rejected batch attempt", async () => { const store = fixtures.makeStorePath(); @@ -79,7 +88,7 @@ describe("cron outcome receipt finalization", () => { }); await finalizeCompletedCronRunOutcomes(state, [ - { + authorOutcome(state, { jobId: stale.id, job: stale, activeJobMarker: markCronJobActive(stale.id), @@ -87,8 +96,8 @@ describe("cron outcome receipt finalization", () => { status: "ok", startedAt, endedAt: startedAt + 2, - }, - { + }), + authorOutcome(state, { jobId: current.id, job: current, activeJobMarker: markCronJobActive(current.id), @@ -96,7 +105,7 @@ describe("cron outcome receipt finalization", () => { status: "ok", startedAt, endedAt: startedAt + 2, - }, + }), ]); expect(events.filter((event) => event.action === "finished")).toEqual([ @@ -170,7 +179,7 @@ describe("cron outcome receipt finalization", () => { }); await finalizeCompletedCronRunOutcomes(state, [ - { + authorOutcome(state, { jobId: completed.id, job: completed, activeJobMarker: markCronJobActive(completed.id), @@ -178,7 +187,7 @@ describe("cron outcome receipt finalization", () => { status: "ok", startedAt, endedAt: startedAt + 1, - }, + }), ]); const persisted = await loadCronStore(store.storePath); @@ -229,7 +238,7 @@ describe("cron outcome receipt finalization", () => { try { await expect( finalizeCompletedCronRunOutcomes(state, [ - { + authorOutcome(state, { jobId: completed.id, job: completed, activeJobMarker: markCronJobActive(completed.id), @@ -237,7 +246,7 @@ describe("cron outcome receipt finalization", () => { status: "ok", startedAt, endedAt: startedAt + 1, - }, + }), ]), ).resolves.toHaveLength(1); diff --git a/src/cron/service/timer-outcome-finalization.ts b/src/cron/service/timer-outcome-finalization.ts index 1ee3783813f5..949db1b9630c 100644 --- a/src/cron/service/timer-outcome-finalization.ts +++ b/src/cron/service/timer-outcome-finalization.ts @@ -6,7 +6,7 @@ import { releaseLocalCronRunReceiptOwnership, type CronRunReceiptHandle, } from "../store/run-receipt-store.js"; -import type { CronJob } from "../types.js"; +import type { CronCompletionStatus, CronJob } from "../types.js"; import { locked } from "./locked.js"; import { releaseQueuedCronRun, supersedeActivatedCronRun } from "./run-admission.js"; import { cronRunReceiptPersistHooks, supersedeServiceCronRunReceipt } from "./run-receipts.js"; @@ -27,6 +27,7 @@ type CronTaskRunFinalizationOutcome = { jobId: string; taskRunId?: string; status: "ok" | "error" | "skipped"; + completionStatus?: CronCompletionStatus; error?: unknown; endedAt: number; summary?: string; diff --git a/src/cron/service/timer-outcomes.ts b/src/cron/service/timer-outcomes.ts index d7d02069a0c2..a8f8202269b3 100644 --- a/src/cron/service/timer-outcomes.ts +++ b/src/cron/service/timer-outcomes.ts @@ -1,5 +1,6 @@ import { resolveCronTriggerMinIntervalMs } from "../../config/cron-limits.js"; import type { CronActiveJobMarker } from "../active-jobs.js"; +import { resolveAdmittedCronCompletionStatus } from "../completion-status.js"; import { resolvePacedNextRunAtMs } from "../pacing.js"; import { normalizeCronRunDiagnostics, summarizeCronRunDiagnostics } from "../run-diagnostics.js"; import { resolveCronRunErrorReason } from "../run-error-reason.js"; @@ -123,18 +124,16 @@ export function applyJobResult( "cron: job run returned error status", ); } - const deliveryState = resolveDeliveryState({ - job, - runStatus: result.status, - delivered: result.delivered, - deliveryAttempted: result.deliveryAttempted, - // A successful run keeps `error` empty but may carry a dedicated - // `deliveryError` when post-run delivery failed (#94058/#95419); prefer it - // so `lastDeliveryError` is populated without conflating it with a - // run-level failure. Error runs fall back to the run error as before. - error: result.deliveryError ?? result.error, - globalFailureDestination: state.deps.cronConfig?.failureAlert, - }); + const deliveryState = + result.deliveryState ?? + resolveDeliveryState({ + job, + runStatus: result.status, + 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; job.state.lastDeliveryError = @@ -206,7 +205,9 @@ export function applyJobResult( isOneShotSchedule && !preserveOneShotSchedule && job.deleteAfterRun === true && - result.status === "ok"; + (result.completionStatus ?? + resolveAdmittedCronCompletionStatus(job, result.status, deliveryState.status)) === + "succeeded"; const retryDisabledHeartbeatOneShot = shouldRetryDisabledHeartbeatOneShot(job, result); if (!ownsSchedule) { @@ -738,6 +739,7 @@ function cronOutcomeEvent(job: CronJob, result: TimedCronRunOutcome, runAtMs: nu action: "finished", job, status: result.status, + completionStatus: result.completionStatus, error: result.error, summary: result.summary, diagnostics: result.diagnostics, diff --git a/src/cron/service/timer-trigger.ts b/src/cron/service/timer-trigger.ts index 6824de17ec5f..dfee9df76989 100644 --- a/src/cron/service/timer-trigger.ts +++ b/src/cron/service/timer-trigger.ts @@ -4,9 +4,9 @@ import { resolveCronDeliveryPlan, resolveFailureDestination } from "../delivery- import { type CronRetryOn, resolveCronExecutionRetryHint } from "../retry-hint.js"; import { createCronStreamSourceIdentity } from "../stream-schedule.js"; import type { - CronDeliveryStatus, CronFailureNotificationDelivery, CronJob, + CronResolvedDeliveryState, CronRunErrorClassification, CronRunStatus, } from "../types.js"; @@ -296,12 +296,7 @@ export function resolveDeliveryState(params: { deliveryAttempted?: boolean; error?: string; globalFailureDestination?: CronConfig["failureAlert"]; -}): { - delivered?: boolean; - status: CronDeliveryStatus; - error?: string; - failureNotification: CronFailureNotificationDelivery; -} { +}): CronResolvedDeliveryState { const primaryDeliveryPlan = resolveCronDeliveryPlan(params.job); const primaryDeliveryRequested = primaryDeliveryPlan.requested; // Failure destinations can receive alerts even when the primary delivery diff --git a/src/cron/service/timer.batch-finalization.test.ts b/src/cron/service/timer.batch-finalization.test.ts index 80101eb05a72..ff8f474baf56 100644 --- a/src/cron/service/timer.batch-finalization.test.ts +++ b/src/cron/service/timer.batch-finalization.test.ts @@ -20,11 +20,12 @@ import type { CronJob } from "../types.js"; import { start, stop } from "./ops-lifecycle.js"; import { add, remove } from "./ops-mutations.js"; import { createCronServiceState } from "./state.js"; +import type { TimedCronRunOutcome } from "./timer-execution-timeout.js"; import { createCompletedCronRunOutcomeDrain, finalizeCompletedCronRunOutcomes, } from "./timer-outcome-finalization.js"; -import { runMissedJobs } from "./timer.js"; +import { authorCronRunCompletion, runMissedJobs } from "./timer.js"; import { onTimer } from "./timer.test-support.js"; const fixtures = setupCronRegressionFixtures({ @@ -76,6 +77,13 @@ function findCronTask(jobId: string) { ); } +function authorOutcome( + state: ReturnType, + outcome: Omit, +) { + return authorCronRunCompletion(state, outcome.job, outcome); +} + describe("cron batch outcome finalization", () => { it.each([ { trigger: "scheduled", installSuccessor: false }, @@ -360,14 +368,16 @@ describe("cron batch outcome finalization", () => { const outcomeDrain = createCompletedCronRunOutcomeDrain(state); for (const job of jobs) { - outcomeDrain.enqueue({ - jobId: job.id, - job, - activeJobMarker: markCronJobActive(job.id), - status: "ok", - startedAt: dueAt, - endedAt: dueAt, - }); + outcomeDrain.enqueue( + authorOutcome(state, { + jobId: job.id, + job, + activeJobMarker: markCronJobActive(job.id), + status: "ok", + startedAt: dueAt, + endedAt: dueAt, + }), + ); } expect(await outcomeDrain.flush()).toHaveLength(jobs.length); @@ -425,7 +435,7 @@ describe("cron batch outcome finalization", () => { runIsolatedAgentJob: vi.fn(), }); await finalizeCompletedCronRunOutcomes(state, [ - { + authorOutcome(state, { jobId: job.id, job: structuredClone(job), activeJobMarker: markCronJobActive(job.id), @@ -433,7 +443,7 @@ describe("cron batch outcome finalization", () => { error: "cron: job execution timed out at /private/agent/work", startedAt: dueAt, endedAt: dueAt + 10, - }, + }), ]); expect(order).toEqual(["notify", "heartbeat"]); @@ -507,7 +517,7 @@ describe("cron batch outcome finalization", () => { runIsolatedAgentJob: vi.fn(), }); const finalized = await finalizeCompletedCronRunOutcomes(state, [ - { + authorOutcome(state, { jobId: job.id, job: structuredClone(job), activeJobMarker: markCronJobActive(job.id), @@ -515,7 +525,7 @@ describe("cron batch outcome finalization", () => { startedAt: dueAt, endedAt: dueAt + 10, nextCheck: { delayMs: MAX_DATE_TIMESTAMP_MS }, - }, + }), ]); expect(finalized).toHaveLength(1); @@ -578,7 +588,7 @@ describe("cron batch outcome finalization", () => { try { await expect( finalizeCompletedCronRunOutcomes(state, [ - { + authorOutcome(state, { jobId: job.id, job: structuredClone(job), activeJobMarker: markCronJobActive(job.id), @@ -586,7 +596,7 @@ describe("cron batch outcome finalization", () => { error: "tenth failure", startedAt: dueAt, endedAt: dueAt + 10, - }, + }), ]), ).rejects.toThrow("terminal write failed"); expect(state.deps.enqueueSystemEvent).not.toHaveBeenCalled(); @@ -622,7 +632,7 @@ describe("cron batch outcome finalization", () => { await finalizeCompletedCronRunOutcomes( state, [ - { + authorOutcome(state, { jobId: job.id, job, activeJobMarker, @@ -630,7 +640,7 @@ describe("cron batch outcome finalization", () => { error: "setup timed out before runner start", startedAt: dueAt, endedAt: dueAt, - }, + }), ], { clearOnFailure: false, discardWhenStopped: true }, ), diff --git a/src/cron/service/timer.pacing.test.ts b/src/cron/service/timer.pacing.test.ts index d02dd243c1c6..1c7012af046e 100644 --- a/src/cron/service/timer.pacing.test.ts +++ b/src/cron/service/timer.pacing.test.ts @@ -10,8 +10,9 @@ import { createNoopLogger } from "../service.test-harness.js"; import type { CronJob, CronPacing } from "../types.js"; import { recomputeNextRunsForMaintenance } from "./jobs-scheduling.js"; import { createCronServiceState } from "./state.js"; +import type { TimedCronRunOutcome } from "./timer-execution-timeout.js"; import { applyOutcomeToStoredJob, applyTriggerNoFireResult } from "./timer-outcomes.js"; -import { applyJobResult } from "./timer.js"; +import { applyJobResult, authorCronRunCompletion } from "./timer.js"; const ENDED_AT = Date.parse("2026-07-18T12:00:00.000Z"); const STARTED_AT = ENDED_AT - 1_000; @@ -36,6 +37,13 @@ function makePacedJob(pacing: CronPacing, everyMs = 60 * 60_000): CronJob { }); } +function applyAuthoredOutcome( + state: ReturnType, + outcome: Omit, +) { + applyOutcomeToStoredJob(state, authorCronRunCompletion(state, outcome.job, outcome)); +} + describe("cron trigger evaluation ownership", () => { it("keeps a replacement once trigger armed after an obsolete fired payload", () => { const state = makeState(); @@ -46,7 +54,7 @@ describe("cron trigger evaluation ownership", () => { job.state.triggerState = { owner: "replacement" }; state.store = { version: 1, jobs: [job] }; - applyOutcomeToStoredJob(state, { + applyAuthoredOutcome(state, { jobId: job.id, job: admittedJob, status: "ok", @@ -74,7 +82,7 @@ describe("cron trigger evaluation ownership", () => { job.state.scheduleErrorCount = 2; state.store = { version: 1, jobs: [job] }; - applyOutcomeToStoredJob(state, { + applyAuthoredOutcome(state, { jobId: job.id, job: admittedJob, status: "ok", @@ -101,7 +109,7 @@ describe("cron trigger evaluation ownership", () => { noteActiveCronJobTriggerMutation(job.id); try { - applyOutcomeToStoredJob(state, { + applyAuthoredOutcome(state, { jobId: job.id, job: admittedJob, activeJobMarker, @@ -253,7 +261,7 @@ describe("applyJobResult dynamic cadence", () => { state.store = { version: 1, jobs: [job] }; const admittedJob = structuredClone(job); - applyOutcomeToStoredJob(state, { + applyAuthoredOutcome(state, { jobId: job.id, job: admittedJob, status: "ok", @@ -279,7 +287,7 @@ describe("applyJobResult dynamic cadence", () => { job.state.forcePreservedNextRunAtMs = marker; state.store = { version: 1, jobs: [job] }; - applyOutcomeToStoredJob(state, { + applyAuthoredOutcome(state, { jobId: job.id, job: admittedJob, status: "ok", diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts index 394203151acc..021ba61c444d 100644 --- a/src/cron/service/timer.ts +++ b/src/cron/service/timer.ts @@ -4,7 +4,7 @@ import { onTimer } from "./timer-scheduler.js"; export type { CronTriggerEvalOutcome } from "./timer-execution-timeout.js"; export type { IsolatedAgentSetupTimeoutSignal } from "./timer-execution-timeout.js"; export { runsDetachedFromMainSession } from "./timer-execution-timeout.js"; -export { executeJobCoreWithTimeout } from "./timer-job-runner.js"; +export { authorCronRunCompletion, executeJobCoreWithTimeout } from "./timer-job-runner.js"; export { applyJobResult } from "./timer-outcomes.js"; export { applyTriggerRunResult } from "./timer-trigger.js"; export { applyScriptRunResult } from "./timer-outcomes.js"; diff --git a/src/cron/task-run-detail.ts b/src/cron/task-run-detail.ts index 743f802c4cea..4bff46e48432 100644 --- a/src/cron/task-run-detail.ts +++ b/src/cron/task-run-detail.ts @@ -12,6 +12,7 @@ import { FAILOVER_REASONS, type FailoverReason, } from "../../packages/gateway-protocol/src/failover-reasons.js"; +import { resolveCronCompletionStatus } from "./completion-status.js"; import { isCronTimeoutErrorText } from "./execution-error-constants.js"; import { normalizeCronRunDiagnosticsCore } from "./run-diagnostics-normalize.js"; @@ -25,6 +26,7 @@ type CronRunStatus = import("./types.js").CronRunStatus; const CRON_TASK_DETAIL_KIND = "cron-run"; const CRON_FAILOVER_REASONS = new Set(FAILOVER_REASONS); const cronRunStatusSchema = z.enum(["ok", "error", "skipped"]); +const cronCompletionStatusSchema = z.enum(["succeeded", "failed", "unknown"]); const cronDeliveryStatusSchema = z.enum(["delivered", "not-delivered", "unknown", "not-requested"]); const optionalCronStringSchema = z.string().optional().catch(undefined); const optionalNonBlankCronStringSchema = z @@ -78,6 +80,7 @@ const cronRunLogEntrySchema = z.looseObject({ .transform((value) => normalizeTimestamp(value)) .pipe(z.number()), status: cronRunStatusSchema.optional().catch(undefined), + completionStatus: cronCompletionStatusSchema.optional().catch(undefined), error: optionalCronStringSchema, errorReason: z .custom( @@ -149,6 +152,13 @@ export function parseCronRunLogEntryObject( jobId: entryObj.jobId, action: "finished", status: entryObj.status, + completionStatus: + entryObj.completionStatus ?? + resolveCronCompletionStatus({ + status: entryObj.status, + delivered: entryObj.delivered, + deliveryStatus: entryObj.deliveryStatus, + }), error: entryObj.error, errorReason: entryObj.errorReason, summary: entryObj.summary, @@ -198,6 +208,7 @@ export function cronRunLogEntryToTaskDetail( const detail = toJsonValue({ kind: CRON_TASK_DETAIL_KIND, status: entry.status, + completionStatus: entry.completionStatus, storeKey: options.storeKey, errorReason: entry.errorReason, diagnostics: entry.diagnostics, @@ -294,7 +305,14 @@ export function cronRunStatusToTaskStatus( entry: Pick & Partial, ): Extract { if (entry.status === "ok") { - return "succeeded"; + const completionStatus = + entry.completionStatus ?? + resolveCronCompletionStatus({ + status: entry.status, + delivered: entry.delivered, + deliveryStatus: entry.deliveryStatus, + }); + return completionStatus === "succeeded" ? "succeeded" : "failed"; } return entry.status === "error" && isCronTimeoutErrorText(entry.error) ? "timed_out" : "failed"; } diff --git a/src/cron/task-run-event-codec.test.ts b/src/cron/task-run-event-codec.test.ts index fc95c40c376d..a5ede2014d10 100644 --- a/src/cron/task-run-event-codec.test.ts +++ b/src/cron/task-run-event-codec.test.ts @@ -23,12 +23,13 @@ describe("cronRunLogEntryFromEvent", () => { jobId: "script-job", action: "finished", status: "error", + completionStatus: "failed", error: "cron script failed", }, 1, { kind: "reason", reason: "timeout" }, ); - expect(entry.errorReason).toBe("timeout"); + expect(entry).toMatchObject({ errorReason: "timeout", completionStatus: "failed" }); }); }); diff --git a/src/cron/task-run-event-codec.ts b/src/cron/task-run-event-codec.ts index 63b1d9983bd5..11508cb11ff4 100644 --- a/src/cron/task-run-event-codec.ts +++ b/src/cron/task-run-event-codec.ts @@ -33,6 +33,7 @@ export function cronRunLogEntryFromEvent( jobId: event.jobId, action: "finished", status: event.status, + completionStatus: event.completionStatus, error: event.error, errorReason, summary: event.summary, diff --git a/src/cron/task-run-history.test.ts b/src/cron/task-run-history.test.ts index 15bf8fce91fb..b8b4bf757e7e 100644 --- a/src/cron/task-run-history.test.ts +++ b/src/cron/task-run-history.test.ts @@ -99,6 +99,25 @@ describe("cron task run history", () => { ).toBe("failed"); }); + it.each([ + { completionStatus: "succeeded" as const, expected: "succeeded" }, + { completionStatus: "failed" as const, expected: "failed" }, + { completionStatus: "unknown" as const, expected: "failed" }, + ])( + "maps execution ok with completion $completionStatus to task status $expected", + ({ completionStatus, expected }) => { + expect( + cronRunStatusToTaskStatus({ + ts: 100, + jobId: JOB_ID, + action: "finished", + status: "ok", + completionStatus, + }), + ).toBe(expected); + }, + ); + it("reads executions produced by the cron service from the ledger", async () => { await withOpenClawTestState( { layout: "state-only", prefix: "openclaw-cron-task-service-history-" }, @@ -211,6 +230,7 @@ describe("cron task run history", () => { jobId: JOB_ID, action: "finished", status: "ok", + completionStatus: "succeeded", summary: "delivered\n needle", diagnostics: { summary: "healthy", @@ -254,6 +274,7 @@ describe("cron task run history", () => { jobId: JOB_ID, action: "finished", status: "error", + completionStatus: "failed", error: "provider overloaded", errorReason: "overloaded", deliveryStatus: "not-delivered", @@ -269,6 +290,7 @@ describe("cron task run history", () => { jobId: JOB_ID, action: "finished", status: "error", + completionStatus: "failed", error: "cron: job execution timed out", errorReason: "timeout", runId: "manual:history:timeout", @@ -281,6 +303,7 @@ describe("cron task run history", () => { jobId: JOB_ID, action: "finished", status: "skipped", + completionStatus: "failed", error: "trigger condition not met", summary: "", runId: "manual:history:skipped", @@ -306,6 +329,12 @@ describe("cron task run history", () => { "error", "ok", ]); + expect(ledger.entries.map((entry) => entry.completionStatus)).toEqual([ + "failed", + "failed", + "failed", + "succeeded", + ]); }, ); }); @@ -467,6 +496,43 @@ describe("cron task run history", () => { expect(entry?.failureNotificationDelivery).toBeUndefined(); }); + it.each([ + { status: "error", delivered: undefined, deliveryStatus: undefined, expected: "failed" }, + { status: "ok", delivered: undefined, deliveryStatus: "delivered", expected: "succeeded" }, + { status: "ok", delivered: true, deliveryStatus: undefined, expected: "succeeded" }, + { status: "ok", delivered: undefined, deliveryStatus: "not-requested", expected: "succeeded" }, + { status: "ok", delivered: undefined, deliveryStatus: "not-delivered", expected: "unknown" }, + { status: "ok", delivered: undefined, deliveryStatus: "unknown", expected: "unknown" }, + { status: "ok", delivered: undefined, deliveryStatus: undefined, expected: "unknown" }, + ] as const)( + "derives legacy $status/$deliveryStatus completion as $expected", + ({ status, delivered, deliveryStatus, expected }) => { + expect( + parseCronRunLogEntryObject({ + ts: 100, + jobId: JOB_ID, + action: "finished", + status, + ...(delivered === undefined ? {} : { delivered }), + ...(deliveryStatus === undefined ? {} : { deliveryStatus }), + })?.completionStatus, + ).toBe(expected); + }, + ); + + it("normalizes invalid completion status from immutable stored facts", () => { + expect( + parseCronRunLogEntryObject({ + ts: 100, + jobId: JOB_ID, + action: "finished", + status: "ok", + deliveryStatus: "not-delivered", + completionStatus: "partial", + })?.completionStatus, + ).toBe("unknown"); + }); + it("keeps quiet-trigger recovery detail out of run history", () => { const task = taskFromEntry( { ts: 100, jobId: JOB_ID, action: "finished", status: "ok" }, @@ -544,6 +610,7 @@ describe("cron task run history", () => { ).toEqual({ ...base, status: undefined, + completionStatus: "unknown", error: undefined, errorReason: undefined, summary: undefined, diff --git a/src/cron/types.ts b/src/cron/types.ts index eb1e11e54930..f8d178ec8cb6 100644 --- a/src/cron/types.ts +++ b/src/cron/types.ts @@ -11,6 +11,7 @@ import type { import type { CronJobBase, CronPacing } from "./types-shared.js"; export type { CronPacing } from "./types-shared.js"; +export type { CronCompletionStatus } from "./completion-status.js"; /** Supported schedule forms persisted in cron job specs. */ export type CronSchedule = @@ -149,6 +150,14 @@ export type CronFailureNotificationDelivery = { error?: string; }; +/** Resolved delivery state recorded with a completed cron run. */ +export type CronResolvedDeliveryState = { + delivered?: boolean; + status: CronDeliveryStatus; + error?: string; + failureNotification: CronFailureNotificationDelivery; +}; + /** Human-readable delivery target preview for list/detail surfaces. */ export type CronDeliveryPreview = { label: string; diff --git a/src/gateway/server-cron-notifications.delivery-failure.test.ts b/src/gateway/server-cron-notifications.delivery-failure.test.ts index 07246bbab4a5..5e18c1b8ce56 100644 --- a/src/gateway/server-cron-notifications.delivery-failure.test.ts +++ b/src/gateway/server-cron-notifications.delivery-failure.test.ts @@ -12,7 +12,7 @@ vi.mock("../cron/delivery.js", async (importOriginal) => { import { dispatchGatewayCronFinishedNotifications } from "./server-cron-notifications.js"; -function createThreadedJob(withFailureDestination: boolean): CronJob { +function createThreadedJob(withFailureDestination: boolean, bestEffort?: boolean): CronJob { return { id: "cron-delivery-failure", name: "threaded report", @@ -28,6 +28,7 @@ function createThreadedJob(withFailureDestination: boolean): CronJob { channel: "telegram", to: "-1001234567890", threadId: 42, + ...(bestEffort === undefined ? {} : { bestEffort }), ...(withFailureDestination ? { failureDestination: { @@ -56,6 +57,7 @@ describe("cron primary delivery failure notifications", () => { 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", }; @@ -86,4 +88,29 @@ describe("cron primary delivery failure notifications", () => { "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); + }, + ); }); diff --git a/src/gateway/server-cron-notifications.ts b/src/gateway/server-cron-notifications.ts index 2e8abdb8da21..0dddec49ce25 100644 --- a/src/gateway/server-cron-notifications.ts +++ b/src/gateway/server-cron-notifications.ts @@ -158,6 +158,7 @@ function buildCronFailureWebhookPayload(params: { evt: CronEvent; job: CronJob } 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, @@ -515,14 +516,21 @@ function dispatchCronFailureDestinationNotifications(params: { ssrfPolicy?: SsrFPolicy; globalFailureDestination?: CronFailureDestinationConfig; }): void { - if (!params.job || params.job.delivery?.bestEffort === true) { + 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); - const deliveryFailed = params.evt.deliveryStatus === "not-delivered"; - if (params.evt.status !== "error" && (!deliveryFailed || !failureDest)) { + if (deliveryOnlyFailed && !failureDest) { return; } const deliverySessionKey = resolveCronDeliverySessionKey(job); diff --git a/src/gateway/server-cron.test.ts b/src/gateway/server-cron.test.ts index 6650c23da4b4..939441a4bd0c 100644 --- a/src/gateway/server-cron.test.ts +++ b/src/gateway/server-cron.test.ts @@ -1761,7 +1761,7 @@ describe("buildGatewayCronService", () => { } }); - it("fails and retains a one-shot command when required delivery fails", async () => { + it("retains a one-shot command without changing execution status when required delivery fails", async () => { const cfg = createCronConfig("server-cron-command-required-delivery-failure"); loadConfigMock.mockReturnValue(cfg); const deliveryError = "network unavailable while delivering command output"; @@ -1790,14 +1790,23 @@ describe("buildGatewayCronService", () => { await state.cron.run(job.id, "force"); const updated = state.cron.getJob(job.id); - expect(updated?.state.lastRunStatus).toBe("error"); - expect(updated?.state.lastError).toBe(deliveryError); - expect(updated?.state.consecutiveErrors).toBe(1); + expect(updated?.enabled).toBe(false); + expect(updated?.state.lastRunStatus).toBe("ok"); + expect(updated?.state.lastError).toBeUndefined(); + expect(updated?.state.consecutiveErrors).toBe(0); expect(updated?.state.lastDeliveryStatus).toBe("not-delivered"); expect(updated?.state.lastDeliveryError).toBe(deliveryError); - expect(updated?.state.nextRunAtMs).toBeGreaterThanOrEqual( - (updated?.updatedAtMs ?? 0) + 30_000, - ); + expect(updated?.state.nextRunAtMs).toBeUndefined(); + expect( + runCronChangedMock.mock.calls + .map((_, index) => + requireRecord( + callArg(runCronChangedMock, index, 0, "cron_changed event"), + "cron_changed event", + ), + ) + .find((event) => event.action === "finished" && event.jobId === job.id), + ).toMatchObject({ status: "ok", completionStatus: "failed" }); } finally { state.cron.stop(); } @@ -1928,6 +1937,80 @@ describe("buildGatewayCronService", () => { } }); + it.each([ + { + name: "default best-effort", + bestEffort: undefined, + retained: false, + completion: "succeeded", + }, + { name: "explicit required", bestEffort: false, retained: true, completion: "failed" }, + ])( + "keeps script execution successful after $name announce failure", + async ({ name, bestEffort, retained, completion }) => { + const cfg = createCronConfig(`server-cron-script-${name}`); + cfg.cron = { ...cfg.cron, triggers: { enabled: true } }; + loadConfigMock.mockReturnValue(cfg); + cronScriptExecutorMock.mockResolvedValueOnce({ + kind: "completed", + notify: "queue changed", + stateChanged: false, + }); + sendCronAnnouncePayloadStrictMock.mockRejectedValueOnce(new Error("delivery rejected")); + + const state = buildGatewayCronService({ + cfg, + deps: {} as CliDeps, + broadcast: () => {}, + }); + try { + const job = await state.cron.add({ + name: `script ${name}`, + enabled: true, + deleteAfterRun: true, + schedule: { kind: "at", at: new Date(1).toISOString() }, + sessionTarget: "isolated", + wakeMode: "next-heartbeat", + payload: { kind: "script", script: "return { notify: 'queue changed' }" }, + delivery: { + mode: "announce", + channel: "telegram", + to: "123", + ...(bestEffort === undefined ? {} : { bestEffort }), + }, + }); + + await state.cron.run(job.id, "force"); + + const updated = state.cron.getJob(job.id); + expect(Boolean(updated)).toBe(retained); + if (updated) { + expect(updated).toMatchObject({ + enabled: false, + state: { + lastRunStatus: "ok", + lastDeliveryStatus: "not-delivered", + consecutiveErrors: 0, + }, + }); + expect(updated.state.lastError).toBeUndefined(); + } + expect( + runCronChangedMock.mock.calls + .map((_, index) => + requireRecord( + callArg(runCronChangedMock, index, 0, "cron_changed event"), + "cron_changed event", + ), + ) + .find((event) => event.action === "finished" && event.jobId === job.id), + ).toMatchObject({ status: "ok", completionStatus: completion }); + } finally { + state.cron.stop(); + } + }, + ); + it("delivers isolated script notify through the cron webhook path", async () => { const cfg = createCronConfig("server-cron-script-webhook"); cfg.cron = { ...cfg.cron, triggers: { enabled: true } }; diff --git a/src/gateway/server-cron.ts b/src/gateway/server-cron.ts index 2643ebce926a..bf8f662f85b4 100644 --- a/src/gateway/server-cron.ts +++ b/src/gateway/server-cron.ts @@ -871,18 +871,6 @@ export function buildGatewayCronService(params: { label: "command", traceResolvedFailure: true, }); - if ("deliveryError" in completion) { - const { deliveryError, ...deliveryResult } = completion; - const requiredDeliveryFailed = job.delivery?.bestEffort === false && result.status === "ok"; - return { - ...result, - // Default announce delivery is best-effort, but an explicit - // bestEffort:false keeps delivery inside the job's success contract. - status: requiredDeliveryFailed ? ("error" as const) : result.status, - ...(requiredDeliveryFailed ? { error: deliveryError } : { deliveryError }), - ...deliveryResult, - }; - } return { ...result, ...completion }; }, sendCronWebhook: async ({ job, event, abortSignal, deadlineAtMs, onDeliveryAccepted }) => { @@ -948,15 +936,6 @@ export function buildGatewayCronService(params: { logger: cronLogger, label: "script payload", }); - if ("deliveryError" in completion) { - const { deliveryError, ...deliveryResult } = completion; - return { - ...base, - status: job.delivery?.bestEffort ? ("ok" as const) : ("error" as const), - ...(job.delivery?.bestEffort ? { deliveryError } : { error: deliveryError }), - ...deliveryResult, - }; - } return { ...base, ...completion }; }, cleanupTimedOutAgentRun: async ({ job, execution }) => { @@ -1045,6 +1024,7 @@ export function buildGatewayCronService(params: { "runAtMs", "durationMs", "status", + "completionStatus", "error", "delivered", "deliveryStatus", diff --git a/src/plugins/hook-types.ts b/src/plugins/hook-types.ts index 3634f5167da8..0580fc7d98b4 100644 --- a/src/plugins/hook-types.ts +++ b/src/plugins/hook-types.ts @@ -985,6 +985,7 @@ export type PluginHookCronChangedEvent = { runAtMs?: number; durationMs?: number; status?: PluginHookGatewayCronRunStatus; + completionStatus?: "succeeded" | "failed" | "unknown"; error?: string; summary?: string; delivered?: boolean;