An agent has just opened a ticket when its process stops. The ticket exists, but the agent has not saved the result of that step. After restarting, it sees unfinished work and runs the step again. There are now two tickets.
This is where I would start when evaluating a stateful workflow. Recovering work is useful; knowing which actions recovery will repeat matters even more. A checkpoint restores the process state. The API receiving a request also needs to recognise when we are asking it to complete the same operation.
I built a Docker lab that exposes this difference. The process is killed immediately after the fake API responds and before the node returns its result. The first version produces two tickets. With a stable operation identity, recovery returns the ticket that already exists. No paid model is involved: the interesting behaviour sits around the agent's decision.
What state and a graph mean here
A workflow is a procedure made of steps. Its state holds the data those steps exchange: the requested operation, available results and information needed to continue. Opening a ticket may require a title, an operation identifier and, after creation, the identifier returned by the service.
A graph makes the path explicit. Nodes are functions that perform steps; edges specify which function runs next. A conditional edge can send a proposal for approval, route an error to a handler or send an incomplete result back for another attempt. That is what graph engineering means here: designing the steps and transitions of a process. Knowledge graphs and a model's internal representations are separate subjects.
The lab deliberately uses a linear path:
START -> prepare -> create_ticket -> ENDprepare establishes the operation key; create_ticket calls the service. I use LangGraph to inspect state and node recovery without adding branches that obscure the defect. Two sequential functions would be enough to open a ticket. A graph earns its place when branching paths, waits and loops become part of the problem.
A node can contain ordinary code, a model call or both. The LangGraph Graph API defines these building blocks. The LangChain, LangGraph, OpenRouter and Langfuse map explains how orchestration differs from the other components.
The checkpoint comes after the step
A checkpoint is a persisted snapshot of execution state. In LangGraph, a thread_id selects it: reusing the original identifier finds the same process; changing it identifies another thread.
The lab's checkpointer writes to SQLite. The file lives in a Docker volume, so replacing the container does not remove it. With durability="sync", LangGraph persists the checkpoint before starting the next step. The checkpointer documentation distinguishes this from async, which writes while the next step runs, and exit, which persists state on a managed graph exit, without saving the intermediate states needed to recover from a process crash.
The critical point sits inside create_ticket:
result = call("/tickets", {"title": state["title"]}, key)
emit("effect_ack", **result)
if args.crash:
os.kill(os.getpid(), signal.SIGKILL)
return {"ticket_id": result["ticket_id"]}The service has already acknowledged the ticket. The node dies before its return. The preceding checkpoint still says the next node is create_ticket. Recovery calls that function from its beginning, including the HTTP request.
Smaller nodes reduce the amount of work that must be repeated, but the gap between two systems remains: the receiver has committed the action, while the caller has not saved its result. A task with a persisted result must also handle an effect that occurs before that result is recorded.
Set up the lab
You need a working Docker Engine and Docker Compose, a terminal and internet access to build the image. You do not need Python installed on your computer, API keys or an AI service account. The test ran on Linux ARM64 in Docker Desktop with Python 3.13.17, LangGraph 1.2.17 and langgraph-checkpoint-sqlite 3.1.2.
Create an empty directory and open a terminal there:
mkdir stateful-agent-lab
cd stateful-agent-lab
docker version
docker compose versionCopy the five complete files at the end of this article into that directory. You write these files; Docker creates the image, network and two volumes:
stateful-agent-lab/
Dockerfile
requirements.txt
compose.yaml
api.py
workflow.pyapi is a fake receiver that records tickets in its own database. workflow runs the nodes and stores checkpoints in a separate database. The network is internal to Docker, with no ports exposed on the computer. Run every command below from this same directory.
docker compose -p stateful-lab up -d --build api
docker compose -p stateful-lab run --rm workflow ticketsThe initial ticket list should be empty. If you are repeating an earlier test and want a clean start, the cleanup command at the end also removes the lab's data.
Test 1: state recovers, the ticket is duplicated
Start the path without idempotency:
docker compose -p stateful-lab run --rm workflow start --thread naive --operation case-naive:open-ticket:v1 --mode naive --crashThe command exits with code 137, after effect_ack and crash_before_node_return. This is the interruption we requested. The Compose setting init: true runs Python as a child process: it is needed for this controlled termination, rather than relying on the special behaviour of container PID 1.
Inspect both systems:
docker compose -p stateful-lab run --rm workflow state --thread naive
docker compose -p stateful-lab run --rm workflow ticketsState shows next: ["create_ticket"] without a ticket_id. The service already lists ticket 1. The gap between those two responses is the important observation.
Resume the same thread:
docker compose -p stateful-lab run --rm workflow resume --thread naive
docker compose -p stateful-lab run --rm workflow ticketsThe node starts again and the service creates ticket 2. prepare does not run again; the action inside the unfinished node does. Checkpoint recovery worked, but the receiver interpreted the second HTTP request as a new creation.
Test 2: one operation, even after restarting the receiver
The fix uses an idempotency key: an identifier that stays unchanged across attempts at the same operation. The receiver atomically binds the key, request content and created ticket. A repeated request returns the same ticket.
The key case-safe:open-ticket:v1 identifies a specific intent. It is saved in the workflow input and sent in the Idempotency-Key header. Generating a new UUID inside the node on every attempt would break this agreement. Using only the title would also be wrong: two legitimate requests can have identical text. This is why the AWS Builders' Library distinguishes request identity from merely comparing parameters.
docker compose -p stateful-lab run --rm workflow start --thread safe --operation case-safe:open-ticket:v1 --mode safe --crash
docker compose -p stateful-lab run --rm workflow state --thread safe
docker compose -p stateful-lab run --rm workflow ticketsAfter exit 137, the checkpoint still points to create_ticket; the service contains the new ticket 3. Restart the receiver as well, then resume:
docker compose -p stateful-lab restart api
docker compose -p stateful-lab run --rm workflow resume --thread safe
docker compose -p stateful-lab run --rm workflow state --thread safe
docker compose -p stateful-lab run --rm workflow ticketsThe action responds with created: false, ticket_id: 3. The node can finally save that result; next becomes an empty list. The service still has three tickets in total: two from the earlier test and one from this one.
In the receiver, a UNIQUE constraint on the key and a BEGIN IMMEDIATE transaction protect the check and creation. If we only wrote a local log after receiving the HTTP response, a crash between the response and the log would recreate the same problem.
Checks that separate a retry from a new request
The positive test alone could let us build a service that rejects every subsequent request. Add two checks:
docker compose -p stateful-lab run --rm workflow post --operation case-safe:open-ticket:v1 --title "Different payload"
docker compose -p stateful-lab run --rm workflow post --operation case-new:open-ticket:v1The first returns HTTP 409, with exit code 1: the key exists but the content differs. The second creates ticket 4, because it identifies another operation. Its title can match the previous title without preventing a new creation.
You can also call resume --thread safe again: the workflow has already completed, and no further effect_ack appears.
| Step | Node still to run | Tickets for that individual operation |
|---|---|---|
| Crash without a key | create_ticket | 1 |
| Resume without a key | none | 2 |
| Crash with a key | create_ticket | 1 |
| Restart receiver and resume with a key | none | 1 |
These are the results observed in the lab. Deduplication depends on the receiver's implemented contract and retention of its database. For an API without idempotency support, use an operation identifier to look up its result or explicitly handle an uncertain outcome before retrying. Adding a caller-side ledger alone does not make a write to another system atomic.
Recovery, replay and approval have different consequences
The test recovers the latest available state. Replay from a historical checkpoint instead asks the graph to execute subsequent nodes again; it can repeat API and model calls. Using history therefore also requires a decision about which effects should be reproduced. In graphs with parallel nodes, LangGraph can recover saved writes from completed nodes; our node died before completing that write.
Human approval is another step. LangGraph provides interrupts and resumption with Command(resume=...). Approval should refer to the action the person reviewed: changing its recipient or payload requires reconsideration. This lab restarts with the same code and data; workflow migrations, concurrent retries across hosts and stream reconnection are outside the test.
For a real service, I would first define how long keys remain available relative to the permitted retry window and how uncertain results are reconciled. Removing an operation's receipt too early makes duplicates possible again.
When I would choose something simpler
A linear procedure may be enough to read data, produce a proposal and return it immediately. If the requirement is to execute independent jobs, retry them and distribute them across workers, a queue with database state avoids modelling a larger application graph.
A graph becomes useful when different paths need to be understood and controlled: a rejected proposal returns for revision, a failed check changes strategy or a person approves days later. It expresses those transitions, while adding a state schema, migration rules and checkpoint maintenance.
A durable orchestrator such as Temporal is an alternative for coordinating activities, waits and recovery. Its Activity documentation retains the idempotency requirement: an activity can perform an effect and die before recording completion.
My starting question would be concrete: after an interruption, which operation can I repeat, and which must I look up in the receiving system first? The graph follows that answer.
Complete sources and cleanup
The following five files belong to a teaching lab rather than a publicly exposed service. The Python image is pinned by digest and the two direct dependencies by version. Transitive dependencies are still resolved by pip.
After testing, this command removes only the containers, network and volumes of project stateful-lab. It therefore also deletes the tickets and checkpoints created here:
docker compose -p stateful-lab down -vThe local image stateful-agent-lab:local can be removed with docker image rm stateful-agent-lab:local when no other lab uses it. Other projects may share the Python base image.
Dockerfile
FROM python:3.13-slim@sha256:57b29a7c050c4e679e4bea572e14b3965cf4b5839de2da49d55e49ffc4024f39
ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1 LANGGRAPH_STRICT_MSGPACK=true
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY api.py workflow.py ./
CMD ["python", "workflow.py", "--help"]requirements.txt
langgraph==1.2.17
langgraph-checkpoint-sqlite==3.1.2compose.yaml
name: stateful-agent-lab
services:
api:
image: stateful-agent-lab:local
build: .
command: ["python", "api.py"]
volumes:
- receiver-data:/data
healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=2)"]
interval: 2s
timeout: 3s
retries: 15
read_only: true
cap_drop: [ALL]
security_opt: ["no-new-privileges:true"]
networks: [lab]
workflow:
init: true
image: stateful-agent-lab:local
entrypoint: ["python", "workflow.py"]
environment:
API_URL: http://api:8000
volumes:
- checkpoint-data:/data
depends_on:
api:
condition: service_healthy
read_only: true
cap_drop: [ALL]
security_opt: ["no-new-privileges:true"]
networks: [lab]
volumes:
receiver-data:
checkpoint-data:
networks:
lab:
internal: trueapi.py
"""Local teaching receiver. No external service calls or credentials."""
import json
import sqlite3
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
DB = "/data/receiver.sqlite"
with sqlite3.connect(DB) as db:
db.execute("""CREATE TABLE IF NOT EXISTS tickets (
id INTEGER PRIMARY KEY,
operation_key TEXT UNIQUE,
payload TEXT NOT NULL
)""")
class Handler(BaseHTTPRequestHandler):
def respond(self, status, value):
data = json.dumps(value, sort_keys=True).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def do_GET(self):
if self.path == "/health":
return self.respond(200, {"ok": True})
if self.path != "/tickets":
return self.respond(404, {"error": "not found"})
with sqlite3.connect(DB) as db:
rows = db.execute(
"SELECT id, operation_key, payload FROM tickets ORDER BY id"
).fetchall()
return self.respond(200, {"tickets": [
{"ticket_id": row[0], "key": row[1],
"payload": json.loads(row[2])} for row in rows
]})
def do_POST(self):
if self.path != "/tickets":
return self.respond(404, {"error": "not found"})
try:
length = int(self.headers.get("Content-Length", "0"))
if not 0 < length <= 4096:
return self.respond(400, {"error": "invalid size"})
payload = json.loads(self.rfile.read(length))
if (not isinstance(payload, dict)
or set(payload) != {"title"}
or not isinstance(payload["title"], str)
or not 1 <= len(payload["title"]) <= 200):
return self.respond(400, {"error": "invalid payload"})
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
key = self.headers.get("Idempotency-Key")
if key is not None and not 1 <= len(key) <= 200:
return self.respond(400, {"error": "invalid key"})
with sqlite3.connect(DB, timeout=10) as db:
db.execute("BEGIN IMMEDIATE")
prior = (db.execute(
"SELECT id, payload FROM tickets WHERE operation_key = ?",
(key,)
).fetchone() if key else None)
if prior:
if prior[1] != canonical:
return self.respond(409, {"error": "key reused with different payload"})
ticket_id, created = prior[0], False
else:
cursor = db.execute(
"INSERT INTO tickets(operation_key, payload) VALUES (?, ?)",
(key, canonical)
)
ticket_id, created = cursor.lastrowid, True
# The database transaction commits BEFORE the HTTP response.
return self.respond(201 if created else 200, {
"ticket_id": ticket_id, "created": created
})
except (ValueError, json.JSONDecodeError):
return self.respond(400, {"error": "invalid JSON"})
ThreadingHTTPServer(("0.0.0.0", 8000), Handler).serve_forever()workflow.py
"""Reproducible checkpoint/effect gap, without an LLM or paid API."""
import argparse
import json
import os
import signal
import urllib.error
import urllib.request
from typing import NotRequired, TypedDict
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
API = os.environ.get("API_URL", "http://api:8000")
def emit(event, **values):
print(json.dumps({"event": event, **values}, sort_keys=True), flush=True)
def call(path, payload=None, key=None):
headers = {"Content-Type": "application/json"}
if key is not None:
headers["Idempotency-Key"] = key
data = None if payload is None else json.dumps(payload).encode()
request = urllib.request.Request(API + path, data=data, headers=headers)
with urllib.request.urlopen(request, timeout=5) as response:
return json.load(response)
class State(TypedDict):
operation_id: str
mode: str
title: str
key: NotRequired[str]
ticket_id: NotRequired[int]
parser = argparse.ArgumentParser()
parser.add_argument("action", choices=["start", "resume", "state", "tickets", "post"])
parser.add_argument("--thread", default="demo")
parser.add_argument("--operation", default="case-001:open-ticket:v1")
parser.add_argument("--mode", choices=["naive", "safe"], default="safe")
parser.add_argument("--title", default="Check supplier configuration")
parser.add_argument("--crash", action="store_true")
args = parser.parse_args()
if args.action == "tickets":
emit("tickets", **call("/tickets"))
elif args.action == "post":
try:
emit("post", **call("/tickets", {"title": args.title}, args.operation))
except urllib.error.HTTPError as error:
emit("http_error", status=error.code, body=json.loads(error.read()))
raise SystemExit(1)
else:
def prepare(state: State):
emit("prepare", operation=state["operation_id"])
return {"key": state["operation_id"]}
def create_ticket(state: State):
key = state["key"] if state["mode"] == "safe" else None
result = call("/tickets", {"title": state["title"]}, key)
emit("effect_ack", **result)
if args.crash:
emit("crash_before_node_return")
os.kill(os.getpid(), signal.SIGKILL)
return {"ticket_id": result["ticket_id"]}
builder = StateGraph(State)
builder.add_node("prepare", prepare)
builder.add_node("create_ticket", create_ticket)
builder.add_edge(START, "prepare")
builder.add_edge("prepare", "create_ticket")
builder.add_edge("create_ticket", END)
config = {"configurable": {"thread_id": args.thread}}
with SqliteSaver.from_conn_string("/data/checkpoints.sqlite") as saver:
graph = builder.compile(checkpointer=saver)
saved = graph.get_state(config)
if args.action == "state":
emit("state", values=saved.values, next=list(saved.next))
elif args.action == "start":
if saved.values:
parser.error("thread already exists; resume it or choose a new thread")
result = graph.invoke({
"operation_id": args.operation, "mode": args.mode,
"title": args.title
}, config, durability="sync")
emit("done", **result)
else:
if not saved.values:
parser.error("thread not found; use the original thread id")
result = graph.invoke(None, config, durability="sync")
emit("done", **result)