diff --git a/docs/automation/taskflow.md b/docs/automation/taskflow.md index 8f708b832c3c..8f514f71c71e 100644 --- a/docs/automation/taskflow.md +++ b/docs/automation/taskflow.md @@ -44,21 +44,30 @@ OpenClaw creates a mirrored one-task flow automatically when a detached ACP or s ## Flow statuses -| Status | Meaning | -| ----------- | -------------------------------------------------------------------------- | -| `queued` | Created, not yet progressing | -| `running` | Flow is actively progressing | -| `waiting` | Managed flow is parked on wait metadata (timer, external event) | -| `blocked` | A step finished without a usable result; `blockedTaskId`/summary say which | -| `succeeded` | Completed successfully | -| `failed` | Completed with an error | -| `cancelled` | Cancel requested and all child tasks settled | -| `lost` | Flow lost its authoritative backing state | +| Status | Meaning | +| ----------- | ----------------------------------------------------------------- | +| `queued` | Created, not yet progressing | +| `running` | Flow is actively progressing | +| `waiting` | Managed flow is parked on wait metadata (timer, external event) | +| `blocked` | Waiting on a blocking condition, or ended without a usable result | +| `succeeded` | Completed successfully | +| `failed` | Completed with an error | +| `cancelled` | Cancel requested and all child tasks settled | +| `lost` | Flow lost its authoritative backing state | + +`blocked` is the only status whose terminal meaning depends on the record. A +managed flow with no `endedAt` remains resumable. A `blocked` flow with +`endedAt` is finished, including mirrored flows whose backing task completed +with a blocked outcome. ## Durable state and revision tracking Flow records persist in the shared SQLite state database (`~/.openclaw/state/openclaw.sqlite`, `flow_runs` table) alongside task records, so progress survives gateway restarts. Each write bumps the flow's `revision`; concurrent writers that pass a stale expected revision get a conflict and must re-read. WAL growth is bounded by SQLite autocheckpointing plus periodic passive checkpoints, with truncate checkpoints on shutdown. The legacy `flows/registry.sqlite` sidecar from older installs is imported by `openclaw doctor`. +Gateway maintenance retains finished flows for 7 days, then prunes them. This +includes `blocked` flows with `endedAt`; resumable managed `blocked` flows are +retained regardless of age. + ## Cancel behavior `openclaw tasks flow cancel` sets a sticky cancel intent on the flow, cancels its active child tasks, and refuses new managed child tasks. Once no child task remains active, the flow finalizes as `cancelled` - immediately, or via the maintenance sweep if children take longer to settle. The intent is persisted, so a cancelled flow stays cancelled even if the gateway restarts before all child tasks have terminated. diff --git a/docs/automation/tasks.md b/docs/automation/tasks.md index 21e19fdac1fb..baedc71cf980 100644 --- a/docs/automation/tasks.md +++ b/docs/automation/tasks.md @@ -366,7 +366,7 @@ Legacy sidecar stores from older installs (`tasks/runs.sqlite`, `flows/registry. ### Automatic maintenance -A sweeper runs every **60 seconds** (first pass about 5 seconds after gateway start) and handles four things: +A sweeper runs every **60 seconds** (first pass about 5 seconds after gateway start) and handles five things: @@ -381,6 +381,9 @@ A sweeper runs every **60 seconds** (first pass about 5 seconds after gateway st Deletes records past their `cleanupAfter` date. + + Deletes terminal Task Flow records after 7 days. A `blocked` flow is terminal only when it has `endedAt`; resumable managed `blocked` flows remain registered. + diff --git a/src/commands/flows.test.ts b/src/commands/flows.test.ts index d0db45d3ac93..46c04a2fe35d 100644 --- a/src/commands/flows.test.ts +++ b/src/commands/flows.test.ts @@ -225,12 +225,22 @@ describe("flows commands", () => { createdAt: 100, updatedAt: 200, }); + createManagedTaskFlow({ + ownerKey: "agent:main:main", + controllerId: "tests/flows-command-ended-blocked", + goal: "Completed blocked work", + status: "blocked", + cancelRequestedAt: 150, + createdAt: 100, + updatedAt: 150, + endedAt: 150, + }); const runtime = createRuntime(); await flowsListCommand({}, runtime); expect(vi.mocked(runtime.log).mock.calls.map(([line]) => String(line))).toContain( - "TaskFlow pressure: 1 active · 0 blocked · 1 cancel-requested · 1 total", + "TaskFlow pressure: 1 active · 1 blocked · 1 cancel-requested · 2 total", ); }); }); diff --git a/src/commands/flows.ts b/src/commands/flows.ts index 0b04e2467a37..4e95e535d1b7 100644 --- a/src/commands/flows.ts +++ b/src/commands/flows.ts @@ -14,6 +14,7 @@ import { type RuntimeEnv, writeRuntimeJson } from "../runtime.js"; import { listTasksForFlowId } from "../tasks/runtime-internal.js"; import { cancelFlowById, getFlowTaskSummary } from "../tasks/task-executor.js"; import { + isTerminalTaskFlow, TASK_FLOW_STATUSES, type TaskFlowRecord, type TaskFlowStatus, @@ -23,7 +24,6 @@ import { listTaskFlowRecords, resolveTaskFlowForLookupToken, } from "../tasks/task-flow-runtime-internal.js"; -import { isTerminalFlowStatus } from "../tasks/task-registry-common.js"; import { formatTaskStatus, formatTaskStatusDetail, @@ -128,9 +128,7 @@ function formatFlowListSummary(flows: TaskFlowRecord[]) { counts.waiting += Number(flow.status === "waiting"); counts.blocked += Number(flow.status === "blocked"); counts.issues += Number(flow.status === "failed" || flow.status === "lost"); - counts.cancelRequested += Number( - flow.cancelRequestedAt != null && !isTerminalFlowStatus(flow.status), - ); + counts.cancelRequested += Number(flow.cancelRequestedAt != null && !isTerminalTaskFlow(flow)); } const waiting = counts.waiting ? ` · ${counts.waiting} waiting` : ""; const issues = counts.issues ? ` · ${counts.issues} issues` : ""; diff --git a/src/tasks/task-executor.test.ts b/src/tasks/task-executor.test.ts index 709884ccd315..f6d51c8cf317 100644 --- a/src/tasks/task-executor.test.ts +++ b/src/tasks/task-executor.test.ts @@ -610,6 +610,44 @@ describe("task-executor", () => { }); }); + it("rejects child tasks and cancellation for ended blocked TaskFlows", async () => { + await withTaskExecutorStateDir(async () => { + const flow = createManagedTaskFlow({ + ownerKey: "agent:main:main", + controllerId: "tests/managed-flow", + goal: "Completed without a usable result", + status: "blocked", + updatedAt: 20, + endedAt: 20, + }); + + const created = runTaskInFlow({ + flowId: flow.flowId, + runtime: "acp", + childSessionKey: "agent:codex:acp:child", + runId: "run-flow-after-blocked-completion", + task: "Should be denied", + }); + const cancelled = await cancelFlowById({ + cfg: {} as never, + flowId: flow.flowId, + }); + + expect(created).toMatchObject({ + found: true, + created: false, + reason: "Flow is already blocked.", + }); + expect(cancelled).toMatchObject({ + found: true, + cancelled: false, + reason: "Flow is already blocked.", + }); + expect(getTaskFlowById(flow.flowId)).toMatchObject({ status: "blocked", endedAt: 20 }); + expect(listTasksForFlowId(flow.flowId)).toStrictEqual([]); + }); + }); + it("refuses to add child tasks once cancellation is requested on a managed TaskFlow", async () => { await withTaskExecutorStateDir(async () => { const flow = createManagedTaskFlow({ diff --git a/src/tasks/task-executor.ts b/src/tasks/task-executor.ts index 5c5d7d323d59..d0e8453dcc85 100644 --- a/src/tasks/task-executor.ts +++ b/src/tasks/task-executor.ts @@ -26,7 +26,7 @@ import { isTaskFlowCancellationPending, } from "./task-cancellation-state.js"; import { getTaskFlowByIdForOwner } from "./task-flow-owner-access.js"; -import type { TaskFlowRecord } from "./task-flow-registry.types.js"; +import { isTerminalTaskFlow, type TaskFlowRecord } from "./task-flow-registry.types.js"; import { createTaskFlowForTask, deleteTaskFlowRecordById, @@ -226,12 +226,6 @@ type RunTaskInFlowResult = { task?: TaskRecord; }; -function isTerminalFlowStatus(status: TaskFlowRecord["status"]): boolean { - return ( - status === "succeeded" || status === "failed" || status === "cancelled" || status === "lost" - ); -} - function markFlowCancelRequested(flow: TaskFlowRecord): TaskFlowRecord | FlowUpdateFailure { if (flow.cancelRequestedAt != null) { return flow; @@ -353,7 +347,7 @@ function runTaskInFlow(params: RunTaskInFlowParams): RunTaskInFlowResult { flow, }; } - if (isTerminalFlowStatus(flow.status)) { + if (isTerminalTaskFlow(flow)) { return { found: true, created: false, @@ -468,7 +462,7 @@ export async function cancelFlowById(params: { reason: "Flow not found.", }; } - if (isTerminalFlowStatus(flow.status)) { + if (isTerminalTaskFlow(flow)) { const provisionalTasks = listTasksForFlowId(flow.flowId).filter(isProvisionalSubagentKillTask); if (flow.status === "cancelled" && provisionalTasks.length > 0) { for (const task of provisionalTasks) { @@ -535,7 +529,7 @@ export async function cancelFlowById(params: { } const now = Date.now(); const refreshedFlow = getTaskFlowById(flow.flowId) ?? cancelRequestedFlow; - if (isTerminalFlowStatus(refreshedFlow.status)) { + if (isTerminalTaskFlow(refreshedFlow)) { return { found: true, cancelled: refreshedFlow.status === "cancelled", diff --git a/src/tasks/task-flow-registry.maintenance.ts b/src/tasks/task-flow-registry.maintenance.ts index 6e36627f0c28..cffb0d2face1 100644 --- a/src/tasks/task-flow-registry.maintenance.ts +++ b/src/tasks/task-flow-registry.maintenance.ts @@ -13,7 +13,7 @@ import { listTaskFlowRecords, updateFlowRecordByIdExpectedRevision, } from "./task-flow-registry.js"; -import type { TaskFlowRecord } from "./task-flow-registry.types.js"; +import { isTerminalTaskFlow, type TaskFlowRecord } from "./task-flow-registry.types.js"; const TASK_FLOW_RETENTION_MS = 7 * 24 * 60 * 60_000; @@ -32,16 +32,6 @@ export function assertTaskFlowRegistryMaintenanceReady(): void { } } -function isTerminalFlow(flow: TaskFlowRecord): boolean { - return ( - flow.status === "succeeded" || - (flow.status === "blocked" && flow.endedAt != null) || - flow.status === "failed" || - flow.status === "cancelled" || - flow.status === "lost" - ); -} - function hasActiveLinkedTasks(flowId: string): boolean { return listTasksForFlowId(flowId).some(isTaskFlowCancellationPending); } @@ -51,7 +41,7 @@ function resolveTerminalAt(flow: TaskFlowRecord): number { } function shouldPruneFlow(flow: TaskFlowRecord, now: number): boolean { - if (!isTerminalFlow(flow)) { + if (!isTerminalTaskFlow(flow)) { return false; } if (hasActiveLinkedTasks(flow.flowId)) { @@ -64,7 +54,7 @@ function shouldFinalizeCancelledFlow(flow: TaskFlowRecord): boolean { if (flow.syncMode !== "managed") { return false; } - if (flow.cancelRequestedAt == null || isTerminalFlow(flow)) { + if (flow.cancelRequestedAt == null || isTerminalTaskFlow(flow)) { return false; } return !hasActiveLinkedTasks(flow.flowId); @@ -101,7 +91,7 @@ function finalizeCancelledFlow(flow: TaskFlowRecord, now: number): boolean { } function shouldRepairTerminalMirroredFlowTimestamp(flow: TaskFlowRecord): boolean { - if (flow.syncMode !== "task_mirrored" || !isTerminalFlow(flow)) { + if (flow.syncMode !== "task_mirrored" || !isTerminalTaskFlow(flow)) { return false; } if (flow.endedAt == null || flow.endedAt < flow.createdAt) { diff --git a/src/tasks/task-flow-registry.types.ts b/src/tasks/task-flow-registry.types.ts index 31ecccbe9848..53b4b8ab0e5c 100644 --- a/src/tasks/task-flow-registry.types.ts +++ b/src/tasks/task-flow-registry.types.ts @@ -64,3 +64,15 @@ export type TaskFlowRecord = { updatedAt: number; endedAt?: number; }; + +// Managed `blocked` flows remain resumable until endedAt is set; mirrored +// `blocked` flows carry endedAt because they project a terminal task outcome. +export function isTerminalTaskFlow(flow: Pick): boolean { + return ( + flow.status === "succeeded" || + (flow.status === "blocked" && flow.endedAt != null) || + flow.status === "failed" || + flow.status === "cancelled" || + flow.status === "lost" + ); +} diff --git a/src/tasks/task-registry-common.ts b/src/tasks/task-registry-common.ts index 12168054ab08..271e4fb00742 100644 --- a/src/tasks/task-registry-common.ts +++ b/src/tasks/task-registry-common.ts @@ -5,7 +5,7 @@ import { } from "../agents/agent-run-terminal-outcome.js"; import { SUBAGENT_KILL_TASK_ERROR } from "./detached-task-runtime-contract.js"; import { isTerminalTaskStatus } from "./task-executor-policy.js"; -import type { TaskFlowRecord } from "./task-flow-registry.types.js"; +import { isTerminalTaskFlow, type TaskFlowRecord } from "./task-flow-registry.types.js"; import { ensureTaskFlowRegistryReady, getTaskFlowById } from "./task-flow-runtime-internal.js"; import type { TaskDeliveryState, @@ -55,12 +55,6 @@ export function isActiveTaskStatus(status: TaskStatus): boolean { return status === "queued" || status === "running"; } -export function isTerminalFlowStatus(status: TaskFlowRecord["status"]): boolean { - return ( - status === "succeeded" || status === "failed" || status === "cancelled" || status === "lost" - ); -} - export function assertTaskOwner(params: { ownerKey: string; scopeKind: TaskScopeKind }) { const ownerKey = params.ownerKey.trim(); if (!ownerKey && params.scopeKind !== "system") { @@ -104,7 +98,7 @@ export function assertParentFlowLinkAllowed(params: { { flowId, status: flow.status }, ); } - if (isTerminalFlowStatus(flow.status)) { + if (isTerminalTaskFlow(flow)) { throw new ParentFlowLinkError("terminal", `Parent flow is already ${flow.status}.`, { flowId, status: flow.status, diff --git a/src/tasks/task-registry-mutation.ts b/src/tasks/task-registry-mutation.ts index 88320e749687..3429397e5f2c 100644 --- a/src/tasks/task-registry-mutation.ts +++ b/src/tasks/task-registry-mutation.ts @@ -3,13 +3,14 @@ import { runWithGatewayIndependentRootWorkAdmission } from "../process/gateway-w import { normalizeDeliveryContext } from "../utils/delivery-context.shared.js"; import { isTaskFlowCancellationPending } from "./task-cancellation-state.js"; import { isTerminalTaskStatus } from "./task-executor-policy.js"; +import { isTerminalTaskFlow } from "./task-flow-registry.types.js"; import { getTaskFlowById, syncFlowFromTaskResult, updateFlowRecordByIdExpectedRevision, } from "./task-flow-runtime-internal.js"; import { clearTaskActivity, flushTaskActivity } from "./task-registry-activity.js"; -import { ensureLinkedTaskFlowRegistryReady, isTerminalFlowStatus } from "./task-registry-common.js"; +import { ensureLinkedTaskFlowRegistryReady } from "./task-registry-common.js"; import { findLatestTaskForFlowId, listTasksForFlowId } from "./task-registry-query.js"; import { cloneTaskDeliveryState, @@ -46,7 +47,7 @@ function syncManagedFlowCancellationFromTask(task: TaskRecord): void { !flow || flow.syncMode !== "managed" || flow.cancelRequestedAt == null || - isTerminalFlowStatus(flow.status) + isTerminalTaskFlow(flow) ) { return; } @@ -75,7 +76,7 @@ function syncManagedFlowCancellationFromTask(task: TaskRecord): void { !flow || flow.syncMode !== "managed" || flow.cancelRequestedAt == null || - isTerminalFlowStatus(flow.status) + isTerminalTaskFlow(flow) ) { return; } diff --git a/src/tasks/task-registry.maintenance.ts b/src/tasks/task-registry.maintenance.ts index 4b6c7f6ea56a..3d372239a567 100644 --- a/src/tasks/task-registry.maintenance.ts +++ b/src/tasks/task-registry.maintenance.ts @@ -53,6 +53,7 @@ import { resolveTaskForLookupToken, setTaskCleanupAfterById, } from "./runtime-internal.js"; +import { runTaskFlowRegistryMaintenance } from "./task-flow-registry.maintenance.js"; import { configureTaskAuditTaskProvider, listTaskAuditFindings, @@ -1013,7 +1014,10 @@ function startScheduledSweep() { sweepInProgress = false; }; void runWithGatewayIndependentRootWorkAdmission(async () => { + // Flow retention reads linked task activity, so reconcile the task owner first. + // Reversing this order can preserve phantom active work for another sweep. await sweepTaskRegistry(); + await runTaskFlowRegistryMaintenance(); }).then(clearSweepInProgress, clearSweepInProgress); } diff --git a/src/tasks/task-registry.test.ts b/src/tasks/task-registry.test.ts index 2826c5376a4f..e55ae086d922 100644 --- a/src/tasks/task-registry.test.ts +++ b/src/tasks/task-registry.test.ts @@ -1924,7 +1924,10 @@ describe("task-registry", () => { }); }); - it("rejects parent flow links for terminal flows", async () => { + it.each([ + { status: "cancelled" as const, endedAt: undefined }, + { status: "blocked" as const, endedAt: 42 }, + ])("rejects parent flow links for $status flows", async ({ status, endedAt }) => { await withTaskRegistryTempDir(async () => { resetTaskRegistryMemoryForTest({ persist: false }); resetTaskFlowRegistryForTests({ persist: false }); @@ -1934,7 +1937,8 @@ describe("task-registry", () => { ownerKey: "agent:main:main", controllerId: "tests/task-registry", goal: "Completed flow", - status: "cancelled", + status, + endedAt, }); expect(() => @@ -1945,7 +1949,7 @@ describe("task-registry", () => { runId: "terminal-flow-link", task: "Should be denied", }), - ).toThrow("Parent flow is already cancelled."); + ).toThrow(`Parent flow is already ${status}.`); }); }); @@ -3563,6 +3567,35 @@ describe("task-registry", () => { }); }); + it("prunes expired ended TaskFlows during scheduled maintenance", async () => { + await withTaskRegistryTempDir( + async () => { + vi.useFakeTimers(); + const endedAt = Date.now() - 8 * 24 * 60 * 60_000; + const flow = createManagedTaskFlow({ + ownerKey: "agent:main:main", + controllerId: "tests/scheduled-task-flow-maintenance", + goal: "Completed without a usable result", + status: "blocked", + createdAt: endedAt, + updatedAt: endedAt, + endedAt, + }); + resetTaskRegistryForTests({ persist: false }); + resetTaskFlowRegistryForTests({ persist: false }); + + try { + startTaskRegistryMaintenance(); + await vi.advanceTimersByTimeAsync(5_000); + await waitForFast(() => expect(getTaskFlowById(flow.flowId)).toBeUndefined()); + } finally { + stopTaskRegistryMaintenance(); + } + }, + { durableStore: true }, + ); + }); + it("keeps scheduled maintenance root-admitted until session cleanup inspection settles", async () => { await withTaskRegistryTempDir(async () => { vi.useFakeTimers();