Tasks:持久化任務

源码版本v2026.6.11

職責

task-executor.ts 和 detached-task-runtime.ts 構成 OpenClaw 的「任務 (task)」子系統:把任何可能跨進程、跨會話、跨重啟的非同步工作建模成一條帶生命週期的記錄,無論它在哪個 runtime(subagent / acp / cli / cron)裡跑,都共享同一份狀態機 (state machine)、同一份 SQLite 持久化 (persistence)、同一份取消與回收語意。CronService 負責排程時機,task-executor.ts 負責「一條記錄怎麼從 queued 走到 succeeded/failed/timed_out/cancelled/lost」。

任務不是執行緒也不是 worker——它只是一個狀態物件 (TaskRecord,TaskRecord:116-146) 加上對應的 runtime 自己實作的生命週期鉤子。runtime(subagent/acp/cli/cron,TaskRuntime:5)決定怎麼真正執行,task 系統只管:存記錄、轉狀態、投遞回原會話、定期清理過期記錄。

設計動機

為什麼 cron 不夠,還要單獨抽一層 task?

1. 跨 runtime:cron 是定時,ACP 是 IDE 觸發的 prompt,subagent 是 agent 主循環派生的子任務,CLI 是使用者在終端機敲的指令。四種 runtime 的觸發時機完全不同,但它們都要面對同樣的問題:進程崩潰後怎麼知道某條工作沒跑完?gateway 重啟後怎麼回收?結果怎麼回到原會話?把這些問題抽成 task registry,每種 runtime 只實作自己的 DetachedTaskLifecycleRuntime(DEFAULT_DETACHED_TASK_LIFECYCLE_RUNTIME:35-49),其他邏輯共享。
2. flow 概念:很多任務屬於一個更大的 flow(比如一個 agent run 觸發三個子任務,主任務取消時三個子任務也要取消)。task-flow-registry(src/tasks/task-flow-registry.ts) 給任務一個父 flow,可以批次取消、批次查狀態。對於 detached ACP/subagent run,executor 自動給它包一層「one-task flow」(isOneTaskFlowEligible + ensureSingleTaskFlow:46-92),讓這些單次 run 也能享受 flow 的狀態/重試面。
3. delivery 狀態:任務結束之後,結果不一定要回到原會話——可能使用者已經關了 IDE、可能 cron 是 isolated session 沒有原會話。TaskDeliveryStatus(TaskDeliveryStatus:16-23)單獨追蹤這條狀態:pending/delivered/session_queued/failed/parent_missing/not_applicable,把「任務執行成功」和「結果送到用戶端」解耦。

關鍵檔案

• task-registry.types.ts:1-146 — TaskRuntime/TaskStatus/TaskDeliveryStatus/TaskNotifyPolicy/TaskRecord 全部型別宣告。
• task-executor.ts:46-127 — ensureSingleTaskFlow + createQueuedTaskRun + createRunningTaskRun,任務建立入口(含 one-task flow 包裝)。
• detached-task-runtime.ts:32-66 — DetachedTaskLifecycleRuntime 介面、預設 runtime、外掛註冊機制。
• tryRecoverTaskBeforeMarkLost:134-171 — 回收 hook:best-effort,5 秒閾值告警,任何拋錯都 fallback 到 mark-lost。
• task-registry.process-state.ts:1-32 — 進程級 in-memory 索引(tasks/taskIdsByRunId/taskIdsByOwnerKey/taskIdsByParentFlowId/taskIdsByRelatedSessionKey/tasksWithPendingDelivery),Symbol.for("openclaw.taskRegistry.state") 掛到 globalThis。
• task-registry.store.sqlite.ts:48-77 — Kysely 映射的 task_runs + task_delivery_state 表,鏡像到 openclaw-state.db。
• task-registry.maintenance.ts:73-100 — 背景 sweeper,週期 60s,TASK_STALE_RUNNING_MS=30*60_000,TASK_RECONCILE_GRACE_MS=5*60_000。
• task-retention.ts:4-29 — 終態任務保留 7 天,lost 任務只保留 1 天。
• src/tasks/task-flow-registry.store.sqlite.ts — flow 持久化,支援 parentFlowId 級聯。
• cron-task-cancel.ts:12-77 — 進程級 active cron task run 取消控制代碼,settlement grace 60s。
• task-registry.reconcile.ts — 公共 facade:reconcileInspectableTasks / reconcileTaskLookupToken。

資料流

任務的狀態機只有 7 個狀態(TaskStatus:7-14):

export type TaskStatus =
  | "queued"       // 已落庫,等 runtime 拉起
  | "running"      // runtime 報告開始執行
  | "succeeded"    // 終態:成功
  | "failed"       // 終態:失敗(帶 error)
  | "timed_out"    // 終態:逾時
  | "cancelled"    // 終態:使用者/系統取消
  | "lost";        // 終態:sweeper 在 grace 後仍未觀察到活躍,標記丟失

createQueuedTaskRun / createRunningTaskRun 是建立入口,兩者都會呼叫 ensureSingleTaskFlow(ensureSingleTaskFlow:56):

function isOneTaskFlowEligible(task: TaskRecord): boolean {
  if (task.parentFlowId?.trim() || task.scopeKind !== "session") return false;
  if (task.deliveryStatus === "not_applicable") return false;
  return task.runtime === "acp" || task.runtime === "subagent";
}

