Tasks: tareas persistentes

源码版本v2026.6.11

Responsabilidad

task-executor.ts y detached-task-runtime.ts forman el subsistema de «tareas (task)» de OpenClaw: modelan cualquier trabajo asincrónico que pueda cruzar proceso, sesión y reinicio como un registro con ciclo de vida, y sin importar en qué runtime (subagent / acp / cli / cron) corra, comparten la misma máquina de estados (state machine), la misma persistencia SQLite (persistence), y las mismas semánticas de cancelación y recolección. CronService se ocupa del scheduling, task-executor.ts se ocupa de «cómo un registro pasa de queued a succeeded/failed/timed_out/cancelled/lost».

Una task no es un thread ni un worker — es un objeto de estado (TaskRecord, TaskRecord:116-146) más los hooks de ciclo de vida que el runtime correspondiente implementa. El runtime (subagent/acp/cli/cron, TaskRuntime:5) decide cómo ejecutar de verdad; el sistema de task solo se ocupa de: almacenar el registro, transicionar estados, entregar de vuelta a la sesión original, y limpiar periódicamente registros expirados.

Motivación de diseño

¿Por qué no basta con cron y hace falta una capa task adicional?

1. Cross-runtime: cron es por tiempo, ACP es un prompt disparado por IDE, subagent es un subtask derivado del bucle principal del agent, CLI es un comando que el usuario teclea. Los cuatro runtimes se disparan en momentos completamente distintos, pero todos enfrentan los mismos problemas: ¿cómo saber tras un crash de proceso que algo no se corrió? ¿cómo recolectar tras un reinicio del gateway? ¿cómo devolver el resultado a la sesión original? Extraer estos problemas en un task registry permite que cada runtime implemente solo su DetachedTaskLifecycleRuntime (DEFAULT_DETACHED_TASK_LIFECYCLE_RUNTIME:35-49) y compartir el resto.
2. Concepto de flow: muchas tareas pertenecen a un flow mayor (por ejemplo, un agent run dispara tres subtasks; al cancelar la principal, las tres subtasks también se cancelan). task-flow-registry (src/tasks/task-flow-registry.ts) da a cada task un flow padre, permitiendo cancelación y consulta de estado batch. Para detached ACP/subagent run, el executor automáticamente los envuelve en un «one-task flow» (isOneTaskFlowEligible + ensureSingleTaskFlow:46-92), dejando que estos runs puntuales también gocen de la cara de estado/retry del flow.
3. Estado de delivery: cuando termina una task, el resultado no necesariamente vuelve a la sesión original — puede que el usuario haya cerrado el IDE, o que cron sea una isolated session sin sesión original. TaskDeliveryStatus (TaskDeliveryStatus:16-23) trackea este estado aparte: pending/delivered/session_queued/failed/parent_missing/not_applicable, desacoplando «task ejecutada con éxito» de «resultado entregado al cliente».

Archivos clave

• task-registry.types.ts:1-146 — todas las declaraciones de tipos TaskRuntime/TaskStatus/TaskDeliveryStatus/TaskNotifyPolicy/TaskRecord.
• task-executor.ts:46-127 — ensureSingleTaskFlow + createQueuedTaskRun + createRunningTaskRun, entradas de creación de task (incluye envoltorio one-task flow).
• detached-task-runtime.ts:32-66 — interfaz DetachedTaskLifecycleRuntime, runtime por defecto, mecanismo de registro de plugin.
• tryRecoverTaskBeforeMarkLost:134-171 — hook de recuperación: best-effort, warn a las 5s, cualquier throw cae a mark-lost.
• task-registry.process-state.ts:1-32 — índices in-memory a nivel proceso (tasks/taskIdsByRunId/taskIdsByOwnerKey/taskIdsByParentFlowId/taskIdsByRelatedSessionKey/tasksWithPendingDelivery), con Symbol.for("openclaw.taskRegistry.state") colgado en globalThis.
• task-registry.store.sqlite.ts:48-77 — tablas task_runs + task_delivery_state mapeadas con Kysely, espejadas en openclaw-state.db.
• task-registry.maintenance.ts:73-100 — sweeper en background, período 60s, TASK_STALE_RUNNING_MS=30*60_000, TASK_RECONCILE_GRACE_MS=5*60_000.
• task-retention.ts:4-29 — las tasks en estado terminal se conservan 7 días, las lost solo 1 día.
• src/tasks/task-flow-registry.store.sqlite.ts — persistencia de flow, soporta cascada por parentFlowId.
• cron-task-cancel.ts:12-77 — handles de cancelación de active cron task run a nivel proceso, settlement grace 60s.
• task-registry.reconcile.ts — facade común: reconcileInspectableTasks / reconcileTaskLookupToken.

Flujo de datos

