--- title: "Stateful agents: recovering from a crash without duplicating actions" source: "https://www.matteovincenti.com/en/blog/stateful-agents-crash-checkpoints-idempotency" language: en author: "Matteo Vincenti" published: 2026-10-24 category: "AI" tags: ["LangGraph", "Workflow", "Idempotency", "Docker"] --- # Stateful agents: recovering from a crash without duplicating actions A Docker lab with LangGraph shows what runs again after a crash and how checkpoints and idempotency keys prevent duplicate tickets. ![A ticket retained in a tray beside a green dispenser, a key and an interrupted return rail.](https://cdn.sanity.io/images/xrqkwry7/production/ed787f69189ec750982585d4f65d3d298d1276df-1728x910.png?w=1400&fit=max&auto=format) Category: AI ยท Tags: LangGraph, Workflow, Idempotency, Docker 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: ```txt START -> prepare -> create_ticket -> END ``` `prepare` 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](https://docs.langchain.com/oss/python/langgraph/graph-api) defines these building blocks. The [LangChain, LangGraph, OpenRouter and Langfuse map](https://www.matteovincenti.com/en/blog/langchain-langgraph-openrouter-langfuse-which-to-choose) 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](https://docs.langchain.com/oss/python/langgraph/checkpointers) 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`: ```py 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: ```bash mkdir stateful-agent-lab cd stateful-agent-lab docker version docker compose version ``` Copy 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: ```txt stateful-agent-lab/ Dockerfile requirements.txt compose.yaml api.py workflow.py ``` `api` 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. ```bash docker compose -p stateful-lab up -d --build api docker compose -p stateful-lab run --rm workflow tickets ``` The 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: ```bash docker compose -p stateful-lab run --rm workflow start --thread naive --operation case-naive:open-ticket:v1 --mode naive --crash ``` The 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: ```bash docker compose -p stateful-lab run --rm workflow state --thread naive docker compose -p stateful-lab run --rm workflow tickets ``` State 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: ```bash docker compose -p stateful-lab run --rm workflow resume --thread naive docker compose -p stateful-lab run --rm workflow tickets ``` The 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](https://aws.amazon.com/builders-library/making-retries-safe-with-idempotent-APIs/) distinguishes request identity from merely comparing parameters. ```bash 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 tickets ``` After exit 137, the checkpoint still points to `create_ticket`; the service contains the new ticket **3**. Restart the receiver as well, then resume: ```bash 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 tickets ``` The 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: ```bash 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:v1 ``` The 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. _Checks that separate a retry from a new request_ | 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=...)`](https://docs.langchain.com/oss/python/langgraph/interrupts). 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](https://docs.temporal.io/activity-definition) 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: ```bash docker compose -p stateful-lab down -v ``` The 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` ```docker 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` ```txt langgraph==1.2.17 langgraph-checkpoint-sqlite==3.1.2 ``` ### `compose.yaml` ```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: true ``` ### `api.py` ```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` ```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) ```