企业官网建设流程全解析

1. 项目概述：为OpenClaw构建一个轻量级的任务恢复层

如果你用过OpenClaw这类AI智能体平台，肯定遇到过这种头疼的情况：一个需要跑好几个小时甚至通宵的复杂任务，比如批量分析数据、生成长篇报告或者执行多步骤的代码审查，跑着跑着就“卡住”了。第二天早上回来一看，终端没响应了，日志停在某个地方，你完全不知道任务到底成功了没有，中间产出的数据还在不在，更别提从断掉的地方继续了。这种“黑盒”式的长任务执行，是很多自动化工作流从“玩具”走向“生产可用”的最大障碍。

openclaw-task-recovery这个项目，就是为了解决这个痛点而生的。它不是要取代OpenClaw现有的技能（Skill）系统，也不是要引入一个臃肿的工作流编排引擎。你可以把它理解成给OpenClaw工作空间（Workspace）加装的一个“行车记录仪”和“自动启停”功能。它的核心目标很简单：让那些长时间运行、可以分阶段、有明确检查点的任务，变得可观测、可恢复、可管理。当任务因为网络波动、资源限制或底层服务超时而意外中断时，这个恢复层能自动检测到“失联”，并尝试从最后一个安全点继续执行，而不是让你从头再来。

这个工具特别适合那些已经开始用OpenClaw处理严肃工作的开发者、数据分析师或运维工程师。它通过几个非常轻量的概念——运行卡片（Run Card）、检查点（Checkpoint）和技能适配器（Skill Adapter）——构建了一套完整的恢复机制。最棒的是，你甚至不需要先有一个成熟的“技能”，对于一次性的长任务，它也能现场为你搭建一个临时的恢复适配器，让你立刻享受到任务持久化的好处。接下来，我会带你深入这套机制的内部，看看它是如何工作的，以及如何将它无缝集成到你自己的OpenClaw工作流中。

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

2.1 问题根源：为什么长任务在AI智能体环境中容易“失联”？

要理解openclaw-task-recovery的设计，首先得明白它要解决什么问题。在传统的脚本或后台服务中，我们可以用supervisord、systemd或者更现代的工作流引擎（如 Airflow、Prefect）来管理进程状态和重试。但OpenClaw这类AI智能体平台的工作模式有所不同：任务往往是由AI根据自然语言指令动态生成和执行的，执行路径可能不确定，中间状态也常常是保存在内存或临时文件中的。

这就导致了几个典型问题：

1. 状态黑盒：任务一旦开始，其内部阶段（Phase）和进度对外不可见。是卡在数据下载，还是报告渲染？
2. 中断即丢失：进程崩溃、会话超时或网络断开，意味着所有中间状态和上下文全部丢失。
3. 恢复成本高：想从断点继续？你得手动去翻日志，找到最后一个成功的输出文件，然后重新构造执行命令和参数，几乎等于重跑。
4. 静默故障：任务可能因为某个子步骤卡死而不再输出任何日志，但进程并未退出，形成“僵尸任务”，直到你手动发现。

openclaw-task-recovery的核心理念，是引入一个“持久化运行时状态”的薄层。它不试图接管OpenClaw的任务调度（那是OpenClaw自己或cron的工作），而是在任务执行的生命周期中，插入记录点和恢复钩子。这个设计借鉴了Durable Execution（持久化执行）的思想，类似Temporal或Inngest，但极大地简化了，目标是能直接放在一个OpenClaw工作空间目录里运行，无需外部服务依赖。

2.2 架构总览：心跳、看门狗与恢复器

整个恢复层的运转，围绕一个核心的观察-决策-执行循环，主要由三个脚本文件驱动：

[OpenClaw 主进程 / Cron任务] | v (定期触发) [心跳检查 HEARTBEAT.md] | v (调用看门狗) [task_runtime_watch.py] —— 检测“陈旧”任务 | v (发现需恢复的任务) [task_runtime_resume.py] —— 通用恢复调度器 | v (调用具体技能的逻辑) [技能适配器 (task_resume.py)] —— 安全恢复特定任务

1. 心跳（Heartbeat）与看门狗（Watchdog）这是整个系统的触发器。OpenClaw 通常有一个HEARTBEAT.md文件或类似机制，用于定期执行健康检查。install.py脚本会在其中添加一个“任务运行时恢复检查”部分。这个检查会定期（比如每分钟）调用task_runtime_watch.py。