La máquina de estados de task solo tiene 7 estados (TaskStatus:7-14):

export type TaskStatus =
  | "queued"       // ya en store, esperando que el runtime lo levante
  | "running"      // el runtime reporta inicio de ejecución
  | "succeeded"    // estado terminal: éxito
  | "failed"       // estado terminal: fallo (con error)
  | "timed_out"    // estado terminal: timeout
  | "cancelled"    // estado terminal: cancelado por usuario o sistema
  | "lost";        // estado terminal: sweeper no observó actividad tras grace, se marca perdido

createQueuedTaskRun / createRunningTaskRun son las entradas de creación, ambas llaman 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;
  }
}

Juicio clave: scopeKind === "session" y runtime ∈ {acp, subagent} y deliveryStatus !== "not_applicable" — cuando estos tres se cumplen, se envuelve automáticamente en un flow. Las tasks de sistema (scopeKind: "system") y las de cron tienen rutas propias de cancel/delivery y no reutilizan flow. Ante fallo solo se hace warn, no throw, garantizando que la creación de task no se bloquee por un fallo del envoltorio flow.

detached-task-runtime.ts (getDetachedTaskLifecycleRuntime:47) expone una API unificada, pero la implementación puede ser sustituida por plugin:

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() devuelve primero la implementación registrada por plugin, si no, la default — deja que un plugin tome todo el ciclo de vida de task (por ejemplo, correr tasks en un proceso worker separado) sin tocar task-executor.

Límites y fallos

• globalThis symbol-keyed para índices a nivel proceso: getTaskRegistryProcessState() (getTaskRegistryProcessState:18) cuelga el state en globalThis con Symbol.for("openclaw.taskRegistry.state"), así tras hot reload de módulos (dev watch) o en aislamiento de tests el índice sigue compartido — si no, distintas instancias de módulo mantendrían cada una su índice in-memory, y las tasks en SQLite no casarían con el índice en memoria.
• sweeper batch yield: SWEEP_YIELD_BATCH_SIZE = 25 (SWEEP_YIELD_BATCH_SIZE:83), cada 25 tasks procesadas se cede el event loop, evitando bloquear el thread principal en limpiezas grandes.
• ** juicio de stale running**: TASK_STALE_RUNNING_MS = 30 * 60_000 — una task en running sin ningún evento de actualización (incluido recordTaskRunProgressByRunId) durante más de 30 minutos solo entonces el sweeper empieza a recolectar. Da ventana holgada a tareas lentas, no mata runs normales largos por error.
• Tolerancia a fallos del hook de recovery: tryRecoverTaskBeforeMarkLost (tryRecoverTaskBeforeMarkLost:134) envuelve todo el hook en try/catch — cualquier throw de plugin, objeto inválido devuelto, o superación del umbral de 5s solo suelta warn y luego continúa hacia mark-lost. Razón: recovery es best-effort, un plugin roto no puede bloquear la limpieza.
• retention dual: las tasks en estado terminal se conservan 7 días, las lost solo 1 día (resolveTaskRetentionMs:5-9). lost es un «estado terminal inferido por el sweeper», con menor confianza que el succeeded/failed explícitamente reportado por runtime, así que se conserva menos — da ventana para investigar sin que «tareas fantasma» ocupen DB a largo plazo.
• Degradación ante fallo de one-task flow: cualquier paso de ensureSingleTaskFlow que lance devuelve la task original (sin reenvoltorio), con warn en log. Que la task se cree es más importante que el envoltorio flow completo.
• Distinción scopeKind: las tasks con scopeKind: "system" no tienen requesterSessionKey (system-scope requesterSessionKey:92-94), ownerKey es el único ancla de búsqueda — evita que una task de sistema colisione por ownerKey con alguna sesión.
• cron-task-cancel settlement grace: al reiniciar el gateway, abortActiveCronTaskRuns (abortActiveCronTaskRunSettlementGrace:50) arranca un período de gracia de 60 segundos, dejando que las promises abortadas terminen de limpiar recursos y escribir estado terminal en finally — si se limpia la tabla directamente, los cleanups se pierden.

Resumen

El subsistema Tasks abstrae «un trabajo asincrónico» en un TaskRecord con 7 estados; cuatro runtimes (subagent/acp/cli/cron) comparten la misma máquina de estados + persistencia SQLite + recolector sweeper. Tiene una relación bidireccional con Cron: tareas programadas: cron se cuelga al índice proceso vía registerActiveCronTaskRun, y task da a cron la capacidad de «interrumpir active run al reiniciar» vía cron-task-cancel.ts. Los prompts detached disparados por ACP también usan la misma maquinaria (ver ACP: puente con el IDE), solo con runtime marcado como "acp". El estado asociado a sesión y el delivery de vuelta al cliente usan los paths de session store de Archivos de memoria.