OpenClaw任务恢复工具：为AI智能体长任务提供断点续传与自动恢复

1. 项目概述

如果你用过OpenClaw这类AI智能体工作空间，肯定遇到过这样的场景：一个需要跑通宵的数据分析任务，第二天早上发现它卡在某个步骤不动了；或者一个生成长篇报告的流程，在渲染最终PDF时因为内存不足崩了，前面几个小时的数据收集和清洗工作全白费了。这种“长任务执行不可靠”的问题，是当前许多AI自动化工具在实际生产环境落地时最大的痛点之一。openclaw-task-recovery这个项目，就是为了解决这个痛点而生的。它不是要取代OpenClaw本身的任务调度，也不是要引入一个臃肿的工作流引擎，而是在现有工作空间之上，增加一个轻量级、持久化的任务恢复层。你可以把它理解为你工作空间里的一个“任务保险丝”或“断点续传”系统，专门负责盯住那些长时间运行的任务，一旦发现它们“失联”或“卡住”，就尝试从最近的检查点安全地恢复执行，而不是让整个任务从头再来。

这个工具的核心用户，是那些已经在用OpenClaw进行自动化工作，但苦于长任务稳定性问题的开发者和自动化工程师。它特别适合处理那些超过10分钟、可以划分为多个阶段、并且每个阶段都能安全地设置检查点的任务。比如，一个典型的应用场景是：你让OpenClaw去爬取某个网站的所有产品信息，进行数据清洗，然后生成一份市场分析报告。这个过程可能持续数小时。有了openclaw-task-recovery，你可以在数据爬取完成、数据清洗完成等关键节点设置检查点。即使中间网络波动或者OpenClaw进程意外重启，系统也能从最后一个成功的检查点继续，而不是重新爬取所有数据。

2. 核心设计思路与架构拆解

2.1 问题根源：为什么长任务容易“掉链子”？

在深入代码之前，我们先要搞清楚长任务失败的常见原因。这不仅仅是OpenClaw的问题，而是所有基于AI智能体的自动化任务面临的共性挑战。

第一，状态丢失。很多AI智能体在设计上是“无状态”或“弱状态”的。它们接收输入，产生输出，但任务执行的中间状态（比如已经处理了100条数据中的哪30条）通常只存在于内存或临时的上下文中。一旦进程中断，这些状态就灰飞烟灭。

第二，执行黑盒。一个复杂的多步骤任务，对外部观察者来说就像一个黑盒。你只知道它开始了，但不知道它现在在哪个阶段，是否健康，或者已经卡死了多久。缺乏可见性使得问题无法被及时发现和干预。

第三，缺乏安全的恢复点。即使你知道任务卡住了，如何安全地恢复也是个难题。有些操作（比如发送邮件、删除文件）是不可逆的，不能简单地重试。你需要明确知道任务可以安全地从哪个点重新开始。

openclaw-task-recovery的设计正是针对这三个痛点。它通过“运行卡片”来持久化任务状态，通过“心跳看门狗”来提供任务健康度可见性，通过“检查点”和“技能适配器”来定义安全的恢复边界。

2.2 架构总览：一个轻量级的恢复层

这个项目的架构非常清晰，它没有试图接管OpenClaw的调度，而是巧妙地“寄生”在现有工作流之上。整个系统的运行流程可以概括为以下几步：

1. 任务启动：当一个长任务开始时，首先通过task_runtime.py脚本创建一个“运行卡片”。这个卡片是一个JSON文件，记录了任务ID、类型、当前阶段、状态、允许的自动恢复策略等信息，并保存在data/task-runs/目录下。这解决了状态持久化的问题。
2. 阶段推进与检查点：在任务执行过程中，每当完成一个可以安全重试的阶段（比如“数据下载完成”），就调用task_runtime.py checkpoint命令，更新运行卡片中的阶段、状态，并可以关联产出物（如文件路径）。这相当于在游戏里手动存了个档。
3. 心跳监控：OpenClaw工作空间通常有一个HEARTBEAT.md文件或类似机制，用于定期报告状态。openclaw-task-recovery安装后，会向这个心跳机制注入一个检查项。这个检查项会定期（比如每分钟）执行task_runtime_watch.py脚本。
4. 看门狗工作：task_runtime_watch.py这个看门狗脚本每次被心跳触发时，就会扫描data/task-runs/目录下所有状态为“运行中”但最近没有更新的运行卡片。如果某个卡片超过预设的“陈旧时间”（例如30分钟），看门狗就判定这个任务“失联”了。
5. 安全恢复尝试：对于失联的任务，看门狗不会盲目重启整个任务。它会读取该任务运行卡片中指定的resume_adapter（恢复适配器）路径。这个适配器是一个你为具体任务编写的Python脚本，它知道这个任务的业务逻辑，并且明确知道哪些阶段（比如“数据收集”）是安全且可以重试的。看门狗调用task_runtime_resume.py，由这个通用分发器去执行对应的适配器脚本。
6. 适配器执行恢复：你的适配器脚本被调用，它接收任务ID，加载对应的运行卡片，根据卡片中记录的最后一个检查点阶段，执行相应的恢复逻辑（例如，如果最后一个检查点是“数据清洗完成”，那么适配器就从“生成报告”阶段开始）。恢复成功后，适配器更新运行卡片的状态和时间戳。
7. 信号反馈：整个监控和恢复过程的结果，会被归类到三个清晰的信号桶中，并反馈到心跳输出里：
   • alerts：本次心跳检查发现了新的失联任务。
   • recoveries：本次心跳成功自动恢复了一个或多个任务。
   • needs_attention：有任务失联了，但看门狗无法安全恢复（例如适配器恢复失败、重试次数用尽），需要人工介入。

