From f4b01fb00c2c1f5804b4e16ba9f91233b363a987 Mon Sep 17 00:00:00 2001
From: unknown <3058704216@qq.com>
Date: Sat, 1 Aug 2026 23:53:58 +0800
Subject: [PATCH 1/4] perf: eliminate full message-history deep clone in
 add_messages reducer
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The messages reducer is the default for the `messages` channel in every
agent state. Each LLM turn previously deep-cloned the entire accumulated
history (all message content + tool args) — O(n^2) over a run.

Change the reducer signature from `fn(&JsonValue, &JsonValue) -> JsonValue`
to `fn(JsonValue, &JsonValue) -> JsonValue` and have
BinaryOperatorAggregate::update hand ownership of the accumulated value to
the reducer via `guard.take()`, so existing messages are moved instead of
cloned. `update` stays borrowed (usually a single new message).

add_messages now:
- consumes `current` and moves existing messages into the result
- collects remove-ids as borrowed `&str` (no per-message String alloc)
- takes a fast path when there are no removals

A benchmark is added (tests/bench_messages.rs, #[ignore]d) measuring the
reducer hot path: ~100x faster at 250 steps up to ~330x at 1000 steps,
and scaling drops from quadratic to near-linear.

This is a breaking API change for reducers: they now receive `current` by
value.
---
 crates/langgraph-core/src/channels/binop.rs   | 19 ++++--
 crates/langgraph-prebuilt/src/chat_agent.rs   |  8 +--
 crates/langgraph-prebuilt/src/types.rs        | 49 +++++++++-----
 .../tests/bench_messages.rs                   | 66 +++++++++++++++++++
 4 files changed, 113 insertions(+), 29 deletions(-)
 create mode 100644 crates/langgraph-prebuilt/tests/bench_messages.rs

diff --git a/crates/langgraph-core/src/channels/binop.rs b/crates/langgraph-core/src/channels/binop.rs
index bd64115..3a28376 100644
--- a/crates/langgraph-core/src/channels/binop.rs
+++ b/crates/langgraph-core/src/channels/binop.rs
@@ -4,7 +4,10 @@ use parking_lot::RwLock;
 use serde_json::Value as JsonValue;
 
 /// Reducer function type: (current, update) -> new
-pub type ReducerFn = fn(&JsonValue, &JsonValue) -> JsonValue;
+///
+/// Takes ownership of `current` so reducers can move existing state instead of
+/// deep-cloning it on every update; `update` is borrowed (usually small).
+pub type ReducerFn = fn(JsonValue, &JsonValue) -> JsonValue;
 
 /// Applies a binary operator to accumulate values.
 ///
@@ -67,7 +70,9 @@ impl Channel for BinaryOperatorAggregate {
                 continue;
             }
 
-            match guard.as_ref() {
+            // Move the accumulated value out so the reducer can consume it
+            // instead of deep-cloning the whole state on every update.
+            match guard.take() {
                 Some(current) => {
                     let new_val = (self.reducer)(current, val);
                     *guard = Some(new_val);
@@ -102,10 +107,10 @@ impl Channel for BinaryOperatorAggregate {
 }
 
 /// Common reducer: append arrays
-pub fn append_reducer(current: &JsonValue, update: &JsonValue) -> JsonValue {
+pub fn append_reducer(current: JsonValue, update: &JsonValue) -> JsonValue {
     let mut result = match current {
-        JsonValue::Array(arr) => arr.clone(),
-        other => vec![other.clone()],
+        JsonValue::Array(arr) => arr,
+        other => vec![other],
     };
     match update {
         JsonValue::Array(arr) => result.extend(arr.iter().cloned()),
@@ -115,10 +120,10 @@ pub fn append_reducer(current: &JsonValue, update: &JsonValue) -> JsonValue {
 }
 
 /// Common reducer: merge objects
-pub fn merge_reducer(current: &JsonValue, update: &JsonValue) -> JsonValue {
+pub fn merge_reducer(current: JsonValue, update: &JsonValue) -> JsonValue {
     match (current, update) {
         (JsonValue::Object(curr), JsonValue::Object(upd)) => {
-            let mut merged = curr.clone();
+            let mut merged = curr;
             for (k, v) in upd {
                 merged.insert(k.clone(), v.clone());
             }
diff --git a/crates/langgraph-prebuilt/src/chat_agent.rs b/crates/langgraph-prebuilt/src/chat_agent.rs
index 661247e..15e4f4f 100644
--- a/crates/langgraph-prebuilt/src/chat_agent.rs
+++ b/crates/langgraph-prebuilt/src/chat_agent.rs
@@ -66,8 +66,8 @@ impl ReActAgent {
 }
 
 /// Reducer for messages channel: appends new messages to existing ones.
-fn messages_reducer(current: &JsonValue, update: &JsonValue) -> JsonValue {
-    add_messages(current.clone(), update.clone())
+fn messages_reducer(current: JsonValue, update: &JsonValue) -> JsonValue {
+    add_messages(current, update)
 }
 
 /// Create a ReAct agent with the given model and tools.
@@ -206,7 +206,7 @@ mod tests {
             {"type": "ai", "content": "Hello"}
         ]);
 
-        let merged = messages_reducer(&current, &update);
+        let merged = messages_reducer(current, &update);
         let messages = merged.as_array().unwrap();
         assert_eq!(messages.len(), 2);
     }
@@ -220,7 +220,7 @@ mod tests {
             "result": "done"
         });
 
-        let _merged = messages_reducer(&current, &update);
+        let _merged = messages_reducer(current, &update);
         // add_messages merges the messages arrays
         // "result" is not messages, so it gets appended as a message
     }
diff --git a/crates/langgraph-prebuilt/src/types.rs b/crates/langgraph-prebuilt/src/types.rs
index f1281fd..97a60e8 100644
--- a/crates/langgraph-prebuilt/src/types.rs
+++ b/crates/langgraph-prebuilt/src/types.rs
@@ -337,7 +337,11 @@ impl From<&str> for MessageContent {
 
 /// Merge function for messages: appends new messages to existing ones.
 /// This is the default reducer for the `messages` field in agent states.
-pub fn add_messages(current: JsonValue, update: JsonValue) -> JsonValue {
+///
+/// Consumes `current` by value so existing messages are moved in place instead
+/// of deep-cloning the whole history on every step. `update` is borrowed and is
+/// usually a single new message.
+pub fn add_messages(current: JsonValue, update: &JsonValue) -> JsonValue {
     // Check if the update is a "Reset" signal: {"reset": true, "messages": [...]}
     if let Some(obj) = update.as_object() {
         if obj.get("reset").and_then(|v| v.as_bool()) == Some(true) {
@@ -352,30 +356,37 @@ pub fn add_messages(current: JsonValue, update: JsonValue) -> JsonValue {
         _ => vec![],
     };
 
-    let new_messages: Vec<JsonValue> = match update {
-        JsonValue::Array(arr) => arr,
+    let new_messages: Vec<&JsonValue> = match update {
+        JsonValue::Array(arr) => arr.iter().collect(),
         other => vec![other],
     };
 
-    // Handle RemoveMessage by filtering out messages with matching IDs
-    let mut result: Vec<JsonValue> = Vec::new();
-    let mut remove_ids: Vec<String> = Vec::new();
-
-    // Collect IDs to remove
+    // Collect IDs to remove, borrowing from `update` to avoid allocating a
+    // String per message just for the removal check.
+    let mut remove_ids: Vec<&str> = Vec::new();
     for msg in &new_messages {
         if let Some(obj) = msg.as_object() {
             if obj.get("type").and_then(|v| v.as_str()) == Some("remove") {
                 if let Some(id) = obj.get("id").and_then(|v| v.as_str()) {
-                    remove_ids.push(id.to_string());
+                    remove_ids.push(id);
                 }
             }
         }
     }
 
+    let mut result: Vec<JsonValue> = Vec::with_capacity(messages.len() + new_messages.len());
+
+    if remove_ids.is_empty() {
+        // Fast path: no removals, just append. Existing messages are moved in.
+        result.extend(messages);
+        result.extend(new_messages.into_iter().cloned());
+        return JsonValue::Array(result);
+    }
+
     // Add existing messages, skipping removed ones
     for msg in messages {
         if let Some(id) = msg.get("id").and_then(|v| v.as_str()) {
-            if remove_ids.contains(&id.to_string()) {
+            if remove_ids.contains(&id) {
                 continue;
             }
         }
@@ -389,16 +400,18 @@ pub fn add_messages(current: JsonValue, update: JsonValue) -> JsonValue {
                 continue;
             }
         }
-        result.push(msg);
+        result.push(msg.clone());
     }
 
     JsonValue::Array(result)
 }
 
-/// Merge function for messages with reference signature.
+/// Merge function for messages, compatible with the `#[channel(...)]` reducer
+/// signature (`fn(JsonValue, &JsonValue) -> JsonValue`).
 ///
-/// This is the version compatible with `#[channel(reducer = "...")]` in the
-/// derive macro, which expects `fn(&JsonValue, &JsonValue) -> JsonValue`.
+/// Kept as a thin wrapper over [`add_messages`] so existing
+/// `#[channel(reducer = "add_messages_ref")]` references and the derive macro's
+/// built-in `#[channel(messages)]` keep working.
 ///
 /// ```ignore
 /// #[derive(StateGraph)]
@@ -407,8 +420,8 @@ pub fn add_messages(current: JsonValue, update: JsonValue) -> JsonValue {
 ///     messages: Vec<Message>,
 /// }
 /// ```
-pub fn add_messages_ref(current: &JsonValue, update: &JsonValue) -> JsonValue {
-    add_messages(current.clone(), update.clone())
+pub fn add_messages_ref(current: JsonValue, update: &JsonValue) -> JsonValue {
+    add_messages(current, update)
 }
 
 #[cfg(test)]
@@ -443,7 +456,7 @@ mod tests {
         let update = serde_json::json!([
             {"type": "ai", "content": "Hello"},
         ]);
-        let result = add_messages(existing, update);
+        let result = add_messages(existing, &update);
         assert_eq!(result.as_array().unwrap().len(), 2);
     }
 
@@ -456,7 +469,7 @@ mod tests {
         let update = serde_json::json!([
             {"type": "remove", "id": "msg1"},
         ]);
-        let result = add_messages(existing, update);
+        let result = add_messages(existing, &update);
         let arr = result.as_array().unwrap();
         assert_eq!(arr.len(), 1);
         assert_eq!(arr[0]["id"], "msg2");
diff --git a/crates/langgraph-prebuilt/tests/bench_messages.rs b/crates/langgraph-prebuilt/tests/bench_messages.rs
new file mode 100644
index 0000000..56d0023
--- /dev/null
+++ b/crates/langgraph-prebuilt/tests/bench_messages.rs
@@ -0,0 +1,66 @@
+//! Benchmark: message-history accumulation through the channel reducer.
+//!
+//! This is the per-agent-turn hot path: a growing `messages` array is merged by
+//! `add_messages_ref` on every LLM step. Run explicitly (it is `#[ignore]`d so
+//! normal `cargo test` runs stay fast):
+//!
+//! ```text
+//! cargo test --release -p langgraph-prebuilt --test bench_messages -- --ignored --nocapture
+//! ```
+
+use langgraph::channels::{BinaryOperatorAggregate, Channel};
+use langgraph_prebuilt::add_messages_ref;
+use serde_json::Value as JsonValue;
+use std::time::{Duration, Instant};
+
+fn make_message(i: usize) -> JsonValue {
+    // A realistic AI message with tool call args, sized so deep clones cost real time.
+    serde_json::json!({
+        "type": "ai",
+        "content": format!("Assistant reply number {i}: {}", "x".repeat(300)),
+        "tool_calls": [{
+            "name": "search_tool",
+            "args": {
+                "query": format!("query-{i}"),
+                "filters": {"category": "test", "limit": 10, "extra": "data"}
+            },
+            "id": format!("call_{i}")
+        }],
+        "id": format!("msg_{i}")
+    })
+}
+
+/// Time appending `steps` messages to a growing history, best of `iterations`.
+fn run_case(steps: usize, iterations: u32) -> Duration {
+    let messages: Vec<JsonValue> = (0..steps).map(make_message).collect();
+
+    let mut best = Duration::MAX;
+    for _ in 0..iterations {
+        let ch = BinaryOperatorAggregate::new("messages", add_messages_ref);
+        // Seed with an empty array, exactly like a real graph initializes the
+        // messages channel, so each update goes through the reducer as an append.
+        ch.update(&[serde_json::json!([])]).unwrap();
+        let start = Instant::now();
+        for msg in &messages {
+            ch.update(std::slice::from_ref(msg)).unwrap();
+        }
+        best = best.min(start.elapsed());
+    }
+    best
+}
+
+#[test]
+#[ignore]
+fn bench_message_accumulation() {
+    for steps in [250usize, 500, 1000] {
+        let elapsed = run_case(steps, 3);
+        println!("add_messages: {steps:>4} steps (with tool calls) => best {elapsed:?}");
+    }
+
+    // Sanity: the channel really accumulates one message per update.
+    let ch = BinaryOperatorAggregate::new("messages", add_messages_ref);
+    ch.update(&[serde_json::json!([])]).unwrap();
+    ch.update(&[make_message(0)]).unwrap();
+    ch.update(&[make_message(1)]).unwrap();
+    assert_eq!(ch.get().unwrap().as_array().unwrap().len(), 2);
+}

From f2d231c9e99b1f9be09056f2b3611060b31549a5 Mon Sep 17 00:00:00 2001
From: unknown <3058704216@qq.com>
Date: Sun, 2 Aug 2026 12:20:55 +0800
Subject: [PATCH 2/4] =?UTF-8?q?perf:=20incremental=20checkpoint=20blob=20w?=
 =?UTF-8?q?rites=20=E2=80=94=20dump=20only=20changed=20channels?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Checkpoint saves re-encoded and re-inserted a blob row for every channel
on every super-step. With a large static channel (e.g. embedded context)
that meant re-serializing and re-writing the same value each step — pure
write amplification.

save_checkpoint now takes the channel versions as of the start of the run
and derives new_versions = channels whose version moved, passing only the
delta to the saver. SqliteSaver::put writes blob rows only for
new_versions; reads merge version-joined blob values over the checkpoint
row body (extend, not replace) so every channel resolves through the
version join regardless of which checkpoint wrote the blob.

Adds test_incremental_blob_writes (a second checkpoint with one updated
channel writes one new blob row; both new and old checkpoints read back
fully) and a benchmark suite (tests/bench_pregel.rs, #[ignore]d) covering
checkpointed growth and a static-context scenario.

Measured (release, 800 steps, per-step): sqlite linear 12.651->12.230ms
(~3%), static 200KB context 12.723->12.200ms (~4%). The win is small on
these micro-benches because the dominant per-step cost is in-process
state handling (read_channels deep clone), not blob I/O; write
amplification is eliminated, which matters far more for remote savers and
large static channels.
---
 .../langgraph-checkpoint-sqlite/src/saver.rs  |  98 ++++++-
 crates/langgraph-core/src/graph/state.rs      |  35 ++-
 tests/bench_pregel.rs                         | 260 ++++++++++++++++++
 3 files changed, 380 insertions(+), 13 deletions(-)
 create mode 100644 tests/bench_pregel.rs

diff --git a/crates/langgraph-checkpoint-sqlite/src/saver.rs b/crates/langgraph-checkpoint-sqlite/src/saver.rs
index d46ac5b..a2efa28 100644
--- a/crates/langgraph-checkpoint-sqlite/src/saver.rs
+++ b/crates/langgraph-checkpoint-sqlite/src/saver.rs
@@ -341,14 +341,13 @@ impl SqliteSaver {
         let mut results = Vec::with_capacity(rows.len());
         for row in rows {
             let mut tuple = Self::row_to_tuple(&row)?;
-            // Reconcile channel values from blobs.
+            // Merge blob values (version-joined) over the row's channel_values
+            // (see aget_tuple for the rationale).
             let thread_id = row.get::<String, _>("thread_id");
             let ns = row.get::<String, _>("checkpoint_ns");
             let cid = tuple.checkpoint.id.clone();
             let blob_values = self.load_blobs(&thread_id, &ns, &cid).await?;
-            if !blob_values.is_empty() {
-                tuple.checkpoint.channel_values = blob_values;
-            }
+            tuple.checkpoint.channel_values.extend(blob_values);
             tuple.pending_writes = Some(self.load_writes(&thread_id, &ns, &cid).await?);
             results.push(tuple);
         }
@@ -517,10 +516,12 @@ impl BaseCheckpointSaver for SqliteSaver {
 
         let mut tuple = Self::row_to_tuple(&row)?;
         let cid = tuple.checkpoint.id.clone();
+        // Blob rows are version-addressed and written incrementally, so the
+        // join in SELECT_BLOBS_SQL resolves every channel's value regardless
+        // of which checkpoint created the blob. Merge (not replace) so values
+        // still resolve even if a blob row is ever missing.
         let blob_values = self.load_blobs(thread_id, checkpoint_ns, &cid).await?;
-        if !blob_values.is_empty() {
-            tuple.checkpoint.channel_values = blob_values;
-        }
+        tuple.checkpoint.channel_values.extend(blob_values);
         tuple.pending_writes = Some(self.load_writes(thread_id, checkpoint_ns, &cid).await?);
         Ok(Some(tuple))
     }
@@ -548,8 +549,11 @@ impl BaseCheckpointSaver for SqliteSaver {
 
         let next_config = Self::make_config(thread_id, checkpoint_ns, &checkpoint.id);
 
-        // Strip channel_values from the JSON checkpoint payload to avoid
-        // duplicating them in the row body — they live in checkpoint_blobs.
+        // Strip channel_values from the JSON checkpoint payload — they live in
+        // checkpoint_blobs. Blob rows are keyed by (channel, version) and are
+        // written incrementally (only channels in `new_versions`), so on load
+        // every channel's value resolves through the version join regardless
+        // of which checkpoint wrote the blob.
         let mut checkpoint_value = serde_json::to_value(checkpoint)
             .map_err(|e| CheckpointError::Storage(e.to_string()))?;
         if let Some(obj) = checkpoint_value.as_object_mut() {
@@ -967,6 +971,82 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_incremental_blob_writes() {
+        // cp1: channels a and b both written at version 1.
+        let saver = fresh_saver().await;
+        let cfg = config_for("thread-INC");
+        let mut cp1 = Checkpoint::empty();
+        cp1.channel_values
+            .insert("a".into(), JsonValue::Number(1.into()));
+        cp1.channel_values
+            .insert("b".into(), JsonValue::Number(1.into()));
+        cp1.channel_versions
+            .insert("a".into(), JsonValue::String("1".into()));
+        cp1.channel_versions
+            .insert("b".into(), JsonValue::String("1".into()));
+        let mut versions1: ChannelVersions = HashMap::new();
+        versions1.insert("a".into(), JsonValue::String("1".into()));
+        versions1.insert("b".into(), JsonValue::String("1".into()));
+        let next1 = saver
+            .aput(&cfg, &cp1, &CheckpointMetadata::default(), &versions1)
+            .await
+            .unwrap();
+
+        // cp2: only channel a is updated — new_versions carries just the delta.
+        let mut cp2 = Checkpoint::empty();
+        cp2.channel_values
+            .insert("a".into(), JsonValue::Number(2.into()));
+        cp2.channel_values
+            .insert("b".into(), JsonValue::Number(1.into()));
+        cp2.channel_versions
+            .insert("a".into(), JsonValue::String("2".into()));
+        cp2.channel_versions
+            .insert("b".into(), JsonValue::String("1".into()));
+        let mut versions2: ChannelVersions = HashMap::new();
+        versions2.insert("a".into(), JsonValue::String("2".into()));
+        saver
+            .aput(&next1, &cp2, &CheckpointMetadata::default(), &versions2)
+            .await
+            .unwrap();
+
+        // Blob rows are keyed by (channel, version): cp2 only adds a@2, so
+        // the table holds exactly a@1, b@1, a@2 (a full rewrite would add b@2).
+        let blob_count: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM checkpoint_blobs WHERE thread_id = 'thread-INC'",
+        )
+        .fetch_one(&saver.pool)
+        .await
+        .unwrap();
+        assert_eq!(blob_count, 3, "expected blobs a@1, b@1, a@2, got {blob_count}");
+
+        // cp2 reads back both channels: a@2 via cp2's blob, b@1 via the
+        // version-addressed blob written by cp1 (b's version never moved).
+        let cfg_cp2 = config_with_id("thread-INC", &cp2.id);
+        let tuple = saver.aget_tuple(&cfg_cp2).await.unwrap().unwrap();
+        assert_eq!(
+            tuple.checkpoint.channel_values.get("a"),
+            Some(&JsonValue::Number(2.into()))
+        );
+        assert_eq!(
+            tuple.checkpoint.channel_values.get("b"),
+            Some(&JsonValue::Number(1.into()))
+        );
+
+        // cp1 remains fully readable (older rows rely on blobs, and b@1 still
+        // matches cp2's join as well since b's version never moved).
+        let cfg_cp1 = config_with_id("thread-INC", &cp1.id);
+        let earlier = saver.aget_tuple(&cfg_cp1).await.unwrap().unwrap();
+        assert_eq!(
+            earlier.checkpoint.channel_values.get("a"),
+            Some(&JsonValue::Number(1.into()))
+        );
+        assert_eq!(
+            earlier.checkpoint.channel_values.get("b"),
+            Some(&JsonValue::Number(1.into()))
+        );
+    }
+
     #[tokio::test]
     async fn test_metadata_filter_returns_only_matching_rows() {
         let saver = fresh_saver().await;
diff --git a/crates/langgraph-core/src/graph/state.rs b/crates/langgraph-core/src/graph/state.rs
index 750f202..8b53285 100644
--- a/crates/langgraph-core/src/graph/state.rs
+++ b/crates/langgraph-core/src/graph/state.rs
@@ -482,6 +482,11 @@ impl CompiledStateGraph {
     }
 
     /// Save a checkpoint from current channel state.
+    ///
+    /// `previous_versions` holds the channel versions as of the start of this
+    /// run; only channels whose version moved since then are passed to the
+    /// saver as `new_versions`, so persistent savers can write delta blobs
+    /// instead of re-encoding every channel on every super-step.
     fn save_checkpoint(
         &self,
         checkpointer: &Arc<dyn BaseCheckpointSaver>,
@@ -489,6 +494,7 @@ impl CompiledStateGraph {
         channels: &HashMap<String, Box<dyn Channel>>,
         channel_versions: &ChannelVersions,
         versions_seen: &HashMap<String, HashMap<String, JsonValue>>,
+        previous_versions: &ChannelVersions,
     ) -> Option<RunnableConfig> {
         use chrono::Utc;
         use langgraph_checkpoint::checkpoint::id::uuid6;
@@ -509,9 +515,17 @@ impl CompiledStateGraph {
             updated_channels: None,
         };
 
+        // Delta vs. the versions this run started from: only channels whose
+        // version changed since then need new blob rows.
+        let new_versions: ChannelVersions = channel_versions
+            .iter()
+            .filter(|(k, v)| previous_versions.get(k.as_str()) != Some(*v))
+            .map(|(k, v)| (k.clone(), v.clone()))
+            .collect();
+
         let metadata = CheckpointMetadata::default();
         checkpointer
-            .put(config, &checkpoint, &metadata, channel_versions)
+            .put(config, &checkpoint, &metadata, &new_versions)
             .ok()
     }
 
@@ -789,6 +803,7 @@ impl CompiledStateGraph {
             .as_ref()
             .map(|s| s.checkpoint.versions_seen.clone())
             .unwrap_or_default();
+        let previous_versions = channel_versions.clone();
 
         // Apply the update values to channels
         if let Some(obj) = values.as_object() {
@@ -817,6 +832,7 @@ impl CompiledStateGraph {
             &channels,
             &channel_versions,
             &versions_seen,
+            &previous_versions,
         );
 
         Ok(config.clone())
@@ -1405,6 +1421,10 @@ impl CompiledStateGraph {
                 )
             };
 
+        // Baseline for incremental checkpoint writes: the versions as of the
+        // start of this run. Only channels that move past these get blob rows.
+        let previous_versions = channel_versions.clone();
+
         // BSP loop counters
         let mut step: u64 = 0;
         let max_steps = config.get_recursion_limit().unwrap_or(self.recursion_limit);
@@ -1528,6 +1548,7 @@ impl CompiledStateGraph {
                             &channels,
                             &channel_versions,
                             &versions_seen,
+                            &previous_versions,
                         ) {
                             config = new_config;
                         }
@@ -1600,6 +1621,7 @@ impl CompiledStateGraph {
                             &channels,
                             &channel_versions,
                             &versions_seen,
+                            &previous_versions,
                         ) {
                             config = new_config;
                         }
@@ -1683,9 +1705,14 @@ impl CompiledStateGraph {
 
             // Save "loop" checkpoint after each completed super-step
             if let Some(ref cp) = self.checkpointer {
-                if let Some(new_config) =
-                    self.save_checkpoint(cp, &config, &channels, &channel_versions, &versions_seen)
-                {
+                if let Some(new_config) = self.save_checkpoint(
+                    cp,
+                    &config,
+                    &channels,
+                    &channel_versions,
+                    &versions_seen,
+                    &previous_versions,
+                ) {
                     config = new_config;
                 }
             }
diff --git a/tests/bench_pregel.rs b/tests/bench_pregel.rs
new file mode 100644
index 0000000..2a5ffc7
--- /dev/null
+++ b/tests/bench_pregel.rs
@@ -0,0 +1,260 @@
+//! Benchmark: end-to-end Pregel loop hot paths.
+//!
+//! Run explicitly (it is `#[ignore]`d so normal `cargo test` runs stay fast):
+//!
+//! ```text
+//! cargo test --release --test bench_pregel -- --ignored --nocapture
+//! ```
+
+use langgraph::channels::{BinaryOperatorAggregate, Channel, LastValue};
+use langgraph::checkpoint::{BaseCheckpointSaver, InMemorySaver};
+use langgraph::prelude::*;
+use langgraph_checkpoint_sqlite::SqliteSaver;
+use langgraph_prebuilt::add_messages_ref;
+use serde_json::json;
+use serde_json::Value as JsonValue;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+fn make_message(i: usize) -> JsonValue {
+    serde_json::json!({
+        "type": "ai",
+        "content": format!("Assistant reply number {i}: {}", "x".repeat(300)),
+        "tool_calls": [{
+            "name": "search_tool",
+            "args": {
+                "query": format!("query-{i}"),
+                "filters": {"category": "test", "limit": 10, "extra": "data"}
+            },
+            "id": format!("call_{i}")
+        }],
+        "id": format!("msg_{i}")
+    })
+}
+
+/// Single-node graph: `messages` accumulates with the `add_messages` reducer.
+/// Every invoke appends one message and saves a fresh checkpoint, so the
+/// checkpointed state grows by one message per super-step.
+fn build_linear_graph(checkpointer: Arc<dyn BaseCheckpointSaver>) -> CompiledStateGraph {
+    let mut channels: HashMap<String, Box<dyn Channel>> = HashMap::new();
+    channels.insert(
+        "messages".to_string(),
+        Box::new(BinaryOperatorAggregate::new("messages", add_messages_ref))
+            as Box<dyn Channel>,
+    );
+
+    let mut graph = StateGraph::new(channels);
+    graph
+        .add_node(
+            "append",
+            |input: JsonValue, _config: RunnableConfig| async move {
+                let n = input
+                    .get("messages")
+                    .and_then(|m| m.as_array())
+                    .map(|a| a.len())
+                    .unwrap_or(0);
+                Ok(json!({"messages": [make_message(n)]}))
+            },
+        )
+        .unwrap();
+    graph.add_edge(START, "append").unwrap();
+    graph.add_edge("append", END).unwrap();
+
+    graph
+        .compile_builder()
+        .checkpointer(checkpointer)
+        .build()
+        .unwrap()
+}
+
+/// Per-step cost of a checkpointed run as the message history grows.
+///
+/// If checkpointing re-serialized the full state every step this shows clear
+/// super-linear growth (total work ~ O(steps^2)).
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn bench_linear_checkpoint_growth() {
+    for steps in [100usize, 200, 400, 800] {
+        let app = build_linear_graph(Arc::new(InMemorySaver::new()));
+        let mut config = RunnableConfig::new();
+        config.insert(
+            "configurable".to_string(),
+            json!({"thread_id": "bench-linear"}),
+        );
+
+        let start = Instant::now();
+        for i in 0..steps {
+            let input = json!({"messages": [make_message(i)]});
+            app.ainvoke(&input, &config).await.unwrap();
+        }
+        let elapsed = start.elapsed();
+        println!(
+            "linear checkpointed (InMemory): {steps:>4} steps, history->{steps} => {elapsed:?}  ({:?}/step)",
+            elapsed / steps as u32
+        );
+    }
+}
+
+/// Same growth benchmark against the SQLite saver — the persistent path where
+/// incremental blob writes matter most.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn bench_linear_checkpoint_sqlite() {
+    for steps in [100usize, 200, 400, 800] {
+        let saver = SqliteSaver::from_conn_string("sqlite::memory:")
+            .await
+            .unwrap();
+        saver.setup().await.unwrap();
+        let app = build_linear_graph(Arc::new(saver));
+        let mut config = RunnableConfig::new();
+        config.insert(
+            "configurable".to_string(),
+            json!({"thread_id": "bench-sqlite"}),
+        );
+
+        let start = Instant::now();
+        for i in 0..steps {
+            let input = json!({"messages": [make_message(i)]});
+            app.ainvoke(&input, &config).await.unwrap();
+        }
+        let elapsed = start.elapsed();
+        println!(
+            "linear checkpointed (SQLite): {steps:>4} steps, history->{steps} => {elapsed:?}  ({:?}/step)",
+            elapsed / steps as u32
+        );
+    }
+}
+
+/// Wall-clock time to run `branches` parallel nodes in a single super-step.
+///
+/// Every branch sleeps for a fixed time. A serial runner takes
+/// `branches * sleep`; a parallel runner takes ~`sleep`.
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+#[ignore]
+async fn bench_parallel_fanout() {
+    const BRANCH_SLEEP_MS: u64 = 40;
+
+    for branches in [2usize, 4, 8] {
+        let channels: HashMap<String, Box<dyn Channel>> = HashMap::new();
+        let mut graph = StateGraph::new(channels);
+        for i in 0..branches {
+            graph
+                .add_node(
+                    format!("branch{i}"),
+                    |_input: JsonValue, _config: RunnableConfig| async move {
+                        tokio::time::sleep(Duration::from_millis(BRANCH_SLEEP_MS)).await;
+                        Ok(json!({}))
+                    },
+                )
+                .unwrap();
+            graph.add_edge(START, format!("branch{i}")).unwrap();
+        }
+        let app = graph.compile().unwrap();
+
+        let start = Instant::now();
+        app.ainvoke(&json!({}), &RunnableConfig::new())
+            .await
+            .unwrap();
+        let elapsed = start.elapsed();
+
+        println!(
+            "parallel fan-out: {branches} branches x {BRANCH_SLEEP_MS}ms => {elapsed:?}  (serial {BRANCH_SLEEP_MS}ms*n, parallel ~{BRANCH_SLEEP_MS}ms)"
+        );
+    }
+}
+
+/// The win case for incremental writes: a large static channel (e.g. embedded
+/// knowledge base) written once at thread start, then untouched while
+/// `messages` grows every step. Without delta writes the static channel's
+/// blob gets re-encoded and re-inserted every single step.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn bench_sqlite_static_context() {
+    const CONTEXT_SIZE: usize = 200 * 1024;
+
+    for steps in [200usize, 400, 800] {
+        let saver = SqliteSaver::from_conn_string("sqlite::memory:")
+            .await
+            .unwrap();
+        saver.setup().await.unwrap();
+
+        let mut channels: HashMap<String, Box<dyn Channel>> = HashMap::new();
+        channels.insert(
+            "messages".to_string(),
+            Box::new(BinaryOperatorAggregate::new("messages", add_messages_ref))
+                as Box<dyn Channel>,
+        );
+        channels.insert(
+            "context".to_string(),
+            Box::new(LastValue::new("context")) as Box<dyn Channel>,
+        );
+
+        let mut graph = StateGraph::new(channels);
+        graph
+            .add_node(
+                "append",
+                |input: JsonValue, _config: RunnableConfig| async move {
+                    let n = input
+                        .get("messages")
+                        .and_then(|m| m.as_array())
+                        .map(|a| a.len())
+                        .unwrap_or(0);
+                    Ok(json!({"messages": [make_message(n)]}))
+                },
+            )
+            .unwrap();
+        graph.add_edge(START, "append").unwrap();
+        graph.add_edge("append", END).unwrap();
+        let app = graph
+            .compile_builder()
+            .checkpointer(Arc::new(saver))
+            .build()
+            .unwrap();
+
+        let context = "x".repeat(CONTEXT_SIZE);
+        let mut config = RunnableConfig::new();
+        config.insert(
+            "configurable".to_string(),
+            json!({"thread_id": "bench-ctx"}),
+        );
+
+        // Seed the static channel once, then grow messages every step.
+        app.ainvoke(&json!({"messages": [make_message(0)], "context": context}), &config)
+            .await
+            .unwrap();
+
+        let start = Instant::now();
+        for i in 1..steps {
+            let input = json!({"messages": [make_message(i)]});
+            app.ainvoke(&input, &config).await.unwrap();
+        }
+        let elapsed = start.elapsed();
+        println!(
+            "sqlite static context: {steps:>4} steps, {CONTEXT_SIZE}-byte static channel => {elapsed:?}  ({:?}/step)",
+            elapsed / (steps - 1) as u32
+        );
+    }
+}
+
+/// Guards against the benchmark graphs silently becoming no-ops.
+#[tokio::test]
+async fn sanity_bench_graphs_work() {
+    let app = build_linear_graph(Arc::new(InMemorySaver::new()));
+    let mut config = RunnableConfig::new();
+    config.insert(
+        "configurable".to_string(),
+        json!({"thread_id": "bench-sanity"}),
+    );
+    app.ainvoke(&json!({"messages": [make_message(0)]}), &config)
+        .await
+        .unwrap();
+    app.ainvoke(&json!({"messages": [make_message(1)]}), &config)
+        .await
+        .unwrap();
+    let snapshot = app.get_state(&config).unwrap();
+    assert_eq!(
+        snapshot.values.get("messages").and_then(|m| m.as_array()).map(|a| a.len()),
+        Some(4)
+    );
+}

From 1498db23c676f973c4618ff558bfecc6b19f9b0f Mon Sep 17 00:00:00 2001
From: unknown <3058704216@qq.com>
Date: Sun, 2 Aug 2026 15:03:31 +0800
Subject: [PATCH 3/4] style: run rustfmt on checkpoint sqlite test and pregel
 bench

---
 .../langgraph-checkpoint-sqlite/src/saver.rs   |  5 ++++-
 tests/bench_pregel.rs                          | 18 ++++++++++++------
 2 files changed, 16 insertions(+), 7 deletions(-)

diff --git a/crates/langgraph-checkpoint-sqlite/src/saver.rs b/crates/langgraph-checkpoint-sqlite/src/saver.rs
index a2efa28..b39b711 100644
--- a/crates/langgraph-checkpoint-sqlite/src/saver.rs
+++ b/crates/langgraph-checkpoint-sqlite/src/saver.rs
@@ -1018,7 +1018,10 @@ mod tests {
         .fetch_one(&saver.pool)
         .await
         .unwrap();
-        assert_eq!(blob_count, 3, "expected blobs a@1, b@1, a@2, got {blob_count}");
+        assert_eq!(
+            blob_count, 3,
+            "expected blobs a@1, b@1, a@2, got {blob_count}"
+        );
 
         // cp2 reads back both channels: a@2 via cp2's blob, b@1 via the
         // version-addressed blob written by cp1 (b's version never moved).
diff --git a/tests/bench_pregel.rs b/tests/bench_pregel.rs
index 2a5ffc7..f1fc994 100644
--- a/tests/bench_pregel.rs
+++ b/tests/bench_pregel.rs
@@ -40,8 +40,7 @@ fn build_linear_graph(checkpointer: Arc<dyn BaseCheckpointSaver>) -> CompiledSta
     let mut channels: HashMap<String, Box<dyn Channel>> = HashMap::new();
     channels.insert(
         "messages".to_string(),
-        Box::new(BinaryOperatorAggregate::new("messages", add_messages_ref))
-            as Box<dyn Channel>,
+        Box::new(BinaryOperatorAggregate::new("messages", add_messages_ref)) as Box<dyn Channel>,
     );
 
     let mut graph = StateGraph::new(channels);
@@ -220,9 +219,12 @@ async fn bench_sqlite_static_context() {
         );
 
         // Seed the static channel once, then grow messages every step.
-        app.ainvoke(&json!({"messages": [make_message(0)], "context": context}), &config)
-            .await
-            .unwrap();
+        app.ainvoke(
+            &json!({"messages": [make_message(0)], "context": context}),
+            &config,
+        )
+        .await
+        .unwrap();
 
         let start = Instant::now();
         for i in 1..steps {
@@ -254,7 +256,11 @@ async fn sanity_bench_graphs_work() {
         .unwrap();
     let snapshot = app.get_state(&config).unwrap();
     assert_eq!(
-        snapshot.values.get("messages").and_then(|m| m.as_array()).map(|a| a.len()),
+        snapshot
+            .values
+            .get("messages")
+            .and_then(|m| m.as_array())
+            .map(|a| a.len()),
         Some(4)
     );
 }

From 2c5d99c904abbb41146bed7a7f291429523d3274 Mon Sep 17 00:00:00 2001
From: unknown <3058704216@qq.com>
Date: Sun, 2 Aug 2026 18:31:00 +0800
Subject: [PATCH 4/4] fix(bench): bound checkpoint retention with
 LatestOnlySaver and fix the multi-step loop measurement
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- LatestOnlySaver retains only the newest checkpoint per thread. InMemorySaver
  keeps every checkpoint, so the growth benches retained O(steps²) serialized
  state and OOM'd after a few hundred steps; retention is now O(latest state).
- The multi-step loop's conditional routing only sees the node output, so
  routing on messages.len() always saw a single-element array and the loop ran
  to the recursion limit (25 default) — silently truncating every run. Route on
  an explicit count channel so the loop runs the real target super-steps.
- Same bench content as PR #31's correction commit (6006a43) so either PR
  merging first lands the identical corrected bench.
---
 Cargo.lock            |  45 ++---
 Cargo.toml            |   1 +
 tests/bench_pregel.rs | 422 +++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 436 insertions(+), 32 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index d3dc339..87d5ecb 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -615,18 +615,6 @@ dependencies = [
  "wasi",
 ]
 
-[[package]]
-name = "getrandom"
-version = "0.3.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
-dependencies = [
- "cfg-if",
- "libc",
- "r-efi 5.3.0",
- "wasip2",
-]
-
 [[package]]
 name = "getrandom"
 version = "0.4.2"
@@ -635,7 +623,7 @@ checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555"
 dependencies = [
  "cfg-if",
  "libc",
- "r-efi 6.0.0",
+ "r-efi",
  "wasip2",
  "wasip3",
 ]
@@ -1053,6 +1041,7 @@ dependencies = [
  "langgraph-prebuilt",
  "langgraph-providers",
  "langgraph-tracing",
+ "mimalloc",
  "serde",
  "serde_json",
  "tokio",
@@ -1213,6 +1202,15 @@ version = "0.2.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
 
+[[package]]
+name = "libmimalloc-sys"
+version = "0.1.49"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6a45a52f43e1c16f667ccfe4dd8c85b7f7c204fd5e3bf46c5b0db9a5c3c0b8e9"
+dependencies = [
+ "cc",
+]
+
 [[package]]
 name = "libredox"
 version = "0.1.16"
@@ -1285,6 +1283,15 @@ version = "2.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79"
 
+[[package]]
+name = "mimalloc"
+version = "0.1.52"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d4139bb28d14ad1facf21d5eb8825051b326e172d216b39f6d31df53cc97862"
+dependencies = [
+ "libmimalloc-sys",
+]
+
 [[package]]
 name = "mime"
 version = "0.3.17"
@@ -1570,12 +1577,6 @@ dependencies = [
  "proc-macro2",
 ]
 
-[[package]]
-name = "r-efi"
-version = "5.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
-
 [[package]]
 name = "r-efi"
 version = "6.0.0"
@@ -1752,7 +1753,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -2272,10 +2273,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
  "fastrand",
- "getrandom 0.3.4",
+ "getrandom 0.4.2",
  "once_cell",
  "rustix",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
diff --git a/Cargo.toml b/Cargo.toml
index f874538..07ba66a 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -70,6 +70,7 @@ dotenvy = "0.15.7"
 uuid = { workspace = true }
 
 [dev-dependencies]
+mimalloc = "0.1"
 langgraph = { path = ".", features = ["full"] }
 langgraph-prebuilt = { workspace = true }
 langgraph-providers = { workspace = true }
diff --git a/tests/bench_pregel.rs b/tests/bench_pregel.rs
index f1fc994..635c78a 100644
--- a/tests/bench_pregel.rs
+++ b/tests/bench_pregel.rs
@@ -5,16 +5,41 @@
 //! ```text
 //! cargo test --release --test bench_pregel -- --ignored --nocapture
 //! ```
+//!
+//! Note: this file installs mimalloc as the global allocator. On Windows the
+//! default system heap is pathological for checkpointed workloads — large
+//! (500KB+) per-super-step state blocks are allocated and freed each step, and
+//! the heap's large-block path can slow the same binary by ~4x depending on
+//! machine state, making results non-reproducible. mimalloc's size-classed,
+//! thread-cached allocation keeps the numbers stable and representative of the
+//! library's actual (allocation-bound) costs.
+//!
+//! Memory note: the growth benches use `LatestOnlySaver`, which retains only
+//! the newest checkpoint per thread. `InMemorySaver` keeps every checkpoint, so
+//! a growing-history run retains O(steps²) serialized state and OOMs after a
+//! few hundred steps; latest-only keeps retention O(latest state) while
+//! preserving the same per-step cost — each step still loads the newest
+//! checkpoint and saves a fresh one. (The sqlite benches are disk-backed and
+//! stay bounded at the caps below.)
+
+use mimalloc::MiMalloc;
+#[global_allocator]
+static GLOBAL: MiMalloc = MiMalloc;
 
 use langgraph::channels::{BinaryOperatorAggregate, Channel, LastValue};
-use langgraph::checkpoint::{BaseCheckpointSaver, InMemorySaver};
+use langgraph::checkpoint::config::RunnableConfigExt;
+use langgraph::checkpoint::error::CheckpointError;
+use langgraph::checkpoint::{
+    BaseCheckpointSaver, ChannelVersions, Checkpoint, CheckpointMetadata, CheckpointTuple,
+    InMemorySaver,
+};
 use langgraph::prelude::*;
 use langgraph_checkpoint_sqlite::SqliteSaver;
 use langgraph_prebuilt::add_messages_ref;
 use serde_json::json;
 use serde_json::Value as JsonValue;
 use std::collections::HashMap;
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
 
 fn make_message(i: usize) -> JsonValue {
@@ -33,6 +58,284 @@ fn make_message(i: usize) -> JsonValue {
     })
 }
 
+/// A checkpoint saver that retains only the newest checkpoint per thread.
+///
+/// `InMemorySaver` keeps every checkpoint forever, so a benchmark that runs `N`
+/// super-steps with a growing message history retains O(N²) serialized state
+/// and OOMs after a few hundred steps. Production savers prune; this one does
+/// the same — each `put` replaces the thread's previous checkpoint, so retained
+/// memory is O(latest state). The serde round-trip (to_value on `put`,
+/// from_value on `get_tuple`) mirrors `InMemorySaver`, so the measured per-step
+/// cost stays comparable.
+/// (thread_id, checkpoint_ns) -> (checkpoint_id, checkpoint_json, metadata_json, parent_cid)
+type StorageEntry = (String, JsonValue, JsonValue, Option<String>);
+/// (thread_id, checkpoint_ns) -> the thread's newest checkpoint
+type StorageMap = HashMap<(String, String), StorageEntry>;
+/// (thread_id, checkpoint_ns, checkpoint_id) -> pending writes (interrupt-only path)
+type WritesMap = HashMap<(String, String, String), Vec<(String, String, JsonValue)>>;
+
+struct LatestOnlySaver {
+    storage: RwLock<StorageMap>,
+    writes: RwLock<WritesMap>,
+}
+
+impl LatestOnlySaver {
+    fn new() -> Self {
+        Self {
+            storage: RwLock::new(HashMap::new()),
+            writes: RwLock::new(HashMap::new()),
+        }
+    }
+
+    /// Number of retained checkpoints across all threads (1 per live thread).
+    fn retained_count(&self) -> usize {
+        self.storage.read().map(|s| s.len()).unwrap_or(0)
+    }
+
+    fn config_to_ids(config: &RunnableConfig) -> (String, String, Option<String>) {
+        let configurable = config.get("configurable");
+        let thread_id = configurable
+            .and_then(|c| c.get("thread_id"))
+            .and_then(|v| v.as_str())
+            .unwrap_or("")
+            .to_string();
+        let checkpoint_ns = configurable
+            .and_then(|c| c.get("checkpoint_ns"))
+            .and_then(|v| v.as_str())
+            .unwrap_or("")
+            .to_string();
+        let checkpoint_id = configurable
+            .and_then(|c| c.get("checkpoint_id"))
+            .and_then(|v| v.as_str())
+            .map(|s| s.to_string());
+        (thread_id, checkpoint_ns, checkpoint_id)
+    }
+}
+
+impl Default for LatestOnlySaver {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl BaseCheckpointSaver for LatestOnlySaver {
+    fn get_tuple(
+        &self,
+        config: &RunnableConfig,
+    ) -> Result<Option<CheckpointTuple>, CheckpointError> {
+        let (thread_id, checkpoint_ns, requested_id) = Self::config_to_ids(config);
+        let storage = self.storage.read().unwrap();
+        let Some((cid, checkpoint_json, metadata_json, parent_cid)) =
+            storage.get(&(thread_id.clone(), checkpoint_ns.clone()))
+        else {
+            return Ok(None);
+        };
+        // A specific older checkpoint was requested — it's been pruned.
+        if let Some(req) = &requested_id {
+            if req != cid {
+                return Ok(None);
+            }
+        }
+
+        let checkpoint: Checkpoint = serde_json::from_value(checkpoint_json.clone())
+            .map_err(|e| CheckpointError::Storage(e.to_string()))?;
+        let metadata: CheckpointMetadata = serde_json::from_value(metadata_json.clone())
+            .map_err(|e| CheckpointError::Storage(e.to_string()))?;
+
+        let parent_config = parent_cid.as_ref().map(|pid| {
+            let mut c = RunnableConfig::new();
+            c.insert(
+                "configurable".to_string(),
+                serde_json::json!({
+                    "thread_id": thread_id.clone(),
+                    "checkpoint_ns": checkpoint_ns.clone(),
+                    "checkpoint_id": pid,
+                }),
+            );
+            c
+        });
+
+        let pending_writes: Vec<(String, String, JsonValue)> = self
+            .writes
+            .read()
+            .unwrap()
+            .get(&(thread_id.clone(), checkpoint_ns.clone(), cid.clone()))
+            .cloned()
+            .unwrap_or_default();
+
+        Ok(Some(CheckpointTuple {
+            config: {
+                let mut c = RunnableConfig::new();
+                c.insert(
+                    "configurable".to_string(),
+                    serde_json::json!({
+                        "thread_id": thread_id,
+                        "checkpoint_ns": checkpoint_ns,
+                        "checkpoint_id": cid,
+                    }),
+                );
+                c
+            },
+            checkpoint,
+            metadata,
+            parent_config,
+            pending_writes: if pending_writes.is_empty() {
+                None
+            } else {
+                Some(pending_writes)
+            },
+        }))
+    }
+
+    fn list(
+        &self,
+        config: Option<&RunnableConfig>,
+        _filter: Option<&HashMap<String, JsonValue>>,
+        _before: Option<&RunnableConfig>,
+        limit: Option<usize>,
+    ) -> Result<Vec<CheckpointTuple>, CheckpointError> {
+        let (thread_id, checkpoint_ns) = match config {
+            Some(c) => {
+                let (tid, ns, _) = Self::config_to_ids(c);
+                (tid, ns)
+            }
+            None => (String::new(), String::new()),
+        };
+        let storage = self.storage.read().unwrap();
+        let mut entries: Vec<_> = storage
+            .iter()
+            .filter(|((tid, ns), _)| {
+                (thread_id.is_empty() || tid == &thread_id)
+                    && (checkpoint_ns.is_empty() || ns == &checkpoint_ns)
+            })
+            .collect();
+        // Newest checkpoint id first, mirroring InMemorySaver's ordering.
+        entries.sort_by(|a, b| b.1 .0.cmp(&a.1 .0));
+        if let Some(limit) = limit {
+            entries.truncate(limit);
+        }
+
+        let mut results = Vec::new();
+        for ((tid, ns), (cid, checkpoint_json, metadata_json, parent_cid)) in entries {
+            let checkpoint: Checkpoint = serde_json::from_value(checkpoint_json.clone())
+                .map_err(|e| CheckpointError::Storage(e.to_string()))?;
+            let metadata: CheckpointMetadata = serde_json::from_value(metadata_json.clone())
+                .map_err(|e| CheckpointError::Storage(e.to_string()))?;
+            let parent_config = parent_cid.as_ref().map(|pid| {
+                let mut c = RunnableConfig::new();
+                c.insert(
+                    "configurable".to_string(),
+                    serde_json::json!({
+                        "thread_id": tid,
+                        "checkpoint_ns": ns,
+                        "checkpoint_id": pid,
+                    }),
+                );
+                c
+            });
+            results.push(CheckpointTuple {
+                config: {
+                    let mut c = RunnableConfig::new();
+                    c.insert(
+                        "configurable".to_string(),
+                        serde_json::json!({
+                            "thread_id": tid,
+                            "checkpoint_ns": ns,
+                            "checkpoint_id": cid,
+                        }),
+                    );
+                    c
+                },
+                checkpoint,
+                metadata,
+                parent_config,
+                pending_writes: None,
+            });
+        }
+        Ok(results)
+    }
+
+    fn put(
+        &self,
+        config: &RunnableConfig,
+        checkpoint: &Checkpoint,
+        metadata: &CheckpointMetadata,
+        _new_versions: &ChannelVersions,
+    ) -> Result<RunnableConfig, CheckpointError> {
+        let (thread_id, checkpoint_ns, _) = Self::config_to_ids(config);
+
+        let checkpoint_json = serde_json::to_value(checkpoint)
+            .map_err(|e| CheckpointError::Storage(e.to_string()))?;
+        let metadata_json =
+            serde_json::to_value(metadata).map_err(|e| CheckpointError::Storage(e.to_string()))?;
+
+        // The new checkpoint's parent is the thread's current newest.
+        let parent_id = self
+            .storage
+            .read()
+            .unwrap()
+            .get(&(thread_id.clone(), checkpoint_ns.clone()))
+            .map(|(cid, _, _, _)| cid.clone());
+
+        // Replace the thread's checkpoint — only the newest is retained.
+        self.storage.write().unwrap().insert(
+            (thread_id.clone(), checkpoint_ns.clone()),
+            (
+                checkpoint.id.clone(),
+                checkpoint_json,
+                metadata_json,
+                parent_id,
+            ),
+        );
+        // Prune pending writes down to the newest checkpoint id.
+        self.writes.write().unwrap().retain(|(tid, ns, cid), _| {
+            tid != &thread_id || ns != &checkpoint_ns || cid == &checkpoint.id
+        });
+
+        let mut new_config = RunnableConfig::new();
+        new_config.insert(
+            "configurable".to_string(),
+            serde_json::json!({
+                "thread_id": thread_id,
+                "checkpoint_ns": checkpoint_ns,
+                "checkpoint_id": checkpoint.id,
+            }),
+        );
+        Ok(new_config)
+    }
+
+    fn put_writes(
+        &self,
+        config: &RunnableConfig,
+        writes: &[(String, String, JsonValue)],
+        task_id: &str,
+        _task_path: &str,
+    ) -> Result<(), CheckpointError> {
+        let (thread_id, checkpoint_ns, checkpoint_id) = Self::config_to_ids(config);
+        let checkpoint_id = checkpoint_id.unwrap_or_default();
+        let mut writes_guard = self.writes.write().unwrap();
+        let entry = writes_guard
+            .entry((thread_id, checkpoint_ns, checkpoint_id))
+            .or_default();
+        for write in writes {
+            entry.push((task_id.to_string(), write.1.clone(), write.2.clone()));
+        }
+        Ok(())
+    }
+
+    fn delete_thread(&self, thread_id: &str) -> Result<(), CheckpointError> {
+        self.storage
+            .write()
+            .unwrap()
+            .retain(|(tid, _), _| tid != thread_id);
+        self.writes
+            .write()
+            .unwrap()
+            .retain(|(tid, _, _), _| tid != thread_id);
+        Ok(())
+    }
+}
+
 /// Single-node graph: `messages` accumulates with the `add_messages` reducer.
 /// Every invoke appends one message and saves a fresh checkpoint, so the
 /// checkpointed state grows by one message per super-step.
@@ -70,12 +373,14 @@ fn build_linear_graph(checkpointer: Arc<dyn BaseCheckpointSaver>) -> CompiledSta
 /// Per-step cost of a checkpointed run as the message history grows.
 ///
 /// If checkpointing re-serialized the full state every step this shows clear
-/// super-linear growth (total work ~ O(steps^2)).
+/// super-linear growth (total work ~ O(steps^2)). Uses `LatestOnlySaver` so the
+/// retained checkpoint set stays O(1) per thread instead of O(steps^2).
 #[tokio::test(flavor = "multi_thread")]
 #[ignore]
 async fn bench_linear_checkpoint_growth() {
-    for steps in [100usize, 200, 400, 800] {
-        let app = build_linear_graph(Arc::new(InMemorySaver::new()));
+    for steps in [100usize, 200, 400] {
+        let saver = Arc::new(LatestOnlySaver::new());
+        let app = build_linear_graph(saver.clone());
         let mut config = RunnableConfig::new();
         config.insert(
             "configurable".to_string(),
@@ -89,8 +394,9 @@ async fn bench_linear_checkpoint_growth() {
         }
         let elapsed = start.elapsed();
         println!(
-            "linear checkpointed (InMemory): {steps:>4} steps, history->{steps} => {elapsed:?}  ({:?}/step)",
-            elapsed / steps as u32
+            "linear checkpointed (latest-only): {steps:>4} steps, history->{steps} => {elapsed:?}  ({:?}/step) [retained checkpoints: {}]",
+            elapsed / steps as u32,
+            saver.retained_count()
         );
     }
 }
@@ -100,7 +406,7 @@ async fn bench_linear_checkpoint_growth() {
 #[tokio::test(flavor = "multi_thread")]
 #[ignore]
 async fn bench_linear_checkpoint_sqlite() {
-    for steps in [100usize, 200, 400, 800] {
+    for steps in [100usize, 200, 400] {
         let saver = SqliteSaver::from_conn_string("sqlite::memory:")
             .await
             .unwrap();
@@ -125,6 +431,102 @@ async fn bench_linear_checkpoint_sqlite() {
     }
 }
 
+/// Multi-super-step loop: one invoke runs exactly `target` super-steps via a
+/// self-looping conditional edge, growing the history each step. This is the
+/// shape that exercises the per-super-step output read (read_channels). The
+/// `count` channel terminates the loop (see `build_loop_graph`); the recursion
+/// limit is just a safety valve so a broken routing can't run forever.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn bench_multi_step_loop() {
+    for target in [50usize, 100, 200] {
+        let saver = Arc::new(LatestOnlySaver::new());
+        let app = build_loop_graph(saver.clone(), target);
+        let mut config = RunnableConfig::new();
+        config.insert(
+            "configurable".to_string(),
+            json!({"thread_id": "bench-loop"}),
+        );
+        // Safety valve: `target` can exceed the default recursion limit of 25.
+        let config = config.with_recursion_limit(100_000);
+
+        let input = json!({"count": 0, "messages": []});
+        let t = Instant::now();
+        app.ainvoke(&input, &config).await.unwrap();
+        let elapsed = t.elapsed();
+        // Sanity: the loop must actually run to `target` messages. get_state
+        // reads the final checkpoint; a truncated run fails here, not silently.
+        let snapshot = app.get_state(&config).unwrap();
+        let got = snapshot
+            .values
+            .get("messages")
+            .and_then(|m| m.as_array())
+            .map(|a| a.len());
+        assert_eq!(
+            got,
+            Some(target),
+            "loop was silently truncated: expected {target} messages, got {got:?}"
+        );
+        println!(
+            "multi-step loop: {target:>3} super-steps in one invoke => {elapsed:?}  ({:?}/super-step) [retained checkpoints: {}]",
+            elapsed / target as u32,
+            saver.retained_count()
+        );
+    }
+}
+
+/// Single-node graph with a self-loop: node "append" routes back to itself
+/// until the history reaches `target` messages, then END.
+///
+/// The loop needs a plain `count` channel to terminate: conditional-edge
+/// routing only sees the *node output* (the combined PregelNode evaluates
+/// `branch.path` on the node's delta, not the accumulated state), so routing
+/// on `messages.len()` always sees a single-element array and loops to the
+/// recursion limit. The node emits the bumped counter alongside the new
+/// message, and routing terminates on the counter.
+fn build_loop_graph(
+    checkpointer: Arc<dyn BaseCheckpointSaver>,
+    target: usize,
+) -> CompiledStateGraph {
+    let mut channels: HashMap<String, Box<dyn Channel>> = HashMap::new();
+    channels.insert(
+        "messages".to_string(),
+        Box::new(BinaryOperatorAggregate::new("messages", add_messages_ref)) as Box<dyn Channel>,
+    );
+    channels.insert("count".to_string(), Box::new(LastValue::new("count")));
+
+    let mut graph = StateGraph::new(channels);
+    graph
+        .add_node(
+            "append",
+            |input: JsonValue, _config: RunnableConfig| async move {
+                let n = input.get("count").and_then(|c| c.as_i64()).unwrap_or(0);
+                Ok(json!({"count": n + 1, "messages": [make_message(n as usize)]}))
+            },
+        )
+        .unwrap();
+    graph.add_edge(START, "append").unwrap();
+    graph
+        .add_conditional_edges(
+            "append",
+            move |input: JsonValue, _config: RunnableConfig| async move {
+                let n = input.get("count").and_then(|c| c.as_i64()).unwrap_or(0);
+                Ok(json!(if n >= target as i64 { "END" } else { "again" }))
+            },
+            Some(HashMap::from([
+                ("again".to_string(), "append".to_string()),
+                ("END".to_string(), END.to_string()),
+            ])),
+        )
+        .unwrap();
+
+    graph
+        .compile_builder()
+        .checkpointer(checkpointer)
+        .build()
+        .unwrap()
+}
+
 /// Wall-clock time to run `branches` parallel nodes in a single super-step.
 ///
 /// Every branch sleeps for a fixed time. A serial runner takes
@@ -170,9 +572,9 @@ async fn bench_parallel_fanout() {
 #[tokio::test(flavor = "multi_thread")]
 #[ignore]
 async fn bench_sqlite_static_context() {
-    const CONTEXT_SIZE: usize = 200 * 1024;
+    const CONTEXT_SIZE: usize = 128 * 1024;
 
-    for steps in [200usize, 400, 800] {
+    for steps in [100usize, 200, 400] {
         let saver = SqliteSaver::from_conn_string("sqlite::memory:")
             .await
             .unwrap();