checkpoint_version

libs/langgraph/channel_versions_explained.md

@@ -0,0 +1,279 @@
+# Understanding `channel_versions` for State Channels in LangGraph
+
+## Overview
+
+While `versions_seen` only tracks trigger channels, `channel_versions` tracks **ALL channels** including state channels like `fieldA` and `fieldB`. This document explains why state channel versions matter.
+
+## The Two Version Tracking Mechanisms
+
+| Mechanism | What it tracks | Purpose |
+|-----------|---------------|---------|
+| `channel_versions` | **All channels** (state + triggers) | Storage, recovery, change tracking |
+| `versions_seen` | **Only triggers** | Scheduling, prevent duplicate execution |
+
+---
+
+## Why Track State Channel Versions?
+
+### Purpose 1: Incremental Storage
+
+When saving a checkpoint, LangGraph only serializes channels that have **changed** since the last checkpoint.
+
+```python
+# In checkpointer.put()
+def put(self, config, checkpoint, metadata, new_versions):
+    for k, v in new_versions.items():  # Only changed channels!
+        self.blobs[(thread_id, ns, k, v)] = serialize(values[k])
+```
+
+The `new_versions` parameter is computed by comparing current versions with previous versions:
+
+```python
+def get_new_channel_versions(previous_versions, current_versions):
+    """Get subset of current_versions that are newer than previous_versions."""
+    return {
+        k: v
+        for k, v in current_versions.items()
+        if v > previous_versions.get(k, null_version)
+    }
+```
+
+**Benefit**: If `fieldA` didn't change in a step, it won't be re-serialized!
+
+### Purpose 2: Version-Keyed Storage
+
+Channel values are stored with their version as part of the key:
+
+```python
+# Storage structure in InMemorySaver
+blobs = {
+    (thread_id, ns, "fieldA", v02): b"Hello",         # Step 0
+    (thread_id, ns, "fieldA", v03): b"Hello->A",      # Step 1  
+    (thread_id, ns, "fieldA", v04): b"Hello->A->B",   # Step 2
+    (thread_id, ns, "fieldB", v02): b"World",         # Step 0
+    (thread_id, ns, "fieldB", v03): b"World->A",      # Step 1
+    ...
+}
+```
+
+**Benefit**: Can restore to ANY historical checkpoint - each version's value is stored independently.
+
+### Purpose 3: Precise Recovery
+
+When restoring a checkpoint, use `channel_versions` to load the correct value:
+
+```python
+# Restoring Step 1's checkpoint
+checkpoint["channel_versions"] = {"fieldA": v03, "fieldB": v03}
+
+# Load correct version of each value
+fieldA = blobs[(thread_id, ns, "fieldA", v03)]  # "Hello->A"
+fieldB = blobs[(thread_id, ns, "fieldB", v03)]  # "World->A"
+```
+
+---
+
+## Example: Step-by-Step Channel Version Changes
+
+Using the same graph:
+
+```
+        ┌──────────┐
+        │  nodeA   │  (reads fieldA + fieldB)
+        └────┬─────┘
+             │
+      ┌──────┴──────┐
+      ▼             ▼
+┌──────────┐  ┌──────────┐
+│  nodeB   │  │  nodeC   │  (nodeB reads fieldA, nodeC reads fieldB)
+└────┬─────┘  └────┬─────┘
+      │             │
+      └──────┬──────┘
+             ▼
+        ┌──────────┐
+        │  nodeD   │  (reads fieldA + fieldB)
+        └──────────┘
+```
+
+### Step -1: Input
+
+```
+channel_versions:
+    __start__: v01
+
+channel_values:
+    __start__: {'fieldA': 'Hello', 'fieldB': 'World'}
+
+new_versions (to save): {__start__: v01}
+  → Only __start__ is saved
+```
+
+### Step 0: `__start__` executes
+
+```
+channel_versions:
+    __start__:        v02
+    branch:to:nodeA:  v02
+    fieldA:           v02  ← NEW!
+    fieldB:           v02  ← NEW!
+
+channel_values:
+    branch:to:nodeA: None
+    fieldA: Hello
+    fieldB: World
+
+new_versions (to save): {__start__: v02, branch:to:nodeA: v02, fieldA: v02, fieldB: v02}
+  → All changed channels are saved
+```
+
+### Step 1: nodeA executes
+
+```
+channel_versions:
+    __start__:        v02  ← unchanged
+    branch:to:nodeA:  v03  ← updated (consumed)
+    branch:to:nodeB:  v03  ← NEW!
+    branch:to:nodeC:  v03  ← NEW!
+    fieldA:           v03  ← updated!
+    fieldB:           v03  ← updated!
+
+channel_values:
+    branch:to:nodeB: None
+    branch:to:nodeC: None
+    fieldA: Hello->A
+    fieldB: World->A
+
+new_versions (to save): {branch:to:nodeA: v03, branch:to:nodeB: v03, branch:to:nodeC: v03, fieldA: v03, fieldB: v03}
+  → __start__ NOT saved (unchanged at v02)
+```
+
+### Step 2: nodeB and nodeC execute (parallel)
+
+```
+channel_versions:
+    __start__:              v02  ← unchanged
+    branch:to:nodeA:        v03  ← unchanged
+    branch:to:nodeB:        v04  ← updated (consumed)
+    branch:to:nodeC:        v04  ← updated (consumed)
+    fieldA:                 v04  ← updated by nodeB!
+    fieldB:                 v04  ← updated by nodeC!
+    join:nodeB+nodeC:nodeD: v04  ← NEW!
+
+channel_values:
+    fieldA: Hello->A->B
+    fieldB: World->A->C
+    join:nodeB+nodeC:nodeD: {'nodeB', 'nodeC'}
+
+new_versions (to save): {branch:to:nodeB: v04, branch:to:nodeC: v04, fieldA: v04, fieldB: v04, join:...: v04}
+  → Only changed channels saved
+```
+
+### Step 3: nodeD executes
+
+```
+channel_versions:
+    __start__:              v02  ← unchanged since Step 0!
+    branch:to:nodeA:        v03  ← unchanged since Step 1
+    branch:to:nodeB:        v04  ← unchanged
+    branch:to:nodeC:        v04  ← unchanged
+    fieldA:                 v05  ← updated by nodeD!
+    fieldB:                 v05  ← updated by nodeD!
+    join:nodeB+nodeC:nodeD: v05  ← updated (consumed)
+
+channel_values:
+    fieldA: Hello->A->B->D
+    fieldB: World->A->C->D
+    join:nodeB+nodeC:nodeD: set()
+
+new_versions (to save): {fieldA: v05, fieldB: v05, join:...: v05}
+  → Only 3 channels saved, not all 7!
+```
+
+---
+
+## Storage Efficiency Visualization
+
+```
+Step 0:  Save [__start__, branch:to:nodeA, fieldA, fieldB]     = 4 channels
+Step 1:  Save [branch:to:nodeA, branch:to:nodeB, branch:to:nodeC, fieldA, fieldB] = 5 channels
+Step 2:  Save [branch:to:nodeB, branch:to:nodeC, fieldA, fieldB, join:...] = 5 channels
+Step 3:  Save [fieldA, fieldB, join:...]                       = 3 channels
+
+Without incremental storage: 7 channels × 4 steps = 28 serializations
+With incremental storage:    4 + 5 + 5 + 3        = 17 serializations
+                                                    = 39% savings!
+```
+
+---
+
+## Time Travel: Restoring Any Checkpoint
+
+Because each version is stored separately, you can restore to any point:
+
+```
+Want to restore Step 1?
+  → checkpoint["channel_versions"] = {fieldA: v03, fieldB: v03, ...}
+  → Load blobs[(thread_id, ns, "fieldA", v03)] = "Hello->A"
+  → Load blobs[(thread_id, ns, "fieldB", v03)] = "World->A"
+
+Want to restore Step 2?
+  → checkpoint["channel_versions"] = {fieldA: v04, fieldB: v04, ...}
+  → Load blobs[(thread_id, ns, "fieldA", v04)] = "Hello->A->B"
+  → Load blobs[(thread_id, ns, "fieldB", v04)] = "World->A->C"
+```
+
+---
+
+## Summary: State Channel Versions vs Trigger Channel Versions
+
+| Aspect | State Channels (fieldA, fieldB) | Trigger Channels (branch:to:*) |
+|--------|--------------------------------|-------------------------------|
+| In `channel_versions`? | ✅ Yes | ✅ Yes |
+| In `versions_seen`? | ❌ No | ✅ Yes |
+| Version increases when? | Value is updated | Written to OR consumed |
+| Used for scheduling? | ❌ No | ✅ Yes |
+| Used for storage? | ✅ Yes (incremental save) | ✅ Yes |
+| Used for recovery? | ✅ Yes (load correct version) | ✅ Yes |
+
+---
+
+## Code Reference
+
+### Where `channel_versions` is updated
+
+```python
+# In apply_writes() - libs/langgraph/langgraph/pregel/_algo.py
+for chan, vals in pending_writes_by_channel.items():
+    if channels[chan].update(vals) and next_version is not None:
+        checkpoint["channel_versions"][chan] = next_version  # ← Update version
+        updated_channels.add(chan)
+```
+
+### Where incremental storage happens
+
+```python
+# In InMemorySaver.put() - libs/checkpoint/langgraph/checkpoint/memory/__init__.py
+def put(self, config, checkpoint, metadata, new_versions):
+    values = checkpoint.pop("channel_values")
+    for k, v in new_versions.items():  # ← Only save changed channels
+        self.blobs[(thread_id, checkpoint_ns, k, v)] = (
+            self.serde.dumps_typed(values[k]) if k in values else ("empty", b"")
+        )
+```
+
+---
+
+## Running the Test
+
+To see channel versions in action:
+
+```bash
+cd libs/langgraph
+uv run python test_versions_seen.py
+```
+
+The output shows `channel_versions` for each step, where you can observe:
+1. All channels (state + triggers) are tracked
+2. Versions increment when values change
+3. Some channels stay at the same version across multiple steps (unchanged)
+

