From cb58073a902b135dcf5a7dea2c4d0fd5aa877532 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 12 Aug 2026 02:46:55 -0700 Subject: [PATCH] fix(ui): paginate all active tasks (#122565) Preserve task-list cursors and drain every active page so supported running work remains visible and cancellable. --- ui/src/lib/tasks/data.test.ts | 15 ++- ui/src/lib/tasks/data.ts | 13 +- .../chat/components/chat-background-tasks.ts | 10 +- ui/src/pages/tasks/tasks-page.test.ts | 118 ++++++++++++++++++ ui/src/pages/tasks/tasks-page.ts | 54 ++++++-- ui/src/pages/tasks/tasks.e2e.test.ts | 108 ++++++++++++++-- 6 files changed, 283 insertions(+), 35 deletions(-) diff --git a/ui/src/lib/tasks/data.test.ts b/ui/src/lib/tasks/data.test.ts index a2dc3a83ac11..c500b8e16c81 100644 --- a/ui/src/lib/tasks/data.test.ts +++ b/ui/src/lib/tasks/data.test.ts @@ -322,13 +322,20 @@ describe("tasks page data", () => { sourceId: "source-1", }; - expect(normalizeTasksListResult({ tasks: [wireTask] })?.[0]).toEqual({ - ...wireTask, - id: "task-1", - taskId: "task-1", + expect(normalizeTasksListResult({ tasks: [wireTask], nextCursor: "page-2" })).toEqual({ + nextCursor: "page-2", + tasks: [ + { + ...wireTask, + id: "task-1", + taskId: "task-1", + }, + ], }); expect(normalizeTasksGetResult({ task: wireTask })?.taskId).toBe("task-1"); expect(normalizeTasksListResult({ tasks: [{ ...wireTask, updatedAt: false }] })).toBeNull(); + expect(normalizeTasksListResult({ tasks: [wireTask], nextCursor: 2 })).toBeNull(); + expect(normalizeTasksListResult({ tasks: "not-a-page" })).toBeNull(); }); it("merges upserts, applies deletes, and requests refetches for restored events", () => { diff --git a/ui/src/lib/tasks/data.ts b/ui/src/lib/tasks/data.ts index 92a8b097c26b..a5169422bdba 100644 --- a/ui/src/lib/tasks/data.ts +++ b/ui/src/lib/tasks/data.ts @@ -170,13 +170,18 @@ export function partitionTasks(tasks: readonly TaskSummary[]): { }; } -export function normalizeTasksListResult(value: unknown): TaskSummary[] | null { +export function normalizeTasksListResult( + value: unknown, +): { tasks: TaskSummary[]; nextCursor?: string } | null { if (!Value.Check(TasksListResultSchema, value)) { return null; } - return sortTasks( - value.tasks.map(normalizeTaskSummary).filter((task): task is TaskSummary => task !== null), - ); + return { + tasks: sortTasks( + value.tasks.map(normalizeTaskSummary).filter((task): task is TaskSummary => task !== null), + ), + ...(value.nextCursor !== undefined ? { nextCursor: value.nextCursor } : {}), + }; } export function normalizeTasksGetResult(value: unknown): TaskSummary | null { diff --git a/ui/src/pages/chat/components/chat-background-tasks.ts b/ui/src/pages/chat/components/chat-background-tasks.ts index 37eda6a462cd..bb8ba5554a9a 100644 --- a/ui/src/pages/chat/components/chat-background-tasks.ts +++ b/ui/src/pages/chat/components/chat-background-tasks.ts @@ -78,9 +78,9 @@ export type BackgroundTasksHost = { requestUpdate?: () => void; }; -// Bounded like the Tasks page: active tasks get their own query because the -// ledger pages newest-first and long-running work can hide behind newer -// terminal records on the first page. +// The chat rail stays bounded to its session while the full Tasks page drains +// every active page. A separate active query still keeps long-running work +// from hiding behind newer terminal records here. const ACTIVE_TASKS_LIMIT = 200; const RECENT_TASKS_LIMIT = 100; @@ -255,10 +255,10 @@ function loadBackgroundTasks( }), client.request("tasks.list", { sessionKey, limit: RECENT_TASKS_LIMIT }), ]); - const active = normalizeTasksListResult(activePayload)?.map((task) => + const active = normalizeTasksListResult(activePayload)?.tasks.map((task) => prepareTaskSnapshot(state, task), ); - const recent = normalizeTasksListResult(recentPayload)?.map((task) => + const recent = normalizeTasksListResult(recentPayload)?.tasks.map((task) => prepareTaskSnapshot(state, task), ); if (!active || !recent) { diff --git a/ui/src/pages/tasks/tasks-page.test.ts b/ui/src/pages/tasks/tasks-page.test.ts index 4ce02a1327a3..095d40d03a74 100644 --- a/ui/src/pages/tasks/tasks-page.test.ts +++ b/ui/src/pages/tasks/tasks-page.test.ts @@ -301,6 +301,124 @@ describe("TasksPage concurrent refresh events", () => { }); }); +describe("TasksPage active pagination", () => { + it("drains active pages with the selected scope and merges each task once", async () => { + const sharedPageOne = createTask("task-shared", "running", { + progressSummary: "Page one progress", + updatedAt: 100, + }); + const sharedPageTwo = createTask("task-shared", "running", { + progressSummary: "Page two progress", + updatedAt: 200, + }); + const request = vi.fn( + ( + method: string, + params?: { + agentId?: string; + cursor?: string; + limit?: number; + status?: readonly string[]; + }, + ) => { + expect(method).toBe("tasks.list"); + if (!params?.status) { + return Promise.resolve({ tasks: [createTask("task-recent", "completed")] }); + } + if (params.cursor === "active-page-2") { + return Promise.resolve({ + tasks: [sharedPageTwo, createTask("task-page-2")], + }); + } + return Promise.resolve({ + tasks: [sharedPageOne, createTask("task-page-1")], + nextCursor: "active-page-2", + }); + }, + ); + const source = createGateway({ request } as unknown as GatewayBrowserClient); + const page = document.createElement("openclaw-tasks-page") as TasksPageTestElement; + page.context = createContext(source.gateway, "writer"); + document.body.append(page); + + await vi.waitFor(() => expect(page.tasks).toHaveLength(4)); + + expect(request).toHaveBeenCalledWith( + "tasks.list", + { + agentId: "writer", + cursor: "active-page-2", + limit: 500, + status: ["queued", "running"], + }, + { signal: expect.any(AbortSignal) }, + ); + expect( + request.mock.calls.filter(([, params]) => !(params as { status?: unknown })?.status), + ).toHaveLength(1); + expect(page.tasks.filter((task) => task.id === "task-shared")).toEqual([sharedPageTwo]); + }); + + it("fails visibly when an active page repeats its cursor", async () => { + let activeCalls = 0; + const request = vi.fn((_method: string, params?: { status?: readonly string[] }) => { + if (!params?.status) { + return Promise.resolve({ tasks: [] }); + } + activeCalls += 1; + return Promise.resolve({ + tasks: [createTask(`task-page-${activeCalls}`)], + nextCursor: "repeated-cursor", + }); + }); + const source = createGateway({ request } as unknown as GatewayBrowserClient); + const page = document.createElement("openclaw-tasks-page") as TasksPageTestElement; + page.context = createContext(source.gateway); + document.body.append(page); + + await vi.waitFor(() => expect(page.error).toBe("The gateway returned an invalid task list.")); + + expect(activeCalls).toBe(2); + expect(request).toHaveBeenCalledTimes(3); + }); + + it("replays buffered events after the final active page resolves", async () => { + const stale = createTask("task-draining", "running", { updatedAt: 100 }); + const finalPage = deferred<{ tasks: TaskSummary[] }>(); + const request = vi.fn( + (_method: string, params?: { cursor?: string; status?: readonly string[] }) => { + if (!params?.status) { + return Promise.resolve({ tasks: [] }); + } + if (params.cursor === "active-page-2") { + return finalPage.promise; + } + return Promise.resolve({ tasks: [stale], nextCursor: "active-page-2" }); + }, + ); + const source = createGateway({ request } as unknown as GatewayBrowserClient); + const page = document.createElement("openclaw-tasks-page") as TasksPageTestElement; + page.context = createContext(source.gateway); + document.body.append(page); + await vi.waitFor(() => + expect(request).toHaveBeenCalledWith( + "tasks.list", + expect.objectContaining({ cursor: "active-page-2" }), + { signal: expect.any(AbortSignal) }, + ), + ); + + source.emitTask({ + action: "upserted", + task: { ...stale, status: "completed", updatedAt: 200 }, + }); + finalPage.resolve({ tasks: [stale] }); + await vi.waitFor(() => expect(page.tasks[0]?.status).toBe("completed")); + + expect(page.tasks).toHaveLength(1); + }); +}); + describe("TasksPage cancellation lifecycle", () => { it("qualifies unscoped task session links with the selected agent", async () => { const request = vi.fn(async () => ({ diff --git a/ui/src/pages/tasks/tasks-page.ts b/ui/src/pages/tasks/tasks-page.ts index 49cbdc629431..cb908cafa4b9 100644 --- a/ui/src/pages/tasks/tasks-page.ts +++ b/ui/src/pages/tasks/tasks-page.ts @@ -62,6 +62,43 @@ type TaskRefreshEventBuffer = { events: TaskRefreshEvent[]; }; +async function loadActiveTaskPages(params: { + client: GatewayBrowserClient; + agentId: string | undefined; + signal: AbortSignal; +}): Promise { + let tasks: TaskSummary[] = []; + let cursor: string | undefined; + const seenCursors = new Set(); + while (true) { + const payload = await params.client.request( + "tasks.list", + { + status: ["queued", "running"], + limit: 500, + ...(params.agentId ? { agentId: params.agentId } : {}), + ...(cursor !== undefined ? { cursor } : {}), + }, + { signal: params.signal }, + ); + const page = normalizeTasksListResult(payload); + if (!page) { + throw new Error(t("tasksPage.invalidResponse")); + } + tasks = mergeTaskLists(tasks, page.tasks); + if (page.nextCursor === undefined) { + return tasks; + } + // Cursors are opaque, so revisiting any prior token is the only safe + // client-side definition of a non-advancing page sequence. + if (!page.nextCursor || seenCursors.has(page.nextCursor)) { + throw new Error(t("tasksPage.invalidResponse")); + } + seenCursors.add(page.nextCursor); + cursor = page.nextCursor; + } +} + class TasksPage extends OpenClawLightDomElement { @consume({ context: applicationContext, subscribe: true }) private context!: ApplicationContext; @@ -115,24 +152,15 @@ class TasksPage extends OpenClawLightDomElement { }; this.taskRefreshEvents = buffer; const agentId = scopeId ?? undefined; - const [activePayload, recentPayload] = await Promise.all([ - client.request( - "tasks.list", - { - status: ["queued", "running"], - limit: 500, - ...(agentId ? { agentId } : {}), - }, - { signal }, - ), + const [active, recentPayload] = await Promise.all([ + loadActiveTaskPages({ client, agentId, signal }), client.request("tasks.list", { limit: 200, ...(agentId ? { agentId } : {}) }, { signal }), ]); - const active = normalizeTasksListResult(activePayload); const recent = normalizeTasksListResult(recentPayload); - if (!active || !recent) { + if (!recent) { throw new Error(t("tasksPage.invalidResponse")); } - return { active, recent, buffer }; + return { active, recent: recent.tasks, buffer }; }, onComplete: ({ active, recent, buffer }) => { // The active query is issued first; a same-millisecond recent page diff --git a/ui/src/pages/tasks/tasks.e2e.test.ts b/ui/src/pages/tasks/tasks.e2e.test.ts index 03b2070e8380..177c519a0943 100644 --- a/ui/src/pages/tasks/tasks.e2e.test.ts +++ b/ui/src/pages/tasks/tasks.e2e.test.ts @@ -65,8 +65,38 @@ const failedTask = { error: "Worker exited", }; +const pageTwoSentinel = { + id: "task-page-two-sentinel", + taskId: "task-page-two-sentinel", + kind: "subagent", + runtime: "subagent", + status: "running", + title: "Page two running sentinel", + agentId: "main", + childSessionKey: "agent:main:subagent:page-two-sentinel", + createdAt: baseTime + 4_000, + updatedAt: baseTime + 5_000, + progressSummary: "Visible only after active pagination", +}; + +const activePageOneTasks = [ + runningTask, + queuedTask, + ...Array.from({ length: 498 }, (_, index) => ({ + id: `task-page-one-${index}`, + taskId: `task-page-one-${index}`, + kind: "cron", + runtime: "cron", + status: "running", + title: `Page one active task ${index + 1}`, + agentId: "main", + createdAt: baseTime - 20_000 - index, + updatedAt: baseTime - 10_000 - index, + })), +]; + suite.define(() => { - it("renders task sections, applies pushed completion, and sends cancel", async () => { + it("renders every active page, applies pushed completion, and cancels a page-two task", async () => { await rm(artifactDir, { force: true, recursive: true }); await mkdir(artifactDir, { recursive: true }); const rawVideoDir = path.join(artifactDir, "raw-video"); @@ -83,12 +113,37 @@ suite.define(() => { const gateway = await installMockGateway(page, { methodResponses: { "tasks.list": { - tasks: [runningTask, queuedTask, completedTask, failedTask], + cases: [ + { + match: { + agentId: "main", + cursor: "active-page-2", + limit: 500, + status: ["queued", "running"], + }, + response: { tasks: [pageTwoSentinel] }, + }, + { + match: { + agentId: "main", + limit: 500, + status: ["queued", "running"], + }, + response: { + tasks: activePageOneTasks, + nextCursor: "active-page-2", + }, + }, + { + match: { agentId: "main", limit: 200 }, + response: { tasks: [completedTask, failedTask] }, + }, + ], }, "tasks.cancel": { found: true, cancelled: true, - task: { ...queuedTask, status: "cancelled", updatedAt: baseTime + 2_000 }, + task: { ...pageTwoSentinel, status: "cancelled", updatedAt: baseTime + 6_000 }, }, }, }); @@ -97,15 +152,39 @@ suite.define(() => { expect(response?.status()).toBe(200); const active = page.locator('[data-task-section="active"]'); const recent = page.locator('[data-task-section="recent"]'); + await active.locator('[data-task-id="task-page-two-sentinel"]').waitFor({ + state: "visible", + }); await active.locator('[data-task-id="task-running"]').waitFor({ state: "visible" }); await active.locator('[data-task-id="task-queued"]').waitFor({ state: "visible" }); await recent.locator('[data-task-id="task-completed"]').waitFor({ state: "visible" }); await recent.locator('[data-task-id="task-failed"]').waitFor({ state: "visible" }); expect(await active.textContent()).toContain("Reading subscription paths"); + expect(await active.textContent()).toContain("Visible only after active pagination"); expect(await recent.textContent()).toContain("Worker exited"); + const listRequests = await gateway.getRequests("tasks.list"); + expect( + listRequests.filter( + (request) => (request.params as { status?: unknown }).status !== undefined, + ), + ).toHaveLength(2); + expect( + listRequests.filter( + (request) => (request.params as { status?: unknown }).status === undefined, + ), + ).toHaveLength(1); + expect(listRequests).toContainEqual({ + id: expect.any(String), + method: "tasks.list", + params: { + agentId: "main", + cursor: "active-page-2", + limit: 500, + status: ["queued", "running"], + }, + }); await page.screenshot({ - path: path.join(artifactDir, "01-task-sections.png"), - fullPage: true, + path: path.join(artifactDir, "01-page-two-sentinel.png"), }); await gateway.emitGatewayEvent("task", { @@ -122,15 +201,26 @@ suite.define(() => { expect(await recent.textContent()).toContain("Review complete"); await page.screenshot({ path: path.join(artifactDir, "02-pushed-completion.png"), - fullPage: true, }); await active - .locator('[data-task-id="task-queued"]') - .getByRole("button", { name: "Cancel Nightly cleanup" }) + .locator('[data-task-id="task-page-two-sentinel"]') + .getByRole("button", { name: "Cancel Page two running sentinel" }) .click(); const cancelRequest = await gateway.waitForRequest("tasks.cancel"); - expect(cancelRequest.params).toEqual({ taskId: "task-queued" }); + expect(cancelRequest.params).toEqual({ taskId: "task-page-two-sentinel" }); + expect(await gateway.getRequests("tasks.cancel")).toHaveLength(1); + const cancelledSentinel = recent.locator('[data-task-id="task-page-two-sentinel"]'); + await cancelledSentinel.waitFor({ + state: "visible", + }); + await active.locator('[data-task-id="task-page-two-sentinel"]').waitFor({ + state: "detached", + }); + await cancelledSentinel.scrollIntoViewIfNeeded(); + await page.screenshot({ + path: path.join(artifactDir, "03-page-two-cancelled.png"), + }); } finally { await context.close(); if (video) {