From efb02a06712a8698f2530cde501a7b2c43aae468 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 9 Aug 2026 09:05:30 -0700 Subject: [PATCH] fix(cron): prevent jobs disappearing during paginated refreshes (#121084) * fix(cron): keep paged consumers on one snapshot * test(cron): type malformed UI fixture * test(cron): update canonical list fixtures --- scripts/control-ui-mock-cron.ts | 3 + src/agents/tools/cron-tool-self-list.ts | 87 +++++++++ src/agents/tools/cron-tool.test.ts | 85 +++++++++ src/agents/tools/cron-tool.ts | 111 +++-------- src/cron/service/list-page-validation.ts | 53 ++++++ .../scheduled-turns.contract.test.ts | 111 ++++++++++- src/plugins/host-hook-scheduled-turns.ts | 68 +++++-- ui/src/api/types.ts | 11 +- ui/src/components/sidebar-attention.test.ts | 20 +- ui/src/e2e/cron-descriptions.e2e.test.ts | 1 + ui/src/e2e/cron-filters.e2e.test.ts | 1 + ui/src/lib/cron/index.test.ts | 172 +++++++++++++++--- ui/src/lib/cron/index.ts | 98 ++++++++-- ui/src/pages/agents/agents-page.test.ts | 49 +++-- ui/src/pages/agents/view.test.ts | 9 +- ui/src/pages/cron/cron-page.test.ts | 35 ++-- 16 files changed, 722 insertions(+), 192 deletions(-) create mode 100644 src/agents/tools/cron-tool-self-list.ts create mode 100644 src/cron/service/list-page-validation.ts diff --git a/scripts/control-ui-mock-cron.ts b/scripts/control-ui-mock-cron.ts index c1dd07dc926d..77edce9864e9 100644 --- a/scripts/control-ui-mock-cron.ts +++ b/scripts/control-ui-mock-cron.ts @@ -6,6 +6,8 @@ import type { CronStatus, } from "../ui/src/api/types.ts"; +const CRON_LIST_SNAPSHOT_REVISION = "control-ui-mock-cron"; + function listResult( jobs: CronJob[], options: { total?: number; limit?: number; offset?: number } = {}, @@ -17,6 +19,7 @@ function listResult( const hasMore = nextOffset < total; return { jobs, + snapshotRevision: CRON_LIST_SNAPSHOT_REVISION, total, offset, limit, diff --git a/src/agents/tools/cron-tool-self-list.ts b/src/agents/tools/cron-tool-self-list.ts new file mode 100644 index 000000000000..54c09ca68bc7 --- /dev/null +++ b/src/agents/tools/cron-tool-self-list.ts @@ -0,0 +1,87 @@ +import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import { + readCanonicalCronListPage, + resolveCronListPageNextOffset, +} from "../../cron/service/list-page-validation.js"; + +const CRON_SELF_LIST_MAX_PAGES = 50; +const CRON_SELF_LIST_MAX_SNAPSHOT_RESTARTS = 3; + +function filterDeliveryPreviewsByJobId(previews: unknown, jobId: string): unknown { + if (!isRecord(previews)) { + return previews; + } + return Object.hasOwn(previews, jobId) ? { [jobId]: previews[jobId] } : {}; +} + +function filterCronListResultToJobId(result: unknown, jobId: string) { + if (!isRecord(result) || !Array.isArray(result.jobs)) { + throw new Error("cron.list returned an invalid inventory page"); + } + const jobs = result.jobs.filter((job) => isRecord(job) && job.id === jobId); + const filteredResult: Record = { + ...result, + jobs, + total: jobs.length, + offset: 0, + limit: jobs.length, + hasMore: false, + nextOffset: null, + ...(Object.hasOwn(result, "deliveryPreviews") + ? { deliveryPreviews: filterDeliveryPreviewsByJobId(result.deliveryPreviews, jobId) } + : {}), + }; + delete filteredResult.snapshotRevision; + return filteredResult; +} + +function cronListPageHasJob(result: { jobs: unknown[] }, jobId: string): boolean { + return result.jobs.some((job) => isRecord(job) && job.id === jobId); +} + +export async function listCronSelfJob(params: { + jobId: string; + pageSize: number; + requestPage: (params: { limit: number; offset: number }) => Promise; +}): Promise { + for (let restart = 0; restart <= CRON_SELF_LIST_MAX_SNAPSHOT_RESTARTS; restart += 1) { + let offset = 0; + let snapshotRevision: string | undefined; + let total: number | undefined; + let snapshotChanged = false; + + for (let pageNumber = 0; pageNumber < CRON_SELF_LIST_MAX_PAGES; pageNumber += 1) { + const page = readCanonicalCronListPage( + await params.requestPage({ limit: params.pageSize, offset }), + params.pageSize, + ); + if ( + (snapshotRevision !== undefined && page.snapshotRevision !== snapshotRevision) || + (total !== undefined && page.total !== total) + ) { + // The current job can move into an already-read offset page. Discard + // the attempt instead of fabricating an empty self view. + snapshotChanged = true; + break; + } + snapshotRevision ??= page.snapshotRevision; + total ??= page.total; + const nextOffset = resolveCronListPageNextOffset(page, offset); + if (cronListPageHasJob(page, params.jobId) || nextOffset === null) { + return filterCronListResultToJobId(page, params.jobId); + } + offset = nextOffset; + } + + if (!snapshotChanged) { + throw new Error( + "cron.list pagination exceeded maximum pages while reading current automation", + ); + } + if (restart === CRON_SELF_LIST_MAX_SNAPSHOT_RESTARTS) { + throw new Error("cron.list inventory changed repeatedly while reading current automation"); + } + } + + throw new Error("cron.list inventory changed repeatedly while reading current automation"); +} diff --git a/src/agents/tools/cron-tool.test.ts b/src/agents/tools/cron-tool.test.ts index 8c8ee20ba300..3dcae3e52225 100644 --- a/src/agents/tools/cron-tool.test.ts +++ b/src/agents/tools/cron-tool.test.ts @@ -363,6 +363,7 @@ describe("cron tool", () => { { id: "job-current", name: "current" }, { id: "job-other", name: "other" }, ], + snapshotRevision: "self-list-one-page", total: 2, offset: 0, limit: 2, @@ -411,6 +412,7 @@ describe("cron tool", () => { id: `job-old-${index}`, name: `old ${index}`, })), + snapshotRevision: "self-list-paged", total: 201, offset: 0, limit: 200, @@ -420,6 +422,7 @@ describe("cron tool", () => { }) .mockResolvedValueOnce({ jobs: [{ id: "job-current", name: "current" }], + snapshotRevision: "self-list-paged", total: 201, offset: 200, limit: 200, @@ -473,9 +476,91 @@ describe("cron tool", () => { }); }); + it("restarts the scoped list when the current job moves behind the page boundary", async () => { + const stableJobs = Array.from({ length: 199 }, (_, index) => ({ + id: `stable-${index}`, + name: `stable ${index}`, + })); + callGatewayMock + .mockResolvedValueOnce({ + jobs: [{ id: "stale-only", name: "stale" }, ...stableJobs], + snapshotRevision: "revision-a", + total: 201, + offset: 0, + limit: 200, + hasMore: true, + nextOffset: 200, + }) + .mockResolvedValueOnce({ + jobs: [], + snapshotRevision: "revision-b", + total: 200, + offset: 200, + limit: 200, + hasMore: false, + nextOffset: null, + }) + .mockResolvedValueOnce({ + jobs: [...stableJobs, { id: "job-current", name: "current" }], + snapshotRevision: "revision-b", + total: 200, + offset: 0, + limit: 200, + hasMore: false, + nextOffset: null, + }); + const tool = createTestCronTool({ selfRemoveOnlyJobId: "job-current" }); + + const result = await tool.execute("call-list-boundary-churn", { action: "list" }); + + expect(callGatewayMock.mock.calls.map((call) => call[0].params.offset)).toEqual([0, 200, 0]); + expect(result.details).toEqual({ + jobs: [{ id: "job-current", name: "current" }], + total: 1, + offset: 0, + limit: 1, + hasMore: false, + nextOffset: null, + }); + }); + + it("rejects a scoped list after repeated snapshot churn", async () => { + callGatewayMock.mockImplementation(async ({ params }: { params: Record }) => { + const callNumber = callGatewayMock.mock.calls.length; + const offset = params.offset as number; + if (offset === 0) { + return { + jobs: Array.from({ length: 200 }, (_, index) => ({ id: `job-${callNumber}-${index}` })), + snapshotRevision: `revision-${callNumber}-a`, + total: 201, + offset: 0, + limit: 200, + hasMore: true, + nextOffset: 200, + }; + } + return { + jobs: [], + snapshotRevision: `revision-${callNumber}-b`, + total: 200, + offset: 200, + limit: 200, + hasMore: false, + nextOffset: null, + }; + }); + const tool = createTestCronTool({ selfRemoveOnlyJobId: "job-current" }); + + await expect(tool.execute("call-list-churn", { action: "list" })).rejects.toThrow( + "cron.list inventory changed repeatedly while reading current automation", + ); + expect(callGatewayMock).toHaveBeenCalledTimes(8); + }); + it("does not let requested pagination bypass the scoped current-job scan", async () => { callGatewayMock.mockResolvedValueOnce({ jobs: [{ id: "job-current", name: "current" }], + snapshotRevision: "self-list-requested-pagination", total: 1, offset: 0, limit: 200, diff --git a/src/agents/tools/cron-tool.ts b/src/agents/tools/cron-tool.ts index 0a02be15bb3e..3020c4d41a36 100644 --- a/src/agents/tools/cron-tool.ts +++ b/src/agents/tools/cron-tool.ts @@ -53,6 +53,7 @@ import { createCronToolSchema, CRON_TOOL_LIST_MAX_LIMIT, } from "./cron-tool-schema.js"; +import { listCronSelfJob } from "./cron-tool-self-list.js"; import { assertCronCreatorAuthorityResolutionAvailable, assertNoCronShellExecution, @@ -111,35 +112,6 @@ function assertCronSelfRemoveScope( throw new Error(CRON_SELF_REMOVE_SCOPE_ERROR); } -function filterCronDeliveryPreviewsByJobId(previews: unknown, jobId: string): unknown { - if (!isRecord(previews)) { - return previews; - } - if (!Object.hasOwn(previews, jobId)) { - return {}; - } - return { [jobId]: previews[jobId] }; -} - -function filterCronListResultToJobId(result: unknown, jobId: string): unknown { - if (!isRecord(result) || !Array.isArray(result.jobs)) { - return result; - } - const jobs = result.jobs.filter((job) => isRecord(job) && job.id === jobId); - return { - ...result, - jobs, - total: jobs.length, - offset: 0, - limit: jobs.length, - hasMore: false, - nextOffset: null, - ...(Object.hasOwn(result, "deliveryPreviews") - ? { deliveryPreviews: filterCronDeliveryPreviewsByJobId(result.deliveryPreviews, jobId) } - : {}), - }; -} - function filterCronStatusResultForSelfScope(result: unknown): unknown { return { enabled: isRecord(result) && result.enabled === true }; } @@ -184,22 +156,6 @@ function formatCronTerminalPresentation( } } -function cronListResultHasJob(result: unknown, jobId: string): boolean { - return ( - isRecord(result) && - Array.isArray(result.jobs) && - result.jobs.some((job) => isRecord(job) && job.id === jobId) - ); -} - -function readCronListNextOffset(result: unknown, currentOffset: number): number | undefined { - if (!isRecord(result) || result.hasMore !== true || typeof result.nextOffset !== "number") { - return undefined; - } - const nextOffset = Math.floor(result.nextOffset); - return Number.isFinite(nextOffset) && nextOffset > currentOffset ? nextOffset : undefined; -} - function isOlderGatewayWithoutCompactCronList(error: unknown): boolean { return ( error instanceof GatewayClientRequestError && @@ -299,47 +255,40 @@ Job wakeMode (main jobs): "now"(default)|"next-heartbeat". Restricted automation const requestedOffset = selfRemoveOnlyJobId ? undefined : readNonNegativeIntegerParam(params, "offset"); - let offset = requestedOffset ?? 0; - let result: unknown; - let shouldContinue = true; let useCompactList = true; - while (shouldContinue) { - try { - result = await callGateway("cron.list", gatewayOpts, { - includeDisabled, - ...(useCompactList ? { compact: true } : {}), - ...(listAgentId ? { agentId: listAgentId } : {}), - ...(selfRemoveOnlyJobId - ? { limit: CRON_TOOL_LIST_MAX_LIMIT, offset } - : { - ...(requestedLimit !== undefined ? { limit: requestedLimit } : {}), - ...(requestedOffset !== undefined ? { offset: requestedOffset } : {}), - }), - }); - } catch (error) { - if (!useCompactList || !isOlderGatewayWithoutCompactCronList(error)) { - throw error; - } - // Protocol v4 gateways predating compact reject the additive field. - // Retry without it for mixed-version correctness; remove at the next protocol break. - useCompactList = false; - continue; - } - if (!selfRemoveOnlyJobId || cronListResultHasJob(result, selfRemoveOnlyJobId)) { - shouldContinue = false; - } else { - const nextOffset = readCronListNextOffset(result, offset); - if (nextOffset === undefined) { - shouldContinue = false; - } else { - offset = nextOffset; + const requestListPage = async (pageParams: Record) => { + for (;;) { + try { + return await callGateway("cron.list", gatewayOpts, { + includeDisabled, + ...(useCompactList ? { compact: true } : {}), + ...(listAgentId ? { agentId: listAgentId } : {}), + ...pageParams, + }); + } catch (error) { + if (!useCompactList || !isOlderGatewayWithoutCompactCronList(error)) { + throw error; + } + // Protocol v4 gateways predating compact reject the additive field. + // Retry without it for mixed-version correctness; remove at the next protocol break. + useCompactList = false; } } + }; + if (!selfRemoveOnlyJobId) { + const result = await requestListPage({ + ...(requestedLimit !== undefined ? { limit: requestedLimit } : {}), + ...(requestedOffset !== undefined ? { offset: requestedOffset } : {}), + }); + return jsonResult(result); } + return jsonResult( - selfRemoveOnlyJobId - ? filterCronListResultToJobId(result, selfRemoveOnlyJobId) - : result, + await listCronSelfJob({ + jobId: selfRemoveOnlyJobId, + pageSize: CRON_TOOL_LIST_MAX_LIMIT, + requestPage: requestListPage, + }), ); } case "get": { diff --git a/src/cron/service/list-page-validation.ts b/src/cron/service/list-page-validation.ts new file mode 100644 index 000000000000..76187ed4c550 --- /dev/null +++ b/src/cron/service/list-page-validation.ts @@ -0,0 +1,53 @@ +import { isRecord } from "@openclaw/normalization-core/record-coerce"; +import type { CronListPageResult } from "./list-page-types.js"; + +type CanonicalCronListPage = Omit & { jobs: TJob[] }; + +function isSafeNonNegativeInteger(value: unknown): value is number { + return typeof value === "number" && Number.isSafeInteger(value) && value >= 0; +} + +export function readCanonicalCronListPage( + value: unknown, + maxLimit: number, +): CanonicalCronListPage { + if (!isRecord(value) || !Array.isArray(value.jobs)) { + throw new Error("cron.list returned an invalid inventory page"); + } + const page = value; + const jobs = value.jobs as TJob[]; + const limit = typeof page.limit === "number" ? page.limit : 0; + if ( + typeof page.snapshotRevision !== "string" || + page.snapshotRevision.length === 0 || + !isSafeNonNegativeInteger(page.total) || + !isSafeNonNegativeInteger(page.offset) || + !Number.isSafeInteger(limit) || + limit < 1 || + limit > maxLimit || + jobs.length > limit || + typeof page.hasMore !== "boolean" || + (page.nextOffset !== null && !isSafeNonNegativeInteger(page.nextOffset)) + ) { + throw new Error("cron.list returned an invalid inventory page"); + } + return page as CanonicalCronListPage; +} + +export function resolveCronListPageNextOffset( + page: CanonicalCronListPage, + requestedOffset: number, +): number | null { + const nextOffset = requestedOffset + page.jobs.length; + if ( + page.offset !== requestedOffset || + !Number.isSafeInteger(nextOffset) || + nextOffset > page.total || + (page.hasMore + ? page.nextOffset !== nextOffset || nextOffset <= requestedOffset || nextOffset >= page.total + : page.nextOffset !== null || nextOffset !== page.total) + ) { + throw new Error("cron.list returned an invalid inventory page"); + } + return page.hasMore ? nextOffset : null; +} diff --git a/src/plugins/contracts/scheduled-turns.contract.test.ts b/src/plugins/contracts/scheduled-turns.contract.test.ts index 8b19834a60a7..6b55685c47bd 100644 --- a/src/plugins/contracts/scheduled-turns.contract.test.ts +++ b/src/plugins/contracts/scheduled-turns.contract.test.ts @@ -279,17 +279,17 @@ describe("plugin scheduled turns", () => { workflowMocks.cronListPage.mockImplementation(async (body: unknown) => { const offset = (body as { offset?: unknown }).offset; listRequests.push(body); - if (offset === undefined) { + if (offset === 0) { return { - jobs: [ + jobs: Array.from({ length: 200 }, (_, index) => makeCronJob({ - id: "job-page-1", - name: "plugin:workflow-plugin:tag:nudge:agent:main:main:1", + id: `job-page-1-${index}`, + name: `plugin:workflow-plugin:tag:nudge:agent:main:main:${String(index).padStart(3, "0")}`, sessionTarget: "session:agent:main:main", }), - ], + ), snapshotRevision: "fixture", - total: 2, + total: 201, offset: 0, limit: 200, hasMore: true, @@ -305,7 +305,7 @@ describe("plugin scheduled turns", () => { }), ], snapshotRevision: "fixture", - total: 2, + total: 201, offset: 200, limit: 200, hasMore: false, @@ -317,11 +317,12 @@ describe("plugin scheduled turns", () => { return { ok: true, removed: true }; }); - await expect(unscheduleWorkflowTurnsByTag()).resolves.toEqual({ removed: 2, failed: 0 }); + await expect(unscheduleWorkflowTurnsByTag()).resolves.toEqual({ removed: 201, failed: 0 }); expect(listRequests).toEqual([ { includeDisabled: true, limit: 200, + offset: 0, query: "plugin:workflow-plugin:tag:nudge:agent:main:main:", sortBy: "name", sortDir: "asc", @@ -335,7 +336,99 @@ describe("plugin scheduled turns", () => { sortDir: "asc", }, ]); - expect(removed.toSorted()).toEqual(["job-page-1", "job-page-2"]); + expect(new Set(removed).size).toBe(201); + expect(removed).toContain("job-page-2"); + }); + + it("restarts tagged cleanup when a job moves behind the page boundary", async () => { + const prefix = "plugin:workflow-plugin:tag:nudge:agent:main:main:"; + const stableJobs = Array.from({ length: 199 }, (_, index) => + makeCronJob({ + id: `stable-${index}`, + name: `${prefix}${String(index + 1).padStart(3, "0")}`, + }), + ); + const staleJob = makeCronJob({ id: "stale-only", name: `${prefix}000` }); + const currentJob = makeCronJob({ id: "target-current", name: `${prefix}999` }); + const offsets: number[] = []; + workflowMocks.cronListPage.mockImplementation(async (body: unknown) => { + const offset = (body as { offset: number }).offset; + offsets.push(offset); + if (offset === 0 && offsets.length === 1) { + return { + jobs: [staleJob, ...stableJobs], + snapshotRevision: "revision-a", + total: 201, + offset: 0, + limit: 200, + hasMore: true, + nextOffset: 200, + }; + } + if (offset === 200) { + return { + jobs: [], + snapshotRevision: "revision-b", + total: 200, + offset: 200, + limit: 200, + hasMore: false, + nextOffset: null, + }; + } + return { + jobs: [...stableJobs, currentJob], + snapshotRevision: "revision-b", + total: 200, + offset: 0, + limit: 200, + hasMore: false, + nextOffset: null, + }; + }); + const removed: string[] = []; + workflowMocks.cronRemove.mockImplementation(async (id: string) => { + removed.push(id); + return { ok: true, removed: true }; + }); + + await expect(unscheduleWorkflowTurnsByTag()).resolves.toEqual({ removed: 200, failed: 0 }); + expect(offsets).toEqual([0, 200, 0]); + expect(new Set(removed)).toEqual(new Set([...stableJobs.map((job) => job.id), currentJob.id])); + expect(removed).not.toContain(staleJob.id); + }); + + it("fails tagged cleanup without removals after repeated snapshot churn", async () => { + workflowMocks.cronListPage.mockImplementation(async (body: unknown) => { + const offset = (body as { offset: number }).offset; + const attempt = Math.floor(workflowMocks.cronListPage.mock.calls.length / 2); + if (offset === 0) { + return { + jobs: Array.from({ length: 200 }, (_, index) => + makeCronJob({ id: `attempt-${attempt}-${index}` }), + ), + snapshotRevision: `revision-${attempt}-a`, + total: 201, + offset: 0, + limit: 200, + hasMore: true, + nextOffset: 200, + }; + } + return { + jobs: [], + snapshotRevision: `revision-${attempt}-b`, + total: 200, + offset: 200, + limit: 200, + hasMore: false, + nextOffset: null, + }; + }); + + await expect(unscheduleWorkflowTurnsByTag()).resolves.toEqual({ removed: 0, failed: 1 }); + expect(workflowMocks.cronListPage).toHaveBeenCalledTimes(8); + expect(workflowMocks.cronRemove).not.toHaveBeenCalled(); }); it("tracks scheduled session turns using cron.add's top-level job id", async () => { diff --git a/src/plugins/host-hook-scheduled-turns.ts b/src/plugins/host-hook-scheduled-turns.ts index 2eea53b764ef..b6c1395c3fd4 100644 --- a/src/plugins/host-hook-scheduled-turns.ts +++ b/src/plugins/host-hook-scheduled-turns.ts @@ -6,6 +6,10 @@ import { } from "@openclaw/normalization-core/number-coercion"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import type { CronServiceContract } from "../cron/service-contract.js"; +import { + readCanonicalCronListPage, + resolveCronListPageNextOffset, +} from "../cron/service/list-page-validation.js"; import type { CronJob, CronJobCreate } from "../cron/types.js"; import { formatErrorMessage } from "../infra/errors.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; @@ -25,6 +29,9 @@ import type { PluginRegistry } from "./registry-types.js"; const log = createSubsystemLogger("plugins/host-scheduled-turns"); const PLUGIN_CRON_NAME_PREFIX = "plugin:"; const PLUGIN_CRON_TAG_MARKER = ":tag:"; +const PLUGIN_CRON_CLEANUP_PAGE_SIZE = 200; +const PLUGIN_CRON_CLEANUP_MAX_PAGES = 50; +const PLUGIN_CRON_CLEANUP_MAX_SNAPSHOT_RESTARTS = 3; type ResolvedSessionTurnSchedule = | { @@ -176,26 +183,53 @@ async function listAllCronJobsForPluginTagCleanup( cron: CronServiceContract, query: string, ): Promise { - const jobs: CronJob[] = []; - let offset = 0; - for (;;) { - const listResult = await cron.listPage({ - includeDisabled: true, - limit: 200, - query, - sortBy: "name", - sortDir: "asc", - ...(offset > 0 ? { offset } : {}), - }); - jobs.push(...listResult.jobs); - if (!listResult.hasMore) { - return jobs; + for (let restart = 0; restart <= PLUGIN_CRON_CLEANUP_MAX_SNAPSHOT_RESTARTS; restart += 1) { + const jobs: CronJob[] = []; + let offset = 0; + let snapshotRevision: string | undefined; + let total: number | undefined; + let snapshotChanged = false; + + for (let pageNumber = 0; pageNumber < PLUGIN_CRON_CLEANUP_MAX_PAGES; pageNumber += 1) { + const page = readCanonicalCronListPage( + await cron.listPage({ + includeDisabled: true, + limit: PLUGIN_CRON_CLEANUP_PAGE_SIZE, + offset, + query, + sortBy: "name", + sortDir: "asc", + }), + PLUGIN_CRON_CLEANUP_PAGE_SIZE, + ); + if ( + (snapshotRevision !== undefined && page.snapshotRevision !== snapshotRevision) || + (total !== undefined && page.total !== total) + ) { + // Offset pages are independent snapshots. Never carry cleanup targets + // across a revision change because the boundary rows may have moved. + snapshotChanged = true; + break; + } + snapshotRevision ??= page.snapshotRevision; + total ??= page.total; + const nextOffset = resolveCronListPageNextOffset(page, offset); + jobs.push(...page.jobs); + if (nextOffset === null) { + return jobs; + } + offset = nextOffset; } - if (listResult.nextOffset === null || listResult.nextOffset <= offset) { - return jobs; + + if (!snapshotChanged) { + throw new Error("cron.list pagination exceeded maximum pages"); + } + if (restart === PLUGIN_CRON_CLEANUP_MAX_SNAPSHOT_RESTARTS) { + throw new Error("cron.list inventory changed repeatedly during cleanup"); } - offset = listResult.nextOffset; } + + throw new Error("cron.list inventory changed repeatedly during cleanup"); } export async function schedulePluginSessionTurn(params: { diff --git a/ui/src/api/types.ts b/ui/src/api/types.ts index c14eb7aaa707..cb6f460e986f 100644 --- a/ui/src/api/types.ts +++ b/ui/src/api/types.ts @@ -696,11 +696,12 @@ export type CronRunLogEntry = { export type CronJobsListResult = { jobs: CronJob[]; - total?: number; - limit?: number; - offset?: number; - nextOffset?: number | null; - hasMore?: boolean; + snapshotRevision: string; + total: number; + limit: number; + offset: number; + nextOffset: number | null; + hasMore: boolean; }; export type CronRunsResult = { diff --git a/ui/src/components/sidebar-attention.test.ts b/ui/src/components/sidebar-attention.test.ts index 948893ca856a..c8e151f7e9da 100644 --- a/ui/src/components/sidebar-attention.test.ts +++ b/ui/src/components/sidebar-attention.test.ts @@ -2,7 +2,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import type { GatewayBrowserClient } from "../api/gateway.ts"; -import type { CronJob, ModelAuthStatusResult } from "../api/types.ts"; +import type { CronJob, CronJobsListResult, ModelAuthStatusResult } from "../api/types.ts"; import type { ApplicationContext, ApplicationGateway } from "../app/context.ts"; import type { ExecApprovalRequest } from "../app/exec-approval.ts"; import { createApplicationContextProvider } from "../test-helpers/application-context.ts"; @@ -40,6 +40,18 @@ function cronJob(id: string): CronJob { }; } +function cronListResponse(jobs: CronJob[]): CronJobsListResult { + return { + jobs, + snapshotRevision: "sidebar-attention-cron-fixture", + total: jobs.length, + offset: 0, + limit: 50, + hasMore: false, + nextOffset: null, + }; +} + type SidebarAttentionElement = HTMLElement & { updateComplete: Promise; cronJobs: CronJob[]; @@ -220,7 +232,7 @@ describe("sidebar attention refresh ownership", () => { const currentAuth = { ts: 2, providers: [] } as ModelAuthStatusResult; now = 200_000; - secondCron.resolve({ jobs: [cronJob("current")] }); + secondCron.resolve(cronListResponse([cronJob("current")])); secondAuth.resolve(currentAuth); await waitForFast(() => expect(element.loadedAtMs).toBe(200_000)); expect(element.cronJobs.map((job) => job.id)).toEqual(["current"]); @@ -228,7 +240,7 @@ describe("sidebar attention refresh ownership", () => { expect(localStorage.getItem(dismissalStoreKey(gateway.connection.gatewayUrl))).not.toBeNull(); now = 300_000; - firstCron.resolve({ jobs: [cronJob("stale")] }); + firstCron.resolve(cronListResponse([cronJob("stale")])); firstAuth.resolve({ ts: 1, providers: [] }); await Promise.all([firstCron.promise, firstAuth.promise]); await new Promise((resolve) => { @@ -244,7 +256,7 @@ describe("sidebar attention refresh ownership", () => { it("clears a stale failure alert when the gateway reports an automation change", async () => { const responses = { - "cron.list": [{ jobs: [cronJob("failed")] }, { jobs: [] }], + "cron.list": [cronListResponse([cronJob("failed")]), cronListResponse([])], "models.authStatus": [{ ts: 1, providers: [] }], }; const request = vi.fn((method: keyof typeof responses) => { diff --git a/ui/src/e2e/cron-descriptions.e2e.test.ts b/ui/src/e2e/cron-descriptions.e2e.test.ts index fdcd116dccb9..912340ea1ced 100644 --- a/ui/src/e2e/cron-descriptions.e2e.test.ts +++ b/ui/src/e2e/cron-descriptions.e2e.test.ts @@ -64,6 +64,7 @@ suite.define(() => { methodResponses: { "cron.list": { jobs: [...jobs, undescribedJob], + snapshotRevision: "cron-descriptions-fixture", total: jobs.length + 1, offset: 0, limit: 50, diff --git a/ui/src/e2e/cron-filters.e2e.test.ts b/ui/src/e2e/cron-filters.e2e.test.ts index ef573649c85a..0cc1587c0561 100644 --- a/ui/src/e2e/cron-filters.e2e.test.ts +++ b/ui/src/e2e/cron-filters.e2e.test.ts @@ -33,6 +33,7 @@ function cronJob(id: string, name: string, schedule: Record, st function cronListResponse(jobs: unknown[], total = jobs.length) { return { jobs, + snapshotRevision: "cron-filters-fixture", total, offset: 0, limit: 50, diff --git a/ui/src/lib/cron/index.test.ts b/ui/src/lib/cron/index.test.ts index 59a3b42ed751..fa89d055376f 100644 --- a/ui/src/lib/cron/index.test.ts +++ b/ui/src/lib/cron/index.test.ts @@ -7,7 +7,7 @@ import { validateCronUpdateParams, } from "../../../../packages/gateway-protocol/src/index.js"; import { createDeferred } from "../../../../test/helpers/promise.js"; -import type { CronJob, CronRunsResult } from "../../api/types.ts"; +import type { CronJob, CronJobsListResult, CronRunsResult } from "../../api/types.ts"; import { parseCronEveryMs } from "../../lib/cron/decimal.ts"; import { addCronJob, @@ -46,7 +46,7 @@ function createCronRequest(jobId: string, options: { existing?: boolean } = {}) return { id: jobId }; } if (method === "cron.list") { - return { jobs }; + return cronJobsListResponse(jobs as CronJob[]); } if (method === "cron.status") { return { enabled: true, jobs: jobs.length, nextWakeAtMs: null }; @@ -157,15 +157,26 @@ function requestPatch(call: readonly [method: string, payload?: unknown]) { return requireRecord(requestPayload(call).patch, `${call[0]} patch`); } -type EmptyCronListResponse = { - jobs: []; - total: number; - hasMore: boolean; - nextOffset: null; -}; +function cronJobsListResponse( + jobs: CronJob[], + overrides: Partial> = {}, +): CronJobsListResult { + return { + jobs, + snapshotRevision: "cron-jobs-fixture", + total: jobs.length, + offset: 0, + limit: 50, + hasMore: false, + nextOffset: null, + ...overrides, + }; +} -function emptyCronListResponse(): EmptyCronListResponse { - return { jobs: [], total: 0, hasMore: false, nextOffset: null }; +function emptyCronListResponse( + overrides: Partial> = {}, +): CronJobsListResult { + return cronJobsListResponse([], overrides); } function createCronRunsResult( @@ -194,7 +205,7 @@ function createCronRunsRace( } function createCronJobsReloadHarness(stateOverrides: Partial = {}) { - const first = createDeferred(); + const first = createDeferred(); const payloads: unknown[] = []; const request = vi.fn(async (method: string, payload?: unknown) => { if (method !== "cron.list") { @@ -373,7 +384,7 @@ describe("cron controller", () => { for (const response of responses) { const request = createMethodRequest({ "cron.add": response, - "cron.list": { jobs: [] }, + "cron.list": emptyCronListResponse(), "cron.status": { enabled: true, jobs: 0, nextWakeAtMs: null }, }); const state = createStateWithRequest(request, { @@ -744,7 +755,7 @@ describe("cron controller", () => { delivery: { mode: "none" }, }); const request = createMethodRequest({ - "cron.list": { jobs: [scriptJob], total: 1, hasMore: false, nextOffset: null }, + "cron.list": cronJobsListResponse([scriptJob]), "cron.update": { id: scriptJob.id }, "cron.status": { enabled: true, jobs: 1, nextWakeAtMs: null }, }); @@ -1463,8 +1474,8 @@ describe("cron controller", () => { sortBy: "updatedAtMs", sortDir: "desc", }); - return { - jobs: [ + return cronJobsListResponse( + [ { id: "job-1", name: "Daily", @@ -1477,10 +1488,8 @@ describe("cron controller", () => { payload: { kind: "systemEvent", text: "ping" }, }, ], - total: 1, - hasMore: false, - nextOffset: null, - }; + { snapshotRevision: "daily-jobs" }, + ); } return {}; }); @@ -1500,6 +1509,106 @@ describe("cron controller", () => { expect(state.cronJobsHasMore).toBe(false); }); + it("appends jobs only from the accepted snapshot revision", async () => { + const firstJob = createCronJob({ id: "job-1", name: "First" }); + const secondJob = createCronJob({ id: "job-2", name: "Second" }); + const request = vi.fn(async () => + cronJobsListResponse([secondJob], { + snapshotRevision: "stable-revision", + total: 2, + offset: 1, + limit: 1, + }), + ); + const state = createStateWithRequest(request, { + cronJobs: [firstJob], + cronJobsSnapshotRevision: "stable-revision", + cronJobsTotal: 2, + cronJobsHasMore: true, + cronJobsNextOffset: 1, + cronJobsLimit: 1, + }); + + await loadCronJobsPage(state, { append: true }); + + expect(state.cronJobs.map((job) => job.id)).toEqual(["job-1", "job-2"]); + expect(state.cronJobsSnapshotRevision).toBe("stable-revision"); + expect(state.cronJobsHasMore).toBe(false); + expect(state.cronJobsNextOffset).toBeNull(); + }); + + it("restarts at page zero instead of committing an append from a changed snapshot", async () => { + const staleJob = createCronJob({ id: "stale-only", name: "Stale" }); + const stableJob = createCronJob({ id: "stable", name: "Stable" }); + const currentJob = createCronJob({ id: "current", name: "Current" }); + const responses = [ + cronJobsListResponse([staleJob, stableJob], { + snapshotRevision: "revision-a", + total: 3, + limit: 2, + hasMore: true, + nextOffset: 2, + }), + emptyCronListResponse({ + snapshotRevision: "revision-b", + total: 2, + offset: 2, + limit: 2, + }), + cronJobsListResponse([stableJob, currentJob], { + snapshotRevision: "revision-b", + total: 2, + limit: 2, + }), + ]; + const offsets: number[] = []; + const request = vi.fn(async (_method: string, payload?: unknown) => { + offsets.push(requireRecord(payload, "cron.list payload").offset as number); + const response = responses.shift(); + if (!response) { + throw new Error("unexpected cron.list call"); + } + return response; + }); + const state = createStateWithRequest(request, { cronJobsLimit: 2 }); + + await loadCronJobsPage(state); + await loadCronJobsPage(state, { append: true }); + + expect(offsets).toEqual([0, 2, 0]); + expect(state.cronJobs.map((job) => job.id)).toEqual(["stable", "current"]); + expect(state.cronJobsSnapshotRevision).toBe("revision-b"); + expect(state.cronJobsTotal).toBe(2); + expect(state.cronJobsHasMore).toBe(false); + expect(state.cronJobsNextOffset).toBeNull(); + }); + + it("keeps the last coherent jobs page when snapshot metadata is invalid", async () => { + const existingJob = createCronJob({ id: "existing", name: "Existing" }); + const request = vi.fn(async () => ({ + jobs: [], + total: 0, + offset: 0, + limit: 50, + hasMore: false, + nextOffset: null, + })); + const state = createStateWithRequest(request, { + cronJobs: [existingJob], + cronJobsSnapshotRevision: "accepted-revision", + cronJobsTotal: 1, + cronJobsHasMore: false, + cronJobsNextOffset: null, + }); + + await loadCronJobsPage(state); + + expect(state.cronJobs).toEqual([existingJob]); + expect(state.cronJobsSnapshotRevision).toBe("accepted-revision"); + expect(state.cronJobsTotal).toBe(1); + expect(state.cronError).toContain("cron.list returned an invalid inventory page"); + }); + it("keeps table-only filters out of shared cron jobs loads", async () => { const request = vi.fn(async (method: string, payload?: unknown) => { if (method === "cron.list") { @@ -1558,6 +1667,8 @@ describe("cron controller", () => { payload: { kind: "systemEvent", text: "ping" }, }), ], + cronJobsSnapshotRevision: "revision-a", + cronJobsTotal: 2, cronJobsHasMore: true, cronJobsNextOffset: 1, }); @@ -1568,7 +1679,13 @@ describe("cron controller", () => { cronJobsLastStatusFilter: "unknown", }); await loadCronJobsPage(state, { tableFilters: true }); - first.resolve(emptyCronListResponse()); + first.resolve( + emptyCronListResponse({ + snapshotRevision: "revision-b", + total: 1, + offset: 1, + }), + ); await appendLoad; expectRecordFields(requireRecord(payloads[0], "append cron.list payload"), { @@ -1582,6 +1699,7 @@ describe("cron controller", () => { expect(request).toHaveBeenCalledTimes(2); expect(state.cronJobsReloadPending).toBe(false); expect(state.cronJobsReloadPendingTableFilters).toBe(false); + expect(state.cronJobsSnapshotRevision).toBe("cron-jobs-fixture"); }); it("uses the latest queued cron jobs table-filter mode", async () => { @@ -1607,8 +1725,8 @@ describe("cron controller", () => { it("drops malformed cron jobs before they enter UI state", async () => { const request = vi.fn(async (method: string) => { if (method === "cron.list") { - return { - jobs: [ + return cronJobsListResponse( + [ { id: "bad-missing-payload", name: "Broken", enabled: true }, { id: "job-ok", @@ -1621,11 +1739,9 @@ describe("cron controller", () => { wakeMode: "next-heartbeat", payload: { kind: "systemEvent", text: "ping" }, }, - ], - total: 2, - hasMore: false, - nextOffset: null, - }; + ] as unknown as CronJob[], + { snapshotRevision: "malformed-job-page" }, + ); } return {}; }); @@ -2043,7 +2159,7 @@ describe("loadCronFailingCount", () => { return { jobs: [], total: 1, offset: 0, limit: 1 }; } if (method === "cron.list") { - return { jobs: [], total: 0, offset: 0, hasMore: false }; + return emptyCronListResponse(); } if (method === "cron.status") { return { enabled: true, jobs: 0 }; diff --git a/ui/src/lib/cron/index.ts b/ui/src/lib/cron/index.ts index 97c78989dee7..35d4ad292659 100644 --- a/ui/src/lib/cron/index.ts +++ b/ui/src/lib/cron/index.ts @@ -182,6 +182,7 @@ export type CronState = { cronJobsReloadPending: boolean; cronJobsReloadPendingTableFilters: boolean; cronJobs: CronJob[]; + cronJobsSnapshotRevision: string | null; cronJobsTotal: number; cronJobsHasMore: boolean; cronJobsNextOffset: number | null; @@ -239,6 +240,7 @@ export function createInitialCronState( cronJobsReloadPending: false, cronJobsReloadPendingTableFilters: false, cronJobs: [], + cronJobsSnapshotRevision: null, cronJobsTotal: 0, cronJobsHasMore: false, cronJobsNextOffset: null, @@ -500,7 +502,7 @@ async function withCronBusy( } } -function normalizeCronPageMeta(params: { +function normalizeCronRunsPageMeta(params: { totalRaw: unknown; offsetRaw: unknown; nextOffsetRaw: unknown; @@ -528,6 +530,66 @@ function normalizeCronPageMeta(params: { return { total, hasMore, nextOffset }; } +type CanonicalCronJobsPage = { + jobs: CronJob[]; + snapshotRevision: string; + total: number; + offset: number; + limit: number; + hasMore: boolean; + nextOffset: number | null; +}; + +function readCanonicalCronJobsPage(value: unknown, requestedLimit: number): CanonicalCronJobsPage { + if ( + !isRecord(value) || + !Array.isArray(value.jobs) || + typeof value.snapshotRevision !== "string" || + value.snapshotRevision.length === 0 || + typeof value.total !== "number" || + !Number.isSafeInteger(value.total) || + value.total < 0 || + typeof value.offset !== "number" || + !Number.isSafeInteger(value.offset) || + value.offset < 0 || + typeof value.limit !== "number" || + !Number.isSafeInteger(value.limit) || + value.limit < 1 || + value.limit > requestedLimit || + value.jobs.length > value.limit || + typeof value.hasMore !== "boolean" || + (value.nextOffset !== null && + (typeof value.nextOffset !== "number" || + !Number.isSafeInteger(value.nextOffset) || + value.nextOffset < 0)) + ) { + throw new Error("cron.list returned an invalid inventory page"); + } + return value as CanonicalCronJobsPage; +} + +function assertCanonicalCronJobsCursor(page: CanonicalCronJobsPage, requestedOffset: number) { + const nextOffset = requestedOffset + page.jobs.length; + if ( + page.offset !== requestedOffset || + !Number.isSafeInteger(nextOffset) || + nextOffset > page.total || + (page.hasMore + ? page.nextOffset !== nextOffset || nextOffset <= requestedOffset || nextOffset >= page.total + : page.nextOffset !== null || nextOffset !== page.total) + ) { + throw new Error("cron.list returned an invalid inventory page"); + } +} + +function queueCronJobsSnapshotRecovery(state: CronState, tableFilters: boolean) { + if (state.cronJobsReloadPending) { + return; + } + state.cronJobsReloadPending = true; + state.cronJobsReloadPendingTableFilters = tableFilters; +} + async function drainPendingCronJobsReload(state: CronState) { if (!state.cronJobsReloadPending) { return; @@ -581,19 +643,25 @@ export async function loadCronJobsPage( sortBy: state.cronJobsSortBy, sortDir: state.cronJobsSortDir, }); - const rawJobs = Array.isArray(res.jobs) ? res.jobs : []; - const jobs = rawJobs.filter(hasCronJobPayload); - state.cronJobs = append ? [...state.cronJobs, ...jobs] : jobs; - const meta = normalizeCronPageMeta({ - totalRaw: res.total, - offsetRaw: res.offset, - nextOffsetRaw: res.nextOffset, - hasMoreRaw: res.hasMore, - pageCount: rawJobs.length, - }); - state.cronJobsTotal = Math.max(meta.total, state.cronJobs.length); - state.cronJobsHasMore = meta.hasMore; - state.cronJobsNextOffset = meta.nextOffset; + const page = readCanonicalCronJobsPage(res, state.cronJobsLimit); + if ( + append && + (page.snapshotRevision !== state.cronJobsSnapshotRevision || + page.total !== state.cronJobsTotal) + ) { + // A changed snapshot can move rows behind the append boundary. Preserve + // the coherent table and let one serialized page-zero reload recover it. + queueCronJobsSnapshotRecovery(state, opts?.tableFilters === true); + return; + } + assertCanonicalCronJobsCursor(page, offset); + const jobs = page.jobs.filter(hasCronJobPayload); + const nextJobs = append ? [...state.cronJobs, ...jobs] : jobs; + state.cronJobs = nextJobs; + state.cronJobsSnapshotRevision = page.snapshotRevision; + state.cronJobsTotal = page.total; + state.cronJobsHasMore = page.hasMore; + state.cronJobsNextOffset = page.nextOffset; if ( state.cronEditingJobId && !state.cronJobs.some((job) => job.id === state.cronEditingJobId) @@ -1308,7 +1376,7 @@ export async function loadCronRuns( } const entries = Array.isArray(res.entries) ? res.entries : []; state.cronRuns = append ? [...state.cronRuns, ...entries] : entries; - const meta = normalizeCronPageMeta({ + const meta = normalizeCronRunsPageMeta({ totalRaw: res.total, offsetRaw: res.offset, nextOffsetRaw: res.nextOffset, diff --git a/ui/src/pages/agents/agents-page.test.ts b/ui/src/pages/agents/agents-page.test.ts index a49c472545f0..15fdbed2e5ac 100644 --- a/ui/src/pages/agents/agents-page.test.ts +++ b/ui/src/pages/agents/agents-page.test.ts @@ -6,6 +6,7 @@ import type { AgentsFilesListResult, AgentsListResult, CronJob, + CronJobsListResult, ModelCatalogEntry, ToolsEffectiveResult, } from "../../api/types.ts"; @@ -122,6 +123,26 @@ function cronJob(id: string, agentId?: string): CronJob { } as CronJob; } +function cronListResponse( + jobs: CronJob[], + options: { total?: number; offset?: number; limit?: number } = {}, +): CronJobsListResult { + const total = options.total ?? jobs.length; + const offset = options.offset ?? 0; + const limit = options.limit ?? 50; + const nextOffset = offset + jobs.length; + const hasMore = nextOffset < total; + return { + jobs, + snapshotRevision: "agents-page-cron-fixture", + total, + offset, + limit, + hasMore, + nextOffset: hasMore ? nextOffset : null, + }; +} + const agentsList: AgentsListResult = { defaultId: "main", mainKey: "main", @@ -494,12 +515,10 @@ describe("AgentsPage gateway lifecycle", () => { } if (method === "cron.list") { const scoped = params?.agentId === "main"; - return { - jobs: scoped ? [implicitDefaultJob] : unrelatedJobs, + return cronListResponse(scoped ? [implicitDefaultJob] : unrelatedJobs, { total: scoped ? 1 : 51, - offset: 0, - hasMore: !scoped, - }; + limit: params?.limit, + }); } throw new Error(`Unexpected gateway method: ${method}`); }); @@ -541,12 +560,12 @@ describe("AgentsPage gateway lifecycle", () => { return { enabled: true, jobs: 80, nextWakeAtMs: null }; } if (params?.limit === 1) { - return { jobs: [jobs[0]], total: 51 }; + return cronListResponse([jobs[0]!], { total: 51, limit: 1 }); } if (params?.offset === 50) { - return { jobs: [lastJob], total: 51, offset: 50, nextOffset: null, hasMore: false }; + return cronListResponse([lastJob], { total: 51, offset: 50 }); } - return { jobs, total: 51, offset: 0, nextOffset: 50, hasMore: true }; + return cronListResponse(jobs, { total: 51 }); }, ); const client = { request } as unknown as GatewayBrowserClient; @@ -585,7 +604,7 @@ describe("AgentsPage gateway lifecycle", () => { if (method === "cron.status") { return { enabled: true, jobs: 2, nextWakeAtMs: null }; } - return { jobs: [cronJob(`${params?.agentId}-job`, params?.agentId)], total: 1 }; + return cronListResponse([cronJob(`${params?.agentId}-job`, params?.agentId)]); }); const client = { request } as unknown as GatewayBrowserClient; const page = document.createElement("openclaw-agents-page") as TestAgentsPage; @@ -610,7 +629,7 @@ describe("AgentsPage gateway lifecycle", () => { it("keeps an in-flight scoped cron request attached to a same-client gateway snapshot", async () => { const job = cronJob("same-client-job", "main"); - const pendingJobs = deferred<{ jobs: CronJob[]; total: number }>(); + const pendingJobs = deferred(); const request = vi.fn((method: string, params?: { limit?: number }) => { if (method === "cron.status") { return Promise.resolve({ enabled: true, jobs: 1, nextWakeAtMs: null }); @@ -618,7 +637,7 @@ describe("AgentsPage gateway lifecycle", () => { if (params?.limit === 50) { return pendingJobs.promise; } - return Promise.resolve({ jobs: [job], total: 1 }); + return Promise.resolve(cronListResponse([job])); }); const client = { request } as unknown as GatewayBrowserClient; const page = document.createElement("openclaw-agents-page") as TestAgentsPage; @@ -634,7 +653,7 @@ describe("AgentsPage gateway lifecycle", () => { setPageGateway(page, client); expect(page.cron).toBe(inFlightState); - pendingJobs.resolve({ jobs: [job], total: 1 }); + pendingJobs.resolve(cronListResponse([job])); await vi.waitFor(() => { expect(page.cron.cronJobs).toEqual([job]); expect(page.cron.cronLoading).toBe(false); @@ -643,7 +662,7 @@ describe("AgentsPage gateway lifecycle", () => { it("immediately publishes cron loading and ignores a second refresh while the first is pending", async () => { const job = cronJob("double-refresh-job", "main"); - const pendingJobs = deferred<{ jobs: CronJob[]; total: number }>(); + const pendingJobs = deferred(); const request = vi.fn((method: string, params?: { limit?: number }) => { if (method === "cron.status") { return Promise.resolve({ enabled: true, jobs: 1, nextWakeAtMs: null }); @@ -651,7 +670,7 @@ describe("AgentsPage gateway lifecycle", () => { if (params?.limit === 50) { return pendingJobs.promise; } - return Promise.resolve({ jobs: [job], total: 1 }); + return Promise.resolve(cronListResponse([job])); }); const client = { request } as unknown as GatewayBrowserClient; const page = document.createElement("openclaw-agents-page") as TestAgentsPage; @@ -670,7 +689,7 @@ describe("AgentsPage gateway lifecycle", () => { ), ).toHaveLength(1); - pendingJobs.resolve({ jobs: [job], total: 1 }); + pendingJobs.resolve(cronListResponse([job])); await firstRefresh; expect(page.cron.cronLoading).toBe(false); diff --git a/ui/src/pages/agents/view.test.ts b/ui/src/pages/agents/view.test.ts index 5d5bdf90e176..e588289a9a57 100644 --- a/ui/src/pages/agents/view.test.ts +++ b/ui/src/pages/agents/view.test.ts @@ -192,6 +192,7 @@ describe("renderAgents", () => { }); it("loads and renders the selected agent's 51st cron job when Load more is clicked", async () => { + const snapshotRevision = "agents-view-cron-fixture"; const jobs = Array.from({ length: 50 }, (_, index) => createCronJob(`main-${index}`, { agentId: "alpha" }), ); @@ -201,8 +202,10 @@ describe("renderAgents", () => { }); const request = vi.fn(async () => ({ jobs: [lastJob], + snapshotRevision, total: 51, offset: 50, + limit: 50, nextOffset: null, hasMore: false, })); @@ -211,6 +214,7 @@ describe("renderAgents", () => { ...createInitialCronState({ client, connected: true }), cronAgentId: "alpha", cronJobs: jobs, + cronJobsSnapshotRevision: snapshotRevision, cronJobsTotal: 51, cronJobsHasMore: true, cronJobsNextOffset: 50, @@ -234,10 +238,7 @@ describe("renderAgents", () => { error: cronState.cronError, }, onCronLoadMore: () => { - const nextPage = loadCronJobsPage(cronState, { - append: true, - tableFilters: true, - }); + const nextPage = loadCronJobsPage(cronState, { append: true, tableFilters: true }); renderCurrentPage(); void nextPage.then(renderCurrentPage); }, diff --git a/ui/src/pages/cron/cron-page.test.ts b/ui/src/pages/cron/cron-page.test.ts index ffe2cf524845..90f90ca6a94a 100644 --- a/ui/src/pages/cron/cron-page.test.ts +++ b/ui/src/pages/cron/cron-page.test.ts @@ -2,7 +2,7 @@ import { nothing } from "lit"; import { afterEach, describe, expect, it, vi } from "vitest"; import { createDeferred } from "../../../../test/helpers/promise.js"; import type { GatewayBrowserClient, GatewayEventListener } from "../../api/gateway.ts"; -import type { CronJob } from "../../api/types.ts"; +import type { CronJob, CronJobsListResult } from "../../api/types.ts"; import type { ApplicationContext, ApplicationGatewaySnapshot } from "../../app/context.ts"; import type { CronState } from "../../lib/cron/index.ts"; import "./cron-page.ts"; @@ -129,10 +129,22 @@ function createPage(context: ApplicationContext, options: { render?: boolean } = return page; } +function cronListResponse(jobs: CronJob[]): CronJobsListResult { + return { + jobs, + snapshotRevision: "cron-page-fixture", + total: jobs.length, + offset: 0, + limit: 50, + hasMore: false, + nextOffset: null, + }; +} + function createRequest() { return vi.fn(async (method: string) => { if (method === "cron.list") { - return { jobs: [], total: 0, offset: 0, hasMore: false }; + return cronListResponse([]); } if (method === "cron.runs") { return { entries: [], total: 0, offset: 0, hasMore: false }; @@ -171,7 +183,7 @@ describe("CronPage editor state sync", () => { }; const request = vi.fn(async (method: string) => { if (method === "cron.list") { - return { jobs: [job], total: 1, offset: 0, hasMore: false }; + return cronListResponse([job]); } if (method === "cron.runs") { return { entries: [], total: 0, offset: 0, hasMore: false }; @@ -316,7 +328,7 @@ describe("CronPage editor state sync", () => { return { id: "job-fresh" }; } if (method === "cron.list") { - return { jobs: [], total: 0, offset: 0, hasMore: false }; + return cronListResponse([]); } if (method === "cron.runs") { return { entries: [], total: 0, offset: 0, hasMore: false }; @@ -371,7 +383,7 @@ describe("CronPage editor state sync", () => { }); it("syncs form enabled after header pause and resets runs scope after remove", async () => { - const job = { + const job: CronJob = { id: "job-1", name: "Nightly digest", enabled: true, @@ -386,12 +398,7 @@ describe("CronPage editor state sync", () => { let removed = false; const request = vi.fn(async (method: string, params?: unknown) => { if (method === "cron.list") { - return { - jobs: removed ? [] : [{ ...job, enabled: serverEnabled }], - total: removed ? 0 : 1, - offset: 0, - hasMore: false, - }; + return cronListResponse(removed ? [] : [{ ...job, enabled: serverEnabled }]); } if (method === "cron.update") { const patch = (params as { patch?: { enabled?: boolean } }).patch; @@ -442,7 +449,7 @@ describe("CronPage editor state sync", () => { }); it("renders read-only controls and rejects a stale admin action after a scope downgrade", async () => { - const job = { + const job: CronJob = { id: "job-1", name: "Nightly digest", enabled: true, @@ -455,7 +462,7 @@ describe("CronPage editor state sync", () => { }; const request = vi.fn(async (method: string) => { if (method === "cron.list") { - return { jobs: [job], total: 1, offset: 0, hasMore: false }; + return cronListResponse([job]); } if (method === "cron.runs") { return { entries: [], total: 0, offset: 0, hasMore: false }; @@ -545,7 +552,7 @@ describe("CronPage lifecycle", () => { return modelRequestCount === 1 ? staleModels.promise : { models: [{ id: "fresh/model" }] }; } if (method === "cron.list") { - return { jobs: [], total: 0, offset: 0, hasMore: false }; + return cronListResponse([]); } if (method === "cron.runs") { return { entries: [], total: 0, offset: 0, hasMore: false };