diff --git a/docs/automation/cron-jobs.md b/docs/automation/cron-jobs.md index 94724b81a894..1323579f922b 100644 --- a/docs/automation/cron-jobs.md +++ b/docs/automation/cron-jobs.md @@ -42,6 +42,7 @@ Cron is the Gateway's built-in scheduler. It persists jobs, wakes the agent at t - Cron runs **inside the Gateway** process (not inside the model). - Job definitions persist at `~/.openclaw/cron/jobs.json` so restarts do not lose schedules. - Runtime execution state persists next to it in `~/.openclaw/cron/jobs-state.json`. If you track cron definitions in git, track `jobs.json` and gitignore `jobs-state.json`. +- If `jobs.json` contains malformed rows, the Gateway keeps valid jobs running, removes the malformed rows from the active store, and saves the raw rows beside it in `jobs-quarantine.json` for later repair or review. - After the split, older OpenClaw versions can read `jobs.json` but may treat jobs as fresh because runtime fields now live in `jobs-state.json`. - When `jobs.json` is edited while the Gateway is running or stopped, OpenClaw compares the changed schedule fields with pending runtime slot metadata and clears stale `nextRunAtMs` values. Pure formatting or key-order-only rewrites preserve the pending slot. - All cron executions create [background task](/automation/tasks) records. diff --git a/docs/cli/cron.md b/docs/cli/cron.md index 054223ff8fde..b49d59491901 100644 --- a/docs/cli/cron.md +++ b/docs/cli/cron.md @@ -118,7 +118,7 @@ Skipped runs are tracked separately from execution errors. They do not affect re For isolated jobs that target a local configured model provider, cron runs a lightweight provider preflight before starting the agent turn. Loopback, private-network, and `.local` `api: "ollama"` providers are probed at `/api/tags`; local OpenAI-compatible providers such as vLLM, SGLang, and LM Studio are probed at `/models`. If the endpoint is unreachable, the run is recorded as `skipped` and retried on a later schedule; matching dead endpoints are cached for 5 minutes to avoid many jobs hammering the same local server. -Note: cron job definitions live in `jobs.json`, while pending runtime state lives in `jobs-state.json`. If `jobs.json` is edited externally, the Gateway reloads changed schedules and clears stale pending slots; formatting-only rewrites do not clear the pending slot. +Note: cron job definitions live in `jobs.json`, while pending runtime state lives in `jobs-state.json`. If `jobs.json` is edited externally, the Gateway reloads changed schedules and clears stale pending slots; formatting-only rewrites do not clear the pending slot. Malformed job rows are removed from active `jobs.json` at load time after their raw contents are copied to `jobs-quarantine.json`. ### Manual runs diff --git a/docs/gateway/doctor.md b/docs/gateway/doctor.md index 1395ab938d5a..3f82539a38ef 100644 --- a/docs/gateway/doctor.md +++ b/docs/gateway/doctor.md @@ -374,6 +374,8 @@ That stages grounded durable candidates into the short-term dreaming store while - payload `provider` delivery aliases → explicit `delivery.channel` - simple legacy `notify: true` webhook fallback jobs → explicit `delivery.mode="webhook"` with `delivery.to=cron.webhook` + The Gateway also sanitizes malformed cron rows at load time so valid jobs keep running. Raw malformed rows are copied to `jobs-quarantine.json` next to the active store before they are removed from `jobs.json`; doctor reports quarantined rows so you can review or repair them manually. + Doctor only auto-migrates `notify: true` jobs when it can do so without changing behavior. If a job combines legacy notify fallback with an existing non-webhook delivery mode, doctor warns and leaves that job for manual review. On Linux, doctor also warns when the user's crontab still invokes legacy `~/.openclaw/bin/ensure-whatsapp.sh`. That host-local script is not maintained by current OpenClaw and can write false `Gateway inactive` messages to `~/.openclaw/logs/whatsapp-health.log` when cron cannot reach the systemd user bus. Remove the stale crontab entry with `crontab -e`; use `openclaw channels status --probe`, `openclaw doctor`, and `openclaw gateway status` for current health checks. diff --git a/src/commands/doctor-cron-store-migration.test.ts b/src/commands/doctor-cron-store-migration.test.ts index 422830bdb7b4..f4652e5fccd5 100644 --- a/src/commands/doctor-cron-store-migration.test.ts +++ b/src/commands/doctor-cron-store-migration.test.ts @@ -159,13 +159,28 @@ describe("normalizeStoredCronJobs", () => { schedule: { kind: "every", everyMs: 60_000, anchorMs: 1 }, payload: { kind: "agentTurn", message: ["tick"] }, }), + makeLegacyJob({ + id: "missing-schedule", + schedule: undefined, + payload: { kind: "systemEvent", text: "tick" }, + }), + makeLegacyJob({ + id: "missing-payload", + schedule: { kind: "every", everyMs: 60_000, anchorMs: 1 }, + payload: undefined, + }), + makeLegacyJob({ + id: "incomplete-system-payload", + schedule: { kind: "cron", expr: "0 9 * * *", tz: "UTC" }, + payload: { kind: "systemEvent" }, + }), ]; const result = normalizeStoredCronJobs(jobs); expect(result.mutated).toBe(true); - expect(result.issues.invalidSchedule).toBe(1); - expect(result.issues.invalidPayload).toBe(1); + expect(result.issues.invalidSchedule).toBe(2); + expect(result.issues.invalidPayload).toBe(3); expect(jobs.map((job) => job.id)).toEqual(["valid"]); expect(result.jobs.map((job) => job.id)).toEqual(["valid"]); }); diff --git a/src/commands/doctor-cron.test.ts b/src/commands/doctor-cron.test.ts index 978fd2718182..340becc1fad3 100644 --- a/src/commands/doctor-cron.test.ts +++ b/src/commands/doctor-cron.test.ts @@ -3,6 +3,7 @@ import os from "node:os"; import path from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { OpenClawConfig } from "../config/config.js"; +import { resolveCronQuarantinePath } from "../cron/store.js"; import { collectLegacyWhatsAppCrontabHealthWarning, maybeRepairLegacyCronStore, @@ -124,6 +125,39 @@ function expectNoNoteContaining(message: string, title: string): void { } describe("maybeRepairLegacyCronStore", () => { + it("reports quarantined cron rows even when the active store is already sanitized", async () => { + const storePath = await makeTempStorePath(); + await writeCronStore(storePath, []); + await fs.writeFile( + resolveCronQuarantinePath(storePath), + JSON.stringify( + { + version: 1, + jobs: [ + { + quarantinedAtMs: Date.parse("2026-05-29T09:00:00.000Z"), + sourceIndex: 1, + reason: "missing-schedule", + job: { id: "bad-cron", name: "Bad cron" }, + }, + ], + }, + null, + 2, + ), + "utf-8", + ); + + await maybeRepairLegacyCronStore({ + cfg: createCronConfig(storePath), + options: {}, + prompter: makePrompter(true), + }); + + expectNoteContaining("Quarantined cron job rows found", "Cron"); + expectNoteContaining("1 row was removed from the active cron store", "Cron"); + }); + it("surfaces cron payload model overrides without rewriting current jobs", async () => { const storePath = await makeTempStorePath(); await writeCronStore(storePath, [ diff --git a/src/commands/doctor-cron.ts b/src/commands/doctor-cron.ts index 970b2a0d69b8..a98afb02be54 100644 --- a/src/commands/doctor-cron.ts +++ b/src/commands/doctor-cron.ts @@ -3,7 +3,13 @@ import { promisify } from "node:util"; import { formatCliCommand } from "../cli/command-format.js"; import { resolveAgentModelPrimaryValue } from "../config/model-input.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; -import { resolveCronStorePath, loadCronStore, saveCronStore } from "../cron/store.js"; +import { + loadCronQuarantineFile, + resolveCronQuarantinePath, + resolveCronStorePath, + loadCronStore, + saveCronStore, +} from "../cron/store.js"; import type { CronJob } from "../cron/types.js"; import { normalizeOptionalLowercaseString, @@ -335,6 +341,7 @@ export async function maybeRepairLegacyCronStore(params: { prompter: Pick; }) { const storePath = resolveCronStorePath(params.cfg.cron?.store); + const quarantinePath = resolveCronQuarantinePath(storePath); let store: Awaited>; try { store = await loadCronStore(storePath); @@ -350,6 +357,28 @@ export async function maybeRepairLegacyCronStore(params: { ); return; } + try { + const quarantine = await loadCronQuarantineFile(quarantinePath); + if (quarantine.jobs.length > 0) { + note( + [ + `Quarantined cron job rows found at ${shortenHomePath(quarantinePath)}.`, + `- ${pluralize(quarantine.jobs.length, "row")} was removed from the active cron store after runtime validation failed.`, + `- Review or repair the quarantined rows manually before copying any job back into ${shortenHomePath(storePath)}.`, + ].join("\n"), + "Cron", + ); + } + } catch (err) { + const reason = err instanceof Error ? err.message : String(err); + note( + [ + `Unable to read quarantined cron rows at ${shortenHomePath(quarantinePath)}.`, + `- ${reason}`, + ].join("\n"), + "Cron", + ); + } const rawJobs = (store.jobs ?? []) as unknown as Array>; if (rawJobs.length === 0) { return; diff --git a/src/cron/service.test-harness.ts b/src/cron/service.test-harness.ts index 41049550459e..96713646649a 100644 --- a/src/cron/service.test-harness.ts +++ b/src/cron/service.test-harness.ts @@ -247,7 +247,8 @@ export function createMockCronStateForJobs(params: { warnedDisabled: false, warnedMissingSessionTargetJobIds: new Set(), warnedInvalidPersistedJobKeys: new Set(), - preservedInvalidPersistedJobs: [], + pendingQuarantineConfigJobs: [], + lastQuarantineFailureWarnKey: null, deps: { storePath: "/mock/path", cronEnabled: true, diff --git a/src/cron/service/state.ts b/src/cron/service/state.ts index f5b270223349..7b1c8465a5f0 100644 --- a/src/cron/service/state.ts +++ b/src/cron/service/state.ts @@ -1,7 +1,7 @@ import type { CronConfig } from "../../config/types.cron.js"; import type { HeartbeatRunResult, HeartbeatWakeRequest } from "../../infra/heartbeat-wake.js"; import type { DeliveryContext } from "../../utils/delivery-context.types.js"; -import type { PreservedCronConfigJob } from "../store.js"; +import type { QuarantinedCronConfigJob } from "../store.js"; import type { CronAgentExecutionPhaseUpdate, CronAgentExecutionStarted, @@ -166,14 +166,11 @@ export type CronServiceState = { warnedMissingSessionTargetJobIds: Set; /** * Persisted job rows with non-canonical storage shape are skipped in memory - * until doctor/fix or an explicit config write repairs the store. + * until the runtime can quarantine and sanitize the active store. */ warnedInvalidPersistedJobKeys: Set; - /** - * Raw persisted config rows that are skipped for runtime safety but must not - * be deleted by routine cron-store writes. - */ - preservedInvalidPersistedJobs: PreservedCronConfigJob[]; + pendingQuarantineConfigJobs: QuarantinedCronConfigJob[]; + lastQuarantineFailureWarnKey: string | null; storeLoadedAtMs: number | null; storeFileMtimeMs: number | null; }; @@ -188,7 +185,8 @@ export function createCronServiceState(deps: CronServiceDeps): CronServiceState warnedDisabled: false, warnedMissingSessionTargetJobIds: new Set(), warnedInvalidPersistedJobKeys: new Set(), - preservedInvalidPersistedJobs: [], + pendingQuarantineConfigJobs: [], + lastQuarantineFailureWarnKey: null, storeLoadedAtMs: null, storeFileMtimeMs: null, }; diff --git a/src/cron/service/store.load-missing-session-target.test.ts b/src/cron/service/store.load-missing-session-target.test.ts index ec9eff81d404..c62415475f2b 100644 --- a/src/cron/service/store.load-missing-session-target.test.ts +++ b/src/cron/service/store.load-missing-session-target.test.ts @@ -2,6 +2,7 @@ import fs from "node:fs/promises"; import path from "node:path"; import { describe, expect, it, vi } from "vitest"; import { setupCronServiceSuite } from "../service.test-harness.js"; +import { resolveCronQuarantinePath } from "../store.js"; import { assertSupportedJobSpec, findJobOrThrow } from "./jobs.js"; import { createCronServiceState } from "./state.js"; import { ensureLoaded } from "./store.js"; @@ -16,7 +17,7 @@ async function writeSingleJobStore(storePath: string, job: Record>) { +async function writeJobStore(storePath: string, jobs: unknown[]) { await fs.mkdir(path.dirname(storePath), { recursive: true }); await fs.writeFile(storePath, JSON.stringify({ version: 1, jobs }, null, 2), "utf8"); } @@ -122,7 +123,7 @@ describe("cron service store load: missing sessionTarget", () => { expect(() => assertSupportedJobSpec(bogus)).toThrow(/missing sessionTarget/); }); - it("skips malformed persisted schedule and payload shapes without rewriting the store", async () => { + it("quarantines malformed persisted schedule and payload shapes while sanitizing the store", async () => { const { storePath } = await makeStorePath(); await writeJobStore(storePath, [ @@ -199,7 +200,6 @@ describe("cron service store load: missing sessionTarget", () => { state: {}, }, ]); - const beforeRaw = await fs.readFile(storePath, "utf-8"); const warnSpy = vi.spyOn(logger, "warn"); const state = createStoreTestState(storePath); @@ -208,11 +208,24 @@ describe("cron service store load: missing sessionTarget", () => { expect(state.store?.jobs.map((job) => job.id)).toEqual(["valid-job"]); expect(findJobOrThrow(state, "valid-job").state.nextRunAtMs).toBe(STORE_TEST_NOW); - await expect(fs.readFile(storePath, "utf-8")).resolves.toBe(beforeRaw); + const sanitized = JSON.parse(await fs.readFile(storePath, "utf-8")) as { + jobs: Array>; + }; + expect(sanitized.jobs.map((job) => job.id)).toEqual(["valid-job"]); + const quarantine = JSON.parse( + await fs.readFile(resolveCronQuarantinePath(storePath), "utf-8"), + ) as { jobs: Array<{ job?: Record }> }; + expect(quarantine.jobs.map((entry) => entry.job?.id)).toEqual([ + "bad-schedule", + "bad-payload", + "bad-cron-expr", + "bad-system-event-text", + "bad-agent-turn-message", + ]); const invalidShapeWarns = warnSpy.mock.calls.filter((call) => { const msg = typeof call[1] === "string" ? call[1] : ""; - return msg.includes("skipped invalid persisted job"); + return msg.includes("quarantined invalid persisted job"); }); expect(invalidShapeWarns).toHaveLength(5); expect(invalidShapeWarns.map((call) => (call[0] as { reason?: string }).reason)).toEqual([ diff --git a/src/cron/service/store.test.ts b/src/cron/service/store.test.ts index e333bbd7ddc4..e820c75fe7db 100644 --- a/src/cron/service/store.test.ts +++ b/src/cron/service/store.test.ts @@ -2,7 +2,7 @@ import fs from "node:fs/promises"; import path from "node:path"; import { describe, expect, it, vi } from "vitest"; import { setupCronServiceSuite } from "../service.test-harness.js"; -import { saveCronStore } from "../store.js"; +import { resolveCronQuarantinePath, saveCronStore } from "../store.js"; import type { CronJob } from "../types.js"; import { findJobOrThrow } from "./jobs.js"; import { createCronServiceState } from "./state.js"; @@ -18,7 +18,7 @@ async function writeSingleJobStore(storePath: string, job: Record>) { +async function writeJobStore(storePath: string, jobs: unknown[]) { await fs.mkdir(path.dirname(storePath), { recursive: true }); await fs.writeFile( storePath, @@ -134,7 +134,7 @@ describe("cron service store seam coverage", () => { expect((state.storeFileMtimeMs ?? 0) >= (firstMtime ?? 0)).toBe(true); }); - it("preserves unsupported payload-kind rows across full persistence without loading them", async () => { + it("quarantines unsupported payload-kind rows and sanitizes active jobs.json", async () => { const { storePath } = await makeStorePath(); await writeJobStore(storePath, [ @@ -189,24 +189,29 @@ describe("cron service store seam coverage", () => { const config = JSON.parse(await fs.readFile(storePath, "utf8")) as { jobs: Array>; }; - expect(config.jobs.map((job) => job.id)).toEqual([ - "valid-job", + expect(config.jobs.map((job) => job.id)).toEqual(["valid-job"]); + expect(config.jobs[0]?.name).toBe("valid job renamed"); + + const quarantine = JSON.parse( + await fs.readFile(resolveCronQuarantinePath(storePath), "utf8"), + ) as { jobs: Array<{ reason?: string; job?: Record }> }; + expect(quarantine.jobs.map((entry) => entry.job?.id)).toEqual([ "legacy-command", "legacy-agentmessage", ]); - expect(config.jobs[0]?.name).toBe("valid job renamed"); - expect(config.jobs[1]).toMatchObject({ + expect(quarantine.jobs[0]?.reason).toBe("invalid-payload"); + expect(quarantine.jobs[0]?.job).toMatchObject({ id: "legacy-command", payload: { kind: "command", command: "echo daily" }, state: { lastRunAtMs: STORE_TEST_NOW - 3_600_000 }, }); - expect(config.jobs[2]).toMatchObject({ + expect(quarantine.jobs[1]?.job).toMatchObject({ id: "legacy-agentmessage", payload: { kind: "agentmessage", message: "summarize" }, metadata: { preserve: { nested: true } }, }); - expect(config.jobs[2]).not.toHaveProperty("state"); - expect(config.jobs[2]).not.toHaveProperty("updatedAtMs"); + expect(quarantine.jobs[1]?.job).not.toHaveProperty("state"); + expect(quarantine.jobs[1]?.job).not.toHaveProperty("updatedAtMs"); const stateFile = JSON.parse( await fs.readFile(storePath.replace(/\.json$/, "-state.json"), "utf8"), @@ -215,7 +220,7 @@ describe("cron service store seam coverage", () => { const invalidPayloadWarns = logger.warn.mock.calls.filter((call) => { const msg = typeof call[1] === "string" ? call[1] : ""; - return msg.includes("skipped invalid persisted job"); + return msg.includes("quarantined invalid persisted job"); }); expect(invalidPayloadWarns.map((call) => (call[0] as { jobId?: string }).jobId)).toEqual([ "legacy-command", @@ -223,7 +228,272 @@ describe("cron service store seam coverage", () => { ]); }); - it("skips preserved unsupported rows that collide with supported jobs by canonical id", async () => { + it("quarantines malformed persisted rows and sanitizes active jobs.json", async () => { + const { storePath } = await makeStorePath(); + + await writeJobStore(storePath, [ + { + id: "valid-job", + name: "valid job", + enabled: true, + createdAtMs: STORE_TEST_NOW - 60_000, + updatedAtMs: STORE_TEST_NOW - 60_000, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "main", + wakeMode: "now", + payload: { kind: "systemEvent", text: "tick" }, + state: {}, + }, + { + id: "missing-schedule-job", + name: "missing schedule job", + enabled: true, + payload: { kind: "systemEvent", text: "tick" }, + state: { lastRunAtMs: STORE_TEST_NOW - 3_600_000 }, + }, + { + id: "missing-schedule-job", + name: "missing schedule job", + enabled: true, + payload: { kind: "systemEvent", text: "tick" }, + state: { lastRunAtMs: STORE_TEST_NOW - 3_600_000 }, + }, + { + id: "missing-system-text-job", + name: "missing system text job", + enabled: true, + schedule: { kind: "cron", expr: "0 9 * * *", tz: "UTC" }, + payload: { kind: "systemEvent" }, + metadata: { preserve: { nested: true } }, + }, + "bad-scalar-row", + ]); + await fs.writeFile( + storePath.replace(/\.json$/, "-state.json"), + JSON.stringify( + { + version: 1, + jobs: { + "missing-system-text-job": { + updatedAtMs: STORE_TEST_NOW - 30_000, + state: { lastStatus: "error", lastRunAtMs: STORE_TEST_NOW - 120_000 }, + }, + }, + }, + null, + 2, + ), + "utf8", + ); + + const state = createStoreTestState(storePath); + await ensureLoaded(state, { skipRecompute: true }); + + expect(state.store?.jobs.map((job) => job.id)).toEqual(["valid-job"]); + expect(() => findJobOrThrow(state, "missing-schedule-job")).toThrow(/unknown cron job id/); + expect(() => findJobOrThrow(state, "missing-system-text-job")).toThrow(/unknown cron job id/); + + const valid = findJobOrThrow(state, "valid-job"); + valid.name = "valid job renamed"; + await persist(state); + + const config = JSON.parse(await fs.readFile(storePath, "utf8")) as { + jobs: Array>; + }; + expect(config.jobs.map((job) => job.id)).toEqual(["valid-job"]); + expect(config.jobs[0]?.name).toBe("valid job renamed"); + + const quarantine = JSON.parse( + await fs.readFile(resolveCronQuarantinePath(storePath), "utf8"), + ) as { + jobs: Array<{ + reason?: string; + job?: Record; + raw?: unknown; + sourceIndex?: number; + state?: Record; + updatedAtMs?: number; + }>; + }; + expect(quarantine.jobs.map((entry) => entry.job?.id ?? entry.raw)).toEqual([ + "missing-schedule-job", + "missing-schedule-job", + "missing-system-text-job", + "bad-scalar-row", + ]); + expect(quarantine.jobs.map((entry) => entry.reason)).toEqual([ + "missing-schedule", + "missing-schedule", + "invalid-payload", + "non-object-row", + ]); + expect(quarantine.jobs.map((entry) => entry.sourceIndex)).toEqual([1, 2, 3, 4]); + expect(quarantine.jobs[0]?.job).toMatchObject({ + id: "missing-schedule-job", + state: { lastRunAtMs: STORE_TEST_NOW - 3_600_000 }, + }); + expect(quarantine.jobs[2]?.job).toMatchObject({ + id: "missing-system-text-job", + metadata: { preserve: { nested: true } }, + }); + expect(quarantine.jobs[2]?.state).toEqual({ + lastStatus: "error", + lastRunAtMs: STORE_TEST_NOW - 120_000, + }); + expect(quarantine.jobs[2]?.updatedAtMs).toBe(STORE_TEST_NOW - 30_000); + expect(quarantine.jobs[2]?.job).not.toHaveProperty("state"); + expect(quarantine.jobs[2]?.job).not.toHaveProperty("updatedAtMs"); + + const stateFile = JSON.parse( + await fs.readFile(storePath.replace(/\.json$/, "-state.json"), "utf8"), + ) as { jobs: Record }; + expect(Object.keys(stateFile.jobs)).toEqual(["valid-job"]); + + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ storePath, jobId: "missing-schedule-job", jobIndex: 1 }), + expect.stringContaining("quarantined invalid persisted job"), + ); + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ storePath, jobId: "missing-system-text-job", jobIndex: 3 }), + expect.stringContaining("quarantined invalid persisted job"), + ); + }); + + it("quarantines legacy jobId rows with split runtime state before pruning state file", async () => { + const { storePath } = await makeStorePath(); + + await writeJobStore(storePath, [ + { + id: "valid-job", + name: "valid job", + enabled: true, + createdAtMs: STORE_TEST_NOW - 60_000, + updatedAtMs: STORE_TEST_NOW - 60_000, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "main", + wakeMode: "now", + payload: { kind: "systemEvent", text: "tick" }, + state: {}, + }, + { + jobId: "legacy-invalid-job", + name: "legacy invalid job", + enabled: true, + schedule: { kind: "cron", expr: "0 9 * * *", tz: "UTC" }, + payload: { kind: "systemEvent" }, + }, + ]); + await fs.writeFile( + storePath.replace(/\.json$/, "-state.json"), + JSON.stringify( + { + version: 1, + jobs: { + "legacy-invalid-job": { + updatedAtMs: STORE_TEST_NOW - 45_000, + scheduleIdentity: "legacy-schedule-identity", + state: { lastStatus: "error", lastRunAtMs: STORE_TEST_NOW - 90_000 }, + }, + }, + }, + null, + 2, + ), + "utf8", + ); + + const state = createStoreTestState(storePath); + await ensureLoaded(state, { skipRecompute: true }); + + const quarantine = JSON.parse( + await fs.readFile(resolveCronQuarantinePath(storePath), "utf8"), + ) as { + jobs: Array<{ + job?: Record; + scheduleIdentity?: string; + state?: Record; + updatedAtMs?: number; + }>; + }; + expect(quarantine.jobs).toHaveLength(1); + expect(quarantine.jobs[0]?.job).toMatchObject({ jobId: "legacy-invalid-job" }); + expect(quarantine.jobs[0]?.state).toEqual({ + lastStatus: "error", + lastRunAtMs: STORE_TEST_NOW - 90_000, + }); + expect(quarantine.jobs[0]?.updatedAtMs).toBe(STORE_TEST_NOW - 45_000); + expect(quarantine.jobs[0]?.scheduleIdentity).toBe("legacy-schedule-identity"); + + const stateFile = JSON.parse( + await fs.readFile(storePath.replace(/\.json$/, "-state.json"), "utf8"), + ) as { jobs: Record }; + expect(Object.keys(stateFile.jobs)).toEqual(["valid-job"]); + }); + + it("blocks later persists until malformed rows are copied to quarantine", async () => { + const { storePath } = await makeStorePath(); + const quarantinePath = resolveCronQuarantinePath(storePath); + + await writeJobStore(storePath, [ + { + id: "valid-job", + name: "valid job", + enabled: true, + createdAtMs: STORE_TEST_NOW - 60_000, + updatedAtMs: STORE_TEST_NOW - 60_000, + schedule: { kind: "every", everyMs: 60_000 }, + sessionTarget: "main", + wakeMode: "now", + payload: { kind: "systemEvent", text: "tick" }, + state: {}, + }, + { + id: "missing-schedule-job", + name: "missing schedule job", + enabled: true, + payload: { kind: "systemEvent", text: "tick" }, + }, + ]); + await fs.writeFile(quarantinePath, "{ not json", "utf8"); + + const state = createStoreTestState(storePath); + await ensureLoaded(state, { skipRecompute: true }); + + const valid = findJobOrThrow(state, "valid-job"); + valid.name = "valid job renamed"; + await persist(state, { stateOnly: true }); + await ensureLoaded(state, { forceReload: true, skipRecompute: true }); + findJobOrThrow(state, "valid-job").name = "valid job renamed"; + await persist(state); + + const quarantineFailureWarns = logger.warn.mock.calls.filter((call) => { + const msg = typeof call[1] === "string" ? call[1] : ""; + return msg.includes("failed to quarantine malformed persisted jobs"); + }); + expect(quarantineFailureWarns).toHaveLength(1); + + let config = JSON.parse(await fs.readFile(storePath, "utf8")) as { + jobs: Array>; + }; + expect(config.jobs.map((job) => job.id)).toEqual(["valid-job", "missing-schedule-job"]); + expect(config.jobs[0]?.name).toBe("valid job"); + + await fs.writeFile(quarantinePath, JSON.stringify({ version: 1, jobs: [] }), "utf8"); + await persist(state); + + config = JSON.parse(await fs.readFile(storePath, "utf8")) as { + jobs: Array>; + }; + expect(config.jobs.map((job) => job.id)).toEqual(["valid-job"]); + expect(config.jobs[0]?.name).toBe("valid job renamed"); + + const quarantine = JSON.parse(await fs.readFile(quarantinePath, "utf8")) as { + jobs: Array<{ job?: Record }>; + }; + expect(quarantine.jobs.map((entry) => entry.job?.id)).toEqual(["missing-schedule-job"]); + }); + + it("keeps canonical jobs when quarantined unsupported rows collide by id", async () => { const { storePath } = await makeStorePath(); await writeJobStore(storePath, [ @@ -496,6 +766,33 @@ describe("cron service store seam coverage", () => { expect(job.state.nextRunAtMs).toBeUndefined(); }); + it("warns once per malformed persisted row across repeated forceReload cycles", async () => { + const { storePath } = await makeStorePath(); + + await writeSingleJobStore(storePath, { + id: "missing-cron-expr-job", + name: "missing cron expr job", + enabled: true, + schedule: { kind: "cron" }, + payload: { kind: "systemEvent", text: "tick" }, + state: {}, + }); + + const warnSpy = vi.spyOn(logger, "warn"); + const state = createStoreTestState(storePath); + + await ensureLoaded(state, { skipRecompute: true }); + await ensureLoaded(state, { forceReload: true, skipRecompute: true }); + await ensureLoaded(state, { forceReload: true, skipRecompute: true }); + + const malformedWarns = warnSpy.mock.calls.filter((call) => { + const msg = typeof call[1] === "string" ? call[1] : ""; + return msg.includes("quarantined invalid persisted job"); + }); + expect(malformedWarns).toHaveLength(1); + warnSpy.mockRestore(); + }); + it("preserves nextRunAtMs after force reload when scheduling inputs are unchanged", async () => { const { storePath } = await makeStorePath(); const originalNextRunAtMs = STORE_TEST_NOW + 3_600_000; diff --git a/src/cron/service/store.ts b/src/cron/service/store.ts index 884407730a1d..a27ab454c99d 100644 --- a/src/cron/service/store.ts +++ b/src/cron/service/store.ts @@ -6,8 +6,9 @@ import { cronSchedulingInputsEqual } from "../schedule-identity.js"; import { isInvalidCronSessionTargetIdError } from "../session-target.js"; import { loadCronStoreWithConfigJobs, + saveCronQuarantineFile, saveCronStore, - type PreservedCronConfigJob, + type QuarantinedCronConfigJob, } from "../store.js"; import type { CronJob } from "../types.js"; import { recomputeNextRuns } from "./jobs.js"; @@ -44,22 +45,7 @@ function warnInvalidPersistedCronJob(params: { jobIndex: params.index, reason: params.reason, }, - "cron: skipped invalid persisted job; run openclaw doctor --fix to repair", - ); -} - -function isRecord(value: unknown): value is Record { - return !!value && typeof value === "object" && !Array.isArray(value); -} - -function hasUnsupportedStringPayloadKind(candidate: Record): boolean { - const payload = candidate.payload; - if (!isRecord(payload)) { - return false; - } - const kind = payload.kind; - return ( - typeof kind === "string" && kind.trim() !== "" && kind !== "systemEvent" && kind !== "agentTurn" + "cron: quarantined invalid persisted job and skipped it from runtime", ); } @@ -72,6 +58,39 @@ async function getFileMtimeMs(path: string): Promise { } } +async function flushPendingQuarantine( + state: CronServiceState, + nowMs: number, +): Promise { + if (state.pendingQuarantineConfigJobs.length === 0) { + return null; + } + try { + const quarantinePath = await saveCronQuarantineFile({ + storePath: state.deps.storePath, + entries: state.pendingQuarantineConfigJobs, + nowMs, + }); + state.pendingQuarantineConfigJobs = []; + state.lastQuarantineFailureWarnKey = null; + return quarantinePath; + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error); + const warnKey = `${state.deps.storePath}\0${errorMessage}`; + if (state.lastQuarantineFailureWarnKey !== warnKey) { + state.lastQuarantineFailureWarnKey = warnKey; + state.deps.log.warn( + { + storePath: state.deps.storePath, + error: errorMessage, + }, + "cron: failed to quarantine malformed persisted jobs; skipping active store sanitization", + ); + } + return null; + } +} + export async function ensureLoaded( state: CronServiceState, opts?: { @@ -97,10 +116,12 @@ export async function ensureLoaded( const loaded = await loadCronStoreWithConfigJobs(state.deps.storePath); const loadedJobs = (loaded.store.jobs ?? []) as unknown as CronJob[]; const jobs: CronJob[] = []; - const preservedInvalidPersistedJobs: PreservedCronConfigJob[] = []; + const quarantinedConfigJobs: QuarantinedCronConfigJob[] = [...loaded.invalidConfigRows]; for (const [index, job] of loadedJobs.entries()) { const raw = job as unknown as Record; const rawConfigJob = loaded.configJobs[index] ?? structuredClone(raw); + const sourceIndex = loaded.configJobIndexes[index] ?? index; + const runtimeEntry = loaded.configJobRuntimeEntries[index]; const { legacyJobIdIssue } = normalizeCronJobIdentityFields(raw); let normalized: Record | null; try { @@ -121,10 +142,24 @@ export async function ensureLoaded( hydrated as unknown as Record, ); if (invalidReason) { - if (invalidReason === "invalid-payload" && hasUnsupportedStringPayloadKind(rawConfigJob)) { - preservedInvalidPersistedJobs.push({ index, job: rawConfigJob }); + const quarantineEntry: QuarantinedCronConfigJob = { + sourceIndex, + reason: invalidReason, + job: rawConfigJob, + }; + const runtimeState = runtimeEntry?.state ?? raw.state; + if (runtimeState && typeof runtimeState === "object" && !Array.isArray(runtimeState)) { + quarantineEntry.state = structuredClone(runtimeState as Record); } - warnInvalidPersistedCronJob({ state, raw, index, reason: invalidReason }); + const updatedAtMs = runtimeEntry?.updatedAtMs ?? raw.updatedAtMs; + if (typeof updatedAtMs === "number" && Number.isFinite(updatedAtMs)) { + quarantineEntry.updatedAtMs = updatedAtMs; + } + if (typeof runtimeEntry?.scheduleIdentity === "string") { + quarantineEntry.scheduleIdentity = runtimeEntry.scheduleIdentity; + } + quarantinedConfigJobs.push(quarantineEntry); + warnInvalidPersistedCronJob({ state, raw, index: sourceIndex, reason: invalidReason }); continue; } jobs.push(hydrated); @@ -183,10 +218,36 @@ export async function ensureLoaded( version: 1, jobs, }; - state.preservedInvalidPersistedJobs = preservedInvalidPersistedJobs; state.storeLoadedAtMs = state.deps.nowMs(); state.storeFileMtimeMs = fileMtimeMs; + if (quarantinedConfigJobs.length > 0) { + state.pendingQuarantineConfigJobs = quarantinedConfigJobs; + const quarantinePath = await flushPendingQuarantine(state, state.storeLoadedAtMs); + if (quarantinePath) { + try { + await saveCronStore(state.deps.storePath, state.store); + state.storeFileMtimeMs = await getFileMtimeMs(state.deps.storePath); + state.deps.log.warn( + { + storePath: state.deps.storePath, + quarantinePath, + quarantinedJobs: quarantinedConfigJobs.length, + }, + "cron: sanitized active jobs.json after quarantining malformed persisted jobs", + ); + } catch (error) { + state.deps.log.warn( + { + storePath: state.deps.storePath, + error: error instanceof Error ? error.message : String(error), + }, + "cron: failed to sanitize malformed persisted jobs after quarantine; continuing with quarantined in-memory view", + ); + } + } + } + if (!opts?.skipRecompute) { recomputeNextRuns(state); } @@ -213,10 +274,16 @@ export async function persist( if (!state.store) { return; } - await saveCronStore(state.deps.storePath, state.store, { - ...opts, - preservedConfigJobs: state.preservedInvalidPersistedJobs, - }); + let flushedPendingQuarantine = false; + if (state.pendingQuarantineConfigJobs.length > 0) { + const quarantinePath = await flushPendingQuarantine(state, state.deps.nowMs()); + if (!quarantinePath) { + return; + } + flushedPendingQuarantine = true; + } + const saveOpts = flushedPendingQuarantine ? { skipBackup: opts?.skipBackup } : opts; + await saveCronStore(state.deps.storePath, state.store, saveOpts); // Update file mtime after save to prevent immediate reload state.storeFileMtimeMs = await getFileMtimeMs(state.deps.storePath); } diff --git a/src/cron/store.test.ts b/src/cron/store.test.ts index 7a1bd46cfaf4..afcbbc0fd8f8 100644 --- a/src/cron/store.test.ts +++ b/src/cron/store.test.ts @@ -3,7 +3,15 @@ import os from "node:os"; import path from "node:path"; import { setTimeout as scheduleNativeTimeout } from "node:timers"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { loadCronStore, loadCronStoreSync, resolveCronStorePath, saveCronStore } from "./store.js"; +import { + loadCronQuarantineFile, + loadCronStore, + loadCronStoreSync, + resolveCronQuarantinePath, + resolveCronStorePath, + saveCronQuarantineFile, + saveCronStore, +} from "./store.js"; import type { CronStoreFile } from "./types.js"; let fixtureRoot = ""; @@ -220,6 +228,45 @@ describe("cron store", () => { expect(loaded.jobs[0]?.state).toStrictEqual({}); }); + it("fails closed instead of overwriting unrecognized quarantine files", async () => { + const { storePath } = await makeStorePath(); + const quarantinePath = resolveCronQuarantinePath(storePath); + await fs.mkdir(path.dirname(storePath), { recursive: true }); + await fs.writeFile( + quarantinePath, + JSON.stringify({ version: 2, jobs: [{ reason: "old-shape", raw: "keep-me" }] }, null, 2), + "utf-8", + ); + + await expect(loadCronQuarantineFile(quarantinePath)).rejects.toThrow( + /Unsupported cron quarantine file shape/, + ); + await expect( + saveCronQuarantineFile({ + storePath, + nowMs: 123, + entries: [{ sourceIndex: 0, reason: "missing-schedule", job: { id: "new-row" } }], + }), + ).rejects.toThrow(/Unsupported cron quarantine file shape/); + + const preserved = JSON.parse(await fs.readFile(quarantinePath, "utf-8")) as { + jobs: Array>; + }; + expect(preserved.jobs[0]?.raw).toBe("keep-me"); + }); + + it("does not rewrite quarantine files when every entry is already present", async () => { + const { storePath } = await makeStorePath(); + const quarantinePath = resolveCronQuarantinePath(storePath); + const entry = { sourceIndex: 0, reason: "missing-schedule", job: { id: "same-row" } }; + + await saveCronQuarantineFile({ storePath, nowMs: 100, entries: [entry] }); + const firstRaw = await fs.readFile(quarantinePath, "utf-8"); + await saveCronQuarantineFile({ storePath, nowMs: 200, entries: [entry] }); + + expect(await fs.readFile(quarantinePath, "utf-8")).toBe(firstRaw); + }); + it("loads split cron state synchronously for task reconciliation", async () => { const { storePath } = await makeStorePath(); await saveCronStore(storePath, makeStore("job-sync", true)); @@ -232,6 +279,54 @@ describe("cron store", () => { expect(loaded.jobs[0]?.updatedAtMs).toBeTypeOf("number"); }); + it("loads split cron state synchronously for legacy jobId rows", async () => { + const { storePath } = await makeStorePath(); + const statePath = storePath.replace(/\.json$/, "-state.json"); + await fs.mkdir(path.dirname(storePath), { recursive: true }); + await fs.writeFile( + storePath, + JSON.stringify( + { + version: 1, + jobs: [ + { + jobId: "legacy-sync-job", + name: "legacy sync job", + enabled: true, + schedule: { kind: "every", everyMs: 60_000 }, + payload: { kind: "systemEvent", text: "tick" }, + }, + ], + }, + null, + 2, + ), + "utf-8", + ); + await fs.writeFile( + statePath, + JSON.stringify( + { + version: 1, + jobs: { + "legacy-sync-job": { + updatedAtMs: 123, + state: { runningAtMs: 456 }, + }, + }, + }, + null, + 2, + ), + "utf-8", + ); + + const loaded = loadCronStoreSync(storePath); + + expect(loaded.jobs[0]?.state).toEqual({ runningAtMs: 456 }); + expect(loaded.jobs[0]?.updatedAtMs).toBe(123); + }); + it("compares split state identity for flat legacy cron rows", async () => { const { storePath } = await makeStorePath(); const statePath = storePath.replace(/\.json$/, "-state.json"); diff --git a/src/cron/store.ts b/src/cron/store.ts index 1019e4d86e73..83d39f635e11 100644 --- a/src/cron/store.ts +++ b/src/cron/store.ts @@ -15,14 +15,27 @@ type SerializedStoreCacheEntry = { needsSplitMigration: boolean; }; -export type PreservedCronConfigJob = { - index: number; - job: Record; +export type QuarantinedCronConfigJob = { + sourceIndex: number; + reason: string; + job?: Record; + raw?: unknown; + state?: Record; + updatedAtMs?: number; + scheduleIdentity?: string; +}; + +export type CronQuarantineFile = { + version: 1; + jobs: Array; }; export type LoadedCronStore = { store: CronStoreFile; configJobs: Array>; + configJobIndexes: number[]; + configJobRuntimeEntries: CronConfigJobRuntimeEntry[]; + invalidConfigRows: QuarantinedCronConfigJob[]; }; const serializedStoreCache = new Map(); @@ -51,29 +64,42 @@ function resolveStatePath(storePath: string): string { return `${storePath}-state.json`; } +export function resolveCronQuarantinePath(storePath: string): string { + if (storePath.endsWith(".json")) { + return storePath.replace(/\.json$/, "-quarantine.json"); + } + return `${storePath}-quarantine.json`; +} + type CronStateFileEntry = { updatedAtMs?: number; scheduleIdentity?: string; state?: Record; }; +export type CronConfigJobRuntimeEntry = CronStateFileEntry; + type CronStateFile = { version: 1; jobs: Record; }; function normalizeCronStoreFile(parsed: unknown): CronStoreFile { - const rawJobs = Array.isArray(parsed) - ? parsed - : isRecord(parsed) && Array.isArray(parsed.jobs) - ? parsed.jobs - : []; + const rawJobs = getRawCronJobs(parsed); return { version: 1, jobs: rawJobs.filter(isRecord) as never as CronStoreFile["jobs"], }; } +function getRawCronJobs(parsed: unknown): unknown[] { + return Array.isArray(parsed) + ? parsed + : isRecord(parsed) && Array.isArray(parsed.jobs) + ? parsed.jobs + : []; +} + function cloneConfigJobs(jobs: Array>): Array> { return jobs.map((job) => structuredClone(job)); } @@ -83,61 +109,11 @@ function stripJobRuntimeFields(job: CronStoreFile["jobs"][number]): Record): string | null { - return normalizeOptionalString(job.id) ?? normalizeOptionalString(job.jobId) ?? null; -} - -function mergePreservedConfigJobs( - jobs: Array>, - preservedConfigJobs: PreservedCronConfigJob[] | undefined, -): Array> { - if (!preservedConfigJobs?.length) { - return jobs; - } - - const occupiedIds = new Set(); - for (const job of jobs) { - const id = persistedJobId(job); - if (id) { - occupiedIds.add(id); - } - } - - const seenPreservedIds = new Set(); - const preserved = preservedConfigJobs - .filter((entry) => { - const id = persistedJobId(entry.job); - if (!id) { - return true; - } - if (occupiedIds.has(id) || seenPreservedIds.has(id)) { - return false; - } - seenPreservedIds.add(id); - return true; - }) - .toSorted((a, b) => a.index - b.index); - - if (preserved.length === 0) { - return jobs; - } - - const merged = jobs.slice(); - for (const entry of preserved) { - const index = Math.max(0, Math.min(entry.index, merged.length)); - merged.splice(index, 0, { ...entry.job }); - } - return merged; -} - -function stripRuntimeOnlyCronFields( - store: CronStoreFile, - preservedConfigJobs?: PreservedCronConfigJob[], -): unknown { +function stripRuntimeOnlyCronFields(store: CronStoreFile): unknown { const jobs = store.jobs.map(stripJobRuntimeFields); return { version: store.version, - jobs: mergePreservedConfigJobs(jobs, preservedConfigJobs), + jobs, }; } @@ -278,6 +254,10 @@ function mergeStateFileEntry(job: CronStoreFile["jobs"][number], entry: unknown) } } +function resolveCronStateId(job: Record): string | undefined { + return normalizeOptionalString(job.id) ?? normalizeOptionalString(job.jobId); +} + export async function loadCronStoreWithConfigJobs(storePath: string): Promise { try { const raw = await fs.promises.readFile(storePath, "utf-8"); @@ -289,9 +269,29 @@ export async function loadCronStoreWithConfigJobs(storePath: string): Promise> = []; + const configJobRuntimeEntries: CronConfigJobRuntimeEntry[] = []; + const invalidConfigRows: QuarantinedCronConfigJob[] = []; + for (const [index, row] of rawJobs.entries()) { + if (isRecord(row)) { + configJobIndexes.push(index); + configRows.push(row); + } else { + invalidConfigRows.push({ + sourceIndex: index, + reason: "non-object-row", + raw: structuredClone(row), + }); + } + } + const store: CronStoreFile = { + version: 1, + jobs: configRows as never as CronStoreFile["jobs"], + }; const jobs = store.jobs as unknown as Array>; - const configJobs = cloneConfigJobs(jobs); + const configJobs = cloneConfigJobs(configRows); // Load state file and merge. const statePath = resolveStatePath(storePath); @@ -301,7 +301,9 @@ export async function loadCronStoreWithConfigJobs(storePath: string): Promise); + const entry = stateId ? stateFile.jobs[stateId] : undefined; + configJobRuntimeEntries.push(isRecord(entry) ? structuredClone(entry) : {}); if (entry) { mergeStateFileEntry(job, entry); } else { @@ -329,11 +331,17 @@ export async function loadCronStoreWithConfigJobs(storePath: string): Promise); + const entry = stateId ? stateFile.jobs[stateId] : undefined; if (entry) { mergeStateFileEntry(job, entry); } else { @@ -391,7 +400,6 @@ export function loadCronStoreSync(storePath: string): CronStoreFile { type SaveCronStoreOptions = { skipBackup?: boolean; stateOnly?: boolean; - preservedConfigJobs?: PreservedCronConfigJob[]; }; async function setSecureFileMode(filePath: string): Promise { @@ -435,11 +443,7 @@ export async function saveCronStore( opts?: SaveCronStoreOptions, ) { const stateOnly = opts?.stateOnly === true; - const configJson = JSON.stringify( - stripRuntimeOnlyCronFields(store, opts?.preservedConfigJobs), - null, - 2, - ); + const configJson = JSON.stringify(stripRuntimeOnlyCronFields(store), null, 2); const stateFile = extractStateFile(store); const stateJson = JSON.stringify(stateFile, null, 2); @@ -485,3 +489,109 @@ export async function saveCronStore( } updatedCache.needsSplitMigration = stateOnly && migrating; } + +export async function loadCronQuarantineFile(path: string): Promise { + try { + const raw = await fs.promises.readFile(path, "utf-8"); + const parsed = parseJsonWithJson5Fallback(raw); + if (!isRecord(parsed) || parsed.version !== 1 || !Array.isArray(parsed.jobs)) { + throw new Error(`Unsupported cron quarantine file shape at ${path}`); + } + const jobs = parsed.jobs.map((entry, index) => { + if ( + !isRecord(entry) || + typeof entry.reason !== "string" || + (!isRecord(entry.job) && !("raw" in entry)) + ) { + throw new Error(`Unsupported cron quarantine entry at ${path} index ${index}`); + } + const sourceIndex = typeof entry.sourceIndex === "number" ? entry.sourceIndex : -1; + const quarantinedAtMs = + typeof entry.quarantinedAtMs === "number" && Number.isFinite(entry.quarantinedAtMs) + ? entry.quarantinedAtMs + : Date.now(); + const quarantined: CronQuarantineFile["jobs"][number] = { + quarantinedAtMs, + sourceIndex, + reason: entry.reason, + }; + if (isRecord(entry.job)) { + quarantined.job = entry.job; + } + if ("raw" in entry) { + quarantined.raw = entry.raw; + } + if (isRecord(entry.state)) { + quarantined.state = entry.state; + } + if (typeof entry.updatedAtMs === "number" && Number.isFinite(entry.updatedAtMs)) { + quarantined.updatedAtMs = entry.updatedAtMs; + } + if (typeof entry.scheduleIdentity === "string") { + quarantined.scheduleIdentity = entry.scheduleIdentity; + } + return quarantined; + }); + return { version: 1, jobs }; + } catch (err) { + if ((err as { code?: unknown })?.code === "ENOENT") { + return { version: 1, jobs: [] }; + } + throw err; + } +} + +function quarantineEntryKey(entry: QuarantinedCronConfigJob): string { + const rawId = entry.job + ? (normalizeOptionalString(entry.job.id) ?? normalizeOptionalString(entry.job.jobId)) + : null; + return JSON.stringify({ + id: rawId ?? null, + sourceIndex: entry.sourceIndex, + reason: entry.reason, + job: entry.job ?? null, + raw: entry.raw ?? null, + state: entry.state ?? null, + updatedAtMs: entry.updatedAtMs ?? null, + scheduleIdentity: entry.scheduleIdentity ?? null, + }); +} + +export async function saveCronQuarantineFile(params: { + storePath: string; + entries: QuarantinedCronConfigJob[]; + nowMs: number; +}) { + if (params.entries.length === 0) { + return null; + } + const quarantinePath = resolveCronQuarantinePath(params.storePath); + const existing = await loadCronQuarantineFile(quarantinePath); + const seen = new Set(existing.jobs.map(quarantineEntryKey)); + const nextJobs = existing.jobs.slice(); + let appended = false; + for (const entry of params.entries.toSorted((a, b) => a.sourceIndex - b.sourceIndex)) { + const key = quarantineEntryKey(entry); + if (seen.has(key)) { + continue; + } + seen.add(key); + appended = true; + nextJobs.push({ + quarantinedAtMs: params.nowMs, + sourceIndex: entry.sourceIndex, + reason: entry.reason, + ...(entry.job ? { job: structuredClone(entry.job) } : {}), + ...("raw" in entry ? { raw: structuredClone(entry.raw) } : {}), + ...(entry.state ? { state: structuredClone(entry.state) } : {}), + ...(entry.updatedAtMs !== undefined ? { updatedAtMs: entry.updatedAtMs } : {}), + ...(entry.scheduleIdentity !== undefined ? { scheduleIdentity: entry.scheduleIdentity } : {}), + }); + } + if (!appended) { + return quarantinePath; + } + const payload = JSON.stringify({ version: 1, jobs: nextJobs }, null, 2); + await atomicWrite(quarantinePath, payload); + return quarantinePath; +} diff --git a/ui/src/ui/controllers/cron-filters.test.ts b/ui/src/ui/controllers/cron-filters.test.ts index efef817868af..caa9ac9a9bb3 100644 --- a/ui/src/ui/controllers/cron-filters.test.ts +++ b/ui/src/ui/controllers/cron-filters.test.ts @@ -42,6 +42,19 @@ describe("getVisibleCronJobs", () => { expect(visible.map((entry) => entry.id)).toEqual(["c"]); }); + it("drops jobs with unsupported schedules before rendering", () => { + const jobs = [ + job("valid", { schedule: { kind: "cron", expr: "0 9 * * *" } }), + job("invalid", { schedule: {} as CronJob["schedule"] }), + ]; + const visible = getVisibleCronJobs({ + cronJobs: jobs, + cronJobsScheduleKindFilter: "all", + cronJobsLastStatusFilter: "all", + }); + expect(visible.map((entry) => entry.id)).toEqual(["valid"]); + }); + it("filters by last status", () => { const jobs = [ job("ok", { state: { lastStatus: "ok", lastRunAtMs: 1 } }), diff --git a/ui/src/ui/controllers/cron.ts b/ui/src/ui/controllers/cron.ts index f3501efef84e..feba735729fd 100644 --- a/ui/src/ui/controllers/cron.ts +++ b/ui/src/ui/controllers/cron.ts @@ -389,9 +389,13 @@ export function getVisibleCronJobs( state: Pick, ): CronJob[] { return state.cronJobs.filter((job) => { + const scheduleKind = resolveCronJobScheduleKind(job); + if (!scheduleKind) { + return false; + } if ( state.cronJobsScheduleKindFilter !== "all" && - job.schedule.kind !== state.cronJobsScheduleKindFilter + scheduleKind !== state.cronJobsScheduleKindFilter ) { return false; } @@ -405,6 +409,14 @@ export function getVisibleCronJobs( }); } +function resolveCronJobScheduleKind(job: CronJob): CronJob["schedule"]["kind"] | null { + const scheduleKind = (job.schedule as { kind?: unknown } | null | undefined)?.kind; + if (scheduleKind === "at" || scheduleKind === "every" || scheduleKind === "cron") { + return scheduleKind; + } + return null; +} + function clearCronEditState(state: CronState) { state.cronEditingJobId = null; }