What a checkpoint does not know
An agent job is running an eval across 16 shards. For each shard it submits a job to the cluster scheduler, waits for it to finish, and writes the shard's scores to the shared results table. It checkpoints its plan as it goes: "shards 1 to 6 done, next: submit shard 7." Then the machine running the agent is preempted. A new worker starts, reads the checkpoint, sees "next: submit shard 7," and submits it.
Did the first worker already submit shard 7? The checkpoint cannot say. It faithfully recorded the job's intention. It recorded nothing about what the scheduler did with a request that was in flight when the worker died. If the scheduler accepted it, you now have two shard-7 jobs burning GPU-hours and racing to write the same scores.
That is the gap between durable task state and durable effect state. LangGraph's persistence documentation describes checkpoints that preserve a graph's execution state so a run can resume after an interruption (LangGraph, Persistence). Saving that state does not show that an external operation happened exactly once. A successor needs the intended operation and its known outcome before it acts.
The book builds the pieces of this in three chapters: the durable inbox and outbox in Chapter 8, reconciling an uncertain send in Chapter 12, and leases and fenced writes in Chapter 13. This article puts them together for a long research job. It adds two things the chapters do not: splitting a job's steps by what repeating them costs, and choosing how often to checkpoint from the crash rate.
Sort the steps by what repeating them costs
Walk through the job and ask of each step: if a successor does this again, what happens?
| Step | Repeating it costs | Recovery rule |
|---|---|---|
| Read the manifest, compute which shards remain | Nothing | Recompute freely |
| Call the model to score an item | Tokens and time; the answer may differ at nonzero temperature | Recompute, or reuse a stored response keyed by item |
| Write a score to the results table | A duplicate row, unless the write is keyed | Upsert by run, item and grader key |
| Submit a shard to the scheduler | A second job and its GPU-hours | Record the send, then reconcile before resending |
| Post the summary to the team channel | A second message people will read | Record the send, then reconcile before resending |
Scroll horizontally to view columns.
Pure steps are safe to replay; replay only costs compute. The keyed write is safe to replay once it is an upsert; the companion article on idempotent writes builds that. The last two rows are the hard ones, because the effect lives in someone else's system.
Three states instead of two
A job modeled as pending and done cannot say "I called the scheduler and do not know what happened." Give each external operation three states:
| State | Meaning | Safe next action |
|---|---|---|
ready | Operation decided, not yet sent | Send it |
sent_unknown | The send was authorized and may have begun; no outcome recorded | Look it up by operation ID before anything else |
confirmed | A receipt that matches this operation is recorded | Move on; never resend |
Scroll horizontally to view columns.
Write sent_unknown to durable storage before the external call. If you write it after, a crash during the call leaves the operation at ready, and the successor sends again. If you write it before, a crash at any point after leaves a record that says: something may have happened here, go and check.
Here is the state machine with a mock scheduler that honors a client operation ID, using only Python's standard library. It runs on Colab's Python 3.12. It is an in-memory model of the protocol, not a storage or concurrency guarantee.
from dataclasses import dataclass, field
@dataclass
class JobStore:
jobs: dict = field(default_factory=dict)
@dataclass
class MockScheduler:
"""Stands in for a scheduler that honors a client operation ID."""
records: dict = field(default_factory=dict)
submit_calls: int = 0
def submit(self, operation_id, payload):
self.submit_calls += 1 # calls to this mock, not jobs created
existing = self.records.get(operation_id)
if existing is not None:
if existing["payload"] != payload:
raise ValueError("operation ID reused with a different payload")
return existing
record = {"operation_id": operation_id, "payload": dict(payload), "job_id": "job-5512"}
self.records[operation_id] = record
return record
def lookup(self, operation_id):
return self.records.get(operation_id)
def record_confirmation(op, receipt):
if receipt.get("operation_id") != op["operation_id"] or receipt.get("payload") != op["payload"]:
raise ValueError("receipt does not confirm this operation")
op.update(state="confirmed", receipt=dict(receipt))
def resume(store, scheduler, key, lose_reply=False):
op = store.jobs[key]
if op["state"] == "ready":
op["state"] = "sent_unknown" # persist this BEFORE the call
receipt = scheduler.submit(op["operation_id"], op["payload"])
if lose_reply:
raise ConnectionError("reply lost after the scheduler accepted the job")
record_confirmation(op, receipt)
return "sent"
if op["state"] == "sent_unknown":
receipt = scheduler.lookup(op["operation_id"])
if receipt is not None:
record_confirmation(op, receipt)
return "reconciled_confirmed"
return "pending_reconciliation"
if op["state"] == "confirmed":
return "already_confirmed"
raise ValueError("unknown state")Now lose the reply after the scheduler accepts the job, and resume:
store = JobStore({"shard-7": {
"operation_id": "run-0930/shard-7/submit",
"state": "ready",
"payload": {"shard": 7, "config": "configs/eval.yaml"},
}})
scheduler = MockScheduler()
try:
resume(store, scheduler, "shard-7", lose_reply=True)
except ConnectionError:
pass
assert store.jobs["shard-7"]["state"] == "sent_unknown"
assert resume(store, scheduler, "shard-7") == "reconciled_confirmed"
assert resume(store, scheduler, "shard-7") == "already_confirmed"
assert scheduler.submit_calls == 1
print(store.jobs["shard-7"]["receipt"]["job_id"], scheduler.submit_calls)
empty = JobStore({"shard-8": {"operation_id": "run-0930/shard-8/submit",
"state": "sent_unknown", "payload": {"shard": 8}}})
print(resume(empty, MockScheduler(), "shard-8"))It prints job-5512 1 and then pending_reconciliation. The successor found the job the first worker created and recorded it without a second submit. The second case matters as much: a sent_unknown operation with no record at the scheduler stays unresolved. It does not become a resend.
The operation ID is derived from the run, the shard and the action, so any successor computes the same one. The exception stands in for a lost reply; this fixture does not kill a process or test disk persistence. A real crash test must restart a worker from the actual store.
An empty lookup can mean the call never arrived, or that the scheduler's record has not reached the place you are checking. Keep the operation pending, give reconciliation a deadline and an owner, and escalate if it expires. Resend only under a receiver that honors the same operation ID and payload.
How often to checkpoint
Everything after the last checkpoint is lost on a crash and must be redone. Checkpointing more often loses less but spends more time checkpointing. Let C be the time to write one checkpoint, τ the time between checkpoints, and M the mean time between crashes. Per unit of time, you spend C/τ on checkpoints. A crash lands, on average, halfway through an interval, so you lose about τ/2 of work per crash, and crashes arrive at rate 1/M. The overhead is approximately
overhead(τ) ≈ C/τ + τ/(2M)Set the derivative −C/τ² + 1/(2M) to zero:
τ* = √(2 · C · M), overhead(τ*) ≈ √(2C/M)This is the first-order approximation published by John Young in 1974 (Young, "A first order approximation to the optimum checkpoint interval," Communications of the ACM 17(9)). It assumes C is much smaller than M.
Two illustrative cases:
- A fine-tuning run on preemptible machines, where writing weights and optimizer state takes C = 60 s and preemptions come on average every M = 24 hours = 86,400 s. Then τ* = √(2 × 60 × 86,400) = √10,368,000 ≈ 3,220 s, about 54 minutes, and the overhead is √(120/86,400) ≈ 3.7%.
- An eval run where committing a batch of keyed results takes C = 2 s and the worker crashes every M = 8 hours = 28,800 s. Then τ* = √(115,200) ≈ 339 s, under six minutes, at about 1.2% overhead.
Measure C and M for your own jobs; these values are not measurements. Note that the formula governs lost compute only. Side effects need the three-state record on every send, however often you checkpoint the rest.
When two successors show up
Recovery systems often start more than one worker, and two of them may both find shard 7 at ready. If both send, the state machine does not help. Claim the operation first:
Read the operation's state and version number in one durable read.
Write the claim only if the version is unchanged, for example UPDATE ... SET owner = ?, version = version + 1 WHERE id = ? AND version = ?. If no row changed, another worker won; reread instead of proceeding.
Carry the claim's version as a fencing token on every write, and have the store reject writes from an older token.
A claim with an expiry, a lease, has a gap. If the first worker is merely paused, its lease expires, a successor claims the operation, and then the first worker wakes up and writes. Martin Kleppmann describes the fix: a fencing token is "a number that increases" every time the lock is acquired, and the storage server rejects a write carrying a lower token than one it has already seen (Kleppmann, "How to do distributed locking"). Chapter 13 derives how to choose the lease from the measured tail of turn lengths, and shows a stale completion refused by the fence.
Read the records, not the transcript
An agent's memory is often a transcript or a summary of one. A successor reading "I will submit shard 7" cannot tell it apart from "shard 7 was submitted." Chapter 13 tested exactly this: given a transcript that stopped after an order call and asked to continue, the smaller of the two local models it tried placed the order again in 14 of 20 samples and never looked it up. Recover from the structured record, with its operation ID and state, and look up the external system before asking a model anything.
Check the decision
Test it before you trust it
Extend the fixture with the cases that catch overconfident recovery: a sent_unknown with no record must stay pending with zero submits; a record with a different payload must be rejected; an unknown state must raise, not default to success. Then repeat the crash points against a real store, restarting from stored bytes: before the send, after the scheduler accepts but before local confirmation, and after confirmation. Race two claimants on one version, and test a worker that wakes after its lease expired. Count received requests and created jobs separately, because an idempotent receiver may accept several attempts and create one job.
Why this matters in a lab
Long jobs are the normal case in research: multi-day training runs on preemptible capacity, evals over tens of thousands of items, sweeps with hundreds of runs, and agents that drive all three. They will crash. What separates a team that trusts its results from one that does not is whether a resumed job repeats its effects.
An interviewer might ask: "Your eval agent was preempted halfway through. How does the replacement know what to redo and what not to touch?" A strong answer sorts steps by the cost of repeating them, writes sent_unknown before each external call, reconciles by operation ID, sets the checkpoint interval from √(2CM) with measured values, and uses a claim with a fencing token when two successors can race. To practice, run the chapter's worker-recovery checkpoint in the course repo (profrod_sovereign_agent_ch13_worker_recovery_checkpoint.py).
Choose a lease from the measured latency tail, fence every controlled write, and recover from records instead of asking the model what happened.