task_runtime_watch.py的工作是扫描所有标记为“可自动恢复”且状态为“运行中”的运行卡片（Run Card）。它会检查每个任务的last_checkpoint_at时间戳。如果当前时间距离最后一次检查点的时间超过了预设的“陈旧阈值”（例如30分钟），看门狗就判定这个任务“失联”了。

2. 运行卡片（Run Card）：任务的身份证和病历本这是持久化状态的核心。一个运行卡片就是一个JSON文件，保存在data/task-runs/目录下，以{task_id}.json命名。它记录了任务的完整元数据和运行时状态：

• 基础信息：任务ID、类型、标题、创建时间。
• 执行上下文：当前阶段（Phase）、状态（Status）、上一次检查点时间。
• 恢复策略：是否允许自动恢复、陈旧阈值、最大重试次数。
• 产出物（Artifacts）：每个阶段生成的关键文件路径，这是恢复的关键依据。
• 检查点历史：记录了任务推进过程中的各个里程碑。

这个卡片文件是任务可恢复的基础。任务每完成一个安全阶段，就应该更新卡片中的阶段和检查点时间。

3. 恢复调度器与技能适配器当看门狗发现一个陈旧任务时，它会调用task_runtime_resume.py，并传入任务ID。恢复调度器是一个通用程序，它只做三件事：加载对应的运行卡片，找到卡片中指定的resume_adapter（技能适配器路径），然后调用这个适配器。

技能适配器是真正包含业务逻辑的地方。每个需要恢复能力的技能（或临时任务），都需要实现一个task_resume.py适配器。这个适配器就像一个“专科医生”，它知道如何诊断自己这个特定任务在当前阶段（比如“数据收集”阶段）卡住后，该如何安全地继续。它会读取运行卡片中记录的阶段和产出物，从最后一个检查点重新执行逻辑，并更新卡片状态。

这种设计实现了关注点分离：恢复层（通用）负责状态管理和调度触发，业务层（技能适配器）负责具体的恢复逻辑。这使得系统非常灵活，任何技能都可以通过实现一个适配器来获得恢复能力。

2.3 设计原则：轻量、非侵入、契约明确

这个项目的设计遵循几个关键原则，这也是它能在OpenClaw生态中保持简洁有效的秘诀：

• 轻量优先（Lightweight First）：所有代码都放在工作空间内，无需数据库、消息队列等外部依赖。状态用JSON文件存储，调度依赖现有的心跳机制。安装即用，卸载无痕。
• 非侵入式集成（Non-invasive Integration）：它不修改OpenClaw核心代码。通过向HEARTBEAT.md添加检查项，和提供脚本、模板供技能调用，它以“插件”方式融入现有工作流。
• 明确的适配器契约（Clear Adapter Contract）：技能适配器需要遵守的接口非常明确（接受--task-id参数，更新指定字段），这降低了集成复杂度，也使得编写临时任务的适配器模板成为可能。
• 安全恢复边界（Safe Resume Boundary）：系统强烈建议（并通过设计引导）只为等幂（Idempotent）且可安全重试的阶段启用自动恢复。对于发送邮件、部署服务等有副作用的操作，适配器逻辑应包含确认或人工干预环节。
• 信号导向的观测性（Signal-oriented Observability）：看门狗的输出不是杂乱的日志，而是归为三类清晰的信号：alerts（告警）、recoveries（已恢复）、needs_attention（需人工介入）。这让用户一眼就能看清系统状态。

3. 核心组件深度拆解与实操要点

3.1 运行卡片（Run Card）：状态持久化的基石

运行卡片是整个系统的核心数据模型。理解它的结构，对于调试和高级用法至关重要。一个典型的运行卡片JSON内容如下：

