Tasks: tâches persistées

源码版本v2026.6.11

Responsabilités

task-executor.ts et detached-task-runtime.ts forment le sous-système de « tâches » (task) d'OpenClaw: modéliser tout travail asynchrone potentiellement cross-process, cross-session, cross-restart comme un enregistrement avec un cycle de vie, quel que soit le runtime (subagent / acp / cli / cron) où il tourne; tous partagent la même machine à états (state machine), la même persistance SQLite, les mêmes sémantiques d'annulation et de récupération. CronService gère le timing de scheduling, task-executor.ts gère « comment un enregistrement passe de queued à succeeded/failed/timed_out/cancelled/lost ».

Une tâche n'est ni un thread ni un worker — c'est juste un objet d'état (TaskRecord, TaskRecord:116-146) plus les hooks de cycle de vie implémentés par le runtime correspondant. Le runtime (subagent/acp/cli/cron, TaskRuntime:5) décide comment exécuter réellement; le système de tâches ne gère que: stocker l'enregistrement, faire les transitions d'état, livrer à la session d'origine, nettoyer périodiquement les enregistrements expirés.

Motivation de conception

Pourquoi cron ne suffit pas et pourquoi extraire une couche task?

1. Cross runtime: cron est temporisé, ACP est un prompt déclenché par IDE, subagent est une sous-tâche dérivée par la boucle principale agent, CLI est une commande tapée par l'utilisateur dans le terminal. Les quatre runtimes ont des timings de déclenchement complètement différents, mais doivent tous affronter les mêmes problèmes: comment savoir après un crash process qu'un travail n'est pas terminé? Comment récupérer après restart gateway? Comment le résultat revient à la session d'origine? Extraire ces problèmes en task registry laisse chaque runtime n'implémenter que son DetachedTaskLifecycleRuntime (DEFAULT_DETACHED_TASK_LIFECYCLE_RUNTIME:35-49), la logique commune partagée.
2. Concept de flow: beaucoup de tâches appartiennent à un flow plus large (par exemple un agent run déclenche trois sous-tâches; à l'annulation de la tâche principale, les trois sous-tâches doivent aussi être annulées). task-flow-registry (src/tasks/task-flow-registry.ts) donne aux tâches un flow parent, permet batch cancel et batch status. Pour les detached ACP/subagent run, l'executor enveloppe automatiquement en « one-task flow » (isOneTaskFlowEligible + ensureSingleTaskFlow:46-92), pour que ces runs ponctuels profitent aussi du surface état/retry du flow.
3. État delivery: après la fin de la tâche, le résultat ne doit pas forcément revenir à la session d'origine — l'utilisateur peut avoir fermé l'IDE, le cron peut être isolated session sans session d'origine. TaskDeliveryStatus (TaskDeliveryStatus:16-23) suit cet état séparément: pending/delivered/session_queued/failed/parent_missing/not_applicable, découplant « tâche exécutée avec succès » et « résultat livré au client ».

Fichiers clés

• task-registry.types.ts:1-146 — Toutes les déclarations de types TaskRuntime/TaskStatus/TaskDeliveryStatus/TaskNotifyPolicy/TaskRecord.
• task-executor.ts:46-127 — ensureSingleTaskFlow + createQueuedTaskRun + createRunningTaskRun, entrées de création de tâche (incluant le wrapping one-task flow).
• detached-task-runtime.ts:32-66 — Interface DetachedTaskLifecycleRuntime, runtime par défaut, mécanisme d'enregistrement plugin.
• tryRecoverTaskBeforeMarkLost:134-171 — Hook de récupération: best-effort, alerte 5s threshold, toute erreur repli vers mark-lost.
• task-registry.process-state.ts:1-32 — Index in-memory au niveau process (tasks/taskIdsByRunId/taskIdsByOwnerKey/taskIdsByParentFlowId/taskIdsByRelatedSessionKey/tasksWithPendingDelivery), accroché à globalThis via Symbol.for("openclaw.taskRegistry.state").
• task-registry.store.sqlite.ts:48-77 — Tables Kysely-mappées task_runs + task_delivery_state, miroir dans openclaw-state.db.
• task-registry.maintenance.ts:73-100 — Sweeper arrière-plan, cycle 60s, TASK_STALE_RUNNING_MS=30*60_000, TASK_RECONCILE_GRACE_MS=5*60_000.
• task-retention.ts:4-29 — Tâches en état terminal conservées 7 jours, lost seulement 1 jour.
• src/tasks/task-flow-registry.store.sqlite.ts — Persistance flow, supporte cascade parentFlowId.
• cron-task-cancel.ts:12-77 — Handles d'annulation au niveau process pour active cron task runs, settlement grace 60s.
• task-registry.reconcile.ts — Facade publique: reconcileInspectableTasks / reconcileTaskLookupToken.

Flux de données

La machine à états des tâches n'a que 7 états (TaskStatus:7-14):

export type TaskStatus =
  | "queued"       // en base, en attente que le runtime le tire
  | "running"      // runtime a signalé le début d'exécution
  | "succeeded"    // état terminal: succès
  | "failed"       // état terminal: échec (avec error)
  | "timed_out"    // état terminal: timeout
  | "cancelled"    // état terminal: annulé par utilisateur/système
  | "lost";        // état terminal: sweeper après grace n'a pas observé d'activité, marqué perdu

createQueuedTaskRun / createRunningTaskRun sont les entrées de création, les deux appellent 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;
  }
}