这样，你就从一个被动的“任务好像挂了”的状态，变成了一个主动的“任务失联了，但系统正在尝试恢复，或者已经恢复成功，或者明确需要你来看看”的状态。整个架构的轻量之处在于，它只是增加了几个脚本和一个数据目录，恢复逻辑的核心（适配器）还是由你自己根据业务来定义，系统只负责触发和协调。

3. 核心组件与实操要点

3.1 运行卡片：任务的“身份证”与“病历本”

运行卡片是openclaw-task-recovery的基石。它本质上是一个JSON文件，但你可以把它理解成任务的“身份证”和“病历本”的结合体。

实操要点：创建一张运行卡片

假设我们有一个名为“夜间市场报告生成”的任务。我们通过命令行来创建它的运行卡片：

python3 ~/.openclaw/workspace/scripts/task_runtime.py create \ --task-id nightly-report-$(date +%Y%m%d) \ --task-type report-generator \ --title "夜间市场数据抓取与分析报告" \ --mode background \ --phase init \ --resume-adapter ~/.openclaw/workspace/skills/market-report/scripts/task_resume.py \ --allow-auto-resume \ --stale-after-minutes 45 \ --max-retries 3

我们来拆解每个参数的作用和背后的考量：

• --task-id: 任务的唯一标识符。这里使用了nightly-report-加当天日期的格式，确保每天的任务ID不同，避免冲突。注意：ID最好具备唯一性和可读性。
• --task-type和--title: 用于分类和人工阅读。task-type可以用于在看门狗中过滤特定类型的任务。
• --mode: 运行模式。示例中用了background，你还可以定义overnight、interactive等，适配器可以根据这个模式决定恢复策略（比如overnight模式的重试间隔可以更长）。
• --phase: 初始阶段。总是从init开始。
• --resume-adapter:这是最关键参数之一。它指向一个具体的Python脚本，这个脚本包含了如何恢复你这个特定任务的知识。路径可以是正式技能目录下的，也可以是临时目录下的。
• --allow-auto-resume: 明确允许看门狗自动恢复此任务。这是一个安全开关，对于绝对不能自动重试的任务（比如涉及支付的），就不要设置这个标志。
• --stale-after-minutes: “陈旧”阈值。这里设为45分钟，意味着如果任务超过45分钟没有更新检查点，看门狗就会认为它失联了。这个值需要根据任务正常执行时长来设定，要留出足够的缓冲，避免误报。
• --max-retries: 最大自动重试次数。防止一个永远失败的任务被无限次重试，浪费资源。达到次数后，任务状态会进入needs_attention。

创建成功后，你会在data/task-runs/目录下找到一个类似nightly-report-20231027.json的文件，里面包含了以上所有信息以及created_at、status（初始为running）、artifacts（空数组）等字段。

注意：运行卡片文件是系统的核心状态存储。严禁手动修改此文件的内容，除非你完全清楚后果。所有更新都应通过task_runtime.py脚本的checkpoint或update命令来完成，以保证状态变更的原子性和日志记录。

3.2 检查点：为你的任务手动“存档”

检查点是你在任务代码中主动插入的“存档”命令。它有两个核心作用：1）告诉系统任务成功推进到了某个阶段；2）保存该阶段的产出物引用。

实操要点：在关键阶段后设置检查点

继续上面的市场报告任务。假设我们的任务流程是：init->data_fetch->data_clean->analysis->report_render->complete。

在数据抓取完成后，我们应该设置一个检查点：

