Durable Agent State Checkpoints: Resume Where You Left Off When the Run Crashes

from agent_state_checkpoint import StateCheckpoint

ckpt = StateCheckpoint(path="./checkpoints/run-abc123.json")

# Load existing state or start fresh
state = ckpt.load() or {
    "documents_processed": [],
    "current_index": 0,
    "findings": [],
}

documents = get_all_documents()

for i, doc in enumerate(documents):
    if i < state["current_index"]:
        continue  # Already processed, skip

    findings = analyze_document(doc)
    state["findings"].extend(findings)
    state["documents_processed"].append(doc["id"])
    state["current_index"] = i + 1

    # Save after each document
    ckpt.save(state)

final_result = compile_report(state["findings"])
ckpt.mark_complete(result=final_result)

import json
import os
import tempfile

class StateCheckpoint:
    def __init__(self, path: str):
        self._path = Path(path)
        self._path.parent.mkdir(parents=True, exist_ok=True)

    def save(self, state: dict) -> None:
        record = {
            "state": state,
            "status": "in_progress",
            "saved_at": time.time(),
        }
        self._atomic_write(record)

    def load(self) -> dict | None:
        if not self._path.exists():
            return None

        with self._path.open() as f:
            record = json.load(f)

        if record.get("status") == "complete":
            return None  # Already done, caller should use mark_complete result

        return record.get("state")

    def mark_complete(self, result: Any = None) -> None:
        record = {
            "state": None,
            "status": "complete",
            "result": result,
            "completed_at": time.time(),
        }
        self._atomic_write(record)

    def _atomic_write(self, record: dict) -> None:
        # Write to temp file, then rename (atomic on POSIX)
        tmp_path = self._path.with_suffix(".tmp")
        try:
            with tmp_path.open("w") as f:
                json.dump(record, f)
                f.flush()
                os.fsync(f.fileno())
            tmp_path.rename(self._path)
        except Exception:
            tmp_path.unlink(missing_ok=True)
            raise

    def get_result(self) -> Any:
        if not self._path.exists():
            return None
        with self._path.open() as f:
            record = json.load(f)
        if record.get("status") == "complete":
            return record.get("result")
        return None

    def clear(self) -> None:
        self._path.unlink(missing_ok=True)

pip install git+https://github.com/MukundaKatta/agent-state-checkpoint

# Or from PyPI
pip install agent-state-checkpoint

from agent_state_checkpoint import StateCheckpoint

def run_research_agent(task_id: str, urls: list[str]) -> dict:
    ckpt = StateCheckpoint(path=f"./checkpoints/{task_id}.json")

    state = ckpt.load() or {
        "processed_urls": [],
        "summaries": [],
        "current_idx": 0,
    }

    print(f"Resuming from index {state['current_idx']} of {len(urls)}")

    for i, url in enumerate(urls):
        if i < state["current_idx"]:
            continue

        try:
            summary = fetch_and_summarize(url)
            state["summaries"].append({
                "url": url,
                "summary": summary,
            })
            state["processed_urls"].append(url)
            state["current_idx"] = i + 1
            ckpt.save(state)
            print(f"Saved checkpoint at {i + 1}/{len(urls)}")
        except Exception as e:
            print(f"Failed on {url}: {e}. Checkpoint saved at {state['current_idx']}.")
            raise

    report = generate_report(state["summaries"])
    ckpt.mark_complete(result=report)
    return report