libs/langgraph/test_versions_seen.py

@@ -0,0 +1,169 @@
+"""
+Test script to demonstrate versions_seen in StateGraph
+
+Graph structure:
+    nodeA -> nodeB + nodeC -> nodeD
+
+State:
+    fieldA: str
+    fieldB: str
+"""
+
+from typing import TypedDict
+from langgraph.graph import StateGraph, START, END
+from langgraph.checkpoint.memory import InMemorySaver
+from pprint import pprint
+import json
+
+
+class State(TypedDict):
+    fieldA: str
+    fieldB: str
+
+
+class StateOnlyA(TypedDict):
+    """Input schema for nodeB - only reads fieldA"""
+    fieldA: str
+
+
+class StateOnlyB(TypedDict):
+    """Input schema for nodeC - only reads fieldB"""
+    fieldB: str
+
+
+def nodeA(state: State) -> dict:
+    """Reads fieldA + fieldB"""
+    print(f"  [nodeA] Reading: fieldA='{state['fieldA']}', fieldB='{state['fieldB']}'")
+    return {"fieldA": state["fieldA"] + "->A", "fieldB": state["fieldB"] + "->A"}
+
+
+def nodeB(state: StateOnlyA) -> dict:
+    """Reads only fieldA"""
+    print(f"  [nodeB] Reading: fieldA='{state['fieldA']}'")
+    return {"fieldA": state["fieldA"] + "->B"}
+
+
+def nodeC(state: StateOnlyB) -> dict:
+    """Reads only fieldB"""
+    print(f"  [nodeC] Reading: fieldB='{state['fieldB']}'")
+    return {"fieldB": state["fieldB"] + "->C"}
+
+
+def nodeD(state: State) -> dict:
+    """Reads fieldA + fieldB"""
+    print(f"  [nodeD] Reading: fieldA='{state['fieldA']}', fieldB='{state['fieldB']}'")
+    return {"fieldA": state["fieldA"] + "->D", "fieldB": state["fieldB"] + "->D"}
+
+
+# Build the graph
+graph = StateGraph(State)
+
+graph.add_node("nodeA", nodeA)  # reads fieldA + fieldB (default: full state)
+graph.add_node("nodeB", nodeB, input_schema=StateOnlyA)  # reads only fieldA
+graph.add_node("nodeC", nodeC, input_schema=StateOnlyB)  # reads only fieldB
+graph.add_node("nodeD", nodeD)  # reads fieldA + fieldB (default: full state)
+
+graph.add_edge(START, "nodeA")
+graph.add_edge("nodeA", "nodeB")
+graph.add_edge("nodeA", "nodeC")
+graph.add_edge(["nodeB", "nodeC"], "nodeD")
+graph.add_edge("nodeD", END)
+
+# Compile with checkpointer
+checkpointer = InMemorySaver()
+app = graph.compile(checkpointer=checkpointer)
+
+# Print compiled graph info
+print("=" * 60)
+print("COMPILED GRAPH INFO")
+print("=" * 60)
+print("\nChannels created:")
+for name, channel in app.channels.items():
+    print(f"  - {name}: {type(channel).__name__}")
+
+print("\nNodes with their triggers and channels:")
+for name, node in app.nodes.items():
+    print(f"  - {name}:")
+    print(f"      triggers: {node.triggers}")
+    print(f"      channels: {node.channels}")
+
+# Run the graph
+print("\n" + "=" * 60)
+print("EXECUTION")
+print("=" * 60)
+
+config = {"configurable": {"thread_id": "test-1"}}
+input_state = {"fieldA": "Hello", "fieldB": "World"}
+
+print(f"\nInput: {input_state}\n")
+
+# Run the graph to completion
+result = app.invoke(input_state, config)
+print(f"Final result: {result}\n")
+
+# Now use get_state_history to get all checkpoints in order
+print("=" * 60)
+print("CHECKPOINT HISTORY (using get_state_history)")
+print("=" * 60)
+
+# get_state_history returns checkpoints in reverse order (newest first)
+history = list(app.get_state_history(config))
+history.reverse()  # Reverse to get oldest first
+
+for idx, state_snapshot in enumerate(history):
+    metadata = state_snapshot.metadata
+    
+    print(f"\n{'='*60}")
+    print(f"Step {metadata.get('step', '?')} - Source: {metadata.get('source', '?')}")
+    print(f"{'='*60}")
+    
+    # Show which node(s) wrote this checkpoint
+    if "writes" in metadata and metadata["writes"]:
+        print(f"Writes by: {list(metadata['writes'].keys())}")
+    
+    print(f"\nState values: {state_snapshot.values}")
+    
+    # Access the actual checkpoint data
+    checkpoint_tuple = checkpointer.get_tuple(state_snapshot.config)
+    if checkpoint_tuple:
+        cp = checkpoint_tuple.checkpoint
+        
+        # Helper to simplify version string
+        def simplify_version(ver):
+            return str(ver).split(".")[0][-2:] if "." in str(ver) else str(ver)
+        
+        # Pretty print checkpoint with simplified versions
+        print("\nCheckpoint (raw):")
+        print(f"  v: {cp['v']}")
+        print(f"  id: {cp['id'][:20]}...")
+        print(f"  ts: {cp['ts']}")
+        print(f"  updated_channels: {cp.get('updated_channels')}")
+        
+        print(f"\n  channel_values:")
+        for ch, val in sorted(cp["channel_values"].items()):
+            val_str = str(val)[:50] + "..." if len(str(val)) > 50 else str(val)
+            print(f"    {ch}: {val_str}")
+        
+        print(f"\n  channel_versions:")
+        for ch, ver in sorted(cp["channel_versions"].items()):
+            print(f"    {ch}: v{simplify_version(ver)}")
+        
+        print(f"\n  versions_seen:")
+        for node_name, seen in sorted(cp["versions_seen"].items()):
+            if seen:
+                print(f"    {node_name}:")
+                for ch, ver in sorted(seen.items()):
+                    print(f"        {ch}: v{simplify_version(ver)}")
+            else:
+                print(f"    {node_name}: {{}}")
+
+print("\n" + "=" * 60)
+print("SUMMARY")
+print("=" * 60)
+print("""
+Key observations:
+1. versions_seen only records TRIGGER channels (branch:to:*, join:*)
+2. State channels (fieldA, fieldB) are NEVER in versions_seen
+3. Each node only records the trigger channel that activated it
+""")
+