python3 ~/.openclaw/workspace/scripts/task_runtime.py checkpoint nightly-report-20231027 \ --phase data_clean \ --status running \ --artifact raw_data=/workspace/data/raw_market_20231027.json \ --message "原始数据抓取完成，共获取5000条记录，即将进入清洗阶段"

参数解析：

• 第一个参数是task_id。
• --phase: 更新任务当前阶段为data_clean。这意味着任务已经从data_fetch阶段成功过渡到了data_clean阶段。
• --status: 通常保持running。如果某个阶段是最终成功或失败，可以更新为succeeded或failed。
• --artifact: 关联产出物。这里我们把原始数据文件的路径保存下来。注意：系统只保存路径字符串，不保存文件内容本身。确保路径是工作空间内的相对路径或绝对路径，并且文件确实存在。
• --message: 人类可读的日志信息，便于后期排查。

这个命令会更新运行卡片文件，将phase改为data_clean，在artifacts数组中加入{"name": "raw_data", "path": "/workspace/..."}，并更新last_checkpoint_at时间戳。这个时间戳正是看门狗判断任务是否“陈旧”的依据。

何时设置检查点？这是一个需要仔细权衡的设计决策。检查点设置得太频繁（比如每处理10条数据就设一个）会产生大量IO，可能影响性能，且恢复逻辑复杂。设置得太稀疏（比如整个任务只有一个检查点），则失去断点续传的意义。

• 最佳实践：在完成一个耗时较长、资源消耗大且结果可持久化的阶段后立即设置。例如，“数据抓取”、“模型训练一轮结束”、“文件转换完成”。
• 必须设置：在进入一个具有副作用或不可逆操作的阶段之前。例如，在“发送邮件通知”之前，必须有一个检查点。这样如果恢复，适配器可以跳过已完成的耗时阶段，直接从“发送邮件”开始，或者因为该阶段不安全而不执行自动恢复。

3.3 心跳看门狗：系统的“巡检员”

task_runtime_watch.py是系统的守护进程，它依靠OpenClaw的HEARTBEAT.md机制定期执行。安装脚本会自动在HEARTBEAT.md中添加或更新如下部分：

## Task Runtime Watchdog - Checks for stale resumable tasks and attempts auto-recovery. - Run: `python3 /absolute/path/to/workspace/scripts/task_runtime_watch.py` - Signals: `alerts`, `recoveries`, `needs_attention`

每次心跳触发时，这个脚本就会执行。它的内部逻辑如下：

