diff --git a/docs/automation/tasks.md b/docs/automation/tasks.md index a9078f616f6..6573762dc82 100644 --- a/docs/automation/tasks.md +++ b/docs/automation/tasks.md @@ -287,6 +287,10 @@ When the current session has no visible linked tasks, `/tasks` falls back to age For the full operator ledger, use the CLI: `openclaw tasks list`. +### Control UI + +The web Control UI has a **Tasks** page in the sidebar with live active and recent background tasks. Use it to inspect progress, open linked sessions, refresh the ledger, or cancel queued and running tasks. + ## Status integration (task pressure) `openclaw status` includes an at-a-glance task line: diff --git a/docs/docs_map.md b/docs/docs_map.md index 4748c8e185c..05c124e4445 100644 --- a/docs/docs_map.md +++ b/docs/docs_map.md @@ -199,6 +199,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H3: Notification policies - H2: CLI reference - H2: Chat task board (/tasks) + - H3: Control UI - H2: Status integration (task pressure) - H2: Storage and maintenance - H3: Where tasks live diff --git a/docs/web/control-ui.md b/docs/web/control-ui.md index 3434b957369..7dfb60ee2b9 100644 --- a/docs/web/control-ui.md +++ b/docs/web/control-ui.md @@ -142,8 +142,9 @@ The compact footer keeps connection status, **Settings**, **Docs**, and mobile p - Dreams: dreaming status, enable/disable toggle, and Dream Diary reader (`doctor.memory.status`, `doctor.memory.dreamDiary`, `config.patch`). - + - Cron jobs: list/add/edit/run/enable/disable plus run history (`cron.*`). + - Tasks: live active and recent background task ledger with linked sessions and cancellation (`tasks.*`). - Skills: status, enable/disable, install, API key updates (`skills.*`). - Nodes: list plus caps (`node.list`), create mobile setup codes, and approve device pairing (`device.pair.*`). - Exec approvals: edit gateway or node allowlists and ask policy for `exec host=gateway/node` (`exec.approvals.*`). diff --git a/scripts/protocol-event-coverage.allowlist.json b/scripts/protocol-event-coverage.allowlist.json index 2e2dbda85f7..b40191715c8 100644 --- a/scripts/protocol-event-coverage.allowlist.json +++ b/scripts/protocol-event-coverage.allowlist.json @@ -8,6 +8,7 @@ "shutdown": "iOS relies on socket close plus reconnect/backoff instead of the shutdown notice.", "heartbeat": "iOS liveness uses tick and WebSocket-level ping; heartbeat is unused.", "cron": "Cron run activity is not surfaced in the iOS app.", + "task": "Background task activity is not surfaced in the iOS app.", "node.pair.requested": "Node pairing state is fetched on demand; no push consumer on iOS yet.", "node.pair.resolved": "Node pairing state is fetched on demand; no push consumer on iOS yet.", "device.pair.requested": "Device pairing flows poll via device.pair.* methods on iOS.", @@ -27,6 +28,7 @@ "shutdown": "Android relies on socket close plus reconnect/backoff instead of the shutdown notice.", "heartbeat": "Android liveness uses tick and WebSocket-level ping; heartbeat is unused.", "cron": "Cron run activity is not surfaced in the Android app.", + "task": "Background task activity is not surfaced in the Android app.", "node.pair.requested": "Android's bounded operator session lacks operator.pairing; onboarding refreshes node approval with explicit node.list requests.", "node.pair.resolved": "Android's bounded operator session lacks operator.pairing; onboarding refreshes node approval with explicit node.list requests.", "device.pair.requested": "Device pairing flows poll via device.pair.* methods on Android.", diff --git a/src/gateway/gateway-misc.test.ts b/src/gateway/gateway-misc.test.ts index 9d351325e76..06744702817 100644 --- a/src/gateway/gateway-misc.test.ts +++ b/src/gateway/gateway-misc.test.ts @@ -439,6 +439,19 @@ describe("gateway broadcaster", () => { expectSentEvents(adminSocket, expectedEvents); }); + it("requires operator.read for task ledger broadcast events", () => { + const { pairingSocket, nodeSocket, readSocket, writeSocket, adminSocket, broadcast } = + makeScopedBroadcastContext(); + + broadcast("task", { action: "deleted", taskId: "task-1" }); + + expect(pairingSocket.send).not.toHaveBeenCalled(); + expect(nodeSocket.send).not.toHaveBeenCalled(); + expectSentEvents(readSocket, ["task"]); + expectSentEvents(writeSocket, ["task"]); + expectSentEvents(adminSocket, ["task"]); + }); + it("allows plugin.* broadcast events for operator.write and operator.admin", () => { const { pairingSocket, nodeSocket, readSocket, writeSocket, adminSocket, broadcast } = makeScopedBroadcastContext(); diff --git a/src/gateway/server-broadcast.ts b/src/gateway/server-broadcast.ts index 4500e507889..374e1953560 100644 --- a/src/gateway/server-broadcast.ts +++ b/src/gateway/server-broadcast.ts @@ -37,6 +37,7 @@ const EVENT_SCOPE_GUARDS: Record = { tick: [], "talk.event": [READ_SCOPE], "talk.mode": [WRITE_SCOPE], + task: [READ_SCOPE], "update.available": [], "voicewake.changed": [READ_SCOPE], "voicewake.routing.changed": [READ_SCOPE], diff --git a/src/gateway/server-close.test.ts b/src/gateway/server-close.test.ts index 5fc866c08a0..2cae7a01e93 100644 --- a/src/gateway/server-close.test.ts +++ b/src/gateway/server-close.test.ts @@ -118,6 +118,7 @@ function createGatewayCloseTestDeps( mediaCleanup: null, worktreeCleanup: null, agentUnsub: null, + taskUnsub: null, heartbeatUnsub: null, transcriptUnsub: null, lifecycleUnsub: null, @@ -1297,12 +1298,14 @@ describe("createGatewayCloseHandler", () => { it("unsubscribes lifecycle listeners and disposes bundle runtimes during shutdown", async () => { const lifecycleUnsub = vi.fn(); + const taskUnsub = vi.fn(); const transcriptUnsub = vi.fn(); const stopTaskRegistryMaintenance = vi.fn(); const close = createGatewayCloseHandler( createGatewayCloseTestDeps({ stopTaskRegistryMaintenance, lifecycleUnsub, + taskUnsub, transcriptUnsub, }), ); @@ -1310,6 +1313,7 @@ describe("createGatewayCloseHandler", () => { await close({ reason: "test shutdown" }); expect(lifecycleUnsub).toHaveBeenCalledTimes(1); + expect(taskUnsub).toHaveBeenCalledTimes(1); expect(transcriptUnsub).toHaveBeenCalledTimes(1); expect(stopTaskRegistryMaintenance).toHaveBeenCalledTimes(1); expect(mocks.disposeAgentHarnesses).toHaveBeenCalledTimes(1); diff --git a/src/gateway/server-close.ts b/src/gateway/server-close.ts index 60c7efeb470..1541b50af85 100644 --- a/src/gateway/server-close.ts +++ b/src/gateway/server-close.ts @@ -692,6 +692,7 @@ export function createGatewayCloseHandler( heartbeatUnsub: (() => void) | null; transcriptUnsub: (() => void) | null; lifecycleUnsub: (() => void) | null; + taskUnsub: (() => void) | null; getPendingReplyCount?: () => number; clients: Set<{ socket: { close: (code: number, reason: string) => void } }>; configReloader: { stop: () => Promise }; @@ -909,6 +910,9 @@ export function createGatewayCloseHandler( if (params.lifecycleUnsub) { await shutdownStep("lifecycle-unsub", () => params.lifecycleUnsub!(), warnings); } + if (params.taskUnsub) { + await shutdownStep("task-unsub", () => params.taskUnsub!(), warnings); + } params.chatRunState.clear(); let clientCloseFailures = 0; for (const c of params.clients) { diff --git a/src/gateway/server-methods-list.ts b/src/gateway/server-methods-list.ts index 843ab8358ef..b43cd085333 100644 --- a/src/gateway/server-methods-list.ts +++ b/src/gateway/server-methods-list.ts @@ -52,6 +52,7 @@ export const GATEWAY_EVENTS = [ "health", "heartbeat", "cron", + "task", "node.pair.requested", "node.pair.resolved", "node.invoke.request", diff --git a/src/gateway/server-methods/task-summary.ts b/src/gateway/server-methods/task-summary.ts new file mode 100644 index 00000000000..d069ab33146 --- /dev/null +++ b/src/gateway/server-methods/task-summary.ts @@ -0,0 +1,70 @@ +// Public task summaries keep task-registry internals and unbounded status text +// out of gateway responses and events. +import type { TaskSummary } from "../../../packages/gateway-protocol/src/index.js"; +import type { TaskRecord, TaskStatus } from "../../tasks/task-registry.types.js"; +import { + TASK_STATUS_DETAIL_MAX_CHARS, + formatTaskStatusTitle, + sanitizeTaskStatusText, +} from "../../tasks/task-status.js"; + +type TaskLedgerStatus = TaskSummary["status"]; + +export const TASK_STATUS_TO_LEDGER_STATUS: Record = { + queued: "queued", + running: "running", + succeeded: "completed", + failed: "failed", + timed_out: "timed_out", + cancelled: "cancelled", + lost: "failed", +}; + +export type TaskEventPayload = + | { action: "upserted"; task: TaskSummary } + | { action: "deleted"; taskId: string } + | { action: "restored" }; + +export function taskUpdatedAt(task: TaskRecord): number { + return task.lastEventAt ?? task.endedAt ?? task.startedAt ?? task.createdAt; +} + +function sanitizeOptionalTaskText( + value: unknown, + opts?: { errorContext?: boolean }, +): string | undefined { + const sanitized = sanitizeTaskStatusText(value, { + errorContext: opts?.errorContext, + maxChars: TASK_STATUS_DETAIL_MAX_CHARS, + }); + return sanitized || undefined; +} + +export function mapTaskSummary(task: TaskRecord): TaskSummary { + const progressSummary = sanitizeOptionalTaskText(task.progressSummary); + const terminalSummary = sanitizeOptionalTaskText(task.terminalSummary, { errorContext: true }); + const error = sanitizeOptionalTaskText(task.error, { errorContext: true }); + return { + id: task.taskId, + taskId: task.taskId, + kind: task.taskKind ?? task.runtime, + runtime: task.runtime, + status: TASK_STATUS_TO_LEDGER_STATUS[task.status], + title: formatTaskStatusTitle(task), + ...(task.agentId ? { agentId: task.agentId } : {}), + sessionKey: task.requesterSessionKey, + ...(task.childSessionKey ? { childSessionKey: task.childSessionKey } : {}), + ownerKey: task.ownerKey, + ...(task.runId ? { runId: task.runId } : {}), + ...(task.parentFlowId ? { flowId: task.parentFlowId } : {}), + ...(task.parentTaskId ? { parentTaskId: task.parentTaskId } : {}), + ...(task.sourceId ? { sourceId: task.sourceId } : {}), + createdAt: task.createdAt, + updatedAt: taskUpdatedAt(task), + ...(task.startedAt !== undefined ? { startedAt: task.startedAt } : {}), + ...(task.endedAt !== undefined ? { endedAt: task.endedAt } : {}), + ...(progressSummary ? { progressSummary } : {}), + ...(terminalSummary ? { terminalSummary } : {}), + ...(error ? { error } : {}), + }; +} diff --git a/src/gateway/server-methods/tasks.test.ts b/src/gateway/server-methods/tasks.test.ts index 67c14c36f25..dda00478ca7 100644 --- a/src/gateway/server-methods/tasks.test.ts +++ b/src/gateway/server-methods/tasks.test.ts @@ -168,6 +168,39 @@ describe("tasks gateway handlers", () => { expect(listedTask?.runId).toBe("run-running"); }); + it("orders the ledger by last activity, not creation time", async () => { + // The registry lists newest-created first; the wire must page by last + // activity so an old task that just finished is not hidden behind + // newer-created records. + const base = Date.now(); + const oldButJustFinished = createTaskRecord({ + runtime: "subagent", + requesterSessionKey: "agent:main:main", + ownerKey: "agent:main:main", + scopeKind: "session", + task: "Old long-running task", + status: "succeeded", + deliveryStatus: "not_applicable", + lastEventAt: base + 60_000, + }); + const newerQuietTask = createTaskRecord({ + runtime: "cli", + requesterSessionKey: "agent:main:main", + ownerKey: "agent:main:main", + scopeKind: "session", + task: "Newer quiet task", + status: "succeeded", + deliveryStatus: "not_applicable", + lastEventAt: base + 1_000, + }); + + const { payload } = await runTaskHandler("tasks.list", {}); + const ids = payload?.tasks?.map((task) => task.id); + expect(ids?.indexOf(oldButJustFinished.taskId)).toBeLessThan( + ids?.indexOf(newerQuietTask.taskId) ?? -1, + ); + }); + it("treats explicit task agentId as authoritative over the session-key fallback", async () => { // Cross-agent subagent task: the registry derives agentId=worker from the // child session key, while owner/requester keys belong to main. tasks.list diff --git a/src/gateway/server-methods/tasks.ts b/src/gateway/server-methods/tasks.ts index 7c9f3c6b706..1634876fbad 100644 --- a/src/gateway/server-methods/tasks.ts +++ b/src/gateway/server-methods/tasks.ts @@ -15,11 +15,7 @@ import { parseAgentSessionKey } from "../../routing/session-key.js"; import { cancelDetachedTaskRunById } from "../../tasks/detached-task-runtime.js"; import { getTaskById, listTaskRecords } from "../../tasks/runtime-internal.js"; import type { TaskRecord, TaskStatus } from "../../tasks/task-registry.types.js"; -import { - TASK_STATUS_DETAIL_MAX_CHARS, - formatTaskStatusTitle, - sanitizeTaskStatusText, -} from "../../tasks/task-status.js"; +import { mapTaskSummary, taskUpdatedAt } from "./task-summary.js"; import type { GatewayRequestHandlers } from "./types.js"; const DEFAULT_TASKS_LIST_LIMIT = 100; @@ -27,18 +23,6 @@ const MAX_TASKS_LIST_LIMIT = 500; type TaskLedgerStatus = TaskSummary["status"]; -// Gateway task APIs preserve the older ledger status vocabulary while the -// runtime registry tracks finer-grained task states such as `lost`. -const TASK_STATUS_TO_LEDGER_STATUS: Record = { - queued: "queued", - running: "running", - succeeded: "completed", - failed: "failed", - timed_out: "timed_out", - cancelled: "cancelled", - lost: "failed", -}; - const LEDGER_STATUS_TO_TASK_STATUSES: Record = { queued: ["queued"], running: ["running"], @@ -48,52 +32,6 @@ const LEDGER_STATUS_TO_TASK_STATUSES: Record = { cancelled: ["cancelled"], }; -function taskUpdatedAt(task: TaskRecord): number { - return task.lastEventAt ?? task.endedAt ?? task.startedAt ?? task.createdAt; -} - -// Status text can originate from providers, shells, and subprocesses. Keep the -// public task shape bounded before it reaches control-plane clients. -function sanitizeOptionalTaskText( - value: unknown, - opts?: { errorContext?: boolean }, -): string | undefined { - const sanitized = sanitizeTaskStatusText(value, { - errorContext: opts?.errorContext, - maxChars: TASK_STATUS_DETAIL_MAX_CHARS, - }); - return sanitized || undefined; -} - -function mapTaskSummary(task: TaskRecord): TaskSummary { - const progressSummary = sanitizeOptionalTaskText(task.progressSummary); - const terminalSummary = sanitizeOptionalTaskText(task.terminalSummary, { errorContext: true }); - const error = sanitizeOptionalTaskText(task.error, { errorContext: true }); - return { - id: task.taskId, - taskId: task.taskId, - kind: task.taskKind ?? task.runtime, - runtime: task.runtime, - status: TASK_STATUS_TO_LEDGER_STATUS[task.status], - title: formatTaskStatusTitle(task), - ...(task.agentId ? { agentId: task.agentId } : {}), - sessionKey: task.requesterSessionKey, - ...(task.childSessionKey ? { childSessionKey: task.childSessionKey } : {}), - ownerKey: task.ownerKey, - ...(task.runId ? { runId: task.runId } : {}), - ...(task.parentFlowId ? { flowId: task.parentFlowId } : {}), - ...(task.parentTaskId ? { parentTaskId: task.parentTaskId } : {}), - ...(task.sourceId ? { sourceId: task.sourceId } : {}), - createdAt: task.createdAt, - updatedAt: taskUpdatedAt(task), - ...(task.startedAt !== undefined ? { startedAt: task.startedAt } : {}), - ...(task.endedAt !== undefined ? { endedAt: task.endedAt } : {}), - ...(progressSummary ? { progressSummary } : {}), - ...(terminalSummary ? { terminalSummary } : {}), - ...(error ? { error } : {}), - }; -} - function normalizeTaskStatusFilter(status: TasksListParams["status"]): Set | null { if (!status) { return null; @@ -171,12 +109,25 @@ export const tasksHandlers: GatewayRequestHandlers = { } const statusFilter = normalizeTaskStatusFilter(params.status); const limit = Math.min(params.limit ?? DEFAULT_TASKS_LIST_LIMIT, MAX_TASKS_LIST_LIMIT); - const filtered = listTaskRecords().filter((task) => { - if (statusFilter && !statusFilter.has(task.status)) { - return false; - } - return taskMatchesAgent(task, params.agentId) && taskMatchesSession(task, params.sessionKey); - }); + // The registry lists newest-created first; the ledger view pages by last + // activity so an old long-running task that just finished still surfaces + // on the first page instead of hiding behind newer-created records. + const filtered = listTaskRecords() + .filter((task) => { + if (statusFilter && !statusFilter.has(task.status)) { + return false; + } + return ( + taskMatchesAgent(task, params.agentId) && taskMatchesSession(task, params.sessionKey) + ); + }) + .toSorted((left, right) => { + const updatedDiff = taskUpdatedAt(right) - taskUpdatedAt(left); + if (updatedDiff !== 0) { + return updatedDiff; + } + return left.taskId < right.taskId ? -1 : left.taskId > right.taskId ? 1 : 0; + }); const page = filtered.slice(cursor, cursor + limit); const nextOffset = cursor + page.length; respond(true, { diff --git a/src/gateway/server-runtime-handles.ts b/src/gateway/server-runtime-handles.ts index e605600b418..571972789a7 100644 --- a/src/gateway/server-runtime-handles.ts +++ b/src/gateway/server-runtime-handles.ts @@ -40,6 +40,7 @@ export type GatewayServerMutableState = { heartbeatUnsub: (() => void) | null; transcriptUnsub: (() => void) | null; lifecycleUnsub: (() => void) | null; + taskUnsub: (() => void) | null; }; /** Creates gateway mutable state with inert handles that are safe to stop before startup finishes. */ @@ -77,5 +78,6 @@ export function createGatewayServerMutableState(): GatewayServerMutableState { heartbeatUnsub: null as (() => void) | null, transcriptUnsub: null as (() => void) | null, lifecycleUnsub: null as (() => void) | null, + taskUnsub: null as (() => void) | null, }; } diff --git a/src/gateway/server-runtime-subscriptions.test.ts b/src/gateway/server-runtime-subscriptions.test.ts index 0dd2033601a..c21e2f95a8c 100644 --- a/src/gateway/server-runtime-subscriptions.test.ts +++ b/src/gateway/server-runtime-subscriptions.test.ts @@ -7,12 +7,16 @@ import { emitInternalSessionTranscriptUpdate, type InternalSessionTranscriptUpdate, } from "../sessions/transcript-events.js"; +import { createTaskRecord, resetTaskRegistryForTests } from "../tasks/task-registry.js"; +import { getTaskRegistryObservers } from "../tasks/task-registry.store.js"; +import { installInMemoryTaskRegistryRuntime } from "../test-utils/task-registry-runtime.js"; import { createChatRunState, createSessionEventSubscriberRegistry, createSessionMessageSubscriberRegistry, createToolEventRecipientRegistry, } from "./server-chat-state.js"; +import type { TaskEventPayload } from "./server-methods/task-summary.js"; const warn = vi.fn(); const mockLog: SubsystemLogger = { @@ -95,6 +99,7 @@ describe("startGatewayEventSubscriptions", () => { auditTestState.enabled = true; auditTestState.created = 0; auditTestState.stopped = 0; + installInMemoryTaskRegistryRuntime(); }); afterEach(async () => { @@ -102,7 +107,9 @@ describe("startGatewayEventSubscriptions", () => { unsubs?.heartbeatUnsub(); unsubs?.transcriptUnsub(); unsubs?.lifecycleUnsub(); + void unsubs?.taskUnsub(); resetAgentEventsForTest(); + resetTaskRegistryForTests({ persist: false }); }); it("records audit events by default and stops the recorder on unsubscribe", async () => { @@ -161,4 +168,115 @@ describe("startGatewayEventSubscriptions", () => { expect.objectContaining({ sessionKey: "agent:main:main" }), ); }); + + it("broadcasts bounded public task summaries with ledger statuses", async () => { + const broadcast = vi.fn(); + unsubs = startGatewayEventSubscriptions({ ...createParams(), broadcast }); + await vi.waitFor(() => expect(getTaskRegistryObservers()).not.toBeNull()); + + const completed = createTaskRecord({ + runtime: "subagent", + requesterSessionKey: "agent:main:main", + ownerKey: "agent:main:main", + scopeKind: "session", + task: "Completed task", + status: "succeeded", + deliveryStatus: "not_applicable", + notifyPolicy: "silent", + terminalSummary: "x".repeat(10_000), + }); + const lost = createTaskRecord({ + runtime: "cli", + requesterSessionKey: "agent:main:main", + ownerKey: "agent:main:main", + scopeKind: "session", + task: "Lost task", + status: "lost", + deliveryStatus: "not_applicable", + notifyPolicy: "silent", + }); + + if (!completed || !lost) { + throw new Error("expected task records to be created"); + } + const taskUpsertsById = new Map( + broadcast.mock.calls + .filter(([event]) => event === "task") + .map(([, payload]) => payload as TaskEventPayload) + .filter( + (payload): payload is Extract => + payload.action === "upserted", + ) + .map((payload) => [payload.task.id, payload.task]), + ); + expect(broadcast).toHaveBeenCalledWith("task", expect.anything(), { dropIfSlow: true }); + // Runtime registry statuses translate to the public ledger vocabulary. + expect(taskUpsertsById.get(completed.taskId)?.status).toBe("completed"); + expect(taskUpsertsById.get(lost.taskId)?.status).toBe("failed"); + // Unbounded status text from providers/shells must be truncated on the wire. + const wireTerminalSummary = taskUpsertsById.get(completed.taskId)?.terminalSummary; + expect(wireTerminalSummary).toBeTruthy(); + expect(wireTerminalSummary?.length ?? 0).toBeLessThan(10_000); + + void unsubs?.taskUnsub(); + await vi.waitFor(() => expect(getTaskRegistryObservers()).toBeNull()); + broadcast.mockClear(); + createTaskRecord({ + runtime: "cli", + requesterSessionKey: "agent:main:main", + ownerKey: "agent:main:main", + scopeKind: "session", + task: "After dispose", + status: "queued", + deliveryStatus: "not_applicable", + notifyPolicy: "silent", + }); + expect(broadcast).not.toHaveBeenCalled(); + }); + + it("keeps a replacement gateway's task observer when a stale unsub runs late", async () => { + const staleBroadcast = vi.fn(); + const staleSubs = startGatewayEventSubscriptions({ + ...createParams(), + broadcast: staleBroadcast, + }); + await vi.waitFor(() => expect(getTaskRegistryObservers()).not.toBeNull()); + const staleObservers = getTaskRegistryObservers(); + + const replacementBroadcast = vi.fn(); + unsubs = startGatewayEventSubscriptions({ + ...createParams(), + broadcast: replacementBroadcast, + }); + await vi.waitFor(() => { + const current = getTaskRegistryObservers(); + expect(current).not.toBeNull(); + expect(current).not.toBe(staleObservers); + }); + + // The stale dispose must not clear the replacement's observer slot. + await staleSubs.taskUnsub(); + await staleSubs.agentUnsub(); + staleSubs.heartbeatUnsub(); + staleSubs.transcriptUnsub(); + staleSubs.lifecycleUnsub(); + expect(getTaskRegistryObservers()).not.toBeNull(); + + createTaskRecord({ + runtime: "cli", + requesterSessionKey: "agent:main:main", + ownerKey: "agent:main:main", + scopeKind: "session", + task: "After stale dispose", + status: "queued", + deliveryStatus: "not_applicable", + notifyPolicy: "silent", + }); + expect(replacementBroadcast).toHaveBeenCalledWith("task", expect.anything(), { + dropIfSlow: true, + }); + expect(staleBroadcast).not.toHaveBeenCalledWith("task", expect.anything(), { + dropIfSlow: true, + }); + }); }); diff --git a/src/gateway/server-runtime-subscriptions.ts b/src/gateway/server-runtime-subscriptions.ts index 5a275522e83..d42339810a8 100644 --- a/src/gateway/server-runtime-subscriptions.ts +++ b/src/gateway/server-runtime-subscriptions.ts @@ -10,6 +10,7 @@ import type { SubsystemLogger } from "../logging/subsystem.js"; import { onSessionLifecycleEvent } from "../sessions/session-lifecycle-events.js"; import { onInternalSessionTranscriptUpdate } from "../sessions/transcript-events.js"; import { createLazyPromise } from "../shared/lazy-runtime.js"; +import type { TaskRegistryObserverEvent } from "../tasks/task-registry.store.js"; import { type ChatAbortControllerEntry, removeChatAbortControllerEntry, @@ -22,6 +23,7 @@ import type { ToolEventRecipientRegistry, } from "./server-chat-state.js"; import { resolveVisibleActiveSessionRunState } from "./server-methods/session-active-runs.js"; +import { mapTaskSummary, type TaskEventPayload } from "./server-methods/task-summary.js"; function dispatchEventHandler(params: { loadHandler: () => Promise<(event: TEvent) => unknown>; @@ -316,10 +318,53 @@ export function startGatewayEventSubscriptions(params: { }); }); + let taskObserverDisposed = false; + const taskObservers = { + onEvent: (event: TaskRegistryObserverEvent) => { + let payload: TaskEventPayload; + switch (event.kind) { + case "upserted": + payload = { action: "upserted", task: mapTaskSummary(event.task) }; + break; + case "deleted": + payload = { action: "deleted", taskId: event.taskId }; + break; + case "restored": + payload = { action: "restored" }; + break; + } + params.broadcast("task", payload, { dropIfSlow: true }); + }, + }; + const taskObserverRuntimePromise = import("../tasks/task-registry.store.js").then((module) => { + if (!taskObserverDisposed) { + module.configureTaskRegistryRuntime({ observers: taskObservers }); + } + return module; + }); + void taskObserverRuntimePromise.catch((error: unknown) => { + params.log.warn("Task registry observer registration failed", { error }); + }); + // The observer slot is a process-wide singleton. Cleanup returns its promise + // so shutdown can await it, and only clears the slot when it still holds + // this subscription's observer — a replacement gateway may have registered + // its own observer before a stale deferred dispose runs. + const taskUnsub = () => { + taskObserverDisposed = true; + return taskObserverRuntimePromise + .then((module) => { + if (module.getTaskRegistryObservers() === taskObservers) { + module.configureTaskRegistryRuntime({ observers: null }); + } + }) + .catch(() => undefined); + }; + return { agentUnsub, heartbeatUnsub, transcriptUnsub, lifecycleUnsub, + taskUnsub, }; } diff --git a/src/gateway/server-startup-post-attach.test.ts b/src/gateway/server-startup-post-attach.test.ts index ab7797e601a..f85d97fd728 100644 --- a/src/gateway/server-startup-post-attach.test.ts +++ b/src/gateway/server-startup-post-attach.test.ts @@ -1915,6 +1915,7 @@ describe("startGatewayPostAttachRuntime", () => { mediaCleanup: null, worktreeCleanup: null, agentUnsub: null, + taskUnsub: null, heartbeatUnsub: null, transcriptUnsub: null, lifecycleUnsub: null, diff --git a/src/gateway/server.impl.ts b/src/gateway/server.impl.ts index 102712b2944..5743862916d 100644 --- a/src/gateway/server.impl.ts +++ b/src/gateway/server.impl.ts @@ -1070,6 +1070,7 @@ export async function startGatewayServer( heartbeatUnsub: runtimeState.heartbeatUnsub, transcriptUnsub: runtimeState.transcriptUnsub, lifecycleUnsub: runtimeState.lifecycleUnsub, + taskUnsub: runtimeState.taskUnsub, chatRunState, chatAbortControllers, chatQueuedTurns, diff --git a/ui/src/app-navigation.test.ts b/ui/src/app-navigation.test.ts index 532efc18edd..96ee081b92b 100644 --- a/ui/src/app-navigation.test.ts +++ b/ui/src/app-navigation.test.ts @@ -42,6 +42,7 @@ describe("navigationIconForRoute", () => { sessions: "fileText", usage: "barChart", cron: "loader", + tasks: "loader", agents: "folder", skills: "zap", "skill-workshop": "wrench", @@ -81,6 +82,7 @@ describe("titleForRoute", () => { sessions: "Sessions", usage: "Usage", cron: "Cron Jobs", + tasks: "Tasks", agents: "Agents", skills: "Skills", "skill-workshop": "Skill Workshop", @@ -114,6 +116,7 @@ describe("subtitleForRoute", () => { sessions: "Active sessions and defaults.", usage: "API usage and costs.", cron: "Wakeups and recurring runs.", + tasks: "Background tasks: subagents, cron runs, CLI.", agents: "Workspaces, tools, identities.", skills: "Skills and API keys.", "skill-workshop": "Review, refine, and apply proposals before they become live skills.", diff --git a/ui/src/app-navigation.ts b/ui/src/app-navigation.ts index b5bad654d1b..baad3ec3f15 100644 --- a/ui/src/app-navigation.ts +++ b/ui/src/app-navigation.ts @@ -21,6 +21,7 @@ export const SIDEBAR_NAV_ROUTES = [ "sessions", "usage", "cron", + "tasks", "agents", "skills", "skill-workshop", @@ -86,6 +87,7 @@ const NAVIGATION_ICONS: NavigationItem = { sessions: "fileText", usage: "barChart", cron: "loader", + tasks: "loader", skills: "zap", "skill-workshop": "wrench", nodes: "monitor", @@ -170,6 +172,7 @@ const NAVIGATION_COPY: Record & Pick): TaskSummary { + return { + taskId: overrides.id, + updatedAt: 100, + ...overrides, + }; +} + +describe("tasks page data", () => { + it("normalizes valid task summaries and rejects invalid statuses", () => { + expect( + normalizeTaskSummary({ + id: " task-1 ", + status: "running", + runtime: "subagent", + title: " Build release ", + updatedAt: "2026-07-05T12:00:00.000Z", + }), + ).toEqual({ + id: "task-1", + taskId: "task-1", + status: "running", + runtime: "subagent", + title: "Build release", + updatedAt: "2026-07-05T12:00:00.000Z", + }); + expect(normalizeTaskSummary({ id: "task-2", status: "lost" })).toBeNull(); + expect(normalizeTasksListResult({ tasks: "not-an-array" })).toBeNull(); + }); + + it("sorts by updated time descending with an id tiebreak", () => { + const sorted = sortTasks([ + task({ id: "b", status: "queued", updatedAt: 200 }), + task({ id: "c", status: "completed", updatedAt: 300 }), + task({ id: "a", status: "running", updatedAt: 200 }), + ]); + expect(sorted.map((entry) => entry.id)).toEqual(["c", "a", "b"]); + }); + + it("partitions active tasks and caps recent terminal tasks at 50", () => { + const terminals = Array.from({ length: 55 }, (_, index) => + task({ id: `terminal-${index}`, status: "completed", updatedAt: index }), + ); + const result = partitionTasks([ + task({ id: "running", status: "running", updatedAt: 1000 }), + task({ id: "queued", status: "queued", updatedAt: 999 }), + ...terminals, + ]); + expect(result.active.map((entry) => entry.id)).toEqual(["running", "queued"]); + expect(result.recent).toHaveLength(50); + expect(result.recent[0]?.id).toBe("terminal-54"); + }); + + it("merges task lists by id with later lists winning", () => { + const recentPage = [ + task({ id: "new-terminal", status: "completed", updatedAt: 900 }), + task({ id: "shared", status: "running", updatedAt: 800 }), + ]; + const activePage = [ + task({ id: "shared", status: "running", updatedAt: 850 }), + task({ id: "old-running", status: "running", updatedAt: 10 }), + ]; + const merged = mergeTaskLists(recentPage, activePage); + expect(merged.map((entry) => entry.id)).toEqual(["new-terminal", "shared", "old-running"]); + expect(merged.find((entry) => entry.id === "shared")?.updatedAt).toBe(850); + }); + + it("normalizes cancel results including refusals with reasons", () => { + expect( + normalizeTasksCancelResult({ + found: true, + cancelled: false, + reason: "task already finished", + task: { id: "task-1", taskId: "task-1", status: "completed" }, + }), + ).toEqual({ + found: true, + cancelled: false, + reason: "task already finished", + task: { id: "task-1", taskId: "task-1", status: "completed" }, + }); + expect(normalizeTasksCancelResult({ found: true, cancelled: true })).toEqual({ + found: true, + cancelled: true, + }); + expect(normalizeTasksCancelResult({ found: true })).toBeNull(); + expect(normalizeTasksCancelResult("nope")).toBeNull(); + }); + + it("merges upserts, applies deletes, and requests refetches for restored events", () => { + const initial = [task({ id: "task-1", status: "running", updatedAt: 100 })]; + const completed = task({ id: "task-1", status: "completed", updatedAt: 200 }); + + const upserted = applyTaskEvent(initial, { action: "upserted", task: completed }); + expect(upserted).toEqual({ tasks: [completed], refetch: false }); + + const deleted = applyTaskEvent(upserted.tasks, { action: "deleted", taskId: "task-1" }); + expect(deleted).toEqual({ tasks: [], refetch: false }); + expect(applyTaskEvent(initial, { action: "restored" })).toEqual({ + tasks: initial, + refetch: true, + }); + expect(applyTaskEvent(initial, { action: "upserted", task: { id: "broken" } })).toEqual({ + tasks: initial, + refetch: true, + }); + }); +}); diff --git a/ui/src/pages/tasks/data.ts b/ui/src/pages/tasks/data.ts new file mode 100644 index 00000000000..ddfa8ae54f8 --- /dev/null +++ b/ui/src/pages/tasks/data.ts @@ -0,0 +1,231 @@ +export type TaskStatus = "queued" | "running" | "completed" | "failed" | "cancelled" | "timed_out"; + +export type TaskRuntime = "subagent" | "cron" | "acp" | "cli"; +export type TaskTimestamp = number | string; + +export type TaskSummary = { + id: string; + taskId: string; + status: TaskStatus; + kind?: string; + runtime?: TaskRuntime; + title?: string; + agentId?: string; + sessionKey?: string; + childSessionKey?: string; + createdAt?: TaskTimestamp; + updatedAt?: TaskTimestamp; + startedAt?: TaskTimestamp; + endedAt?: TaskTimestamp; + progressSummary?: string; + terminalSummary?: string; + error?: string; +}; + +export type TaskEventPayload = + | { action: "upserted"; task: TaskSummary } + | { action: "deleted"; taskId: string } + | { action: "restored" }; + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function optionalString(value: unknown): string | undefined { + return typeof value === "string" && value.trim() ? value.trim() : undefined; +} + +function normalizeTaskStatus(value: unknown): TaskStatus | null { + switch (value) { + case "queued": + case "running": + case "completed": + case "failed": + case "cancelled": + case "timed_out": + return value; + default: + return null; + } +} + +function normalizeTaskRuntime(value: unknown): TaskRuntime | undefined { + switch (value) { + case "subagent": + case "cron": + case "acp": + case "cli": + return value; + default: + return undefined; + } +} + +function normalizeTimestamp(value: unknown): TaskTimestamp | undefined { + if (typeof value === "number" && Number.isFinite(value) && value >= 0) { + return value; + } + if (typeof value === "string" && Number.isFinite(Date.parse(value))) { + return value; + } + return undefined; +} + +export function normalizeTaskSummary(value: unknown): TaskSummary | null { + if (!isRecord(value)) { + return null; + } + const id = optionalString(value.id); + const taskId = optionalString(value.taskId) ?? id; + const status = normalizeTaskStatus(value.status); + if (!id || !taskId || !status) { + return null; + } + const runtime = normalizeTaskRuntime(value.runtime); + const kind = optionalString(value.kind); + const title = optionalString(value.title); + const agentId = optionalString(value.agentId); + const sessionKey = optionalString(value.sessionKey); + const childSessionKey = optionalString(value.childSessionKey); + const createdAt = normalizeTimestamp(value.createdAt); + const updatedAt = normalizeTimestamp(value.updatedAt); + const startedAt = normalizeTimestamp(value.startedAt); + const endedAt = normalizeTimestamp(value.endedAt); + const progressSummary = optionalString(value.progressSummary); + const terminalSummary = optionalString(value.terminalSummary); + const error = optionalString(value.error); + return { + id, + taskId, + status, + ...(kind ? { kind } : {}), + ...(runtime ? { runtime } : {}), + ...(title ? { title } : {}), + ...(agentId ? { agentId } : {}), + ...(sessionKey ? { sessionKey } : {}), + ...(childSessionKey ? { childSessionKey } : {}), + ...(createdAt !== undefined ? { createdAt } : {}), + ...(updatedAt !== undefined ? { updatedAt } : {}), + ...(startedAt !== undefined ? { startedAt } : {}), + ...(endedAt !== undefined ? { endedAt } : {}), + ...(progressSummary ? { progressSummary } : {}), + ...(terminalSummary ? { terminalSummary } : {}), + ...(error ? { error } : {}), + }; +} + +export function taskTimestampMs(value: TaskTimestamp | undefined): number { + if (typeof value === "number") { + return value; + } + if (typeof value === "string") { + const parsed = Date.parse(value); + return Number.isFinite(parsed) ? parsed : 0; + } + return 0; +} + +export function sortTasks(tasks: readonly TaskSummary[]): TaskSummary[] { + return tasks.toSorted((left, right) => { + const timeDelta = taskTimestampMs(right.updatedAt) - taskTimestampMs(left.updatedAt); + if (timeDelta !== 0) { + return timeDelta; + } + return left.id < right.id ? -1 : left.id > right.id ? 1 : 0; + }); +} + +export function partitionTasks(tasks: readonly TaskSummary[]): { + active: TaskSummary[]; + recent: TaskSummary[]; +} { + const sorted = sortTasks(tasks); + return { + active: sorted.filter((task) => task.status === "queued" || task.status === "running"), + recent: sorted + .filter((task) => task.status !== "queued" && task.status !== "running") + .slice(0, 50), + }; +} + +export function normalizeTasksListResult(value: unknown): TaskSummary[] | null { + if (!isRecord(value) || !Array.isArray(value.tasks)) { + return null; + } + return sortTasks( + value.tasks.map(normalizeTaskSummary).filter((task): task is TaskSummary => task !== null), + ); +} + +// The ledger pages newest-first, so one page can hide long-running tasks behind +// newer terminal records; callers fetch active tasks separately and merge here. +export function mergeTaskLists(...lists: readonly (readonly TaskSummary[])[]): TaskSummary[] { + const byId = new Map(); + for (const list of lists) { + for (const task of list) { + byId.set(task.id, task); + } + } + return sortTasks([...byId.values()]); +} + +export type TasksCancelResult = { + found: boolean; + cancelled: boolean; + reason?: string; + task?: TaskSummary; +}; + +// Cancellation refusals (already-terminal, missing handle, stale id) arrive as +// successful responses with cancelled=false + reason, not thrown errors. +export function normalizeTasksCancelResult(value: unknown): TasksCancelResult | null { + if (!isRecord(value) || typeof value.cancelled !== "boolean") { + return null; + } + const reason = optionalString(value.reason); + const task = normalizeTaskSummary(value.task); + return { + found: value.found === true, + cancelled: value.cancelled, + ...(reason ? { reason } : {}), + ...(task ? { task } : {}), + }; +} + +export function normalizeTaskEventPayload(value: unknown): TaskEventPayload | null { + if (!isRecord(value)) { + return null; + } + if (value.action === "restored") { + return { action: "restored" }; + } + if (value.action === "deleted") { + const taskId = optionalString(value.taskId); + return taskId ? { action: "deleted", taskId } : null; + } + if (value.action === "upserted") { + const task = normalizeTaskSummary(value.task); + return task ? { action: "upserted", task } : null; + } + return null; +} + +export function applyTaskEvent( + tasks: readonly TaskSummary[], + value: unknown, +): { tasks: TaskSummary[]; refetch: boolean } { + const event = normalizeTaskEventPayload(value); + if (!event || event.action === "restored") { + return { tasks: [...tasks], refetch: true }; + } + if (event.action === "deleted") { + return { + tasks: sortTasks(tasks.filter((task) => task.id !== event.taskId)), + refetch: false, + }; + } + return { + tasks: sortTasks([event.task, ...tasks.filter((task) => task.id !== event.task.id)]), + refetch: false, + }; +} diff --git a/ui/src/pages/tasks/route.ts b/ui/src/pages/tasks/route.ts new file mode 100644 index 00000000000..1c31294c6f6 --- /dev/null +++ b/ui/src/pages/tasks/route.ts @@ -0,0 +1,12 @@ +import { definePage } from "@openclaw/uirouter"; +import { html } from "lit"; + +export const page = definePage({ + id: "tasks", + path: "/tasks", + component: () => + import("./tasks-page.ts").then(() => ({ + header: true, + render: () => html``, + })), +}); diff --git a/ui/src/pages/tasks/tasks-page.ts b/ui/src/pages/tasks/tasks-page.ts new file mode 100644 index 00000000000..65a0a993d82 --- /dev/null +++ b/ui/src/pages/tasks/tasks-page.ts @@ -0,0 +1,191 @@ +import { consume } from "@lit/context"; +import { html, LitElement } from "lit"; +import { state } from "lit/decorators.js"; +import type { GatewayBrowserClient } from "../../api/gateway.ts"; +import { subtitleForRoute, titleForRoute } from "../../app-navigation.ts"; +import { applicationContext, type ApplicationContext } from "../../app/context.ts"; +import { hasOperatorWriteAccess } from "../../app/operator-access.ts"; +import { t } from "../../i18n/index.ts"; +import { searchForSession } from "../../lib/sessions/index.ts"; +import { + applyTaskEvent, + mergeTaskLists, + normalizeTasksCancelResult, + normalizeTasksListResult, + type TaskSummary, +} from "./data.ts"; +import { renderTasks } from "./view.ts"; + +function formatTaskError(error: unknown, fallback: string): string { + if (error instanceof Error && error.message.trim()) { + return error.message.trim(); + } + return typeof error === "string" && error.trim() ? error.trim() : fallback; +} + +export class TasksPage extends LitElement { + override createRenderRoot() { + return this; + } + + @consume({ context: applicationContext, subscribe: false }) + private context!: ApplicationContext; + + @state() private tasks: TaskSummary[] = []; + @state() private connected = false; + @state() private loading = false; + @state() private error: string | null = null; + @state() private cancellingTaskIds = new Set(); + + private client: GatewayBrowserClient | null = null; + private loadGeneration = 0; + private stopGatewaySubscription?: () => void; + private stopGatewayEvents?: () => void; + + override connectedCallback() { + super.connectedCallback(); + this.syncGatewayState(); + this.stopGatewaySubscription = this.context.gateway.subscribe(() => { + const wasConnected = this.connected; + const previousClient = this.client; + this.syncGatewayState(); + if (this.connected && (this.client !== previousClient || !wasConnected)) { + void this.refreshTasks(); + } + }); + this.stopGatewayEvents = this.context.gateway.subscribeEvents((event) => { + if (event.event !== "task") { + return; + } + const result = applyTaskEvent(this.tasks, event.payload); + if (result.refetch) { + void this.refreshTasks(); + return; + } + this.tasks = result.tasks; + }); + if (this.connected) { + void this.refreshTasks(); + } + } + + override disconnectedCallback() { + this.loadGeneration += 1; + this.stopGatewaySubscription?.(); + this.stopGatewaySubscription = undefined; + this.stopGatewayEvents?.(); + this.stopGatewayEvents = undefined; + super.disconnectedCallback(); + } + + private syncGatewayState() { + const gateway = this.context.gateway.snapshot; + if (this.client !== gateway.client) { + this.loadGeneration += 1; + this.client = gateway.client; + this.tasks = []; + this.loading = false; + this.error = null; + this.cancellingTaskIds = new Set(); + } + this.connected = gateway.connected; + } + + private async refreshTasks() { + const client = this.client; + if (!this.connected || !client) { + return; + } + const generation = ++this.loadGeneration; + this.loading = true; + this.error = null; + try { + // Active tasks need their own query: the ledger pages newest-first, so a + // long-running task can hide behind newer terminal records on page one. + const [activePayload, recentPayload] = await Promise.all([ + client.request("tasks.list", { status: ["queued", "running"], limit: 500 }), + client.request("tasks.list", { limit: 200 }), + ]); + const active = normalizeTasksListResult(activePayload); + const recent = normalizeTasksListResult(recentPayload); + if (!active || !recent) { + throw new Error(t("tasksPage.invalidResponse")); + } + const tasks = mergeTaskLists(recent, active); + if (generation === this.loadGeneration && client === this.client) { + this.tasks = tasks; + } + } catch (error) { + if (generation === this.loadGeneration && client === this.client) { + this.error = formatTaskError(error, t("tasksPage.loadFailed")); + } + } finally { + if (generation === this.loadGeneration && client === this.client) { + this.loading = false; + } + } + } + + private async cancelTask(taskId: string) { + const client = this.client; + if (!this.connected || !client || this.cancellingTaskIds.has(taskId)) { + return; + } + this.cancellingTaskIds = new Set([...this.cancellingTaskIds, taskId]); + this.error = null; + try { + const payload = await client.request("tasks.cancel", { taskId }); + const result = normalizeTasksCancelResult(payload); + if (result?.task) { + this.tasks = applyTaskEvent(this.tasks, { action: "upserted", task: result.task }).tasks; + } + // Refusals (already terminal, stale id, no cancellation handle) are + // successful responses with cancelled=false; surface them like errors. + if (!result?.cancelled) { + this.error = result?.reason?.trim() || t("tasksPage.cancelFailed"); + } + } catch (error) { + this.error = formatTaskError(error, t("tasksPage.cancelFailed")); + } finally { + const next = new Set(this.cancellingTaskIds); + next.delete(taskId); + this.cancellingTaskIds = next; + } + } + + override render() { + return html` +
+
+
${titleForRoute("tasks")}
+
${subtitleForRoute("tasks")}
+
+ +
+ ${renderTasks({ + basePath: this.context.basePath, + connected: this.connected, + // tasks.cancel needs operator.write; read-only operators get no button. + canCancel: hasOperatorWriteAccess(this.context.gateway.snapshot.hello?.auth ?? null), + loading: this.loading, + error: this.error, + tasks: this.tasks, + cancellingTaskIds: this.cancellingTaskIds, + onCancel: (taskId) => void this.cancelTask(taskId), + onNavigateToChat: (sessionKey) => + this.context.navigate("chat", { search: searchForSession(sessionKey) }), + })} + `; + } +} + +if (!customElements.get("openclaw-tasks-page")) { + customElements.define("openclaw-tasks-page", TasksPage); +} diff --git a/ui/src/pages/tasks/tasks.e2e.test.ts b/ui/src/pages/tasks/tasks.e2e.test.ts new file mode 100644 index 00000000000..894fa1a015a --- /dev/null +++ b/ui/src/pages/tasks/tasks.e2e.test.ts @@ -0,0 +1,163 @@ +import { copyFile, mkdir, rm } from "node:fs/promises"; +import path from "node:path"; +import { chromium, type Browser } from "playwright"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + canRunPlaywrightChromium, + installMockGateway, + resolvePlaywrightChromiumExecutablePath, + startControlUiE2eServer, + type ControlUiE2eServer, +} from "../../test-helpers/control-ui-e2e.ts"; + +const chromiumExecutablePath = resolvePlaywrightChromiumExecutablePath(chromium.executablePath()); +const chromiumAvailable = canRunPlaywrightChromium(chromiumExecutablePath); +const allowMissingChromium = process.env.OPENCLAW_UI_E2E_ALLOW_MISSING_CHROMIUM === "1"; +const describeControlUiE2e = chromiumAvailable || !allowMissingChromium ? describe : describe.skip; +const artifactDir = path.resolve(process.cwd(), ".artifacts/control-ui-e2e/tasks"); +const baseTime = Date.parse("2026-07-05T18:00:00.000Z"); + +const runningTask = { + id: "task-running", + taskId: "task-running", + kind: "subagent", + runtime: "subagent", + status: "running", + title: "Review gateway changes", + agentId: "main", + childSessionKey: "agent:main:subagent:review", + createdAt: baseTime - 5_000, + updatedAt: baseTime, + progressSummary: "Reading subscription paths", +}; + +const queuedTask = { + id: "task-queued", + taskId: "task-queued", + kind: "cron", + runtime: "cron", + status: "queued", + title: "Nightly cleanup", + agentId: "main", + sessionKey: "agent:main:cron:cleanup", + createdAt: baseTime - 10_000, + updatedAt: baseTime - 1_000, +}; + +const completedTask = { + id: "task-completed", + taskId: "task-completed", + kind: "cli", + runtime: "cli", + status: "completed", + title: "Generate media index", + createdAt: baseTime - 30_000, + updatedAt: baseTime - 20_000, + terminalSummary: "Index generated", +}; + +const failedTask = { + id: "task-failed", + taskId: "task-failed", + kind: "acp", + runtime: "acp", + status: "failed", + title: "Run ACP worker", + createdAt: baseTime - 40_000, + updatedAt: baseTime - 30_000, + error: "Worker exited", +}; + +let server: ControlUiE2eServer; +let browser: Browser; + +describeControlUiE2e("Control UI Tasks mocked Gateway E2E", () => { + beforeAll(async () => { + if (!chromiumAvailable) { + throw new Error( + `Playwright Chromium is not installed at ${chromiumExecutablePath}. Run \`pnpm --dir ui exec playwright install chromium\`, or set OPENCLAW_UI_E2E_ALLOW_MISSING_CHROMIUM=1 only when intentionally skipping this lane.`, + ); + } + server = await startControlUiE2eServer(); + browser = await chromium.launch({ executablePath: chromiumExecutablePath }); + }); + + afterAll(async () => { + await browser?.close(); + await server?.close(); + }); + + it("renders task sections, applies pushed completion, and sends cancel", async () => { + await rm(artifactDir, { force: true, recursive: true }); + await mkdir(artifactDir, { recursive: true }); + const rawVideoDir = path.join(artifactDir, "raw-video"); + await mkdir(rawVideoDir, { recursive: true }); + const context = await browser.newContext({ + locale: "en-US", + recordVideo: { dir: rawVideoDir, size: { width: 1440, height: 900 } }, + serviceWorkers: "block", + viewport: { width: 1440, height: 900 }, + }); + const page = await context.newPage(); + const video = page.video(); + try { + const gateway = await installMockGateway(page, { + methodResponses: { + "tasks.list": { + tasks: [runningTask, queuedTask, completedTask, failedTask], + }, + "tasks.cancel": { + found: true, + cancelled: true, + task: { ...queuedTask, status: "cancelled", updatedAt: baseTime + 2_000 }, + }, + }, + }); + + const response = await page.goto(`${server.baseUrl}tasks`); + 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-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 recent.textContent()).toContain("Worker exited"); + await page.screenshot({ + path: path.join(artifactDir, "01-task-sections.png"), + fullPage: true, + }); + + await gateway.emitGatewayEvent("task", { + action: "upserted", + task: { + ...runningTask, + status: "completed", + updatedAt: baseTime + 1_000, + terminalSummary: "Review complete", + }, + }); + await recent.locator('[data-task-id="task-running"]').waitFor({ state: "visible" }); + await active.locator('[data-task-id="task-running"]').waitFor({ state: "detached" }); + 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" }) + .click(); + const cancelRequest = await gateway.waitForRequest("tasks.cancel"); + expect(cancelRequest.params).toEqual({ taskId: "task-queued" }); + } finally { + await context.close(); + if (video) { + await copyFile(await video.path(), path.join(artifactDir, "tasks-flow.webm")); + } + await rm(rawVideoDir, { force: true, recursive: true }); + } + }); +}); diff --git a/ui/src/pages/tasks/view.ts b/ui/src/pages/tasks/view.ts new file mode 100644 index 00000000000..d06a4233717 --- /dev/null +++ b/ui/src/pages/tasks/view.ts @@ -0,0 +1,191 @@ +import { html, nothing } from "lit"; +import { repeat } from "lit/directives/repeat.js"; +import { pathForRoute } from "../../app-route-paths.ts"; +import { t } from "../../i18n/index.ts"; +import { formatMs, formatRelativeTimestamp } from "../../lib/format.ts"; +import { searchForSession } from "../../lib/sessions/index.ts"; +import { partitionTasks, taskTimestampMs, type TaskStatus, type TaskSummary } from "./data.ts"; + +export type TasksProps = { + basePath: string; + connected: boolean; + canCancel: boolean; + loading: boolean; + error: string | null; + tasks: TaskSummary[]; + cancellingTaskIds: ReadonlySet; + onCancel: (taskId: string) => void; + onNavigateToChat: (sessionKey: string) => void; +}; + +const STATUS_LABEL_KEYS = { + queued: "tasksPage.status.queued", + running: "tasksPage.status.running", + completed: "tasksPage.status.completed", + failed: "tasksPage.status.failed", + cancelled: "tasksPage.status.cancelled", + timed_out: "tasksPage.status.timedOut", +} as const satisfies Record; + +const STATUS_CHIP_CLASSES = { + queued: "chip-warn", + running: "chip-warn", + completed: "chip-ok", + failed: "chip-danger", + cancelled: "", + timed_out: "chip-danger", +} as const satisfies Record; + +function statusLabel(status: TaskStatus): string { + return t(STATUS_LABEL_KEYS[status]); +} + +function statusClass(status: TaskStatus): string { + return STATUS_CHIP_CLASSES[status]; +} + +function runtimeLabel(task: TaskSummary): string { + switch (task.runtime) { + case "subagent": + return t("tasksPage.runtime.subagent"); + case "cron": + return t("tasksPage.runtime.cron"); + case "acp": + return t("tasksPage.runtime.acp"); + case "cli": + return t("tasksPage.runtime.cli"); + default: + return t("tasksPage.runtime.unknown"); + } +} + +function taskTitle(task: TaskSummary): string { + return task.title ?? task.kind ?? (task.runtime ? runtimeLabel(task) : t("tasksPage.untitled")); +} + +function taskDetail(task: TaskSummary): string | null { + if (task.status === "queued" || task.status === "running") { + return task.progressSummary ?? null; + } + if (task.status === "failed" || task.status === "timed_out") { + return task.error ?? task.terminalSummary ?? task.progressSummary ?? null; + } + return task.terminalSummary ?? task.error ?? task.progressSummary ?? null; +} + +function renderSessionLink(task: TaskSummary, props: TasksProps) { + const sessionKey = task.childSessionKey ?? task.sessionKey; + if (!sessionKey) { + return nothing; + } + const href = `${pathForRoute("chat", props.basePath)}${searchForSession(sessionKey)}`; + return html` { + if ( + event.defaultPrevented || + event.button !== 0 || + event.metaKey || + event.ctrlKey || + event.shiftKey || + event.altKey + ) { + return; + } + event.preventDefault(); + props.onNavigateToChat(sessionKey); + }} + >${t("tasksPage.openSession")}`; +} + +function renderTask(task: TaskSummary, props: TasksProps) { + const active = task.status === "queued" || task.status === "running"; + const timestamp = taskTimestampMs(task.updatedAt ?? task.createdAt); + const detail = taskDetail(task); + const title = taskTitle(task); + const cancelling = props.cancellingTaskIds.has(task.id); + return html` +
+
+
${title}
+
+ ${statusLabel(task.status)} + ${runtimeLabel(task)} + ${task.agentId + ? html`${t("tasksPage.agent", { agent: task.agentId })}` + : nothing} +
+ ${detail ? html`
${detail}
` : nothing} +
+
+ ${timestamp > 0 + ? html`${formatRelativeTimestamp(timestamp)}` + : html`${t("common.na")}`} + ${renderSessionLink(task, props)} + ${active && props.canCancel + ? html`` + : nothing} +
+
+ `; +} + +function renderSection( + id: "active" | "recent", + title: string, + tasks: readonly TaskSummary[], + emptyText: string, + props: TasksProps, +) { + return html` +
+
+
${title}
+
+ ${tasks.length === 1 + ? t("tasksPage.taskCountOne") + : t("tasksPage.taskCount", { count: String(tasks.length) })} +
+
+ ${tasks.length === 0 + ? html`
${emptyText}
` + : html`
+ ${repeat( + tasks, + (task) => task.id, + (task) => renderTask(task, props), + )} +
`} +
+ `; +} + +export function renderTasks(props: TasksProps) { + const { active, recent } = partitionTasks(props.tasks); + return html` +
+ ${!props.connected + ? html`
${t("tasksPage.disconnected")}
` + : nothing} + ${props.error ? html`
${props.error}
` : nothing} + ${props.loading && props.tasks.length === 0 + ? html`
${t("tasksPage.loading")}
` + : nothing} + ${!props.loading && props.tasks.length === 0 + ? html`
${t("tasksPage.empty")}
` + : nothing} + ${renderSection("active", t("tasksPage.active"), active, t("tasksPage.emptyActive"), props)} + ${renderSection("recent", t("tasksPage.recent"), recent, t("tasksPage.emptyRecent"), props)} +
+ `; +}