libs/langgraph/tests/test_pregel.py

@@ -79,9 +79,151 @@ from tests.messages import (
 
 pytestmark = pytest.mark.anyio
 
+
+def pregel_pretty(data):
+    """Pretty print pregel nodes, channels, or graph with nice formatting."""
+    if not data:
+        return "Empty"
+    
+    # Check if this is a graph object
+    if hasattr(data, 'nodes') and hasattr(data, 'channels'):
+        # This is a graph object, show comprehensive info
+        result = []
+        
+        # Show nodes
+        result.append("="*50)
+        result.append("🔗 GRAPH NODES")
+        result.append("="*50)
+        for name, node in data.nodes.items():
+            node_type = type(node).__name__
+            result.append(f"  📍 {name:<12} → {node_type}")
+        
+        # Show channels
+        result.append("\n" + "="*50)
+        result.append("📡 GRAPH CHANNELS")
+        result.append("="*50)
+        for name, channel in data.channels.items():
+            channel_type = type(channel).__name__
+            if name in ['hello', 'messages']:
+                result.append(f"  🎯 {name:<20} → {channel_type} (user defined)")
+            elif name.startswith('branch:'):
+                result.append(f"  🌿 {name:<20} → {channel_type} (branch)")
+            else:
+                result.append(f"  ⚙️  {name:<20} → {channel_type} (system)")
+        
+        # Show graph structure
+        result.append("\n" + "="*50)
+        result.append("🏗️  GRAPH STRUCTURE")
+        result.append("="*50)
+        try:
+            graph_info = data.get_graph()
+            result.append(f"  Nodes: {len(graph_info.nodes)}")
+            result.append(f"  Edges: {len(graph_info.edges)}")
+            result.append("\n  📊 Execution Flow:")
+            for edge in graph_info.edges:
+                arrow = "  ├─" if edge != graph_info.edges[-1] else "  └─"
+                result.append(f"{arrow} {edge.source} → {edge.target}")
+        except Exception as e:
+            result.append(f"  Could not get graph structure: {e}")
+        
+        result.append("="*50)
+        return "\n".join(result)
+    
+    # Check if this is nodes or channels dict
+    first_key, first_value = next(iter(data.items()))
+    
+    # Detect if this is nodes or channels
+    is_nodes = hasattr(first_value, '__class__') and 'Node' in first_value.__class__.__name__
+    is_channels = hasattr(first_value, '__class__') and ('Channel' in first_value.__class__.__name__ or 
+                                                         'Value' in first_value.__class__.__name__ or
+                                                         'Topic' in first_value.__class__.__name__ or
+                                                         'Aggregate' in first_value.__class__.__name__)
+    
+    result = []
+    
+    if is_nodes:
+        result.append("="*50)
+        result.append("🔗 GRAPH NODES")
+        result.append("="*50)
+        for name, node in data.items():
+            node_type = type(node).__name__
+            result.append(f"  📍 {name:<12} → {node_type}")
+    
+    elif is_channels:
+        result.append("="*50)
+        result.append("📡 GRAPH CHANNELS")
+        result.append("="*50)
+        for name, channel in data.items():
+            channel_type = type(channel).__name__
+            if name in ['hello', 'messages']:
+                result.append(f"  🎯 {name:<20} → {channel_type} (user defined)")
+            elif name.startswith('branch:'):
+                result.append(f"  🌿 {name:<20} → {channel_type} (branch)")
+            else:
+                result.append(f"  ⚙️  {name:<20} → {channel_type} (system)")
+    
+    else:
+        # Fallback for unknown data types
+        result.append("="*50)
+        result.append("🔍 UNKNOWN DATA TYPE")
+        result.append("="*50)
+        for name, item in data.items():
+            item_type = type(item).__name__
+            result.append(f"  ❓ {name:<20} → {item_type}")
+    
+    result.append("="*50)
+    return "\n".join(result)
+
 logger = logging.getLogger(__name__)
 
 
+def test_parallel_nodes() -> None:
+    class State(TypedDict):
+        hello: str
+        messages: Annotated[list[str], add_messages]
+
+    def node_a(state: State) -> State:
+        return {"hello": "world-a", "messages": [_AnyIdHumanMessage(content="hello-a")]}
+
+    def node_b(state: State) -> State:
+        return {"messages": [_AnyIdHumanMessage(content="hello-b")]}
+
+    def node_c(state: State) -> State:
+        return {"messages": [_AnyIdHumanMessage(content="hello-c")]}    
+
+    def node_d(state: State) -> State:
+        return {"hello": "world-d", "messages": [_AnyIdHumanMessage(content="hello-d")]}    
+
+    builder = StateGraph(State)
+    builder.add_node("a", node_a)
+    builder.add_node("b", node_b)
+    builder.add_node("c", node_c)
+    builder.add_node("d", node_d)
+
+    builder.set_entry_point("a")
+    builder.add_edge("a", "b")
+    builder.add_edge("a", "c")
+    builder.add_edge("b", "d")
+    builder.add_edge("c", "d")
+    builder.add_edge("d", END)
+    graph = builder.compile()
+
+    print("\n======COMPLETE GRAPH======\n", pregel_pretty(graph))
+    
+    print(f"\n🔍 CHANNEL CONFIGURATION:")
+    print(f"  📤 output_channels: {graph.output_channels}")
+    print(f"  📡 stream_channels: {graph.stream_channels}")
+    print(f"  📥 input_channels: {graph.input_channels}")
+    print(f"  🌊 stream_channels_asis: {graph.stream_channels_asis}")
+    print(f"  📋 stream_channels_list: {graph.stream_channels_list}")
+
+    result = graph.invoke({"hello": "there"})
+    assert result["hello"] == "world-d"
+    # Only the final message from node_d should be in the result
+    # because each node overwrites the messages field completely
+    assert len(result["messages"]) == 1
+    assert result["messages"][0].content == "hello-d"
+
 def test_graph_validation() -> None:
     class State(TypedDict):
         hello: str

libs/langgraph/versions_seen_explained.md

@@ -0,0 +1,424 @@
+# Understanding `versions_seen` in LangGraph Checkpoints
+
+## Overview
+
+`versions_seen` is a nested dictionary in the checkpoint that tracks which channel versions each node has processed. It's defined in `libs/checkpoint/langgraph/checkpoint/base/__init__.py`:
+
+```python
+versions_seen: dict[str, ChannelVersions]
+"""Map from node ID to map from channel name to version seen.
+This keeps track of the versions of the channels that each node has seen.
+Used to determine which nodes to execute next.
+"""
+```
+
+## Data Structure
+
+```
+versions_seen = {
+    "node_name": {
+        "channel_name": version,
+        ...
+    },
+    ...
+}
+```
+
+## Key Point
+
+**`versions_seen` only records trigger channels, NOT state channels!**
+
+In StateGraph:
+- `triggers` = edge control channels like `branch:to:nodeA`
+- `channels` = state keys like `fieldA`, `fieldB`
+
+So `versions_seen` records `branch:to:*` channels, **NOT** `fieldA` or `fieldB`.
+
+---
+
+## Example Graph
+
+```
+        ┌──────────┐
+        │  nodeA   │
+        └────┬─────┘
+             │
+      ┌──────┴──────┐
+      ▼             ▼
+┌──────────┐  ┌──────────┐
+│  nodeB   │  │  nodeC   │
+└────┬─────┘  └────┬─────┘
+      │             │
+      └──────┬──────┘
+             ▼
+        ┌──────────┐
+        │  nodeD   │
+        └──────────┘
+```
+
+### State Definition
+
+```python
+class State(TypedDict):
+    fieldA: str
+    fieldB: str
+
+class StateOnlyA(TypedDict):
+    """Input schema for nodeB - only reads fieldA"""
+    fieldA: str
+
+class StateOnlyB(TypedDict):
+    """Input schema for nodeC - only reads fieldB"""
+    fieldB: str
+```
+
+---
+
+## Compiled Graph Structure
+
+### Channels Created
+
+| Channel | Type | Purpose |
+|---------|------|---------|
+| `fieldA` | LastValue | State data |
+| `fieldB` | LastValue | State data |
+| `__start__` | EphemeralValue | Input channel |
+| `branch:to:nodeA` | EphemeralValue | Edge control channel |
+| `branch:to:nodeB` | EphemeralValue | Edge control channel |
+| `branch:to:nodeC` | EphemeralValue | Edge control channel |
+| `branch:to:nodeD` | EphemeralValue | Edge control channel |
+| `join:nodeB+nodeC:nodeD` | NamedBarrierValue | Parallel join channel |
+
+### Nodes Configuration
+
+| Node | triggers | channels | Note |
+|------|----------|----------|------|
+| `__start__` | `["__start__"]` | `"__start__"` | Input node |
+| `nodeA` | `["branch:to:nodeA"]` | `["fieldA", "fieldB"]` | Reads full state |
+| `nodeB` | `["branch:to:nodeB"]` | `["fieldA"]` | Only reads fieldA (via `input_schema=StateOnlyA`) |
+| `nodeC` | `["branch:to:nodeC"]` | `["fieldB"]` | Only reads fieldB (via `input_schema=StateOnlyB`) |
+| `nodeD` | `["branch:to:nodeD", "join:nodeB+nodeC:nodeD"]` | `["fieldA", "fieldB"]` | Reads full state |
+
+**Note**: `triggers` are edge control channels, `channels` are state fields the node reads. Use `input_schema` to control which fields a node reads.
+
+---
+
+## Step-by-Step Execution
+
+### Input
+
+```python
+{"fieldA": "Hello", "fieldB": "World"}
+```
+
+---
+
+### Step -1: Input Phase (source: input)
+
+```
+State values: {}
+
+channel_versions:
+    __start__: v01
+
+versions_seen:
+    __input__: {}
+```
+
+Initial checkpoint when input is received.
+
+---
+
+### Step 0: `__start__` executes (source: loop)
+
+```
+State values: {'fieldA': 'Hello', 'fieldB': 'World'}
+
+channel_versions:
+    __start__:        v02
+    branch:to:nodeA:  v02
+    fieldA:           v02
+    fieldB:           v02
+
+versions_seen:
+    __input__: {}
+    __start__:
+        __start__: v01    ← __start__ node saw __start__ channel
+```
+
+**Note**: `fieldA` and `fieldB` are NOT in `versions_seen`!
+
+---
+
+### Step 1: nodeA executes (source: loop)
+
+```
+nodeA reads: fieldA='Hello', fieldB='World'
+State values: {'fieldA': 'Hello->A', 'fieldB': 'World->A'}
+
+channel_versions:
+    __start__:        v02
+    branch:to:nodeA:  v03
+    branch:to:nodeB:  v03
+    branch:to:nodeC:  v03
+    fieldA:           v03
+    fieldB:           v03
+
+versions_seen:
+    __input__: {}
+    __start__:
+        __start__: v01
+    nodeA:
+        branch:to:nodeA: v02    ← nodeA saw its trigger
+```
+
+**Key observation**:
+- `nodeA`'s `versions_seen` only records `branch:to:nodeA`
+- **NO** `fieldA` or `fieldB` because they are NOT triggers!
+
+---
+
+### Step 2: nodeB and nodeC execute in parallel (source: loop)
+
+```
+nodeB reads: fieldA='Hello->A'
+nodeC reads: fieldB='World->A'
+State values: {'fieldA': 'Hello->A->B', 'fieldB': 'World->A->C'}
+
+channel_versions:
+    __start__:              v02
+    branch:to:nodeA:        v03
+    branch:to:nodeB:        v04
+    branch:to:nodeC:        v04
+    fieldA:                 v04
+    fieldB:                 v04
+    join:nodeB+nodeC:nodeD: v04
+
+versions_seen:
+    __input__: {}
+    __start__:
+        __start__: v01
+    nodeA:
+        branch:to:nodeA: v02
+    nodeB:
+        branch:to:nodeB: v03    ← nodeB saw its trigger
+    nodeC:
+        branch:to:nodeC: v03    ← nodeC saw its trigger
+```
+
+---
+
+### Step 3: nodeD executes (source: loop)
+
+```
+nodeD reads: fieldA='Hello->A->B', fieldB='World->A->C'
+State values: {'fieldA': 'Hello->A->B->D', 'fieldB': 'World->A->C->D'}
+
+channel_versions:
+    __start__:              v02
+    branch:to:nodeA:        v03
+    branch:to:nodeB:        v04
+    branch:to:nodeC:        v04
+    fieldA:                 v05
+    fieldB:                 v05
+    join:nodeB+nodeC:nodeD: v05
+
+versions_seen:
+    __input__: {}
+    __start__:
+        __start__: v01
+    nodeA:
+        branch:to:nodeA: v02
+    nodeB:
+        branch:to:nodeB: v03
+    nodeC:
+        branch:to:nodeC: v03
+    nodeD:
+        join:nodeB+nodeC:nodeD: v04    ← nodeD saw join channel
+```
+
+---
+
+## Summary Table
+
+| Node | versions_seen records | Why? |
+|------|----------------------|------|
+| `__start__` | `__start__` | Its trigger is `__start__` |
+| `nodeA` | `branch:to:nodeA` | Its trigger is `branch:to:nodeA` |
+| `nodeB` | `branch:to:nodeB` | Its trigger is `branch:to:nodeB` |
+| `nodeC` | `branch:to:nodeC` | Its trigger is `branch:to:nodeC` |
+| `nodeD` | `join:nodeB+nodeC:nodeD` | One of its triggers (join channel) |
+
+---
+
+## Conclusion
+
+- `versions_seen` **only records triggers**
+- `fieldA` and `fieldB` **never appear** in `versions_seen`
+- In StateGraph, triggers are edge control channels (`branch:to:*`), not state fields
+- The purpose of `versions_seen` is to **prevent duplicate triggering**, so it only needs to track trigger channel versions
+
+---
+
+## Deep Dive: How `versions_seen` Determines the Last Node
+
+### The Problem
+
+When calling `update_state()` without specifying `as_node`, LangGraph needs to figure out which node "last updated" the state. This is done using `versions_seen`.
+
+### The Algorithm
+
+```python
+last_seen_by_node = sorted(
+    (v, n)
+    for n, seen in checkpoint["versions_seen"].items()
+    if n in self.nodes
+    for v in seen.values()
+)
+```
+
+This creates a sorted list of `(version, node_name)` tuples.
+
+### Key Insight: Version = Superstep
+
+**Nodes that execute in the same superstep (parallel execution) will have the same trigger channel version.**
+
+This is because:
+1. Each superstep increments the version counter
+2. All nodes triggered in the same superstep see the same version
+3. So `version` effectively identifies which superstep a node executed in
+
+### Analysis by Step (Using Our Example)
+
+#### Step 1: After nodeA executes
+
+```
+versions_seen:
+    __start__: { __start__: v01 }
+    nodeA:     { branch:to:nodeA: v02 }
+
+last_seen_by_node = [(v01, "__start__"), (v02, "nodeA")]
+
+Check: last[-1][0] != last[-2][0]?
+       v02 != v01? ✅ YES
+
+Result: as_node = "nodeA" (last node in the latest superstep)
+```
+
+#### Step 2: After nodeB and nodeC execute (parallel)
+
+```
+versions_seen:
+    __start__: { __start__: v01 }
+    nodeA:     { branch:to:nodeA: v02 }
+    nodeB:     { branch:to:nodeB: v03 }  ← same version!
+    nodeC:     { branch:to:nodeC: v03 }  ← same version!
+
+last_seen_by_node = [(v01, "__start__"), (v02, "nodeA"), (v03, "nodeB"), (v03, "nodeC")]
+
+Check: last[-1][0] != last[-2][0]?
+       v03 != v03? ❌ NO (same version = same superstep)
+
+Result: AMBIGUOUS! Multiple nodes executed in the last superstep.
+        → Raises InvalidUpdateError("Ambiguous update, specify as_node")
+```
+
+#### Step 3: After nodeD executes
+
+```
+versions_seen:
+    __start__: { __start__: v01 }
+    nodeA:     { branch:to:nodeA: v02 }
+    nodeB:     { branch:to:nodeB: v03 }
+    nodeC:     { branch:to:nodeC: v03 }
+    nodeD:     { join:nodeB+nodeC:nodeD: v04 }
+
+last_seen_by_node = [(v01, "__start__"), (v02, "nodeA"), (v03, "nodeB"), (v03, "nodeC"), (v04, "nodeD")]
+
+Check: last[-1][0] != last[-2][0]?
+       v04 != v03? ✅ YES
+
+Result: as_node = "nodeD" (only node in the latest superstep)
+```
+
+### Summary Table
+
+| Step | Last Two Versions | Same Superstep? | as_node |
+|------|-------------------|-----------------|---------|
+| Step 1 | v02, v01 | No | ✅ nodeA |
+| Step 2 | v03, v03 | **Yes (parallel!)** | ❌ Ambiguous |
+| Step 3 | v04, v03 | No | ✅ nodeD |
+
+### Visual Representation
+
+```
+Superstep Timeline:
+
+  Superstep 0    Superstep 1    Superstep 2    Superstep 3
+  (v01, v02)        (v03)          (v04)          (v05)
+      │               │              │              │
+      ▼               ▼              ▼              ▼
+  ┌────────┐    ┌──────────┐   ┌─────────┐    ┌────────┐
+  │__start__│    │  nodeA   │   │ nodeB   │    │ nodeD  │
+  └────────┘    └──────────┘   │ nodeC   │    └────────┘
+                               │(parallel)│
+                               └─────────┘
+
+When version[-1] == version[-2]:
+  → Multiple nodes in the same superstep
+  → Cannot determine which one was "last"
+  → Ambiguous!
+```
+
+### The Logic Explained
+
+```python
+if last_seen_by_node:
+    if len(last_seen_by_node) == 1:
+        # Only one node ever executed
+        as_node = last_seen_by_node[0][1]
+    elif last_seen_by_node[-1][0] != last_seen_by_node[-2][0]:
+        # Last two have different versions
+        # → Last superstep had only ONE node
+        # → That node is unambiguously the "last" one
+        as_node = last_seen_by_node[-1][1]
+    # else: versions are equal
+    #     → Multiple nodes in the last superstep
+    #     → Ambiguous, will raise error later
+```
+
+---
+
+## How Scheduling Works
+
+```python
+def _triggers(channels, versions, seen, null_version, proc) -> bool:
+    for chan in proc.triggers:  # Only checks triggers!
+        if channels[chan].is_available() and \           # Condition 1: channel has value
+           versions.get(chan, null_version) > seen.get(chan, null_version):  # Condition 2: version updated
+            return True
+    return False
+```
+
+Translation:
+> Trigger the node if ANY trigger channel satisfies **BOTH** conditions:
+> 1. `is_available()` - the channel has a value
+> 2. `current_version > seen_version` - the version is newer than what the node has seen
+
+**Important**: Both conditions must be met! This is why `EphemeralValue` channels (like `branch:to:*`) 
+can have their version increase after being consumed, but won't re-trigger the node because 
+`is_available()` returns `False` after consumption.
+
+Since only triggers are checked, only trigger versions need to be recorded in `versions_seen`.
+
+---
+
+## Running the Test
+
+To run the test script yourself:
+
+```bash
+cd libs/langgraph
+uv run python test_versions_seen.py
+```
+

pregel.learning.md

@@ -0,0 +1,153 @@
+# LangGraph Channel Types Usage Analysis
+
+本文档分析了 LangGraph 中各种通道类型在 StateGraph 中的使用情况，基于对整个仓库的深入分析。
+
+## 通道类型使用总结
+
+### 1. **LastValue** - 最常用的默认通道
+**使用场景：** 普通状态字段的默认通道类型
+**创建方式：** 自动创建（fallback）
+**仓库例子：**
+```python
+class State(TypedDict):
+    hello: str  # 自动创建 LastValue(str) 通道
+    count: int  # 自动创建 LastValue(int) 通道
+```
+
+### 2. **BinaryOperatorAggregate** - 状态聚合通道
+**使用场景：** 使用 reducer 函数进行状态聚合
+**创建方式：** 通过 `Annotated[Type, reducer_function]`
+**仓库例子：**
+```python
+# 例子1: 使用 add_messages
+class State(TypedDict):
+    messages: Annotated[list[str], add_messages]  # 创建 BinaryOperatorAggregate
+
+# 例子2: 使用 operator.add
+StateGraph(Annotated[str, operator.add])  # 整个状态使用聚合
+StateGraph(Annotated[list, operator.add])  # 列表聚合
+
+# 例子3: 使用 operator.or_
+class State(TypedDict):
+    val3: Required[Annotated[dict, operator.or_]]  # 字典合并
+```
+
+### 3. **EphemeralValue** - 临时通道
+**使用场景：** 系统内部使用，节点间的临时通信
+**创建方式：** 系统自动创建
+**仓库例子：**
+```python
+# StateGraph.compile() 中自动创建
+START: EphemeralValue(self.input_schema)  # 输入通道
+
+# CompiledStateGraph.attach_node() 中创建
+EphemeralValue(Any, guard=False)  # 节点分支通道
+```
+
+### 4. **LastValueAfterFinish** - 延迟通道
+**使用场景：** 节点设置了 `defer=True` 时使用
+**创建方式：** 系统根据节点配置自动创建
+**仓库例子：**
+```python
+# CompiledStateGraph.attach_node() 中
+if node.defer:
+    LastValueAfterFinish(Any)  # 延迟节点的分支通道
+```
+
+### 5. **NamedBarrierValue** - 同步屏障通道
+**使用场景：** 多个节点汇聚到一个节点时的同步
+**创建方式：** 系统在处理 waiting_edges 时自动创建
+**仓库例子：**
+```python
+# CompiledStateGraph.attach_edge() 中
+channel_name = f"join:{'+'.join(starts)}:{end}"
+self.channels[channel_name] = NamedBarrierValue(str, set(starts))
+```
+
+### 6. **NamedBarrierValueAfterFinish** - 延迟同步屏障通道
+**使用场景：** 延迟节点的多节点汇聚同步
+**创建方式：** 系统在处理延迟节点的 waiting_edges 时自动创建
+**仓库例子：**
+```python
+# CompiledStateGraph.attach_edge() 中
+if self.builder.nodes[end].defer:
+    self.channels[channel_name] = NamedBarrierValueAfterFinish(str, set(starts))
+```
+
+### 7. **Topic** - 发布订阅通道
+**使用场景：** 仅在直接使用 Pregel 时手动创建，StateGraph 中无实际使用
+**创建方式：** 手动创建或系统内部 TASKS 通道
+**仓库例子：**
+```python
+# Pregel.__init__() 中系统创建
+self.channels[TASKS] = Topic(Send, accumulate=False)
+
+# 直接 Pregel 使用（非 StateGraph）
+app = Pregel(
+    channels={"c": Topic(str, accumulate=True)},  # 手动创建
+    # ...
+)
+```
+
+### 8. **UntrackedValue** - 未追踪通道
+**使用场景：** 在仓库中未找到实际使用例子
+**创建方式：** 需要手动创建
+**状态：** 理论存在但实际未使用
+
+### 9. **AnyValue** - 任意值通道
+**使用场景：** 在仓库中未找到实际使用例子
+**创建方式：** 需要手动创建
+**状态：** 理论存在但实际未使用
+
+## 使用模式总结
+
+### **用户显式创建的通道：**
+1. **BinaryOperatorAggregate** - 通过 `Annotated[Type, reducer]`
+2. **Topic** - 仅在直接 Pregel API 中手动创建
+
+### **系统自动创建的通道：**
+1. **LastValue** - 默认通道类型
+2. **EphemeralValue** - 输入和分支通道
+3. **LastValueAfterFinish** - 延迟节点分支
+4. **NamedBarrierValue** - 多节点汇聚同步
+5. **NamedBarrierValueAfterFinish** - 延迟节点汇聚同步
+
+### **实际使用频率：**
+1. **高频使用：** LastValue, BinaryOperatorAggregate, EphemeralValue
+2. **中频使用：** NamedBarrierValue, LastValueAfterFinish
+3. **低频使用：** Topic（仅系统内部）
+4. **未使用：** UntrackedValue, AnyValue
+
+## 通道创建机制
+
+### 自动创建流程
+1. **StateGraph._add_schema()** - 解析状态类型
+2. **_get_channels()** - 提取类型注解
+3. **_get_channel()** - 判断通道类型：
+   - 检查 Managed Value
+   - 检查 Channel（如 Topic）
+   - 检查 BinaryOperator（如 add_messages）
+   - 默认创建 LastValue
+
+### 编译时创建
+- **START 通道：** `EphemeralValue(input_schema)`
+- **分支通道：** `EphemeralValue(Any, guard=False)` 或 `LastValueAfterFinish(Any)`
+- **汇聚通道：** `NamedBarrierValue` 或 `NamedBarrierValueAfterFinish`
+
+## 设计哲学
+
+**StateGraph 主要关注状态管理，大部分通道类型都是系统自动管理的，用户只需要关心状态结构和聚合逻辑**。只有在需要特殊聚合行为时，用户才需要显式使用 `Annotated` 注解来指定 reducer 函数。
+
+## 关键发现
+
+1. **Topic 通道在 StateGraph 中几乎不使用** - 仓库中没有通过状态 Schema 注解创建 Topic 的实际例子
+2. **BinaryOperatorAggregate 是用户最常显式创建的通道** - 通过 `add_messages` 等 reducer 函数
+3. **大部分通道都是系统内部自动管理** - 用户无需关心底层通道实现
+4. **StateGraph 和直接 Pregel 的使用场景不同** - StateGraph 专注状态管理，Pregel 专注消息传递
+
+## 实际代码位置
+
+- **通道创建逻辑：** `libs/langgraph/langgraph/graph/state.py:1299-1388`
+- **编译时通道管理：** `libs/langgraph/langgraph/graph/state.py:856-894`
+- **节点附加逻辑：** `libs/langgraph/langgraph/graph/state.py:935-1067`
+- **边处理逻辑：** `libs/langgraph/langgraph/graph/state.py:1038-1062`