1. 加载所有data/task-runs/*.json文件。
2. 过滤出status为running且allow_auto_resume为true的卡片。
3. 对每张卡片，计算当前时间与last_checkpoint_at的差值。
4. 如果差值超过stale_after_minutes，则将此任务加入stale_tasks列表（对应alerts信号）。
5. 对于每个陈旧任务，检查其retry_count是否小于max_retries。如果是，则调用task_runtime_resume.py并传入task_id进行恢复尝试。
6. 根据恢复结果，将任务归类到recovered_tasks（恢复成功）或attention_tasks（恢复失败或重试超限）列表。
7. 将这三个列表格式化输出，作为心跳检查的结果。

实操心得：调整看门狗敏感度

• stale_after_minutes是全局的，但在create命令时按任务设置。对于不同时长的任务，应设置不同的值。一个运行2小时的任务，设置stale-after-minutes 30可能太敏感，容易误报。可以设置为120或更长。
• 心跳的执行频率也会影响发现问题的延迟。如果心跳是5分钟一次，那么一个任务卡死，最多需要stale_after_minutes + 5分钟才会被发现。在稳定性和及时性之间需要权衡。

4. 技能适配器开发实战

适配器是openclaw-task-recovery的灵魂，也是需要你投入最多开发精力的部分。它定义了“如何安全地恢复一个特定的任务”。项目提供了一个模板templates/openclaw-task-recovery/task_resume.py，这是你开发的起点。

4.1 适配器合约：你必须遵守的约定

你的适配器脚本必须满足以下合约，才能被task_runtime_resume.py正确调用：

1. 可执行性：它是一个独立的Python脚本。
2. 参数接口：必须能接受--task-id <TASK_ID>命令行参数。可选支持--timeout-seconds <TIMEOUT>。
3. 输入：脚本内部需要通过task_runtime.py提供的工具函数或直接读取data/task-runs/<TASK_ID>.json来加载运行卡片。
4. 逻辑核心：根据运行卡片中的phase、artifacts等信息，判断当前任务处于哪个阶段，并执行该阶段及后续阶段的恢复逻辑。关键原则：只恢复安全（幂等）的阶段。
5. 输出与状态更新：
   • 恢复过程中，需要适时调用checkpoint来更新进度。
   • 恢复成功结束时，应将卡片status更新为succeeded（如果任务完成）或保持running并更新phase（如果只是推进了阶段）。
   • 恢复失败时，应将卡片status更新为failed，并可选地增加retry_count。
   • 脚本应尽可能将结果（如成功信息、错误信息）以JSON格式打印到标准输出(stdout)，方便上层调用者解析。

4.2 实战案例：构建一个报告生成任务的适配器

假设我们有一个market-report技能，其task_resume.py适配器内容如下。我们通过注释来详解每一部分：

#!/usr/bin/env python3 """ 市场报告生成技能的恢复适配器。 根据运行卡片的状态，从断点恢复报告生成流程。 """ import sys import json import subprocess from pathlib import Path import argparse # 假设 task_runtime 模块在 workspace/scripts/ 下，将其加入路径 sys.path.insert(0, str(Path.home() / '.openclaw' / 'workspace' / 'scripts')) try: from task_runtime import load_run_card, update_run_card except ImportError: # 简化处理，如果找不到模块，直接使用文件操作 def load_run_card(task_id): card_path = Path.home() / '.openclaw' / 'workspace' / 'data' / 'task-runs' / f'{task_id}.json' with open(card_path, 'r') as f: return json.load(f) def update_run_card(task_id, updates): # 这里简化为直接覆盖，实际应使用原子操作 card_path = Path.home() / '.openclaw' / 'workspace' / 'data' / 'task-runs' / f'{task_id}.json' with open(card_path, 'r') as f: data = json.load(f) data.update(updates) with open(card_path, 'w') as f: json.dump(data, f, indent=2) def resume_data_clean(task_id, run_card): """恢复数据清洗阶段。此阶段是幂等的，可安全重试。""" print(f"[INFO] 恢复任务 {task_id}: 阶段 data_clean", file=sys.stderr) raw_data_path = None for artifact in run_card.get('artifacts', []): if artifact.get('name') == 'raw_data': raw_data_path = artifact.get('path') break if not raw_data_path or not Path(raw_data_path).exists(): raise FileNotFoundError(f"原始数据文件不存在: {raw_data_path}") # 调用实际的数据清洗脚本（假设是另一个Python脚本） # 这里使用subprocess模拟，实际可能是函数调用 clean_script = Path.home() / '.openclaw' / 'workspace' / 'skills' / 'market-report' / 'scripts' / 'clean_data.py' cmd = [sys.executable, str(clean_script), '--input', raw_data_path] result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) # 5分钟超时 if result.returncode != 0: raise RuntimeError(f"数据清洗失败: {result.stderr}") cleaned_path = '/workspace/data/cleaned_market.json' # 假设清洗脚本将结果输出到了 cleaned_path # 设置检查点，进入下一阶段 subprocess.run([ sys.executable, str(Path.home() / '.openclaw' / 'workspace' / 'scripts' / 'task_runtime.py'), 'checkpoint', task_id, '--phase', 'analysis', '--artifact', f'cleaned_data={cleaned_path}', '--message', '数据清洗完成，进入分析阶段' ], check=True) return {"phase": "analysis", "cleaned_data": cleaned_path} def resume_analysis(task_id, run_card): """恢复数据分析阶段。此阶段也是幂等的。""" print(f"[INFO] 恢复任务 {task_id}: 阶段 analysis", file=sys.stderr) # ... 类似地，加载 cleaned_data，执行分析脚本 ... # 分析完成后，设置检查点进入 report_render subprocess.run([...], check=True) # 调用 checkpoint return {"phase": "report_render"} def resume_report_render(task_id, run_card): """恢复报告渲染阶段。注意：此阶段可能涉及覆盖文件，但通常认为是安全的。""" print(f"[INFO] 恢复任务 {task_id}: 阶段 report_render", file=sys.stderr) # ... 执行报告生成 ... # 完成后，将状态标记为成功 subprocess.run([ sys.executable, str(Path.home() / '.openclaw' / 'workspace' / 'scripts' / 'task_runtime.py'), 'update', task_id, '--status', 'succeeded', '--message', '市场报告生成完成' ], check=True) return {"status": "succeeded"} def main(): parser = argparse.ArgumentParser(description='恢复市场报告生成任务') parser.add_argument('--task-id', required=True, help='要恢复的任务ID') parser.add_argument('--timeout-seconds', type=int, default=600, help='恢复操作超时时间') args = parser.parse_args() task_id = args.task_id try: run_card = load_run_card(task_id) except Exception as e: print(json.dumps({"error": f"加载运行卡片失败: {str(e)}", "needs_attention": True})) sys.exit(1) current_phase = run_card.get('phase', 'init') resume_result = {} # 根据当前阶段，路由到对应的恢复函数 # 这是适配器的核心决策逻辑 try: if current_phase == 'data_clean': resume_result = resume_data_clean(task_id, run_card) elif current_phase == 'analysis': resume_result = resume_analysis(task_id, run_card) elif current_phase == 'report_render': resume_result = resume_report_render(task_id, run_card) elif current_phase == 'init': # 如果还在初始阶段，可能任务根本没开始，或者需要从头开始。 # 这里我们选择从头开始执行数据抓取（如果安全的话）。 # 更保守的做法是标记为 needs_attention，让用户决定。 print(json.dumps({"warning": "任务处于初始阶段，自动恢复可能需从头开始。建议手动检查。", "needs_attention": True})) sys.exit(0) else: print(json.dumps({"error": f"未知或无法恢复的阶段: {current_phase}", "needs_attention": True})) sys.exit(1) # 恢复成功，输出结果 print(json.dumps({"success": True, "task_id": task_id, "result": resume_result})) except subprocess.TimeoutExpired: print(json.dumps({"error": f"恢复操作超时 ({args.timeout-seconds}s)", "needs_attention": True})) sys.exit(1) except Exception as e: # 任何其他异常，更新运行卡片为失败，并增加重试计数 error_msg = str(e) print(json.dumps({"error": error_msg, "needs_attention": True})) # 注意：在实际中，update_run_card 应该处理并发和原子性 new_retry = run_card.get('retry_count', 0) + 1 update_run_card(task_id, {'status': 'failed', 'retry_count': new_retry, 'last_error': error_msg}) sys.exit(1) if __name__ == '__main__': main()

代码解读与避坑指南：

1. 阶段路由逻辑：main()函数中的if-elif链是适配器的“大脑”。它根据run_card['phase']决定从哪个阶段开始恢复。务必确保你的阶段定义是清晰且线性的，避免出现环形依赖。
2. 幂等性检查：每个恢复函数（如resume_data_clean）的开头，都应该检查所需的前置产出物（artifacts）是否存在且有效。如果raw_data文件丢失，恢复应该立即失败并报错，而不是尝试重新抓取（除非你的设计允许）。
3. 安全边界：注意resume_report_render函数的注释。报告渲染可能会覆盖已有的报告文件。你需要评估这个操作对你的业务是否“安全”。如果不安全（比如每次渲染都应生成新文件），那么应该在这个阶段禁止自动恢复（allow-auto-resume: false），或者让适配器在渲染前检查文件是否存在并采取相应措施（例如重命名旧文件）。
4. 错误处理与状态更新：在except Exception as e块中，我们不仅打印错误，还尝试更新运行卡片状态为failed并增加retry_count。这确保了看门狗下次心跳时，能通过重试次数判断是否该放弃并标记为needs_attention。这是一个关键模式，保证了失败状态能被持久化。
5. 输出标准化：适配器使用JSON格式输出结果和错误。这使得task_runtime_resume.py或上层调用者可以轻松解析结果，并根据success、error、needs_attention等字段决定后续动作。

4.3 临时任务适配器：快速原型利器

你并不需要为每一个临时性的长任务都创建一个正式的技能。openclaw-task-recovery鼓励一种“渐进式”的工作流。

实操：为一键式临时任务创建适配器

假设你临时需要让OpenClaw处理一个非常耗时的日志分析，你可以直接使用项目提供的“一键式提示词”：

Run this as a resumable long task under openclaw-task-recovery in the current OpenClaw workspace. If no skill exists yet, scaffold a temporary adapter undertmp/task-runtime/<task-slug>/task_resume.py, create a run card, checkpoint the safe phases, enable heartbeat auto-resume, and at minimum leave me either a final result, a partial result, or a clear block report: <你的具体任务描述，例如：“分析 /var/log/app/ 下所有 .log 文件，找出 ERROR 级别的日志，按小时聚合，并生成摘要报告”>

当你向OpenClaw发送这条指令后，一个典型的自动化过程会发生：

1. OpenClaw会先在tmp/task-runtime/log-analysis-20231027/目录下，根据模板生成一个临时的task_resume.py适配器。
2. 它会为这个任务创建一个运行卡片，resume-adapter字段就指向这个临时适配器。
3. 然后开始执行你的任务。在执行过程中，它会将任务逻辑分解成几个阶段（比如file_discovery,error_extraction,aggregation,reporting），并在每个阶段完成后自动插入checkpoint命令。
4. 如果任务中途失败或停滞，心跳看门狗会检测到，并调用这个临时适配器来恢复。临时适配器里的恢复逻辑，可能就是OpenClaw在最初分解任务时一起生成的。

这样做的好处是：你无需事先设计好完整的恢复逻辑。你可以先快速跑起来，让系统帮你搭建框架。如果这个临时任务后来变成了一个经常需要执行的例行任务，你可以将tmp/task-runtime/log-analysis-20231027/目录下的内容进行整理、优化，然后移动到skills/log-analyzer/scripts/目录下，将其“晋升”为一个正式的、可复用的技能。这完美契合了从原型到产品的迭代路径。

5. 部署、集成与日常运维指南

5.1 安装与初始化配置

手动安装步骤在项目README中已经给出，但我想强调几个容易出错的细节：

# 1. 克隆仓库 - 注意目标路径 # 假设你的OpenClaw工作空间根目录是 ~/my_openclaw_workspace git clone https://github.com/m0x14o/openclaw-task-recovery ~/my_openclaw_workspace/repos/openclaw-task-recovery # 2. 进入目录并安装 cd ~/my_openclaw_workspace/repos/openclaw-task-recovery python3 install.py

安装过程详解：install.py脚本会做以下几件事，你可以打开它查看具体逻辑：

• 复制脚本：将scripts/目录下的task_runtime.py，task_runtime_watch.py，task_runtime_resume.py复制到工作空间的scripts/目录下。确保你的工作空间scripts/目录在系统的PATH环境变量中，或者你总是使用绝对路径调用它们。
• 复制模板和文档：将templates/和docs/下的内容复制到工作空间对应目录，方便查找。
• 修改HEARTBEAT.md：这是最关键的一步。脚本会查找HEARTBEAT.md文件，并在其中添加或更新一个“Task Runtime Watchdog”部分。请务必在安装后检查一下这个文件，确认添加成功，格式正确。如果心跳机制不是通过HEARTBEAT.md，而是其他方式（比如一个定时调用的脚本），你需要手动将python3 /path/to/scripts/task_runtime_watch.py这一行集成到你的心跳触发机制中。

注意：安装脚本通常是幂等的，多次运行不会造成问题。但建议在首次安装后，备份一下你的HEARTBEAT.md文件。

5.2 与现有技能和自动化流程集成

如果你已经有一些运行在OpenClaw上的技能或自动化脚本，想要为它们增加恢复能力，可以遵循以下模式：

1. 识别长任务：审视你的技能，找出那些运行时间超过10分钟，或者虽然短但失败成本高的任务。
2. 定义阶段：将这个任务分解成几个逻辑阶段。阶段划分的原则是：阶段之间有明显的数据或状态产出，且每个阶段内部的操作尽可能幂等。
3. 创建适配器：为这个技能创建scripts/task_resume.py文件。初期可以简单一些，只实现从第一个阶段开始的恢复。随着使用深入，再逐步完善多阶段恢复逻辑。
4. 修改任务启动逻辑：在原来直接调用核心业务代码的地方，改为先创建运行卡片，然后启动业务代码，并在业务代码的关键节点插入检查点调用。
   • 之前：python3 my_skill_main.py --arg1 value
   • 之后：

     # 1. 创建运行卡片 TASK_ID="my-skill-$(date +%s)" python3 scripts/task_runtime.py create \ --task-id $TASK_ID \ --task-type my-skill \ ... # 其他参数 # 2. 执行核心业务，并将任务ID传递进去 python3 my_skill_main.py --arg1 value --task-id $TASK_ID

     在my_skill_main.py内部，在完成每个阶段后调用scripts/task_runtime.py checkpoint $TASK_ID ...。
5. 更新文档：在你的技能README中，添加一小节说明如何利用任务恢复功能，特别是--allow-auto-resume的使用场景和禁忌。

5.3 监控与排查：看懂看门狗的信号

集成成功后，你的HEARTBEAT.md输出会变得更有信息量。你需要学会解读看门狗发出的信号：

## Task Runtime Watchdog - Checks for stale resumable tasks and attempts auto-recovery. - Run: `python3 /.../scripts/task_runtime_watch.py` - Signals: `alerts`, `recoveries`, `needs_attention` **alerts**: - `nightly-report-20231027` (stale for 61m, phase: `data_clean`) **recoveries**: - `nightly-report-20231027` resumed from phase `data_clean` to `analysis`. **needs_attention**: - (空)

解读：

• alerts：发现任务nightly-report-20231027在data_clean阶段已经停滞61分钟（超过设定的45分钟阈值），因此发出了警报。
• recoveries：看门狗成功恢复了该任务，使其从data_clean阶段推进到了analysis阶段。
• needs_attention：当前没有需要人工干预的任务。

这是一个理想的自动恢复场景。如果恢复失败，你可能会看到：

**alerts**: - `nightly-report-20231027` (stale for 61m, phase: `data_clean`) **recoveries**: - (空) **needs_attention**: - `nightly-report-20231027` - Auto-resume failed at phase `data_clean`: [Errno 2] No such file or directory: '/workspace/data/raw_market_20231027.json'. Retry 1/3.

这里显示恢复失败，原因是所需的原始数据文件找不到了。看门狗会按照max-retries配置进行重试。如果重试次数用尽，任务将永久停留在needs_attention列表，等待你手动检查。

日常运维检查清单：

• 定期查看data/task-runs/目录：清理已经完成（succeeded或failed）的旧任务卡片，避免文件堆积。可以写一个简单的清理脚本，定期删除比如7天前的卡片。
• 关注needs_attention：这是你需要介入的主要入口。一旦有任务出现在这里，应立即检查对应卡片的last_error字段和日志，判断是数据问题、环境问题还是适配器逻辑错误。
• 审查适配器日志：确保你的适配器脚本将足够的调试信息输出到标准错误(stderr)，这些信息通常会出现在OpenClaw或你的任务调度器的日志中。
• 调整参数：根据实际运行情况，优化stale-after-minutes和max-retries这两个参数。对于网络依赖重的任务，可以适当调大stale-after-minutes和重试次数。

6. 常见问题与故障排除实录

在实际使用中，你可能会遇到一些典型问题。以下是我在部署和使用过程中踩过的一些坑以及解决方案。

6.1 问题：心跳看门狗没有执行

症状：任务卡住了，但HEARTBEAT.md的输出中没有出现Task Runtime Watchdog的信号部分。排查步骤：

1. 检查安装：确认scripts/task_runtime_watch.py文件是否存在于工作空间的scripts/目录下。
2. 检查HEARTBEAT.md：打开HEARTBEAT.md，查看是否包含了调用task_runtime_watch.py的命令行。确保路径是正确的绝对路径。
3. 手动测试：在终端手动执行HEARTBEAT.md中那条命令（例如python3 /path/to/scripts/task_runtime_watch.py）。观察是否有错误输出。常见的错误是Python依赖缺失，确保工作空间的Python环境已安装所需模块（通常只是标准库，但如果你修改了脚本可能会引入依赖）。
4. 检查心跳机制：确认你的OpenClaw实例确实在定期执行HEARTBEAT.md中的检查项。有些部署可能心跳机制不同，需要你将看门狗调用集成到对应的定时任务中。

6.2 问题：任务被误判为陈旧并恢复，但实际仍在运行

症状：一个本该运行2小时的任务，在运行1小时后就出现在alerts和recoveries中，导致同一个任务被重复执行，产生冲突。原因：stale-after-minutes设置得太短，小于任务单个阶段的正常执行时间。解决方案：

1. 合理设置阈值：在创建运行卡片时，根据任务的历史运行数据或预估，为每个任务设置合理的--stale-after-minutes。对于长时间任务，可以设置几小时甚至更长。
2. 精细化检查点：如果任务整体时间长，但内部有很多短阶段，可以在每个短阶段结束后都设置检查点。这样，即使某个短阶段卡住，也能被及时检测到，而不会因为整个任务阈值设得长而掩盖问题。
3. 使用“心跳”检查点：对于那种长时间没有明显阶段划分的单一操作（例如训练一个大型模型），你可以在任务循环内部，定期（比如每10分钟）调用一个“空”的检查点，只更新last_checkpoint_at时间戳，而不改变phase。这相当于向看门狗报告“我还活着”。你需要稍微修改你的任务代码来支持这一点。

6.3 问题：自动恢复后，任务状态或数据出现不一致

症状：任务恢复后，产生了重复的数据、漏处理了部分数据，或最终结果错误。原因：根本原因在于恢复逻辑（适配器）的幂等性设计有缺陷。排查与解决：

1. 审查阶段划分：检查你的阶段划分是否在逻辑上是“无状态”的。一个理想的阶段，其输出应完全由该阶段的输入决定，且重复执行多次结果不变。例如，“从API获取第X页到第Y页的数据”是幂等的（如果API不变）。“处理队列中的下一条消息”就不是幂等的，因为恢复后“下一条”可能已经变了。
2. 审查适配器逻辑：仔细检查适配器中每个恢复函数的开头。它是否严格依赖运行卡片artifacts中记录的、由上一个检查点产出的文件或数据？它是否假设了某些内存中的状态或外部环境状态？恢复函数应该只基于持久化的artifacts和phase来决定做什么。
3. 引入幂等键：对于无法做到完全幂等的操作，可以引入“幂等键”。例如，在发送通知时，将(任务ID, 阶段名, 接收方)作为唯一键存入一个外部数据库或文件。在恢复函数执行发送前，先检查这个键是否已存在，如果存在则跳过。这需要额外的状态存储，但能保证安全。
4. 彻底测试恢复流程：不要只测试任务的成功路径。主动在任务执行过程中模拟故障（如杀死进程、断网），然后触发看门狗恢复，观察结果是否符合预期。这是保证可靠性的唯一方法。

6.4 问题：needs_attention列表中的任务堆积，手动处理麻烦

症状：很多失败的任务最终都进入了needs_attention，需要人工一个个去查看日志、分析原因、决定是重试还是放弃。解决方案：

1. 优化错误信息：确保你的适配器在失败时，不仅更新status: failed，还在last_error字段或自定义字段中写入足够详细的错误信息，甚至包括堆栈跟踪。这样你无需查看分散的日志文件，在运行卡片里就能初步判断问题。
2. 实现一个简单的管理界面：可以写一个简单的Python脚本或Web页面，读取data/task-runs/目录，过滤出status: failed且retry_count >= max_retries的任务，并以表格形式展示task_id,phase,last_error,created_at等信息。你甚至可以在这个界面中加入“强制重试”（将retry_count清零并重置状态）或“标记为完成并忽略”的按钮。
3. 设置告警：将needs_attention视为一种告警信号。你可以修改task_runtime_watch.py脚本，当needs_attention列表非空时，除了在心跳中显示，还可以发送邮件、Slack消息或其他通知，提醒你及时处理。

6.5 性能与扩展性考量

对于任务量非常大的场景，当前的简单文件系统存储（每个任务一个JSON文件）可能会遇到性能瓶颈。

• 目录扫描：task_runtime_watch.py每次心跳都要扫描整个data/task-runs/目录。如果存在数万个文件，可能会影响心跳速度。可以考虑定期归档已完成的任务卡片，或者为运行中的任务卡片使用单独的目录。
• 状态更新并发：虽然概率不高，但如果多个进程同时尝试更新同一个运行卡片（比如任务主进程和看门狗恢复进程同时写），可能会损坏JSON文件。目前的实现没有处理文件锁。对于高并发场景，你需要增强task_runtime.py中的update_run_card函数，使用文件锁（fcntl或portalocker）或考虑使用更健壮的存储后端（如SQLite，甚至Redis）。不过，对于OpenClaw工作空间这种通常单任务串行或低并发的场景，文件系统的简单性往往是更可取的。

7. 设计哲学与进阶思考

openclaw-task-recovery的设计体现了一种“简约而足够”的哲学。它没有选择去实现一个完整的、功能繁复的工作流引擎，而是精准地抓住了“持久化状态”和“安全恢复”这两个核心需求。这种设计选择带来了几个显著优势：

1. 低侵入性：它不需要你改造现有的OpenClaw技能的核心逻辑。你只需要在任务的外围加上创建卡片、设置检查点的调用，并额外编写一个恢复适配器。原有的业务代码几乎可以保持不变。

2. 技术栈亲和：它完全用Python实现，使用JSON文件存储状态，通过命令行调用。这与OpenClaw工作空间本身的技术栈和操作模式高度一致，学习成本和集成成本极低。

3. 概念清晰：运行卡片、检查点、适配器、看门狗，这几个核心概念非常直观，容易理解和应用。这降低了团队成员协作和维护的心智负担。

4. 渐进式采用：你可以先从最重要的一个长任务开始试用，获得收益后再逐步推广到其他任务。临时任务适配器的设计更是支持了“先用起来，再规范化”的工作流。

当然，这种简约设计也意味着它有一些天然的边界。它不适合需要复杂分支、循环、并行执行的工作流。它不处理任务间的依赖关系。它的调度完全依赖外部机制（如cron、OpenClaw会话）。但正如项目作者所言，它本意就不是一个通用的DAG引擎，而是一个专为OpenClaw工作空间定制的恢复层。

我个人在实际使用中的体会是，这个工具最大的价值在于它改变了你和长任务之间的关系。以前，启动一个耗时任务后，你心里总会有点忐忑，时不时要去看看日志，确认它还在跑。现在，你可以更安心地启动任务然后去忙别的，因为你知道有一个“保险丝”在盯着。即使出了问题，系统也会明确地告诉你“需要你来看看了”，而不是无声无息地失败。这种从“被动监控”到“主动恢复+明确告警”的转变，对于提升自动化流程的可靠性和开发者的心理舒适度，效果是立竿见影的。它就像给你的自动化脚本加了一个贴心的副驾驶，平时不打扰你，但在关键时刻会帮你稳住方向盘。