{ "task_id": "research_20250415_022034", "task_type": "overnight-research", "title": "竞品分析报告生成", "created_at": "2025-04-15T02:20:34Z", "status": "running", // 可能的值：pending, running, completed, failed, stalled "phase": "data_aggregation", "resume_adapter": "/home/user/.openclaw/workspace/skills/research/scripts/task_resume.py", "allow_auto_resume": true, "stale_after_minutes": 30, "max_retries": 3, "retry_count": 0, "last_checkpoint_at": "2025-04-15T03:45:12Z", "artifacts": { "phase_collect": { "raw_data": "/tmp/research/raw_data.json", "source_list": "/tmp/research/sources.txt" }, "phase_data_aggregation": { "normalized_data": "/tmp/research/normalized.json" } }, "checkpoints": [ { "phase": "collect", "at": "2025-04-15T02:45:00Z", "message": "已完成初始数据收集，共获取20个来源。" }, { "phase": "data_aggregation", "at": "2025-04-15T03:45:12Z", "message": "数据标准化完成，已去除重复项。" } ], "metadata": { "user": "alice", "priority": "high", "tags": ["report", "automation"] } }

关键字段解析与实操要点：

1. status与phase的协同：

   • status表示任务的宏观生命周期状态（如运行中、完成）。
   • phase表示任务当前所处的具体业务阶段（如数据收集、分析、渲染）。恢复动作总是针对某个phase进行的。当适配器恢复执行时，它需要知道从哪个phase开始，以及这个phase是否允许重试。

2. artifacts的结构：

   • 这是一个嵌套字典，键通常是阶段名（phase_xxx），值是该阶段产生的关键文件路径。
   • 为什么这是恢复的关键？假设任务在“渲染”阶段失败，恢复适配器需要找到“分析”阶段生成的analysis_result.json才能开始渲染。这个路径就记录在artifacts.phase_analysis下。
   • 实操建议：在创建检查点时，务必使用--artifact参数记录下阶段产出的绝对路径。避免使用相对路径，因为恢复时的工作目录可能不同。

3. allow_auto_resume与stale_after_minutes：

   • 这两个字段共同定义了看门狗的行为。只有allow_auto_resume为true的任务才会被自动检测和恢复。
   • stale_after_minutes需要根据任务类型谨慎设置。一个CPU密集型的模型训练可能2小时没更新也算正常，而一个频繁心跳的API轮询任务可能5分钟没动静就算异常。设置过短会导致不必要的恢复中断，设置过长则意味着故障发现延迟。

4. checkpoints列表：

   • 这不仅是一个日志，更是恢复的“安全点”序列。每次调用checkpoint命令，都会在这里添加一条记录。
   • 最佳实践：在checkpoint的--message参数中，记录足够的信息，以便人工查看时能快速理解进度，例如“已处理1000条记录中的250条”。

注意：运行卡片文件是系统的唯一状态源。切勿手动编辑此文件，除非你完全清楚后果。所有更新都应通过task_runtime.py脚本的命令来完成，以保证状态的一致性。

3.2 技能适配器（Skill Adapter）的契约与实现模式

技能适配器是连接通用恢复层和具体业务逻辑的桥梁。它不是一个复杂的框架，而是一个符合简单契约的Python脚本。

适配器必须遵守的基本契约：

1. 可执行：必须能通过命令行调用，并接受--task-id参数。
2. 可加载运行卡片：内部需调用task_runtime模块的load_run_card函数。
3. 状态驱动：根据运行卡片中的phase和status决定恢复逻辑。
4. 安全恢复：只对等幂的、安全的阶段执行恢复操作。对于有副作用的阶段，应记录日志并等待人工干预。
5. 状态更新：恢复执行后，必须通过task_runtime模块的update_run_card函数更新卡片的phase、status、artifacts和last_checkpoint_at。
6. 输出协议：尽可能将结果以JSON格式打印到标准输出（stdout），供上层调用者解析。

一个适配器的典型代码骨架如下：

#!/usr/bin/env python3 import sys import json import argparse from pathlib import Path # 假设 task_runtime 模块已安装在 workspace 脚本路径下 sys.path.insert(0, str(Path.home() / '.openclaw' / 'workspace' / 'scripts')) from task_runtime import load_run_card, update_run_card, RunCard def resume_phase_data_collection(run_card: RunCard): """恢复‘数据收集’阶段""" print(f"[INFO] 恢复阶段: data_collection for task {run_card.task_id}", file=sys.stderr) # 1. 从 artifacts 中获取上一个检查点的输出（如果有） last_artifact = run_card.artifacts.get('phase_data_collection', {}).get('partial_list') start_from = 0 if last_artifact and Path(last_artifact).exists(): with open(last_artifact, 'r') as f: start_from = int(f.read().strip()) # 2. 执行具体的恢复逻辑（例如：从第 start_from 条开始继续收集） # ... 你的业务逻辑 here ... new_data_point = start_from + 100 # 模拟新收集了100条 # 3. 更新产出物 new_artifact_path = f"/tmp/{run_card.task_id}_partial_{new_data_point}.txt" with open(new_artifact_path, 'w') as f: f.write(str(new_data_point)) # 4. 更新运行卡片（关键步骤！） run_card.artifacts.setdefault('phase_data_collection', {})['partial_list'] = new_artifact_path run_card.phase = 'data_analysis' # 进入下一阶段 run_card.last_checkpoint_at = datetime.utcnow().isoformat() + 'Z' # 注意：status 可能保持 'running'，除非完成或失败 update_run_card(run_card.task_id, run_card) return {"resumed_to": new_data_point, "next_phase": run_card.phase} def resume_phase_data_analysis(run_card: RunCard): """恢复‘数据分析’阶段""" # 此阶段依赖上一阶段的产出物 input_data_path = run_card.artifacts.get('phase_data_collection', {}).get('final_data') if not input_data_path or not Path(input_data_path).exists(): print(f"[ERROR] 依赖的输入文件不存在: {input_data_path}", file=sys.stderr) run_card.status = 'failed' run_card.metadata['error'] = 'Missing input artifact for analysis' update_run_card(run_card.task_id, run_card) return {"error": "Missing input artifact"} # ... 分析逻辑 ... # 更新阶段和产出物 run_card.phase = 'report_generation' run_card.artifacts.setdefault('phase_data_analysis', {})['result_json'] = analysis_output_path update_run_card(run_card.task_id, run_card) return {"status": "analysis_completed"} def main(): parser = argparse.ArgumentParser(description='恢复特定技能的长任务') parser.add_argument('--task-id', required=True, help='要恢复的任务ID') parser.add_argument('--timeout-seconds', type=int, default=300, help='恢复操作超时时间') args = parser.parse_args() # 加载运行卡片 run_card = load_run_card(args.task_id) if not run_card: print(json.dumps({"error": f"Task {args.task_id} not found"}), file=sys.stderr) sys.exit(1) # 根据当前阶段路由到不同的恢复函数 result = {} if run_card.phase == 'data_collection': result = resume_phase_data_collection(run_card) elif run_card.phase == 'data_analysis': result = resume_phase_data_analysis(run_card) elif run_card.phase == 'report_generation': # 报告生成可能涉及外部API调用，需谨慎处理自动恢复 if run_card.metadata.get('report_sent'): print(json.dumps({"warning": "Report already sent, skipping auto-resume to avoid duplicate."}), file=sys.stderr) run_card.status = 'needs_attention' update_run_card(run_card.task_id, run_card) result = {"action": "paused", "reason": "non-idempotent phase"} else: result = resume_phase_report_generation(run_card) else: print(json.dumps({"error": f"Unknown phase: {run_card.phase}"}), file=sys.stderr) sys.exit(1) # 输出结果（JSON格式便于上游处理） print(json.dumps(result)) if __name__ == '__main__': main()

适配器设计的心得与避坑指南：

• 阶段划分的艺术：阶段（Phase）划分要足够细，使得每个阶段都是相对独立、可恢复的单元。但也不能太细，否则会产生大量检查点开销。一个好的经验法则是：一个阶段应该对应一个清晰的、会产生持久化中间产出的子任务。
• 等幂性是生命线：确保每个允许自动恢复的阶段都是等幂的。即重复执行多次与执行一次的效果相同。数据下载、计算、转换通常是等幂的；发送通知、创建订单则不是。对于非等幂阶段，适配器内必须有防护逻辑（如检查metadata中的标志位）。
• 依赖管理：在artifacts中清晰定义阶段间的输入输出依赖。恢复时，适配器首先要检查所需的上游产出物是否存在且有效。
• 超时与资源清理：适配器可能执行很长时间。利用--timeout-seconds参数或在内部实现超时逻辑，避免恢复操作本身卡死。同时，对于临时文件，考虑在任务最终完成或失败后进行清理，可以在metadata中记录需要清理的路径列表。

3.3 看门狗（Watchdog）信号系统：从“静默”到“可对话”

task_runtime_watch.py不仅仅是检测陈旧任务，更重要的是它提供了一套清晰的信号系统，将底层状态转化为用户和上游系统能理解的语义。这是提升运维体验的关键。

看门狗每次执行（由心跳触发），都会遍历所有运行卡片，并生成一个包含三个“桶”的报告：

{ "alerts": [ { "task_id": "task_abc", "phase": "data_fetch", "stale_for_minutes": 45, "last_checkpoint": "2025-04-15T01:00:00Z" } ], "recoveries": [ { "task_id": "task_xyz", "from_phase": "processing", "to_phase": "completed", "recovered_at": "2025-04-15T02:05:00Z" } ], "needs_attention": [ { "task_id": "task_123", "phase": "send_email", "reason": "non_idempotent_phase_blocked", "blocked_since": "2025-04-15T01:30:00Z" } ] }

这三个信号桶的解读与处理策略：

1. alerts（告警）：任务已陈旧（超过stale_after_minutes），但尚未尝试恢复或恢复尚未完成。这通常意味着任务刚刚失联，看门狗已经将其标记，并可能即将触发恢复（如果allow_auto_resume=True）。监控系统可以订阅此信号，用于触发告警（如发送Slack通知），但不必立即人工介入，因为系统可能会自动恢复。

2. recoveries（已恢复）：看门狗在本轮检测中成功调用了恢复适配器，并且适配器报告任务已推进到新的阶段或完成。这是一个积极信号，表明恢复机制正在正常工作。你可以将此信号记录到审计日志中，用于评估系统的自愈能力。

3. needs_attention（需人工介入）：这是最重要的信号。它表示看门狗遇到了无法自动处理的情况，必须由人来决定下一步。常见原因包括：

   • 任务处于非等幂阶段（如send_email），适配器主动暂停。
   • 自动恢复尝试次数已达到max_retries上限，但任务仍未完成。
   • 恢复适配器本身执行失败或出错。
   • 运行卡片状态异常（如status为running但resume_adapter路径无效）。

实操建议：将needs_attention信号集成到你的运维仪表盘或通知系统中。这是避免任务长时间停滞无人知晓的关键。你可以配置一个自动化规则，当needs_attention列表非空时，自动创建一张工单或发送一条高优先级的告警消息给负责人。

4. 完整集成与实操流程

4.1 从零开始：安装与基础配置

假设你的OpenClaw工作空间位于~/.openclaw/workspace。集成openclaw-task-recovery只需要几步。

第一步：克隆仓库并安装

cd ~/.openclaw/workspace git clone https://github.com/m0x14o/openclaw-task-recovery repos/openclaw-task-recovery cd repos/openclaw-task-recovery python3 install.py

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

1. 将scripts/目录下的核心脚本复制到工作空间的scripts/目录。
2. 将docs/下的文档复制到工作空间的docs/openclaw-task-recovery/目录。
3. 将适配器模板templates/task_resume.py复制到工作空间的templates/openclaw-task-recovery/目录。
4. 最关键的一步：在HEARTBEAT.md文件的末尾，添加或更新一个名为“Task Runtime Recovery Check”的章节。内容大致是调用task_runtime_watch.py脚本。脚本会确保这个添加是等幂的，即重复运行安装脚本不会造成重复内容。

第二步：验证安装安装后，检查以下内容：

• 文件是否存在：ls -la ~/.openclaw/workspace/scripts/task_runtime*.py
• 心跳文件是否更新：tail -n 20 ~/.openclaw/workspace/HEARTBEAT.md，你应该能看到类似python3 scripts/task_runtime_watch.py的调用。
• 运行一次看门狗脚本，确保无语法错误：cd ~/.openclaw/workspace && python3 scripts/task_runtime_watch.py。初次运行，由于没有任务，输出应该是一个空的JSON对象{}。

4.2 为现有技能添加恢复能力：以“数据报告生成”技能为例

假设你有一个已有的OpenClaw技能，位于~/.openclaw/workspace/skills/report-generator/，它能根据一个主题进行网络搜索、数据分析并生成PDF报告。现在我们要让它支持断点续做。

第一步：创建技能适配器使用安装时提供的模板，快速创建适配器。

cp ~/.openclaw/workspace/templates/openclaw-task-recovery/task_resume.py \ ~/.openclaw/workspace/skills/report-generator/scripts/task_resume.py

第二步：分析技能阶段，设计检查点你需要审视你的报告生成流程，将其划分为可恢复的阶段。例如：

• phase_1_collect：从多个来源收集原始数据和文章。
• phase_2_analyze：对收集的内容进行总结和关键信息提取。
• phase_3_render：将分析结果渲染成PDF。

其中，phase_1_collect和phase_2_analyze是相对等幂的（重复收集和分析同一数据源，结果一致）。phase_3_render生成最终文件，也是等幂的。但假设渲染后有一个phase_4_email（发送邮件），这个阶段就是非等幂的，不能自动恢复。

第三步：修改技能主逻辑，插入检查点调用在你的技能主脚本（例如report-generator/main.py）中，在完成每个安全阶段后，调用task_runtime.py checkpoint命令。

# 假设在 main.py 中 import subprocess import json import sys def run_report_generation(task_id, topic): # ... 你的原有逻辑 ... # 阶段1: 收集 raw_data_file = do_data_collection(topic) # 创建检查点 subprocess.run([ sys.executable, f"{workspace_path}/scripts/task_runtime.py", "checkpoint", task_id, "--phase", "collect", "--artifact", f"raw_data={raw_data_file}", "--message", f"已完成‘{topic}’的原始数据收集，文件位于 {raw_data_file}" ], check=True) # 阶段2: 分析 analysis_result = do_analysis(raw_data_file) analysis_file = save_analysis(analysis_result) subprocess.run([ sys.executable, f"{workspace_path}/scripts/task_runtime.py", "checkpoint", task_id, "--phase", "analyze", "--artifact", f"analysis_result={analysis_file}", "--message", "数据分析完成，已提取核心观点。" ], check=True) # 阶段3: 渲染 pdf_report = render_pdf(analysis_file) subprocess.run([ sys.executable, f"{workspace_path}/scripts/task_runtime.py", "checkpoint", task_id, "--phase", "render", "--artifact", f"final_report={pdf_report}", "--message", "PDF报告渲染完成。" ], check=True) # 阶段4: 发送邮件 (非等幂，不设置自动恢复检查点，或设置一个特殊标志) # send_email(pdf_report) # 更新任务状态为完成，但不在这个阶段设置自动恢复的检查点 # 或者，设置一个 metadata 标志位，让适配器知道邮件已发送

第四步：实现适配器恢复逻辑编辑刚创建的task_resume.py，根据阶段实现恢复函数。核心是读取运行卡片中的phase和artifacts，从中断的地方继续。 例如，恢复analyze阶段的函数需要读取artifacts.phase_collect.raw_data文件，然后执行do_analysis。

第五步：启动任务时创建运行卡片修改技能启动的入口，在开始长任务前，先创建一个运行卡片。

# 在调用技能前，或在技能初始化时执行 TASK_ID="report_$(date +%s)" python3 ~/.openclaw/workspace/scripts/task_runtime.py create \ --task-id "$TASK_ID" \ --task-type "report-generation" \ --title "生成关于‘AI代理’的报告" \ --phase "collect" \ --resume-adapter "$HOME/.openclaw/workspace/skills/report-generator/scripts/task_resume.py" \ --allow-auto-resume \ --stale-after-minutes 20 \ --max-retries 2

然后将这个TASK_ID传递给技能的主函数。

至此，你的“数据报告生成”技能就具备了基本的任务恢复能力。如果它在分析阶段因为网络问题卡住，20分钟后看门狗会检测到，并自动调用你的适配器从分析阶段重新开始。

4.3 处理一次性临时任务：无需创建正式技能

对于临时性的、一次性的长任务，你不需要创建一个完整的技能目录。openclaw-task-recovery提供了更快捷的路径。

使用“一线提示词”模式这是项目README中提到的精髓。你直接向你的OpenClaw发送这样一条指令：

“运行这个作为可恢复的长任务：帮我扫描项目目录/home/user/myproject下的所有.py文件，统计代码行数、函数个数，并生成一个复杂度报告。如果中途中断，请能从断点恢复。”

一个配置良好的OpenClaw，在接收到这条指令并识别出“可恢复的长任务”这个模式后，可以自动执行以下操作：

1. 在tmp/task-runtime/code-analysis-20250415/下创建一个临时目录。
2. 将适配器模板复制到那里，并基于你的任务描述，生成一个简单的、专门针对代码分析任务的task_resume.py适配器骨架。
3. 创建一个运行卡片，resume_adapter指向这个临时适配器。
4. 开始执行任务，并在关键步骤插入检查点。

手动创建临时任务流程如果OpenClaw没有自动完成，你也可以手动模拟这个过程：

# 1. 创建临时任务目录和适配器 TASK_SLUG="adhoc_code_analysis" TMP_ADAPTER_DIR="$HOME/.openclaw/workspace/tmp/task-runtime/$TASK_SLUG" mkdir -p "$TMP_ADAPTER_DIR" cp ~/.openclaw/workspace/templates/openclaw-task-recovery/task_resume.py "$TMP_ADAPTER_DIR/" # 2. 编辑这个临时适配器，实现你的代码分析逻辑和阶段恢复 # 用你熟悉的编辑器编辑 $TMP_ADAPTER_DIR/task_resume.py # 3. 创建运行卡片 python3 ~/.openclaw/workspace/scripts/task_runtime.py create \ --task-id "$TASK_SLUG" \ --task-type "adhoc-analysis" \ --title "临时代码分析任务" \ --phase "file_discovery" \ --resume-adapter "$TMP_ADAPTER_DIR/task_resume.py" \ --allow-auto-resume \ --stale-after-minutes 10 # 4. 开始你的任务脚本，并在其中调用检查点 # 你的脚本 my_analysis.sh 或 my_analysis.py 内部，需要知道 TASK_SLUG # 并在每个子步骤后调用 `task_runtime.py checkpoint ...`

这种方式非常适合探索性工作或一次性数据处理任务。任务完成后，你可以选择保留这个临时适配器以备后用，或者直接删除tmp/task-runtime/下的目录。

5. 常见问题、排查技巧与高级模式

5.1 故障排查清单

在实际使用中，你可能会遇到以下问题。这里是一个快速排查指南：

5.2 高级模式与最佳实践

1. 混合模式：自动恢复与人工审批结合对于包含非等幂步骤的流程，可以采用“自动恢复+人工闸门”模式。例如，一个发布流程：构建（可自动恢复）-> 测试（可自动恢复）-> 部署（需人工批准）。

• 在运行卡片中设置phase: deployment_pending。
• 适配器检测到此阶段时，不执行任何操作，而是将任务状态改为needs_attention，并在metadata中添加pending_action: “manual_deployment_approval”。
• 看门狗会将其放入needs_attention信号桶。
• 你可以在一个管理界面查看所有needs_attention的任务，手动点击“批准部署”。
• 批准操作触发另一个脚本，该脚本执行部署，并在成功后更新运行卡片状态为completed。

2. 基于产出物校验的增强恢复简单的检查点时间戳可能不够。你可以在适配器中加入对产出物的校验逻辑。例如，在恢复数据分析阶段前，先检查上游的“数据收集”阶段产出的文件是否完整（例如，检查文件大小、校验和或文件头信息）。如果产出物损坏，则恢复失败，并标记为needs_attention，而不是基于一个损坏的中间结果继续执行。

3. 任务依赖链虽然openclaw-task-recovery本身不是DAG引擎，但你可以通过运行卡片的metadata字段实现简单的依赖。例如，任务B的metadata中记录depends_on: [“task_a_id”]。在任务B的适配器开始恢复前，先检查任务A的运行卡片状态是否为completed。如果不是，则将自己标记为pending或needs_attention。你可以编写一个外部的协调脚本，定期检查这些依赖关系并触发后续任务。

4. 运行卡片的生命周期管理

• 归档：对于重要的已完成任务，可以将运行卡片JSON文件连同其artifacts中记录的产出物一起打包存档，用于事后审计或复现。
• 清理：定期清理旧的和失败的任务卡片，避免data/task-runs/目录膨胀。可以写一个简单的清理脚本，根据created_at时间和status来删除文件。
• 备份：由于运行卡片是纯JSON文件，可以很容易地纳入你的工作空间备份策略中。

5. 监控与可视化你可以利用看门狗输出的JSON信号，构建简单的监控面板。例如，用一个Python脚本定期调用task_runtime_watch.py，解析其输出，然后将alerts、recoveries、needs_attention的数量推送到Prometheus、Datadog，或者简单地显示在一个静态网页上。这能让你对整个工作空间的长任务健康状态一目了然。

这套轻量级的任务恢复层，其威力在于它的简单和直接。它没有试图解决所有问题，而是精准地解决了长任务执行中最令人头疼的“失联”和“不可恢复”问题。通过将持久化状态、定期检测和可插拔的恢复逻辑分离，它既提供了足够的可靠性，又保持了极致的灵活性。无论是集成到现有技能，还是管理临时性任务，它都能显著提升你在OpenClaw上运行自动化工作的信心和效率。最关键的是，这一切都发生在一个目录里，无需改变你现有的架构，这种低成本的可靠性提升，在实际工作中往往是最有价值的。