fix(tasks): stop completed TaskFlows from accumulating (#129873)

Co-authored-by: Galin Iliev <galin.iliev@microsoft.com>

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.

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:
 
 <Steps>
   <Step title="Reconciliation">
@@ -381,6 +381,9 @@ A sweeper runs every **60 seconds** (first pass about 5 seconds after gateway st
   <Step title="Pruning">
     Deletes records past their `cleanupAfter` date.
   </Step>
+  <Step title="Task Flow retention">
+    Deletes terminal Task Flow records after 7 days. A `blocked` flow is terminal only when it has `endedAt`; resumable managed `blocked` flows remain registered.
+  </Step>
 </Steps>
 
 <Note>

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",
       );
     });
   });

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` : "";

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({

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",

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) {

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<TaskFlowRecord, "status" | "endedAt">): boolean {
+  return (
+    flow.status === "succeeded" ||
+    (flow.status === "blocked" && flow.endedAt != null) ||
+    flow.status === "failed" ||
+    flow.status === "cancelled" ||
+    flow.status === "lost"
+  );
+}

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,

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;
     }

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);
 }
 

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();