function ensureSingleTaskFlow(params: { task; requesterOrigin? }): TaskRecord {
  if (!isOneTaskFlowEligible(params.task)) return params.task;
  try {
    const flow = createTaskFlowForTask({ task: params.task, requesterOrigin: params.requesterOrigin });
    if (!flow) return params.task;
    const linked = linkTaskToFlowById({ taskId: params.task.taskId, flowId: flow.flowId });
    if (!linked) { deleteTaskFlowRecordById(flow.flowId); return params.task; }
    if (linked.parentFlowId !== flow.flowId) { deleteTaskFlowRecordById(flow.flowId); return linked; }
    return linked;
  } catch (error) {
    log.warn("Failed to create one-task flow for detached run", { ... });
    return params.task;
  }
}

關鍵判斷:scopeKind === "session" 且 runtime ∈ {acp, subagent} 且 deliveryStatus !== "not_applicable"——這三條滿足時才給它自動包一層 flow。系統級任務(scopeKind: "system")和 cron 任務有自己單獨的 cancel/delivery 路徑,不重走 flow。失敗時只 warn 不拋,保證任務建立本身不會因為 flow 包裝失敗被阻塞。

detached-task-runtime.ts(getDetachedTaskLifecycleRuntime:47)對外暴露統一 API,但具體實作可以外掛替換:

const DEFAULT_DETACHED_TASK_LIFECYCLE_RUNTIME: DetachedTaskLifecycleRuntime = {
  createQueuedTaskRun: createQueuedTaskRunFromExecutor,
  createRunningTaskRun: createRunningTaskRunFromExecutor,
  startTaskRunByRunId: startTaskRunByRunIdFromExecutor,
  recordTaskRunProgressByRunId: recordTaskRunProgressByRunIdFromExecutor,
  finalizeTaskRunByRunId: finalizeTaskRunByRunIdFromExecutor,
  completeTaskRunByRunId: completeTaskRunByRunIdFromExecutor,
  failTaskRunByRunId: failTaskRunByRunIdFromExecutor,
  setDetachedTaskDeliveryStatusByRunId: setDetachedTaskDeliveryStatusByRunIdFromExecutor,
  cancelDetachedTaskRunById: cancelDetachedTaskRunByIdInCore,
};

getDetachedTaskLifecycleRuntime() 優先回傳外掛註冊的實作,否則用預設——這讓 plugin 可以接管整個 task 生命週期(例如把 task 跑到獨立 worker 進程)而不必改 task-executor。

邊界與失敗

• 進程級索引的 symbol-keyed globalThis:getTaskRegistryProcessState()(getTaskRegistryProcessState:18)用 Symbol.for("openclaw.taskRegistry.state") 掛到 globalThis,這樣模組熱重載(dev watch)或測試隔離時索引仍然共享——否則不同模組實例會各自維護一份 in-memory 索引,落庫到 SQLite 的任務和記憶體索引會失配。
• sweeper 批次 yield:SWEEP_YIELD_BATCH_SIZE = 25(SWEEP_YIELD_BATCH_SIZE:83),每處理 25 條任務就讓出事件迴圈,避免大批量清理時阻塞主執行緒。
• stale running 判定:TASK_STALE_RUNNING_MS = 30 * 60_000——running 狀態超過 30 分鐘無任何事件更新(包括 recordTaskRunProgressByRunId),sweeper 才會開始回收。這是給慢任務一個寬鬆視窗,不會誤殺正常長跑。
• recovery hook 容錯:tryRecoverTaskBeforeMarkLost(tryRecoverTaskBeforeMarkLost:134)裡 try/catch 包住整個 hook 呼叫——任何 plugin 實作 throw、回傳非法物件、超過 5s 閾值,都只 warn,然後繼續走 mark-lost。理由:recovery 是 best-effort,不能讓壞外掛阻塞清理。
• retention 雙軌:終態任務保留 7 天,lost 任務只保留 1 天(resolveTaskRetentionMs:5-9)。lost 是「sweeper 推測出來的終態」,可信度低於 runtime 顯式上報的 succeeded/failed,所以保留期短——既給排查視窗,又不讓「幽靈任務」長期佔用 DB。
• one-task flow 失敗降級:ensureSingleTaskFlow 任何一步拋錯都回傳原 task(不重新包裝),日誌裡記 warn。任務能建立成功比 flow 包裝完整更重要。
• scopeKind 區分:scopeKind: "system" 的任務沒有 requesterSessionKey(system-scope requesterSessionKey:92-94),ownerKey 是唯一查找錨——這避免了系統任務和某個 session 的 ownerKey 撞車。
• cron-task-cancel settlement grace:gateway 重啟時 abortActiveCronTaskRuns(abortActiveCronTaskRunSettlementGrace:50)會啟動一個 60 秒的「沉降寬限期」,讓被 abort 的 promise 有時間在 finally 裡清完資源、寫完終態——直接 clear 表會讓 cleanup 走丟。

小結

Tasks 子系統把「一條非同步工作」抽象成 7 狀態的 TaskRecord,四種 runtime(subagent/acp/cli/cron)共用一套狀態機 + SQLite 持久化 + sweeper 回收。它和 Cron:定時任務 是雙向關係:cron 透過 registerActiveCronTaskRun 把自己掛到 task 進程級表裡,task 系統又透過 cron-task-cancel.ts 給 cron 提供「重啟時打斷活躍 run」的能力。ACP 啟動的 detached prompt 走的也是同一套(見 ACP:IDE 橋接),只是 runtime 標成 "acp"。session 關聯狀態和 task delivery 回投會用到 Memory 檔案 裡的 session store 路徑。