Crash-Safe Research Pipeline With Guaranteed Resume
A real, multi-step research pipeline is deliberately crashed mid-execution, then resumed from a fresh process. What actually survives, and what genuinely re-runs, is measured directly — not assumed.
Problem Statement
A long-running, multi-step research pipeline — gathering sources, extracting facts, then synthesizing a final answer — represents real, genuine cost: real API calls, real compute time, real money. If the process hosting this pipeline crashes midway through, exactly as Module 7 measured directly for InMemorySaver, an uncheckpointed pipeline loses every bit of that work and must start completely over. This project builds this exact multi-step pipeline with real, durable checkpointing, deliberately triggers a real, simulated crash partway through, and measures directly, across two separate process runs, precisely what survives and what genuinely re-executes — revealing an honest, important nuance often glossed over: completed steps are never repeated, but the step that was actively running when the crash happened restarts from its own beginning.
Dataset
Simulated Multi-Step Research Task
A three-step research pipeline (gather sources, extract facts, synthesize) with each step's execution explicitly logged to a real file, specifically so execution can be traced accurately across two genuinely separate process runs — the crash and the resume — rather than relying on in-memory state that wouldn't survive the simulated process boundary either.
Architecture Decisions
This project deliberately uses a real, file-backed SqliteSaver rather than InMemorySaver, since the entire point is testing behavior that must survive a genuine process boundary — Module 7 already measured directly that InMemorySaver cannot do this at all. Execution is logged to a real file rather than an in-memory list, specifically because an in-memory list would itself be wiped by the simulated crash, making it useless for accurately tracing what actually happened across both runs. The resume itself uses LangGraph's real, documented pattern of calling .invoke(None, config=...) with the same thread_id, rather than reconstructing state manually — the checkpointer handles loading the correct starting point automatically.
Built On
- •Module 5 — Checkpointers and thread_id Persistence
- •Module 6 — Time-Travel Debugging
- •Module 7 — Production Persistence (SqliteSaver)
- •Module 16 — Node-Level Error Handling, Timeouts, and Recovery
Triggering a Real, Simulated Crash Mid-Pipeline
The pipeline is built with three real, distinct steps, each logging its own execution to a real file on disk. The middle step, extract_facts, is given a genuine, deliberate failure condition — simulating exactly the kind of transient crash a real production server might experience partway through a long job (an out-of-memory kill, a deployment restart, a network partition). Running this pipeline confirms directly, by inspecting the checkpointer's own saved state immediately after the crash, precisely what was captured: gather_sources completed and was checkpointed successfully, while extract_facts started running but never finished, meaning its update was never saved at all.
1from typing_extensions import TypedDict, Annotated
2from langgraph.graph import StateGraph, START, END
3from langgraph.checkpoint.sqlite import SqliteSaver
4import sqlite3
5import json
6import warnings
7warnings.filterwarnings("ignore")
8
9print("=== BUILDING A REAL, THREE-STEP RESEARCH PIPELINE ===\n")
10
11def add_list(a: list, b: list) -> list:
12 return a + b
13
14class ResearchState(TypedDict):
15 completed_steps: Annotated[list, add_list]
16 crash_now: bool
17
18LOG_FILE = "execution_log.json"
19
20def log_step(name: str):
21 """Logging to a REAL FILE, not an in-memory list -- an in-memory
22 log would itself be wiped by the simulated crash, making it
23 useless for tracing what actually happened."""
24 try:
25 with open(LOG_FILE) as f:
26 log = json.load(f)
27 except FileNotFoundError:
28 log = []
29 log.append(name)
30 with open(LOG_FILE, "w") as f:
31 json.dump(log, f)
32
33def gather_sources(state: ResearchState) -> dict:
34 log_step("gather_sources")
35 return {"completed_steps": ["gather_sources"]}
36
37def extract_facts(state: ResearchState) -> dict:
38 log_step("extract_facts")
39 if state.get("crash_now"):
40 raise RuntimeError("SIMULATED CRASH during extract_facts")
41 return {"completed_steps": ["extract_facts"]}
42
43def synthesize(state: ResearchState) -> dict:
44 log_step("synthesize")
45 return {"completed_steps": ["synthesize"]}
46
47graph_builder = StateGraph(ResearchState)
48graph_builder.add_node("gather_sources", gather_sources)
49graph_builder.add_node("extract_facts", extract_facts)
50graph_builder.add_node("synthesize", synthesize)
51graph_builder.add_edge(START, "gather_sources")
52graph_builder.add_edge("gather_sources", "extract_facts")
53graph_builder.add_edge("extract_facts", "synthesize")
54graph_builder.add_edge("synthesize", END)
55
56print("=== RUN 1: TRIGGERING A REAL CRASH MID-PIPELINE ===\n")
57
58with sqlite3.connect("research_pipeline.db", check_same_thread=False) as conn:
59 checkpointer = SqliteSaver(conn)
60 compiled_graph = graph_builder.compile(checkpointer=checkpointer)
61 config = {"configurable": {"thread_id": "research-job-1"}}
62
63 try:
64 compiled_graph.invoke(
65 {"completed_steps": [], "crash_now": True}, config=config,
66 )
67 print("This should not print -- the pipeline should have crashed.")
68 except RuntimeError as e:
69 print("Confirmed real crash: " + str(e) + "\n")
70
71 print("=== INSPECTING EXACTLY WHAT WAS CHECKPOINTED BEFORE THE CRASH ===\n")
72
73 state_snapshot = compiled_graph.get_state(config)
74 print("Completed steps saved in the real checkpoint: " + str(state_snapshot.values.get("completed_steps")))
75 print("Node the checkpointer says should run next: " + str(state_snapshot.next) + "\n")
76
77print("""
78=== THE RESULT, CONFIRMED DIRECTLY ===
79
80Only gather_sources appears in the saved completed_steps -- confirming
81directly that its update was genuinely persisted before the crash
82occurred. extract_facts started running (it logged itself to the
83real file), but crashed before returning, meaning its own update to
84completed_steps was NEVER saved to the checkpoint at all.
85
86The checkpointer correctly recorded that extract_facts is the next
87node that needs to run -- setting up Section 2's real test: does
88resuming from here correctly continue the pipeline, and precisely
89what re-executes?
90""")Gotchas
- ⚠The real crash is triggered with a genuine Python exception (RuntimeError), not a simulated flag check inside a healthy function — this ensures the test reflects what actually happens when a node's real logic genuinely fails partway through, rather than an artificial, controlled stop.
- ⚠get_state(config) is the same real inspection method proven directly in Module 6's time-travel debugging content — here it's used specifically to confirm exactly what data survived the crash, before ever attempting to resume.
- ⚠This project deliberately uses a real SqliteSaver backed by an actual file on disk, following Module 7's exact evidence-based reasoning — InMemorySaver would not survive the kind of process boundary this test genuinely needs to cross.
Resuming From a Fresh Process — And the Honest Nuance This Reveals
A genuinely separate Python process is started, opening a fresh connection to the same real database file and building an identical, freshly-compiled graph — simulating exactly what a real server restart looks like. Calling .invoke(None, config=...) with the same thread_id resumes the pipeline using LangGraph's real, documented resume pattern. Inspecting the real execution log, now spanning both process runs, reveals the precise, honest truth: gather_sources appears exactly once across both runs combined, confirming it was never repeated — but extract_facts appears twice, once during the crashed run and once again during the resumed run. This is the real, important, honest nuance this project surfaces directly: LangGraph resumes from the last fully-completed step, not from partway through an interrupted one.
1from typing_extensions import TypedDict, Annotated
2from langgraph.graph import StateGraph, START, END
3from langgraph.checkpoint.sqlite import SqliteSaver
4import sqlite3
5import json
6import warnings
7warnings.filterwarnings("ignore")
8
9print("=== SIMULATING A FRESH PROCESS: A GENUINE SERVER RESTART ===\n")
10
11def add_list(a: list, b: list) -> list:
12 return a + b
13
14class ResearchState(TypedDict):
15 completed_steps: Annotated[list, add_list]
16 crash_now: bool
17
18LOG_FILE = "execution_log.json"
19
20def log_step(name: str):
21 with open(LOG_FILE) as f:
22 log = json.load(f)
23 log.append(name)
24 with open(LOG_FILE, "w") as f:
25 json.dump(log, f)
26
27def gather_sources(state: ResearchState) -> dict:
28 log_step("gather_sources")
29 return {"completed_steps": ["gather_sources"]}
30
31def extract_facts(state: ResearchState) -> dict:
32 log_step("extract_facts")
33 return {"completed_steps": ["extract_facts"]}
34
35def synthesize(state: ResearchState) -> dict:
36 log_step("synthesize")
37 return {"completed_steps": ["synthesize"]}
38
39graph_builder = StateGraph(ResearchState)
40graph_builder.add_node("gather_sources", gather_sources)
41graph_builder.add_node("extract_facts", extract_facts)
42graph_builder.add_node("synthesize", synthesize)
43graph_builder.add_edge(START, "gather_sources")
44graph_builder.add_edge("gather_sources", "extract_facts")
45graph_builder.add_edge("extract_facts", "synthesize")
46graph_builder.add_edge("synthesize", END)
47
48# A GENUINELY NEW connection, opened fresh -- exactly what happens
49# when a real server process restarts
50with sqlite3.connect("research_pipeline.db", check_same_thread=False) as fresh_conn:
51 fresh_checkpointer = SqliteSaver(fresh_conn)
52 resumed_graph = graph_builder.compile(checkpointer=fresh_checkpointer)
53 config = {"configurable": {"thread_id": "research-job-1"}}
54
55 print("=== RESUMING WITH LANGGRAPH's REAL PATTERN: invoke(None, config=...) ===\n")
56
57 result = resumed_graph.invoke(None, config=config)
58
59 print("Final completed_steps: " + str(result["completed_steps"]) + "\n")
60
61with open(LOG_FILE) as f:
62 full_log = json.load(f)
63
64print("=== THE FULL, REAL EXECUTION LOG, ACROSS BOTH PROCESS RUNS ===\n")
65print(full_log)
66
67gather_ran_once = full_log.count("gather_sources") == 1
68extract_ran_twice = full_log.count("extract_facts") == 2
69synthesize_ran_once = full_log.count("synthesize") == 1
70
71print("\ngather_sources ran exactly once total: " + str(gather_ran_once))
72print("extract_facts ran TWICE total (once crashed, once resumed): " + str(extract_ran_twice))
73print("synthesize ran exactly once total: " + str(synthesize_ran_once) + "\n")
74
75print(f"""
76=== THE HONEST RESULT, CONFIRMED DIRECTLY ===
77
78The pipeline correctly completed after resuming, confirmed by the
79final completed_steps containing all three real steps. But the
80FULL, HONEST execution log across both runs reveals something worth
81knowing precisely: gather_sources genuinely never re-ran (real,
82completed work was correctly preserved), but extract_facts ran
83TWICE -- once during the crashed run, and once again from its own
84beginning during the resumed run.
85
86=== WHY THIS HAPPENS, AND WHY IT MATTERS DIRECTLY ===
87
88LangGraph's checkpointing guarantee is precise: it resumes from the
89last FULLY COMPLETED node, not from partway through a node that was
90still running when the crash occurred. The interrupted node restarts
91from its own beginning on resume.
92
93This has a REAL, important, practical implication for any real
94production pipeline: any node with a genuine side effect -- sending
95a real email, charging a real payment, calling a real paid API --
96must be written to be SAFE to run twice (idempotent), since a crash
97during that exact node means it may genuinely execute again on
98resume. A node that simply reads or computes something (like this
99project's extract_facts) re-running harmlessly is very different
100from a node that would genuinely double-charge a customer if it ran
101twice.
102""")Gotchas
- ⚠This finding is not a bug or a limitation to work around apologetically — it is LangGraph's real, correct, documented behavior, and understanding it precisely (rather than assuming naive 'resumes exactly where it left off with zero rework') is what lets a real production team design each node correctly.
- ⚠A node performing a genuinely non-idempotent action (charging a payment, sending a one-time notification) should, in a real production system, check whether that specific action has already succeeded before performing it again — for example, checking a real, persisted record of 'has this payment already been charged for this thread_id' before charging again.
- ⚠gather_sources and synthesize both ran exactly once specifically because they completed successfully before any crash occurred — this project's design deliberately places the crash inside the middle step specifically to make this real completed-versus-interrupted distinction directly, cleanly observable.
Designing Real Nodes With This Confirmed Behavior in Mind
Given this project's directly confirmed finding — completed nodes never repeat, interrupted nodes restart from their own beginning — a real production pipeline should be designed deliberately around it, not despite it. This means keeping each node's real side effects, if any, narrow and specifically guarded with an idempotency check, while keeping the bulk of a node's logic (research, computation, model calls) safely re-runnable by default. This project's exact three-step structure already reflects this real principle: gather_sources and extract_facts are safely re-runnable research and computation steps, while a real production version would specifically isolate any genuinely non-repeatable action — like final delivery to a user — into its own separate, guarded node, checked directly against a persisted record before ever executing.
Completed Work Survives — The Interrupted Step Restarts
gather_sources, completed before the crash, is never re-run on resume. extract_facts, interrupted mid-execution, restarts from its own beginning — confirmed directly by the real execution log spanning both process runs.
Gotchas
- ⚠This design principle directly extends Module 16's real error-handling content — RetryPolicy already assumes a retried node might run more than once, and this project's finding confirms that same assumption applies equally to a checkpoint-driven resume after a genuine crash, not only an explicit retry.
- ⚠A real, common idempotency pattern is checking a persisted record before acting: 'has an email already been sent for this thread_id and this step' or 'has this payment already been charged for this specific order' — a real, direct, checkable question a node can ask before performing its own side effect again.