💾

Long-Running Task Checkpointer

Agents & Workflows · by AIDiveForge · Apr 20, 2026 · Advanced

Persist JSON agent state with atomic checkpoints and explicit recovery limits.

Get a working checkpoint helper and recovery fixtures.

Download skill ZIP See the worked example

Free · 5 KB · SKILL.md, instructions and example files. Python 3.10+ is required. The included helper uses only the standard library.

A worked example

A small example to try before using your own data.

Input

{
  "command": "python scripts/checkpoints.py save --directory ./checkpoints --state examples/input.json --step extract-1",
  "state": {
    "completed": [
      "extract-1"
    ],
    "next": "summarize"
  }
}

Expected result

{
  "loaded_state": {
    "completed": [
      "extract-1"
    ],
    "next": "summarize"
  },
  "last_completed_step": "extract-1",
  "recovery_rule": "A corrupt newer file is reported and the last valid checkpoint remains readable."
}

What to check: The round trip preserves state and step ID; incomplete/corrupt newer files cannot displace a valid checkpoint.

Limits: The helper does not capture an agent automatically or guarantee exactly-once external actions. JSON state must omit secrets you do not want persisted.

Worked example included. Model-specific performance and vendor compatibility have not been benchmarked.

Installation options and other formats

Install this skill

An agent skill starts with SKILL.md. The ZIP also includes its examples and any helper scripts. Drop the file into your tool of choice — or pick a different format if you use Cursor, Windsurf, Copilot, or something else.

Format Download SKILL.md

Quick install — paste into your terminal

python3 - <<'ADF_SKILL_INSTALL'
import hashlib, io, os, stat, tempfile, urllib.request, zipfile
from pathlib import Path, PurePosixPath
folder = "long-running-task-checkpointer"
root = Path(os.environ.get('ADF_SKILL_ROOT', str(Path.home() / '.claude' / 'skills'))).expanduser()
destination = root / folder
if destination.exists() or destination.is_symlink():
    raise SystemExit('An installation already exists. Back it up and merge the ZIP deliberately.')
root.mkdir(parents=True, exist_ok=True)
request = urllib.request.Request("https://aidiveforge.com/skill/long-running-task-checkpointer.zip", headers={'User-Agent': 'AIDiveForge-Resource-Installer/1.0 (+https://aidiveforge.com/skills/)'})
with urllib.request.urlopen(request, timeout=30) as response:
    data = response.read(8 * 1024 * 1024 + 1)
if len(data) > 8 * 1024 * 1024 or hashlib.sha256(data).hexdigest() != "93e26b9580ed63bd7fc08f4dbbf0cf742cd15d6990f87ae8f0835f9b0dfd1f83":
    raise SystemExit('The download did not match this reviewed bundle. No files were installed.')
with zipfile.ZipFile(io.BytesIO(data)) as archive, tempfile.TemporaryDirectory(prefix='.adf-skill-', dir=root) as temporary:
    members = archive.infolist()
    if len(members) > 100 or sum(m.file_size for m in members) > 16 * 1024 * 1024:
        raise SystemExit('Unexpected archive size')
    for member in members:
        name = PurePosixPath(member.filename)
        if name.is_absolute() or '..' in name.parts or not name.parts or name.parts[0] != folder or '\\' in member.filename or stat.S_ISLNK(member.external_attr >> 16):
            raise SystemExit('Unsafe archive entry; nothing was installed')
        target = Path(temporary).joinpath(*name.parts)
        if member.is_dir():
            target.mkdir(parents=True, exist_ok=True)
        else:
            target.parent.mkdir(parents=True, exist_ok=True)
            with target.open('xb') as stream:
                stream.write(archive.read(member))
    staged = Path(temporary) / folder
    if not (staged / 'SKILL.md').is_file() or destination.exists() or destination.is_symlink():
        raise SystemExit('The skill could not be installed without replacing existing data')
    os.rename(staged, destination)
print('Installed the complete skill folder:', destination)
ADF_SKILL_INSTALL

Claude Code: ~/.claude/skills/long-running-task-checkpointer/SKILL.md. Codex: ~/.codex/skills/long-running-task-checkpointer/SKILL.md. Keep the complete ZIP folder when helper scripts are included.

🧠 Why it works

The round trip preserves state and step ID; incomplete/corrupt newer files cannot displace a valid checkpoint.

⚙️ How it works

Use scripts/checkpoints.py to save JSON state after a completed logical step; include the last completed operation identifier.

The helper writes a checksum-protected envelope to a temporary file and atomically publishes a numbered checkpoint.

On recovery, load the newest valid checkpoint; inspect reported invalid files and any operation whose completion is uncertain.

Reconcile uncertain external side effects before retrying them. A state snapshot does not make a payment, email or tool call exactly once.

Run the supplied save/load example before integrating the helper into a real agent loop.

Description

Get a working checkpoint helper and recovery fixtures.

Python 3.10+ is required. The included helper uses only the standard library.

Limits: The helper does not capture an agent automatically or guarantee exactly-once external actions. JSON state must omit secrets you do not want persisted.