From 9bba88dbba15c36cc49fbc01d37e63a3fda209ff Mon Sep 17 00:00:00 2001 From: SunnyShu Date: Thu, 20 Aug 2026 23:38:57 +0800 Subject: [PATCH] fix(tasks): rank terminal tasks by completion and keep Recent terminal-only (#123219) * fix(tasks): rank terminal tasks by completion and keep Recent terminal-only - updateTaskStateByRunId backfills lastEventAt from endedAt for terminal finalizers (mirrors markTaskTerminalById), keeping activity monotonic - both taskUpdatedAt projections rank terminal tasks by the maximum available activity timestamp, healing stale rows while preserving later delivery/terminal-outcome events recorded after completion - Tasks page Recent fetch filters to terminal statuses so queued/running rows cannot starve the Recent section Related to #100911 * refactor(tasks): normalize completion at registry owner Absorb terminal timestamp ordering into the canonical registry lifecycle boundary, remove duplicated projection and writer policy, and prove Recent remains visible behind 200 active tasks in Chromium. Co-authored-by: SunnyShu0925 --------- Co-authored-by: Peter Steinberger --- src/gateway/server-methods/tasks.test.ts | 81 ++++++++++++++++++++++++ src/tasks/task-registry-records.ts | 8 +-- src/tasks/task-registry.test.ts | 42 ++++++++++++ ui/src/pages/tasks/tasks-page.test.ts | 37 ++++++++--- ui/src/pages/tasks/tasks-page.ts | 10 ++- ui/src/pages/tasks/tasks.e2e.test.ts | 70 +++++++++++++++++++- 6 files changed, 230 insertions(+), 18 deletions(-) diff --git a/src/gateway/server-methods/tasks.test.ts b/src/gateway/server-methods/tasks.test.ts index 81c919201818..44311f2dee70 100644 --- a/src/gateway/server-methods/tasks.test.ts +++ b/src/gateway/server-methods/tasks.test.ts @@ -296,6 +296,87 @@ describe("tasks gateway handlers", () => { ); }); + it("ranks terminal tasks by completion time when the progress timestamp is stale", async () => { + const base = Date.now(); + const justFinished = createSnapshotTask({ + taskId: "task-just-finished", + runId: "run-just-finished", + status: "succeeded", + deliveryStatus: "not_applicable", + createdAt: base - 10_000, + startedAt: base - 9_000, + lastEventAt: base - 5_000, + endedAt: base - 1_000, + }); + const finishedEarlier = createSnapshotTask({ + taskId: "task-finished-earlier", + runId: "run-finished-earlier", + status: "succeeded", + deliveryStatus: "not_applicable", + createdAt: base - 8_000, + startedAt: base - 7_000, + lastEventAt: base - 2_000, + endedAt: base - 3_000, + }); + saveTaskRegistryStateToSqlite({ + tasks: new Map([ + [justFinished.taskId, justFinished], + [finishedEarlier.taskId, finishedEarlier], + ]), + deliveryStates: new Map(), + }); + reloadTaskRegistryFromStore(); + + const { payload } = await runTaskHandler("tasks.list", {}); + + expect(payload?.tasks?.map((task) => task.taskId)).toEqual([ + justFinished.taskId, + finishedEarlier.taskId, + ]); + }); + + it("ranks a terminal task by its later activity when completion trails it", async () => { + const base = Date.now(); + const laterActivity = createSnapshotTask({ + taskId: "task-later-activity", + runId: "run-later-activity", + status: "succeeded", + deliveryStatus: "not_applicable", + createdAt: base - 10_000, + startedAt: base - 9_000, + lastEventAt: base - 100, + endedAt: base - 2_000, + }); + const laterCompletion = createSnapshotTask({ + taskId: "task-later-completion", + runId: "run-later-completion", + status: "succeeded", + deliveryStatus: "not_applicable", + createdAt: base - 8_000, + startedAt: base - 7_000, + lastEventAt: base - 4_000, + endedAt: base - 500, + }); + saveTaskRegistryStateToSqlite({ + tasks: new Map([ + [laterActivity.taskId, laterActivity], + [laterCompletion.taskId, laterCompletion], + ]), + deliveryStates: new Map(), + }); + reloadTaskRegistryFromStore(); + + const { payload } = await runTaskHandler("tasks.list", {}); + const byId = new Map(payload?.tasks?.map((task) => [task.taskId, task])); + + expect(payload?.tasks?.map((task) => task.taskId)).toEqual([ + laterActivity.taskId, + laterCompletion.taskId, + ]); + expect(byId.get("task-later-activity")?.updatedAt).toBe(base - 100); + expect(byId.get("task-later-completion")?.updatedAt).toBe(base - 500); + }); + it("preserves activity ordering across cursor pages", async () => { const created = [500, 100, 700, 300, 500].map((lastEventAt, index) => createTaskRecord({ diff --git a/src/tasks/task-registry-records.ts b/src/tasks/task-registry-records.ts index c8a838d61ce4..78e2c2325d6e 100644 --- a/src/tasks/task-registry-records.ts +++ b/src/tasks/task-registry-records.ts @@ -20,14 +20,14 @@ export function normalizeTaskTimestamps(task: TaskRecord): TaskRecord { const startedAt = typeof task.startedAt === "number" ? Math.max(task.startedAt, createdAt) : task.startedAt; - const lastEventAt = - typeof task.lastEventAt === "number" - ? Math.max(task.lastEventAt, startedAt ?? createdAt) - : task.lastEventAt; const endedAt = typeof task.endedAt === "number" ? Math.max(task.endedAt, startedAt ?? createdAt) : task.endedAt; + const lastEventAt = + typeof task.lastEventAt === "number" + ? Math.max(task.lastEventAt, endedAt ?? startedAt ?? createdAt) + : task.lastEventAt; if ( createdAt === task.createdAt && diff --git a/src/tasks/task-registry.test.ts b/src/tasks/task-registry.test.ts index cb164cebe430..f9eea4c5c23f 100644 --- a/src/tasks/task-registry.test.ts +++ b/src/tasks/task-registry.test.ts @@ -686,6 +686,48 @@ describe("task-registry", () => { }); }); + it("fills terminal lastEventAt from endedAt when a finalize omits the progress timestamp", async () => { + await withTaskRegistryTempDir(async () => { + resetTaskRegistryMemoryForTest(); + createTaskFixture("subagent", { + childSessionKey: "agent:main:subagent:terminal-timestamp", + runId: "run-terminal-timestamp", + task: "Finalize without a progress timestamp", + lastEventAt: 1_000, + }); + finalizeSubagentTask(requireTaskByRunId("run-terminal-timestamp"), { + status: "succeeded", + endedAt: 2_000, + }); + expectRecordFields(requireTaskByRunId("run-terminal-timestamp"), { + status: "succeeded", + endedAt: 2_000, + lastEventAt: 2_000, + }); + }); + }); + + it("keeps a newer terminal progress timestamp when endedAt trails it", async () => { + await withTaskRegistryTempDir(async () => { + resetTaskRegistryMemoryForTest(); + createTaskFixture("subagent", { + childSessionKey: "agent:main:subagent:monotonic-timestamp", + runId: "run-monotonic-timestamp", + task: "Preserve the newest activity timestamp", + lastEventAt: 3_000, + }); + finalizeSubagentTask(requireTaskByRunId("run-monotonic-timestamp"), { + status: "failed", + endedAt: 2_000, + }); + expectRecordFields(requireTaskByRunId("run-monotonic-timestamp"), { + status: "failed", + endedAt: 2_000, + lastEventAt: 3_000, + }); + }); + }); + it.each([ { name: "persists an ACP producer timestamp across lifecycle projection and SQLite reload", diff --git a/ui/src/pages/tasks/tasks-page.test.ts b/ui/src/pages/tasks/tasks-page.test.ts index 8678e88fc4c7..13e39aa85a14 100644 --- a/ui/src/pages/tasks/tasks-page.test.ts +++ b/ui/src/pages/tasks/tasks-page.test.ts @@ -95,7 +95,7 @@ async function createDeferredTaskRefresh(initialTasks: TaskSummary[]) { if (method !== "tasks.list" || !deferRefresh) { return Promise.resolve({ tasks: currentTasks }); } - return params?.status ? active.promise : recent.promise; + return params?.status?.includes("completed") ? recent.promise : active.promise; }, ); const source = createGateway({ request } as unknown as GatewayBrowserClient); @@ -164,12 +164,12 @@ afterEach(() => { }); describe("TasksPage concurrent refresh events", () => { - it("keeps the later recent page's equally current running progress", async () => { + it("keeps the later recent snapshot when a task transitions to terminal", async () => { const initial = createTask("task-progress", "running", { toolUseCount: 2, progressSummary: "Preparing the concurrent task report", }); - const recent = createTask("task-progress", "running", { + const recent = createTask("task-progress", "completed", { toolUseCount: 2, progressSummary: "Finishing the concurrent task report", }); @@ -178,7 +178,9 @@ describe("TasksPage concurrent refresh events", () => { const refreshCalls = refresh.request.mock.calls.slice(-2); expect(refreshCalls[0]?.[1]).toMatchObject({ status: ["queued", "running"] }); - expect(refreshCalls[1]?.[1]).not.toHaveProperty("status"); + expect(refreshCalls[1]?.[1]).toMatchObject({ + status: ["completed", "failed", "timed_out", "cancelled"], + }); refresh.active.resolve({ tasks: [initial] }); refresh.recent.resolve({ tasks: [recent] }); await pending; @@ -335,10 +337,10 @@ describe("TasksPage active pagination", () => { }, ) => { expect(method).toBe("tasks.list"); - if (!params?.status) { + if (params?.status?.includes("completed")) { return Promise.resolve({ tasks: [createTask("task-recent", "completed")] }); } - if (params.cursor === "active-page-2") { + if (params?.cursor === "active-page-2") { return Promise.resolve({ tasks: [sharedPageTwo, createTask("task-page-2")], }); @@ -367,15 +369,26 @@ describe("TasksPage active pagination", () => { { signal: expect.any(AbortSignal) }, ); expect( - request.mock.calls.filter(([, params]) => !(params as { status?: unknown })?.status), + request.mock.calls.filter(([, params]) => + (params as { status?: readonly string[] } | undefined)?.status?.includes("completed"), + ), ).toHaveLength(1); + expect(request).toHaveBeenCalledWith( + "tasks.list", + expect.objectContaining({ + agentId: "writer", + limit: 200, + status: ["completed", "failed", "timed_out", "cancelled"], + }), + { signal: expect.any(AbortSignal) }, + ); 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) { + if (!params?.status || params.status.length !== 2) { return Promise.resolve({ tasks: [] }); } activeCalls += 1; @@ -400,7 +413,7 @@ describe("TasksPage active pagination", () => { const finalPage = deferred<{ tasks: TaskSummary[] }>(); const request = vi.fn( (_method: string, params?: { cursor?: string; status?: readonly string[] }) => { - if (!params?.status) { + if (!params?.status || params.status.includes("completed")) { return Promise.resolve({ tasks: [] }); } if (params.cursor === "active-page-2") { @@ -528,7 +541,11 @@ describe("TasksPage cancellation lifecycle", () => { ); expect(request).toHaveBeenCalledWith( "tasks.list", - expect.objectContaining({ agentId: "writer", limit: 200 }), + expect.objectContaining({ + agentId: "writer", + limit: 200, + status: ["completed", "failed", "timed_out", "cancelled"], + }), { signal: expect.any(AbortSignal) }, ); }); diff --git a/ui/src/pages/tasks/tasks-page.ts b/ui/src/pages/tasks/tasks-page.ts index 47ab3ee41093..8b9ead6fd442 100644 --- a/ui/src/pages/tasks/tasks-page.ts +++ b/ui/src/pages/tasks/tasks-page.ts @@ -163,7 +163,15 @@ class TasksPage extends OpenClawLightDomElement { const agentId = scopeId ?? undefined; const [active, recentPayload] = await Promise.all([ loadActiveTaskPages({ client, agentId, signal }), - client.request("tasks.list", { limit: 200, ...(agentId ? { agentId } : {}) }, { signal }), + client.request( + "tasks.list", + { + status: ["completed", "failed", "timed_out", "cancelled"], + limit: 200, + ...(agentId ? { agentId } : {}), + }, + { signal }, + ), ]); const recent = normalizeTasksListResult(recentPayload); if (!recent) { diff --git a/ui/src/pages/tasks/tasks.e2e.test.ts b/ui/src/pages/tasks/tasks.e2e.test.ts index 4aa445d8ea8e..2bf2917aac19 100644 --- a/ui/src/pages/tasks/tasks.e2e.test.ts +++ b/ui/src/pages/tasks/tasks.e2e.test.ts @@ -130,6 +130,57 @@ const activePageOneTasks = [ ]; suite.define(() => { + it("keeps completed tasks visible when active work fills the unfiltered page", async () => { + await mkdir(artifactDir, { recursive: true }); + const context = await suite.browser.newContext({ + locale: "en-US", + serviceWorkers: "block", + viewport: { width: 1440, height: 900 }, + }); + const page = await context.newPage(); + try { + const activeTasks = activePageOneTasks.slice(0, 200); + const terminalStatuses = ["completed", "failed", "timed_out", "cancelled"]; + const gateway = await installMockGateway(page, { + methodResponses: { + "tasks.list": { + cases: [ + { + match: { agentId: "main", limit: 500, status: ["queued", "running"] }, + response: { tasks: activeTasks }, + }, + { + match: { agentId: "main", limit: 200, status: terminalStatuses }, + response: { tasks: [completedTask, failedTask] }, + }, + { + match: { agentId: "main", limit: 200 }, + response: { tasks: activeTasks }, + }, + ], + }, + }, + }); + + await page.goto(`${suite.server.baseUrl}tasks`); + const active = page.locator('[data-task-section="active"]'); + const recent = page.locator('[data-task-section="recent"]'); + await active.locator('[data-task-id="task-running"]').waitFor({ state: "visible" }); + await recent.scrollIntoViewIfNeeded(); + await page.screenshot({ path: path.join(artifactDir, "10-recent-terminal-starvation.png") }); + + expect(await recent.textContent()).toContain("Generate media index"); + expect(await recent.textContent()).toContain("Worker exited"); + expect(await gateway.getRequests("tasks.list")).toContainEqual({ + id: expect.any(String), + method: "tasks.list", + params: { agentId: "main", limit: 200, status: terminalStatuses }, + }); + } finally { + await context.close(); + } + }); + it("keeps retry and dismiss outcomes authoritative across a stale refresh and reconnect", async () => { const actionArtifactDir = path.resolve( process.cwd(), @@ -325,7 +376,11 @@ suite.define(() => { }, }, { - match: { agentId: "main", limit: 200 }, + match: { + agentId: "main", + limit: 200, + status: ["completed", "failed", "timed_out", "cancelled"], + }, response: { tasks: [completedTask, failedTask] }, }, ], @@ -357,12 +412,12 @@ suite.define(() => { listRequests.filter( (request) => (request.params as { status?: unknown }).status !== undefined, ), - ).toHaveLength(2); + ).toHaveLength(3); expect( listRequests.filter( (request) => (request.params as { status?: unknown }).status === undefined, ), - ).toHaveLength(1); + ).toHaveLength(0); expect(listRequests).toContainEqual({ id: expect.any(String), method: "tasks.list", @@ -373,6 +428,15 @@ suite.define(() => { status: ["queued", "running"], }, }); + expect(listRequests).toContainEqual({ + id: expect.any(String), + method: "tasks.list", + params: { + agentId: "main", + limit: 200, + status: ["completed", "failed", "timed_out", "cancelled"], + }, + }); await page.screenshot({ path: path.join(artifactDir, "01-page-two-sentinel.png"), });