Jugement clé: scopeKind === "session" et runtime ∈ {acp, subagent} et deliveryStatus !== "not_applicable" — quand ces trois conditions sont remplies, on enveloppe automatiquement d'un flow. Les tâches système (scopeKind: "system") et cron ont leur propre chemin cancel/delivery, ne repassent pas par flow. En cas d'échec, warn sans throw, garantit que la création de la tâche n'est pas bloquée par un échec de wrapping flow.

detached-task-runtime.ts (getDetachedTaskLifecycleRuntime:47) expose une API unifiée, mais l'implémentation peut être remplacée par 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() privilégie l'implémentation enregistrée par plugin, sinon default — permet à un plugin de prendre en charge tout le cycle de vie task (par exemple faire tourner les tâches dans un worker process séparé) sans modifier task-executor.

Limites et modes d'échec

• globalThis symbol-keyed pour l'index au niveau process: getTaskRegistryProcessState() (getTaskRegistryProcessState:18) utilise Symbol.for("openclaw.taskRegistry.state") accroché à globalThis, ainsi les hot reload de module (dev watch) ou les isolations de tests partagent l'index — sinon différentes instances du module maintiennent chacune leur index in-memory, et les tâches persistées en SQLite ne matchent plus l'index mémoire.
• Sweeper batch yield: SWEEP_YIELD_BATCH_SIZE = 25 (SWEEP_YIELD_BATCH_SIZE:83), tous les 25 tâches traitées on yield l'event loop, pour éviter de bloquer le thread principal lors des nettoyages massifs.
• Jugement stale running: TASK_STALE_RUNNING_MS = 30 * 60_000 — running state sans aucune mise à jour d'événement pendant 30 min (y compris recordTaskRunProgressByRunId), le sweeper déclenche la récupération. Cela laisse une fenêtre large pour les tâches longues, sans tuer par erreur les longs runs normaux.
• Tolérance d'erreur du hook recovery: tryRecoverTaskBeforeMarkLost (tryRecoverTaskBeforeMarkLost:134) enveloppe l'appel de hook dans un try/catch — tout throw de plugin, retour d'objet invalide, dépassement du seuil 5s ne donne lieu qu'à un warn, puis continue vers mark-lost. Raison: recovery est best-effort, un plugin cassé ne doit pas bloquer le nettoyage.
• Rétention double voie: tâches en état terminal 7 jours, lost seulement 1 jour (resolveTaskRetentionMs:5-9). Lost est « un état terminal déduit par le sweeper », moins fiable que les états succeeded/failed explicitement rapportés par runtime, donc durée de rétention plus courte — donner une fenêtre d'investigation sans laisser les « ghost tasks » encombrer la DB.
• Dégradation one-task flow en échec: ensureSingleTaskFlow retourne le task original (sans re-wrap) sur n'importe quelle erreur, warn dans les logs. La création de la tâche réussir est plus important que l'intégrité du wrapping flow.
• Distinction scopeKind: scopeKind: "system" n'a pas de requesterSessionKey (system-scope requesterSessionKey:92-94), ownerKey est le seul ancrage de recherche — évite que les tâches système et l'ownerKey d'une session entrent en collision.
• settlement grace cron-task-cancel: au restart gateway, abortActiveCronTaskRuns (abortActiveCronTaskRunSettlementGrace:50) lance une « période de grâce de settlement » de 60 secondes, pour laisser les promises aborts le temps de cleaner leurs ressources dans le finally et d'écrire leur état terminal — un clear direct de la table ferait perdre les cleanups.

Résumé

Le sous-système Tasks abstrait « un travail asynchrone » en TaskRecord 7 états, quatre runtimes (subagent/acp/cli/cron) partageant une machine à états + persistance SQLite + sweeper de récupération. Relation bidirectionnelle avec Cron: tâches planifiées: cron s'enregistre dans la table au niveau process du task via registerActiveCronTaskRun, le système task fournit à cron via cron-task-cancel.ts la capacité « au restart, interrompre les active runs ». Les detached prompts déclenchés par ACP passent par le même système (voir ACP: pont IDE), juste runtime étiqueté "acp". L'état associé à la session et le delivery task remontent au chemin de session store dans fichiers de mémoire.