fix signature

snippet had a missing decorator that might lead the user to confuse whether it is really a tool

docs/_scripts/add_translation.py

@@ -1,157 +1,154 @@
-"""Add typescript translation to a given markdown file."""
+"""Translate Python markdown to TypeScript and/or consolidate Python-JS markdown into a single document."""
 
 import argparse
-import re
 
 import requests
 from langchain_anthropic import ChatAnthropic
 
+# Load reference TypeScript snippets
 URL = "https://gist.githubusercontent.com/eyurtsev/e7486731415463a9bc5b4682358859c8/raw/b5a5fda9c7e3387cfcb781f25082814d43675d50/gistfile1.txt"
 response = requests.get(URL)
 response.raise_for_status()
 reference_snippets = response.text
 
-model = ChatAnthropic(model="claude-3-5-sonnet-latest")
+# Initialize model
+model = ChatAnthropic(model="claude-sonnet-4-0", max_tokens=64_000)
+
+TRANSLATION_PROMPT = (
+    "You are a helpful assistant that translates Python-based technical "
+    "documentation written in Markdown to equivalent TypeScript-based documentation. "
+    "The input is a Markdown file written in mkdocs format. It contains "
+    "Python code snippets embedded in prose. "
+    "Your task is to rewrite the content by translating the Python code to "
+    "idiomatic TypeScript, using the provided TypeScript reference snippets "
+    "to ensure accurate and consistent usage (e.g., correct imports, function "
+    "names, and patterns). "
+    "Remove the original Python code and replace it with the corresponding "
+    "TypeScript version. "
+    "Do not alter the surrounding prose unless a change is necessary to "
+    "reflect differences between Python and TypeScript. "
+    "Preserve the structure and formatting of the original Markdown document. "
+    "Do not make stylistic or structural changes unless they directly support "
+    "the translation. "
+    "Use the reference TypeScript snippets as guidance whenever possible to "
+    "maintain alignment with existing conventions.\n\n"
+    f"Here are the reference TypeScript snippets:\n\n{reference_snippets}\n\n"
+)
+
+CONSOLIDATION_PROMPT = (
+    "You are a helpful assistant that consolidates parallel Python and JavaScript (TypeScript) technical documentation "
+    "written in Markdown into a single unified Markdown document. "
+    "The input consists of two documents: the first is for Python users, and the second is for JavaScript/TypeScript users. "
+    "Your task is to merge these into one Markdown file using language-specific fenced blocks to separate the content where needed. "
+    "Use the following syntax to distinguish content for each language:\n\n"
+    ":::python\n"
+    "# Python-specific content\n"
+    ":::\n\n"
+    ":::js\n"
+    "# JavaScript/TypeScript-specific content\n"
+    ":::\n\n"
+    "Follow these consolidation rules:\n"
+    "- When content (prose or code) is the same or nearly identical in both versions, include it only once—outside of any fenced block.\n"
+    "- When content differs between the Python and JS versions, wrap each version in its corresponding fenced block.\n"
+    "- Prefer **paragraph-level separation** of language-specific content. Do not combine Python and JS snippets or terminology in the same sentence or paragraph using conditional phrases.\n"
+    "  For example, avoid inline constructs like:\n"
+    "  `The :::python add_messages ::: :::js reducer ::: function...`\n"
+    "  Instead, write two distinct paragraphs:\n\n"
+    "  :::python\n"
+    "  The `add_messages` function in our `State` will append the LLM's response messages to whatever messages are already in the state.\n"
+    "  ::: \n\n"
+    "  :::js\n"
+    "  The `reducer` function in our `StateAnnotation` will append the LLM's response messages to whatever messages are already in the state.\n"
+    "  :::\n\n"
+    "- Preserve the overall structure, ordering, and formatting of the original Markdown documents.\n"
+    "- Do not rephrase or unify content unless it is logically and semantically identical.\n"
+    "- Use the fenced blocks for both prose and code as needed, and ensure output is clean, readable Markdown suitable for tools that parse these directives.\n"
+    "Your goal is to produce a cleanly merged documentation file that serves both Python and JavaScript users without redundancy, while maximizing clarity and separation of language-specific details."
+)
 
 
-def _get_tqdm():
-    try:
-        from tqdm import tqdm
-    except ImportError:
-        # If not available return a simple identity function
-        def tqdm(iterable, *args, **kwargs):
-            return iterable
-
-    return tqdm
-
-
-_tqdm = _get_tqdm()
-
-opening_pattern = re.compile(r"^\s*```python(?:\s+.*)?\s*$")
-closing_pattern = re.compile(r"^\s*```\s*$")
-
-
-def extract_python_snippets(markdown: str) -> list[str]:
-    """
-    Extract all python code blocks (including their fence lines) from the markdown content.
-    A python block is defined as any block that starts with a line containing an opening fence
-    with '```python' (optionally with extra parameters) and ends with a closing fence '```'.
-    """
-    snippets = []
-    inside_block = False
-    current_snippet = []
-
-    for line in markdown.splitlines(keepends=True):
-        if not inside_block:
-            if opening_pattern.match(line):
-                inside_block = True
-                current_snippet = [line]
-        else:
-            current_snippet.append(line)
-            if closing_pattern.match(line):
-                inside_block = False
-                snippets.append("".join(current_snippet))
-                current_snippet = []
-    return snippets
-
-
-def translate_snippet(python_snippet: str) -> str:
-    """Translate a python code block into a TypeScript code block using Langchain.
-    The response is expected to be a properly fenced TypeScript code block (i.e.
-    starting with ```typescript and ending with ```).
-    """
-    ai_message = model.invoke(
+def translate_python_to_ts(markdown_content: str) -> str:
+    response = model.invoke(
         [
             {
                 "role": "system",
-                "content": (
-                    f"You have access to the following up-to-date example TypeScript code "
-                    f"snippets that show examples of building with langgraph "
-                    f"and langchain:\n\n{reference_snippets}\n\n"
-                    "Use this context to translate the following Python code to equivalent "
-                    "TypeScript. Ensure that your output is a valid fenced TypeScript "
-                    "code block (i.e. starts with ```typescript and ends with ```)."
-                ),
-            },
-            {
-                "role": "user",
-                "content": f"Translate this Python snippet to TypeScript:\n\n{python_snippet}",
+                "content": TRANSLATION_PROMPT,
+                "cache_control": {"type": "ephemeral"},
             },
+            {"role": "user", "content": markdown_content},
         ]
     )
-
-    # Use a regular expression to search for a TypeScript code block in the response.
-    pattern = r"```typescript\s*(.*?)\s*```"
-    match = re.search(pattern, ai_message.content, re.DOTALL)
-    if match:
-        # Reconstruct the code block with proper fences.
-        typescript_code = match.group(1).strip()
-        return f"```typescript\n{typescript_code}\n```"
-    else:
-        raise ValueError("No TypeScript code block found in the model's response.")
+    return response.content
 
 
-def insert_translations_into_markdown(
-    markdown: str, typescript_snippets: list[str]
-) -> str:
-    """Walks through the original markdown content and, after each
-    Python snippet block, inserts the corresponding translated TypeScript snippet.
-    It assumes that the ordering of the Python snippets
-    (from extract_python_snippets) matches the order they appear in the markdown.
-    """
-    output_lines = []
-    lines = markdown.splitlines(keepends=True)
-    inside_block = False
-    snippet_index = 0
-
-    for line in lines:
-        output_lines.append(line)
-        if not inside_block and opening_pattern.match(line):
-            # We've encountered the start of a python code block.
-            inside_block = True
-        elif inside_block:
-            if closing_pattern.match(line):
-                # End of a python snippet block.
-                inside_block = False
-                if snippet_index < len(typescript_snippets):
-                    # Insert an extra newline for clarity, then the translated TypeScript snippet.
-                    output_lines.append("\n")
-                    output_lines.append(typescript_snippets[snippet_index])
-                    output_lines.append("\n")
-                    snippet_index += 1
-    return "".join(output_lines)
+def consolidate_python_and_ts(combined_content: str) -> str:
+    response = model.invoke(
+        [
+            {
+                "role": "system",
+                "content": CONSOLIDATION_PROMPT,
+                "cache_control": {"type": "ephemeral"},
+            },
+            {"role": "user", "content": combined_content},
+        ]
+    )
+    return response.content
 
 
-def main(file_path: str) -> None:
-    # Read the markdown file.
-    with open(file_path, "r") as f:
+def main(file_path: str, translate_only: bool, consolidate_only: bool) -> None:
+    with open(file_path, "r", encoding="utf-8") as f:
         markdown_content = f.read()
 
-    # 1. Extract all Python snippets.
-    python_snippets = extract_python_snippets(markdown_content)[:1]
+    if translate_only:
+        translated = translate_python_to_ts(markdown_content)
+        output_path = file_path.replace(".md", ".translated.md")
+        with open(output_path, "w", encoding="utf-8") as f:
+            f.write(translated)
+        print(f"Translated JS/TS version written to: {output_path}")
 
-    # 2. Translate each Python snippet to TypeScript.
-    typescript_snippets = []
-    # Replace with .batch() for faster translation
-    for python_snippet in _tqdm(python_snippets):
-        ts_snippet = translate_snippet(python_snippet)
-        typescript_snippets.append(ts_snippet)
+    elif consolidate_only:
+        consolidated = consolidate_python_and_ts(markdown_content)
+        with open(file_path, "w", encoding="utf-8") as f:
+            f.write(consolidated)
+        print(f"Consolidated content written to: {file_path}")
 
-    # 3. Insert the TypeScript translations after their respective Python snippets.
-    updated_markdown = insert_translations_into_markdown(
-        markdown_content, typescript_snippets
-    )
-
-    # Overwrite the original markdown file with the updated content.
-    with open(file_path, "w") as f:
-        f.write(updated_markdown)
+    else:
+        # Default behavior: translate first, then consolidate both
+        translated = translate_python_to_ts(markdown_content)
+        combined = f"{markdown_content.strip()}\n\n\n{translated.strip()}"
+        consolidated = consolidate_python_and_ts(combined)
+        with open(file_path, "w", encoding="utf-8") as f:
+            f.write(consolidated)
+        print(f"Translated and consolidated content written to: {file_path}")
 
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(
-        description="Translate Python snippets in a markdown file to TypeScript and insert them after each Python snippet."
+        description=(
+            "Translate Python markdown to TypeScript and/or consolidate "
+            "Python-JS markdown into one file."
+        )
     )
     parser.add_argument("file_path", type=str, help="Path to the markdown file.")
+    parser.add_argument(
+        "--translate-only",
+        action="store_true",
+        help="Only generate the JS translation.",
+    )
+    parser.add_argument(
+        "--consolidate-only",
+        action="store_true",
+        help="Only consolidate pre-paired Python and JS content.",
+    )
     args = parser.parse_args()
 
-    main(args.file_path)
+    if args.translate_only and args.consolidate_only:
+        raise ValueError(
+            "Cannot use both --translate-only and --consolidate-only at the same time."
+        )
+
+    main(
+        args.file_path,
+        translate_only=args.translate_only,
+        consolidate_only=args.consolidate_only,
+    )

docs/_scripts/generate_llms_text.py

@@ -3,19 +3,21 @@
 import asyncio
 import glob
 import os
-from typing import TypedDict, List, Optional
-import pydantic
 import re
-from pydantic import BaseModel, Field
-from langchain_core.rate_limiters import InMemoryRateLimiter
+from typing import TypedDict, List, Optional
 
 import yaml
 from langchain.chat_models import init_chat_model
+from langchain_core.rate_limiters import InMemoryRateLimiter
 from mkdocs.structure.files import File
 from mkdocs.structure.pages import Page
+from pydantic import BaseModel, Field
 from yaml import SafeLoader
 
-from _scripts.notebook_hooks import _on_page_markdown_with_config
+from _scripts.notebook_hooks import (
+    _on_page_markdown_with_config,
+    _apply_conditional_rendering,
+)
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 # Get source directory (parent of HERE / docs)
@@ -211,7 +213,9 @@ async def process_nav_items(nav_items: list[NavItem]) -> list[NavItem]:
     # Remove any items that start with http:// or https:// looking only for
     # local file at this stages.
     nav_items = [
-        item for item in nav_items if not item["url"].startswith(("http://", "https://"))
+        item
+        for item in nav_items
+        if not item["url"].startswith(("http://", "https://"))
     ]
     # Process items in parallel
     tasks = [process_single_item(item) for item in nav_items]

docs/_scripts/link_map.py

@@ -0,0 +1,5 @@
+JS_LINK_MAP = {
+    "langgraph.types.interrupt": "https://langchain-ai.github.io/langgraphjs/reference/functions/langgraph.interrupt-2.html",
+    "create_react_agent": "https://langchain-ai.github.io/langgraphjs/reference/functions/langgraph_prebuilt.createReactAgent.html",
+    "langgraph.types.Command": "https://langchain-ai.github.io/langgraphjs/reference/classes/langgraph.Command.html",
+}

docs/_scripts/notebook_hooks.py

@@ -16,6 +16,7 @@ from mkdocs.structure.pages import Page
 
 from _scripts.generate_api_reference_links import update_markdown_with_imports
 from _scripts.notebook_convert import convert_notebook
+from _scripts.link_map import JS_LINK_MAP
 
 logger = logging.getLogger(__name__)
 logging.basicConfig()
@@ -86,7 +87,7 @@ REDIRECT_MAP = {
     "cloud/how-tos/stream_events.md": "cloud/how-tos/streaming.md#stream-events",
     "cloud/how-tos/stream_debug.md": "cloud/how-tos/streaming.md#debug",
     "cloud/how-tos/stream_multiple.md": "cloud/how-tos/streaming.md#stream-multiple-modes",
-    # prebuit redirects
+    # prebuilt redirects
     "how-tos/create-react-agent.ipynb": "agents/agents.md#basic-configuration",
     "how-tos/create-react-agent-memory.ipynb": "agents/memory.md",
     "how-tos/create-react-agent-system-prompt.ipynb": "agents/context.md#prompts",
@@ -158,6 +159,62 @@ def _add_path_to_code_blocks(markdown: str, page: Page) -> str:
     return code_block_pattern.sub(replace_code_block_header, markdown)
 
 
+def _resolve_cross_references(md_text: str, link_map: dict[str, str]) -> str:
+    """Replace [title][identifier] with [title](url) using language-specific link_map.
+
+    Args:
+        md_text: The markdown text to process.
+        link_map: mapping of identifier to URL.
+
+    Returns:
+        The processed markdown text with cross-references resolved.
+    """
+    # Pattern to match [title][identifier]
+    pattern = re.compile(r"\[([^\]]+)\]\[([^\]]+)\]")
+
+    def replace_reference(match: re.Match) -> str:
+        """Replace the matched reference with the corresponding URL."""
+        title, identifier = match.group(1), match.group(2)
+        url = link_map.get(identifier)
+
+        if url:
+            return f"[{title}]({url})"
+        else:
+            # Leave it unchanged if not found
+            return match.group(0)
+
+    return pattern.sub(replace_reference, md_text)
+
+
+def _apply_conditional_rendering(md_text: str, target_language: str) -> str:
+    if target_language not in {"python", "js"}:
+        raise ValueError("target_language must be 'python' or 'js'")
+
+    pattern = re.compile(
+        r"(?P<indent>[ \t]*):::(?P<language>\w+)\s*\n"
+        r"(?P<content>((?:.*\n)*?))"  # Capture the content inside the block
+        r"(?P=indent):::"  # Match closing with the same indentation
+    )
+
+    def replace_conditional_blocks(match: re.Match) -> str:
+        """Keep active conditionals."""
+        language = match.group("language")
+        content = match.group("content")
+
+        if language not in {"python", "js"}:
+            # If the language is not supported, return the original block
+            return match.group(0)
+
+        if language == target_language:
+            return content
+
+        # If the language does not match, return an empty string
+        return ""
+
+    processed = pattern.sub(replace_conditional_blocks, md_text)
+    return processed
+
+
 def _highlight_code_blocks(markdown: str) -> str:
     """Find code blocks with highlight comments and add hl_lines attribute.
 
@@ -257,6 +314,20 @@ def _on_page_markdown_with_config(
     # Apply highlight comments to code blocks
     markdown = _highlight_code_blocks(markdown)
 
+    # Apply conditional rendering for code blocks
+    target_language = kwargs.get("target_language", "python")
+    markdown = _apply_conditional_rendering(markdown, target_language)
+    if target_language == "js":
+        markdown = _resolve_cross_references(markdown, JS_LINK_MAP)
+    elif target_language == "python":
+        # Via a dedicated plugin
+        pass
+    else:
+        raise ValueError(
+            f"Unsupported target language: {target_language}. "
+            "Supported languages are 'python' and 'js'."
+        )
+
     # Add file path as an attribute to code blocks that are executable.
     # This file path is used to associate fixtures with the executable code
     # which can be used in CI to test the docs without making network requests.

docs/docs/cloud/deployment/cloud.md

@@ -62,6 +62,15 @@ Starting from the `LangGraph Platform` view...
 1. In the panel, select the `Server` tab to view server logs for the revision. Server logs are only available after a revision has been deployed.
 1. Within the `Server` tab, adjust the date/time range picker as needed. By default, the date/time range picker is set to the `Last 7 days`.
 
+## View Deployment Metrics
+
+Starting from the <a href="https://smith.langchain.com/" target="_blank">LangSmith UI</a>...
+
+1. In the left-hand navigation panel, select `LangGraph Platform`. The `LangGraph Platform` view contains a list of existing LangGraph Platform deployments.
+1. Select an existing deployment to monitor.
+1. Select the `Monitoring` tab to view the deployment metrics. See a list of [all available metrics](../../concepts/langgraph_control_plane.md#monitoring).
+1. Within the `Monitoring` tab, use the date/time range picker as needed. By default, the date/time range picker is set to the `Last 15 minutes`.
+
 ## Interrupt Revision
 
 Interrupting a revision will stop deployment of the revision.

docs/docs/cloud/deployment/graph_rebuild.md

@@ -20,7 +20,7 @@ my-app/
 |-- openai_agent.py     # code for your graph
 ```
 
-where the graph is defined in `openai_agent.py`.
+where the graph is defined in `openai_agent.py`. 
 
 ### No rebuild
 
@@ -28,11 +28,11 @@ In the standard LangGraph API configuration, the server uses the compiled graph
 
 ```python
 from langchain_openai import ChatOpenAI
-from langgraph.graph import END, START, StateGraph, MessagesState
+from langgraph.graph import END, START, MessageGraph
 
 model = ChatOpenAI(temperature=0)
 
-graph_workflow = StateGraph(MessagesState)
+graph_workflow = MessageGraph()
 
 graph_workflow.add_node("agent", model)
 graph_workflow.add_edge("agent", END)
@@ -61,7 +61,7 @@ To make your graph rebuild on each new run with custom configuration, you need t
 from typing import Annotated
 from typing_extensions import TypedDict
 from langchain_openai import ChatOpenAI
-from langgraph.graph import END, START
+from langgraph.graph import END, START, MessageGraph
 from langgraph.graph.state import StateGraph
 from langgraph.graph.message import add_messages
 from langgraph.prebuilt import ToolNode
@@ -144,4 +144,4 @@ Finally, you need to specify the path to your graph-making function (`make_graph
 }
 ```
 
-See more info on LangGraph API configuration file [here](../reference/cli.md#configuration-file)
+See more info on LangGraph API configuration file [here](../reference/cli.md#configuration-file)

docs/docs/cloud/how-tos/configuration_cloud.md

@@ -212,6 +212,7 @@ We have now created an assistant called "Open AI Assistant" that has `model_name
 
 Output:
 
+    ```
     Receiving event of type: metadata
     {'run_id': '1ef6746e-5893-67b1-978a-0f1cd4060e16'}
 
@@ -219,6 +220,7 @@ Output:
 
     Receiving event of type: updates
     {'agent': {'messages': [{'content': 'I was created by OpenAI, a research organization focused on developing and advancing artificial intelligence technology.', 'additional_kwargs': {}, 'response_metadata': {'finish_reason': 'stop', 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_157b3831f5'}, 'type': 'ai', 'name': None, 'id': 'run-e1a6b25c-8416-41f2-9981-f9cfe043f414', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]}}
+    ```
 
 ### LangGraph Platform UI
 
@@ -231,9 +233,11 @@ Inside your deployment, select the "Assistants" tab. For the assistant you would
 To edit the assistant, use the `update` method. This will create a new version of the assistant with the provided edits. See the [Python](https://langchain-ai.github.io/langgraph/cloud/reference/sdk/python_sdk_ref/#langgraph_sdk.client.AssistantsClient.update) and [JS](https://langchain-ai.github.io/langgraph/cloud/reference/sdk/js_ts_sdk_ref/#update) SDK reference docs for more information.
 
 !!! note "Note"
-You must pass in the ENTIRE config (and metadata if you are using it). The update endpoint creates new versions completely from scratch and does not rely on previous versions.
+
+    You must pass in the ENTIRE config (and metadata if you are using it). The update endpoint creates new versions completely from scratch and does not rely on previous versions.
 
 For example, to update your assistant's system prompt:
+
 === "Python"
 
     ```python

docs/docs/cloud/how-tos/interrupt_concurrent.md

@@ -247,5 +247,7 @@ Verify that the original, interrupted run was interrupted
 
 Output:
 
+    ```
     'interrupted'
+    ```
 

docs/docs/cloud/how-tos/studio/quick_start.md

@@ -73,9 +73,11 @@ langgraph dev --debug-port 5678
 Then attach your preferred debugger:
 
 === "VS Code"
-Add this configuration to `launch.json`:
-`json
-      {
+
+    Add this configuration to `launch.json`:
+
+    ```json
+    {
         "name": "Attach to LangGraph",
         "type": "debugpy",
         "request": "attach",
@@ -83,11 +85,16 @@ Add this configuration to `launch.json`:
           "host": "0.0.0.0",
           "port": 5678
         }
-      }
-    `
-Specify the port number you chose in the previous step.
+    }
+    ```
 
-=== "PyCharm" 1. Go to Run → Edit Configurations 2. Click + and select "Python Debug Server" 3. Set IDE host name: `localhost` 4. Set port: `5678` (or the port number you chose in the previous step) 5. Click "OK" and start debugging
+=== "PyCharm" 
+
+    1. Go to Run → Edit Configurations 
+    2. Click + and select "Python Debug Server" 
+    3. Set IDE host name: `localhost` 
+    4. Set port: `5678` (or the port number you chose in the previous step) 
+    5. Click "OK" and start debugging
 
 ## Troubleshooting
 

docs/docs/cloud/how-tos/webhooks.md

@@ -8,15 +8,15 @@ Currently, the SDK does not provide built-in support for defining webhook endpoi
 
 The following API endpoints accept a `webhook` parameter:
 
-| Operation | HTTP Method | Endpoint |
-|-----------|------------|----------|
-| Create Run | `POST` | `/thread/{thread_id}/runs` |
-| Create Thread Cron | `POST` | `/thread/{thread_id}/runs/crons` |
-| Stream Run | `POST` | `/thread/{thread_id}/runs/stream` |
-| Wait Run | `POST` | `/thread/{thread_id}/runs/wait` |
-| Create Cron | `POST` | `/runs/crons` |
-| Stream Run Stateless | `POST` | `/runs/stream` |
-| Wait Run Stateless | `POST` | `/runs/wait` |
+| Operation            | HTTP Method | Endpoint                          |
+|----------------------|-------------|-----------------------------------|
+| Create Run           | `POST`      | `/thread/{thread_id}/runs`        |
+| Create Thread Cron   | `POST`      | `/thread/{thread_id}/runs/crons`  |
+| Stream Run           | `POST`      | `/thread/{thread_id}/runs/stream` |
+| Wait Run             | `POST`      | `/thread/{thread_id}/runs/wait`   |
+| Create Cron          | `POST`      | `/runs/crons`                     |
+| Stream Run Stateless | `POST`      | `/runs/stream`                    |
+| Wait Run Stateless   | `POST`      | `/runs/wait`                      |
 
 In this guide, we’ll show how to trigger a webhook after streaming a run.
 
@@ -25,36 +25,39 @@ In this guide, we’ll show how to trigger a webhook after streaming a run.
 Before making API calls, set up your assistant and thread.
 
 === "Python"
-```python
-from langgraph_sdk import get_client
 
-client = get_client(url=<DEPLOYMENT_URL>)
-assistant_id = "agent"
-thread = await client.threads.create()
-print(thread)
-```
+    ```python
+    from langgraph_sdk import get_client
+
+    client = get_client(url=<DEPLOYMENT_URL>)
+    assistant_id = "agent"
+    thread = await client.threads.create()
+    print(thread)
+    ```
 
 === "JavaScript"
-```js
-import { Client } from "@langchain/langgraph-sdk";
 
-const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
-const assistantID = "agent";
-const thread = await client.threads.create();
-console.log(thread);
-```
+    ```js
+    import { Client } from "@langchain/langgraph-sdk";
+
+    const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
+    const assistantID = "agent";
+    const thread = await client.threads.create();
+    console.log(thread);
+    ```
 
 === "CURL"
-```bash
-curl --request POST \
-    --url <DEPLOYMENT_URL>/assistants/search \
-    --header 'Content-Type: application/json' \
-    --data '{ "limit": 10, "offset": 0 }' | jq -c 'map(select(.config == null or .config == {})) | .[0]' && \
-curl --request POST \
-    --url <DEPLOYMENT_URL>/threads \
-    --header 'Content-Type: application/json' \
-    --data '{}'
-```
+
+    ```bash
+    curl --request POST \
+        --url <DEPLOYMENT_URL>/assistants/search \
+        --header 'Content-Type: application/json' \
+        --data '{ "limit": 10, "offset": 0 }' | jq -c 'map(select(.config == null or .config == {})) | .[0]' && \
+    curl --request POST \
+        --url <DEPLOYMENT_URL>/threads \
+        --header 'Content-Type: application/json' \
+        --data '{}'
+    ```
 
 Example response:
 
@@ -77,48 +80,51 @@ To use a webhook, specify the `webhook` parameter in your API request. When the
 For example, if your server listens for webhook events at `https://my-server.app/my-webhook-endpoint`, include this in your request:
 
 === "Python"
-```python
-input = { "messages": [{ "role": "user", "content": "Hello!" }] }
 
-async for chunk in client.runs.stream(
-    thread_id=thread["thread_id"],
-    assistant_id=assistant_id,
-    input=input,
-    stream_mode="events",
-    webhook="https://my-server.app/my-webhook-endpoint"
-):
-    pass
-```
+    ```python
+    input = { "messages": [{ "role": "user", "content": "Hello!" }] }
+
+    async for chunk in client.runs.stream(
+        thread_id=thread["thread_id"],
+        assistant_id=assistant_id,
+        input=input,
+        stream_mode="events",
+        webhook="https://my-server.app/my-webhook-endpoint"
+    ):
+        pass
+    ```
 
 === "JavaScript"
-```js
-const input = { messages: [{ role: "human", content: "Hello!" }] };
 
-const streamResponse = client.runs.stream(
-  thread["thread_id"],
-  assistantID,
-  {
-    input: input,
-    webhook: "https://my-server.app/my-webhook-endpoint"
-  }
-);
+    ```js
+    const input = { messages: [{ role: "human", content: "Hello!" }] };
 
-for await (const chunk of streamResponse) {
-  // Handle stream output
-}
-```
+    const streamResponse = client.runs.stream(
+      thread["thread_id"],
+      assistantID,
+      {
+        input: input,
+        webhook: "https://my-server.app/my-webhook-endpoint"
+      }
+    );
+
+    for await (const chunk of streamResponse) {
+      // Handle stream output
+    }
+    ```
 
 === "CURL"
-```bash
-curl --request POST \
-    --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
-    --header 'Content-Type: application/json' \
-    --data '{
-        "assistant_id": <ASSISTANT_ID>,
-        "input": {"messages": [{"role": "user", "content": "Hello!"}]},
-        "webhook": "https://my-server.app/my-webhook-endpoint"
-    }'
-```
+
+    ```bash
+    curl --request POST \
+        --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
+        --header 'Content-Type: application/json' \
+        --data '{
+            "assistant_id": <ASSISTANT_ID>,
+            "input": {"messages": [{"role": "user", "content": "Hello!"}]},
+            "webhook": "https://my-server.app/my-webhook-endpoint"
+        }'
+    ```
 
 ## Webhook payload
 

docs/docs/cloud/reference/cli.md

@@ -50,9 +50,10 @@ The LangGraph CLI requires a JSON configuration file that follows this [schema](
     | <span style="white-space: nowrap;">`python_version`</span>   | `3.11`, `3.12`, or `3.13`. Defaults to `3.11`.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  |
     | <span style="white-space: nowrap;">`node_version`</span>     | Specify `node_version: 20` to use LangGraph.js.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 |
     | <span style="white-space: nowrap;">`pip_config_file`</span>  | Path to `pip` config file.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
+    | <span style="white-space: nowrap;">`pip_installer`</span> | _(Added in v0.3)_ Optional. Python package installer selector. It can be set to `"auto"`, `"pip"`, or `"uv"`. From version&nbsp;0.3 onward the default strategy is to run `uv pip`, which typically delivers faster builds while remaining a drop-in replacement. In the uncommon situation where `uv` cannot handle your dependency graph or the structure of your `pyproject.toml`, specify `"pip"` here to revert to the earlier behaviour. |
     | <span style="white-space: nowrap;">`dockerfile_lines`</span> | Array of additional lines to add to Dockerfile following the import from parent image.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
     | <span style="white-space: nowrap;">`checkpointer`</span>   | Configuration for the checkpointer. Contains a `ttl` field which is an object with the following keys: <ul><li>`strategy`: How to handle expired checkpoints (e.g., `"delete"`).</li><li>`sweep_interval_minutes`: How often to check for expired checkpoints (integer).</li><li>`default_ttl`: Default time-to-live for checkpoints in **minutes** (integer). Defines how long checkpoints are kept before the specified strategy is applied.</li></ul> |
-    | <span style="white-space: nowrap;">`http`</span>            | HTTP server configuration with the following fields: <ul><li>`app`: Path to custom Starlette/FastAPI app (e.g., `"./src/agent/webapp.py:app"`). See [custom routes guide](../../how-tos/http/custom_routes.md).</li><li>`disable_assistants`: Disable `/assistants` routes</li><li>`disable_threads`: Disable `/threads` routes</li><li>`disable_runs`: Disable `/runs` routes</li><li>`disable_store`: Disable `/store` routes</li><li>`disable_meta`: Disable `/ok`, `/info`, `/metrics`, and `/docs` routes</li><li>`cors`: CORS configuration with fields for `allow_origins`, `allow_methods`, `allow_headers`, etc.</li><li>`configurable_headers`: Define which request headers to exclude or include as a run's configurable values.</li></ul> |
+    | <span style="white-space: nowrap;">`http`</span>            | HTTP server configuration with the following fields: <ul><li>`app`: Path to custom Starlette/FastAPI app (e.g., `"./src/agent/webapp.py:app"`). See [custom routes guide](../../how-tos/http/custom_routes.md).</li><li>`disable_assistants`: Disable `/assistants` routes</li><li>`disable_threads`: Disable `/threads` routes</li><li>`disable_runs`: Disable `/runs` routes</li><li>`disable_store`: Disable `/store` routes</li><li>`disable_meta`: Disable `/ok`, `/info`, `/metrics`, and `/docs` routes</li><li>`disable_mcp`: Disable `/mcp` routes</li><li>`cors`: CORS configuration with fields for `allow_origins`, `allow_methods`, `allow_headers`, etc.</li><li>`configurable_headers`: Define which request headers to exclude or include as a run's configurable values.</li></ul> |
 
 === "JS"
 
@@ -128,7 +129,7 @@ The LangGraph CLI requires a JSON configuration file that follows this [schema](
         - `cohere:embed-english-v3.0`: 1024 
         - `cohere:embed-english-light-v3.0`: 384 
         - `cohere:embed-multilingual-v3.0`: 1024 
-        - `cohere:embed-multilingual-light-v3.0`: 384
+        - `cohere:embed-multilingual-light-v3.0`: 384 
 
     #### Semantic search with a custom embedding function
 
@@ -361,8 +362,8 @@ The LangGraph CLI requires a JSON configuration file that follows this [schema](
 
     **Options**
 
-    | Option               | Default          | Description                                                                                                                  |
-    | -------------------- | ---------------- | ---------------------------------------------------------------------------------------------------------------------------- |
+    | Option               | Default          | Description                                                                                                     |
+    | -------------------- | ---------------- | --------------------------------------------------------------------------------------------------------------- |
     | `--platform TEXT`    |                  | Target platform(s) to build the Docker image for. Example: `langgraph build --platform linux/amd64,linux/arm64`              |
     | `-t, --tag TEXT`     |                  | **Required**. Tag for the Docker image. Example: `langgraph build -t my-image`                                               |
     | `--pull / --no-pull` | `--pull`         | Build with latest remote Docker image. Use `--no-pull` for running the LangGraph Platform API server with locally built images. |
@@ -381,8 +382,8 @@ The LangGraph CLI requires a JSON configuration file that follows this [schema](
 
     **Options**
 
-    | Option               | Default          | Description                                                                                                                  |
-    | -------------------- | ---------------- | ---------------------------------------------------------------------------------------------------------------------------- |
+    | Option               | Default          | Description                                                                                                     |
+    | -------------------- | ---------------- | --------------------------------------------------------------------------------------------------------------- |
     | `--platform TEXT`    |                  | Target platform(s) to build the Docker image for. Example: `langgraph build --platform linux/amd64,linux/arm64`              |
     | `-t, --tag TEXT`     |                  | **Required**. Tag for the Docker image. Example: `langgraph build -t my-image`                                               |
     | `--no-pull`          |                  | Use locally built images. Defaults to `false` to build with latest remote Docker image.                                      |

docs/docs/cloud/reference/env_var.md

@@ -50,11 +50,10 @@ Set this environment variable to have a deployment send traces to a self-hosted
 
 ## `LANGSMITH_TRACING`
 
-!!! info "Only for Self-Hosted Data Plane, Self-Hosted Control Plane, and Standalone Container"
-    Disabling LangSmith tracing is only available for [Self-Hosted Data Plane](../../concepts/langgraph_self_hosted_data_plane.md), [Self-Hosted Control Plane](../../concepts/langgraph_self_hosted_control_plane.md), and [Standalone Container](../../concepts/langgraph_standalone_container.md) deployments.
-
 Set `LANGSMITH_TRACING` to `false` to disable tracing to LangSmith.
 
+Defaults to `true`.
+
 ## `LOG_LEVEL`
 
 Configure [log level](https://docs.python.org/3/library/logging.html#logging-levels). Defaults to `INFO`.

docs/docs/concepts/langgraph_cli.md

@@ -9,13 +9,18 @@ search:
 
 ## Installation
 
-The LangGraph CLI can be installed via pip:
+The LangGraph CLI can be installed via pip or [Homebrew](https://brew.sh/):
 
 === "pip" 
     ```bash
     pip install langgraph-cli
     ```
 
+=== "Homebrew"
+    ```bash
+    brew install langgraph-cli
+    ```
+
 ## Commands
 
 LangGraph CLI provides the following core functionality:

docs/docs/concepts/langgraph_control_plane.md

@@ -19,6 +19,7 @@ From the control plane UI, you can:
 - Update a deployment.
 - Update environment variables for a deployment.
 - View build and server logs of a deployment.
+- View deployment metrics such as CPU and memory usage.
 - Delete a deployment.
 
 The Control Plane UI is embedded in [LangSmith](https://docs.smith.langchain.com/langgraph_cloud).
@@ -88,6 +89,17 @@ Infrastructure for deployments and revisions are provisioned and deployed asynch
 
 The control plane and [LangGraph Data Plane](./langgraph_data_plane.md) "listener" application coordinate to achieve asynchronous deployments.
 
+### Monitoring
+
+After a deployment is ready, the control plane monitors the deployment and records various metrics, such as:
+
+- CPU and memory usage of the deployment.
+- Number of container restarts.
+- Number of replicas (this will increase with [autoscaling](../concepts/langgraph_data_plane.md#autoscaling)).
+- [Postgres](../concepts/langgraph_data_plane.md#postgres) CPU, memory usage, and disk usage.
+
+These metrics are displayed as charts in the Control Plane UI.
+
 ### LangSmith Integration
 
 A [LangSmith](https://docs.smith.langchain.com/) tracing project is automatically created for each deployment. The tracing project has the same name as the deployment. When creating a deployment, the `LANGCHAIN_TRACING` and `LANGSMITH_API_KEY`/`LANGCHAIN_API_KEY` environment variables do not need to be specified; they are set automatically by the control plane.

docs/docs/concepts/multi_agent.md

@@ -87,6 +87,7 @@ One of the most common agent types is a [tool-calling agent](../agents/overview.
 ```python
 from langchain_core.tools import tool
 
+@tool
 def transfer_to_bob():
     """Transfer to bob."""
     return Command(
@@ -414,4 +415,4 @@ There are two high-level approaches to achieve that:
 An agent might need to have a different state schema from the rest of the agents. For example, a search agent might only need to keep track of queries and retrieved documents. There are two ways to achieve this in LangGraph:
 
 - Define [subgraph](./subgraphs.md) agents with a separate state schema. If there are no shared state keys (channels) between the subgraph and the parent graph, it’s important to [add input / output transformations](../how-tos/subgraph.ipynb#different-state-schemas) so that the parent graph knows how to communicate with the subgraphs.
-- Define agent node functions with a [private input state schema](../how-tos/graph-api.ipynb/#pass-private-state-between-nodes) that is distinct from the overall graph state schema. This allows passing information that is only needed for executing that particular agent.
+- Define agent node functions with a [private input state schema](../how-tos/graph-api.ipynb/#pass-private-state-between-nodes) that is distinct from the overall graph state schema. This allows passing information that is only needed for executing that particular agent.

docs/docs/concepts/persistence.md

@@ -473,6 +473,23 @@ If the checkpointer is used with asynchronous graph execution (i.e. executing th
 When checkpointers save the graph state, they need to serialize the channel values in the state. This is done using serializer objects.
 `langgraph_checkpoint` defines [protocol][langgraph.checkpoint.serde.base.SerializerProtocol] for implementing serializers provides a default implementation ([JsonPlusSerializer][langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer]) that handles a wide variety of types, including LangChain and LangGraph primitives, datetimes, enums and more.
 
+#### Serialization with `pickle`
+
+The default serializer, [`JsonPlusSerializer`][langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer], uses ormsgpack and JSON under the hood, which is not suitable for all types of objects.
+
+If you want to fallback to pickle for objects not currently supported by our msgpack encoder (such as Pandas dataframes),
+you can use the `pickle_fallback` argument of the `JsonPlusSerializer`:
+
+```python
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
+
+# ... Define the graph ...
+graph.compile(
+    checkpointer=MemorySaver(serde=JsonPlusSerializer(pickle_fallback=True))
+)
+```
+
 #### Encryption
 
 Checkpointers can optionally encrypt all persisted state. To enable this, pass an instance of [`EncryptedSerializer`][langgraph.checkpoint.serde.encrypted.EncryptedSerializer] to the `serde` argument of any `BaseCheckpointSaver` implementation. The easiest way to create an encrypted serializer is via [`from_pycryptodome_aes`][langgraph.checkpoint.serde.encrypted.EncryptedSerializer.from_pycryptodome_aes], which reads the AES key from the `LANGGRAPH_AES_KEY` environment variable (or accepts a `key` argument):

docs/docs/how-tos/http/custom_routes.md

@@ -34,7 +34,7 @@ def read_root():
 
 ## Configure `langgraph.json`
 
-Add the following to your `langgraph.json` configuration file. Make sure the path points to the `app.py` file you created above.
+Add the following to your `langgraph.json` configuration file. Make sure the path points to the FastAPI application instance `app` in the `webapp.py` file you created above.
 
 ```json
 {
@@ -71,4 +71,4 @@ You can deploy this app as-is to LangGraph Platform or to your self-hosted platf
 
 ## Next steps
 
-Now that you've added a custom route to your deployment, you can use this same technique to further customize how your server behaves, such as defining custom [custom middleware](./custom_middleware.md) and [custom lifespan events](./custom_lifespan.md). 
+Now that you've added a custom route to your deployment, you can use this same technique to further customize how your server behaves, such as defining custom [custom middleware](./custom_middleware.md) and [custom lifespan events](./custom_lifespan.md). 

docs/docs/tutorials/extraction/retries.ipynb

@@ -89,7 +89,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": null,
+   "execution_count": 2,
    "id": "baf669a0-04ee-492d-80d8-8fcb658ed128",
    "metadata": {},
    "outputs": [],
@@ -313,8 +313,8 @@
     "\n",
     "    builder.add_edge(\"finalizer\", END)\n",
     "\n",
-    "    # These functions let the step be used in a\n",
-    "    # StateGraph with 'messages' as the key.\n",
+    "    # These functions let the step be used in a MessageGraph\n",
+    "    # or a StateGraph with 'messages' as the key.\n",
     "    def encode(x: Union[Sequence[AnyMessage], PromptValue]) -> dict:\n",
     "        \"\"\"Ensure the input is the correct format.\"\"\"\n",
     "        if isinstance(x, PromptValue):\n",

docs/docs/tutorials/workflows.md

@@ -648,7 +648,7 @@ With orchestrator-worker, an orchestrator breaks down a task and delegates each
     Because orchestrator-worker workflows are common, LangGraph **has the `Send` API to support this**. It lets you dynamically create worker nodes and send each one a specific input. Each worker has its own state, and all worker outputs are written to a *shared state key* that is accessible to the orchestrator graph. This gives the orchestrator access to all worker output and allows it to synthesize them into a final output. As you can see below, we iterate over a list of sections and `Send` each to a worker node. See further documentation [here](https://langchain-ai.github.io/langgraph/how-tos/map-reduce/) and [here](https://langchain-ai.github.io/langgraph/concepts/low_level/#send).
 
     ```python
-    from langgraph.constants import Send
+    from langgraph.types import Send
 
 
     # Graph state

docs/mkdocs.yml

@@ -365,16 +365,6 @@ markdown_extensions:
 hooks:
   - _scripts/notebook_hooks.py
 extra:
-  consent:
-    title: Cookie consent
-    actions:
-      - accept
-      - reject
-    description: >- 
-      We use cookies to recognize your repeated visits and preferences, as well
-      as to measure the effectiveness of our documentation and whether users
-      find what they're searching for. <strong>Clicking "Accept" makes our
-      documentation better. Thank you!</strong> ❤️
   social:
     - icon: fontawesome/brands/js
       link: https://langchain-ai.github.io/langgraphjs/

docs/overrides/language-switcher.css

@@ -0,0 +1,41 @@
+.lang-python,
+.lang-javascript {
+    display: none;
+}
+
+.language-switcher-global {
+    display: flex;
+    align-items: center;
+    padding-left: 0.5rem;
+    margin-right: 0.5rem;
+}
+
+/* Style the select to match the header */
+.language-switcher-global select {
+    appearance: none;
+    font: inherit;
+    border: none;
+    padding: 0.25rem 0.6rem;
+    cursor: pointer;
+    outline: none;
+    font-weight: bolder;
+}
+
+/* Hover/focus effect */
+.language-switcher-global select:hover,
+.language-switcher-global select:focus {
+    text-decoration: underline;
+}
+
+/* Theme-specific overrides */
+html[data-md-color-scheme="default"] .language-switcher-global select,
+html[data-md-color-scheme="default"] .language-switcher-global option {
+    color: #333;
+    background-color: transparent;
+}
+
+html[data-md-color-scheme="slate"] .language-switcher-global select,
+html[data-md-color-scheme="slate"] .language-switcher-global option {
+    color: #eee;
+    background-color: transparent;
+}

docs/overrides/language-switcher.js

@@ -0,0 +1,38 @@
+function applyLanguageSwitching() {
+    const selector = document.getElementById("global-language-selector");
+
+    const langBlocks = {
+        python: document.querySelectorAll(".lang-python"),
+        javascript: document.querySelectorAll(".lang-javascript"),
+    };
+
+    const setLanguage = (lang) => {
+        for (const [key, blocks] of Object.entries(langBlocks)) {
+            blocks.forEach((block) => {
+                block.style.display = key === lang ? "block" : "none";
+            });
+        }
+        localStorage.setItem("preferredLang", lang);
+    };
+
+    const saved = localStorage.getItem("preferredLang") || "python";
+
+    if (selector) {
+        selector.value = saved;
+        selector.addEventListener("change", (e) => setLanguage(e.target.value));
+    }
+
+    setLanguage(saved);
+}
+
+// Run on initial load
+document.addEventListener("DOMContentLoaded", applyLanguageSwitching);
+
+// Re-run after client-side navigation (MkDocs Material)
+document.addEventListener("pjax:success", applyLanguageSwitching);
+
+// Optional: observe DOM changes (e.g., for late-loaded content)
+if (window.MutationObserver) {
+    const observer = new MutationObserver(() => applyLanguageSwitching());
+    observer.observe(document.body, { childList: true, subtree: true });
+}

docs/overrides/partials/language-toggle.html

@@ -0,0 +1,6 @@
+<div class="md-header__button language-switcher-global" title="Select Language">
+  <select id="global-language-selector" aria-label="Select Language">
+    <option value="python">🐍 Python</option>
+    <option value="javascript">⚡️ JavaScript</option>
+  </select>
+</div>

docs/tests/unit_tests/test_conditional_rendering.py

@@ -0,0 +1,22 @@
+from _scripts.notebook_hooks import _apply_conditional_rendering
+
+
+CONDITIONAL_RENDERING = """
+above
+:::js
+js-content
+:::
+between
+:::python
+python-content
+:::
+below
+"""
+
+
+def test_conditional_rendering() -> None:
+    """Test logic for conditional rendering of content."""
+    output = _apply_conditional_rendering(CONDITIONAL_RENDERING, "js")
+    assert output.strip() == "above\njs-content\n\nbetween\n\nbelow"
+    output = _apply_conditional_rendering(CONDITIONAL_RENDERING, "python")
+    assert output.strip() == "above\n\nbetween\npython-content\n\nbelow"

libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

@@ -23,6 +23,7 @@ from langgraph.checkpoint.base import (
 )
 from langgraph.checkpoint.postgres import _internal
 from langgraph.checkpoint.postgres.base import BasePostgresSaver
+from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
 from langgraph.checkpoint.serde.base import SerializerProtocol
 
 Conn = _internal.Conn  # For backward compatibility
@@ -456,4 +457,4 @@ class PostgresSaver(BasePostgresSaver):
         )
 
 
-__all__ = ["PostgresSaver", "BasePostgresSaver", "Conn"]
+__all__ = ["PostgresSaver", "BasePostgresSaver", "ShallowPostgresSaver", "Conn"]

libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

@@ -23,6 +23,7 @@ from langgraph.checkpoint.base import (
 )
 from langgraph.checkpoint.postgres import _ainternal
 from langgraph.checkpoint.postgres.base import BasePostgresSaver
+from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
 from langgraph.checkpoint.serde.base import SerializerProtocol
 
 Conn = _ainternal.Conn  # For backward compatibility
@@ -559,4 +560,4 @@ class AsyncPostgresSaver(BasePostgresSaver):
         ).result()
 
 
-__all__ = ["AsyncPostgresSaver", "Conn"]
+__all__ = ["AsyncPostgresSaver", "AsyncShallowPostgresSaver", "Conn"]

libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py

@@ -168,7 +168,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
         checkpoint["channel_versions"][TASKS] = (
             max(checkpoint["channel_versions"].values())
             if checkpoint["channel_versions"]
-            else self.get_next_version(None)
+            else self.get_next_version(None, None)
         )
 
     def _load_blobs(
@@ -246,7 +246,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
             for idx, (channel, value) in enumerate(writes)
         ]
 
-    def get_next_version(self, current: str | None) -> str:
+    def get_next_version(self, current: str | None, channel: None) -> str:
         if current is None:
             current_v = 0
         elif isinstance(current, int):

libs/checkpoint-postgres/langgraph/checkpoint/postgres/shallow.py

@@ -0,0 +1,959 @@
+import asyncio
+import threading
+import warnings
+from collections.abc import AsyncIterator, Iterator, Sequence
+from contextlib import asynccontextmanager, contextmanager
+from typing import Any, Optional
+
+from langchain_core.runnables import RunnableConfig
+from psycopg import (
+    AsyncConnection,
+    AsyncCursor,
+    AsyncPipeline,
+    Capabilities,
+    Connection,
+    Cursor,
+    Pipeline,
+)
+from psycopg.rows import DictRow, dict_row
+from psycopg.types.json import Jsonb
+from psycopg_pool import AsyncConnectionPool, ConnectionPool
+
+from langgraph.checkpoint.base import (
+    WRITES_IDX_MAP,
+    ChannelVersions,
+    Checkpoint,
+    CheckpointMetadata,
+    CheckpointTuple,
+    get_checkpoint_metadata,
+)
+from langgraph.checkpoint.postgres import _ainternal, _internal
+from langgraph.checkpoint.postgres.base import BasePostgresSaver
+from langgraph.checkpoint.serde.base import SerializerProtocol
+from langgraph.checkpoint.serde.types import TASKS
+
+"""
+To add a new migration, add a new string to the MIGRATIONS list.
+The position of the migration in the list is the version number.
+"""
+MIGRATIONS = [
+    """CREATE TABLE IF NOT EXISTS checkpoint_migrations (
+    v INTEGER PRIMARY KEY
+);""",
+    """CREATE TABLE IF NOT EXISTS checkpoints (
+    thread_id TEXT NOT NULL,
+    checkpoint_ns TEXT NOT NULL DEFAULT '',
+    type TEXT,
+    checkpoint JSONB NOT NULL,
+    metadata JSONB NOT NULL DEFAULT '{}',
+    PRIMARY KEY (thread_id, checkpoint_ns)
+);""",
+    """CREATE TABLE IF NOT EXISTS checkpoint_blobs (
+    thread_id TEXT NOT NULL,
+    checkpoint_ns TEXT NOT NULL DEFAULT '',
+    channel TEXT NOT NULL,
+    type TEXT NOT NULL,
+    blob BYTEA,
+    PRIMARY KEY (thread_id, checkpoint_ns, channel)
+);""",
+    """CREATE TABLE IF NOT EXISTS checkpoint_writes (
+    thread_id TEXT NOT NULL,
+    checkpoint_ns TEXT NOT NULL DEFAULT '',
+    checkpoint_id TEXT NOT NULL,
+    task_id TEXT NOT NULL,
+    idx INTEGER NOT NULL,
+    channel TEXT NOT NULL,
+    type TEXT,
+    blob BYTEA NOT NULL,
+    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
+);""",
+    """
+    CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoints_thread_id_idx ON checkpoints(thread_id);
+    """,
+    """
+    CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoint_blobs_thread_id_idx ON checkpoint_blobs(thread_id);
+    """,
+    """
+    CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoint_writes_thread_id_idx ON checkpoint_writes(thread_id);
+    """,
+    """
+    ALTER TABLE checkpoint_writes ADD COLUMN task_path TEXT NOT NULL DEFAULT '';
+    """,
+]
+
+SELECT_SQL = f"""
+select
+    thread_id,
+    checkpoint,
+    checkpoint_ns,
+    metadata,
+    (
+        select array_agg(array[bl.channel::bytea, bl.type::bytea, bl.blob])
+        from jsonb_each_text(checkpoint -> 'channel_versions')
+        inner join checkpoint_blobs bl
+            on bl.thread_id = checkpoints.thread_id
+            and bl.checkpoint_ns = checkpoints.checkpoint_ns
+            and bl.channel = jsonb_each_text.key
+    ) as channel_values,
+    (
+        select
+        array_agg(array[cw.task_id::text::bytea, cw.channel::bytea, cw.type::bytea, cw.blob] order by cw.task_id, cw.idx)
+        from checkpoint_writes cw
+        where cw.thread_id = checkpoints.thread_id
+            and cw.checkpoint_ns = checkpoints.checkpoint_ns
+            and cw.checkpoint_id = (checkpoint->>'id')
+    ) as pending_writes,
+    (
+        select array_agg(array[cw.type::bytea, cw.blob] order by cw.task_path, cw.task_id, cw.idx)
+        from checkpoint_writes cw
+        where cw.thread_id = checkpoints.thread_id
+            and cw.checkpoint_ns = checkpoints.checkpoint_ns
+            and cw.channel = '{TASKS}'
+    ) as pending_sends
+from checkpoints """
+
+UPSERT_CHECKPOINT_BLOBS_SQL = """
+    INSERT INTO checkpoint_blobs (thread_id, checkpoint_ns, channel, type, blob)
+    VALUES (%s, %s, %s, %s, %s)
+    ON CONFLICT (thread_id, checkpoint_ns, channel) DO UPDATE SET
+        type = EXCLUDED.type,
+        blob = EXCLUDED.blob;
+"""
+
+UPSERT_CHECKPOINTS_SQL = """
+    INSERT INTO checkpoints (thread_id, checkpoint_ns, checkpoint, metadata)
+    VALUES (%s, %s, %s, %s)
+    ON CONFLICT (thread_id, checkpoint_ns)
+    DO UPDATE SET
+        checkpoint = EXCLUDED.checkpoint,
+        metadata = EXCLUDED.metadata;
+"""
+
+UPSERT_CHECKPOINT_WRITES_SQL = """
+    INSERT INTO checkpoint_writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, blob)
+    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
+    ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) DO UPDATE SET
+        channel = EXCLUDED.channel,
+        type = EXCLUDED.type,
+        blob = EXCLUDED.blob;
+"""
+
+INSERT_CHECKPOINT_WRITES_SQL = """
+    INSERT INTO checkpoint_writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, blob)
+    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
+    ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) DO NOTHING
+"""
+
+
+def _dump_blobs(
+    serde: SerializerProtocol,
+    thread_id: str,
+    checkpoint_ns: str,
+    values: dict[str, Any],
+    versions: ChannelVersions,
+) -> list[tuple[str, str, str, str, Optional[bytes]]]:
+    if not versions:
+        return []
+
+    return [
+        (
+            thread_id,
+            checkpoint_ns,
+            k,
+            *(serde.dumps_typed(values[k]) if k in values else ("empty", None)),
+        )
+        for k in versions
+    ]
+
+
+class ShallowPostgresSaver(BasePostgresSaver):
+    """A checkpoint saver that uses Postgres to store checkpoints.
+
+    This checkpointer ONLY stores the most recent checkpoint and does NOT retain any history.
+    It is meant to be a light-weight drop-in replacement for the PostgresSaver that
+    supports most of the LangGraph persistence functionality with the exception of time travel.
+    """
+
+    SELECT_SQL = SELECT_SQL
+    MIGRATIONS = MIGRATIONS
+    UPSERT_CHECKPOINT_BLOBS_SQL = UPSERT_CHECKPOINT_BLOBS_SQL
+    UPSERT_CHECKPOINTS_SQL = UPSERT_CHECKPOINTS_SQL
+    UPSERT_CHECKPOINT_WRITES_SQL = UPSERT_CHECKPOINT_WRITES_SQL
+    INSERT_CHECKPOINT_WRITES_SQL = INSERT_CHECKPOINT_WRITES_SQL
+
+    lock: threading.Lock
+
+    def __init__(
+        self,
+        conn: _internal.Conn,
+        pipe: Optional[Pipeline] = None,
+        serde: Optional[SerializerProtocol] = None,
+    ) -> None:
+        warnings.warn(
+            "ShallowPostgresSaver is deprecated as of version 2.0.20 and will be removed in 3.0.0. "
+            "Use PostgresSaver instead, and invoke the graph with `graph.invoke(..., checkpoint_during=False)`.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        super().__init__(serde=serde)
+        if isinstance(conn, ConnectionPool) and pipe is not None:
+            raise ValueError(
+                "Pipeline should be used only with a single Connection, not ConnectionPool."
+            )
+
+        self.conn = conn
+        self.pipe = pipe
+        self.lock = threading.Lock()
+        self.supports_pipeline = Capabilities().has_pipeline()
+
+    @classmethod
+    @contextmanager
+    def from_conn_string(
+        cls, conn_string: str, *, pipeline: bool = False
+    ) -> Iterator["ShallowPostgresSaver"]:
+        """Create a new ShallowPostgresSaver instance from a connection string.
+
+        Args:
+            conn_string: The Postgres connection info string.
+            pipeline: whether to use Pipeline
+
+        Returns:
+            ShallowPostgresSaver: A new ShallowPostgresSaver instance.
+        """
+        with Connection.connect(
+            conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
+        ) as conn:
+            if pipeline:
+                with conn.pipeline() as pipe:
+                    yield cls(conn, pipe)
+            else:
+                yield cls(conn)
+
+    def setup(self) -> None:
+        """Set up the checkpoint database asynchronously.
+
+        This method creates the necessary tables in the Postgres database if they don't
+        already exist and runs database migrations. It MUST be called directly by the user
+        the first time checkpointer is used.
+        """
+        with self._cursor() as cur:
+            cur.execute(self.MIGRATIONS[0])
+            results = cur.execute(
+                "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
+            )
+            row = results.fetchone()
+            if row is None:
+                version = -1
+            else:
+                version = row["v"]
+            for v, migration in zip(
+                range(version + 1, len(self.MIGRATIONS)),
+                self.MIGRATIONS[version + 1 :],
+            ):
+                cur.execute(migration)
+                cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
+        if self.pipe:
+            self.pipe.sync()
+
+    def list(
+        self,
+        config: Optional[RunnableConfig],
+        *,
+        filter: Optional[dict[str, Any]] = None,
+        before: Optional[RunnableConfig] = None,
+        limit: Optional[int] = None,
+    ) -> Iterator[CheckpointTuple]:
+        """List checkpoints from the database.
+
+        This method retrieves a list of checkpoint tuples from the Postgres database based
+        on the provided config. For ShallowPostgresSaver, this method returns a list with
+        ONLY the most recent checkpoint.
+        """
+        where, args = self._search_where(config, filter, before)
+        query = self.SELECT_SQL + where
+        if limit:
+            query += f" LIMIT {limit}"
+        with self._cursor() as cur:
+            cur.execute(self.SELECT_SQL + where, args, binary=True)
+            for value in cur:
+                checkpoint: Checkpoint = {
+                    **value["checkpoint"],
+                    "channel_values": self._load_blobs(value["channel_values"]),
+                    "pending_sends": [
+                        self.serde.loads_typed((t.decode(), v))
+                        for t, v in value["pending_sends"]
+                    ]
+                    if value["pending_sends"]
+                    else [],
+                }
+                yield CheckpointTuple(
+                    config={
+                        "configurable": {
+                            "thread_id": value["thread_id"],
+                            "checkpoint_ns": value["checkpoint_ns"],
+                            "checkpoint_id": checkpoint["id"],
+                        }
+                    },
+                    checkpoint=checkpoint,
+                    metadata=value["metadata"],
+                    pending_writes=self._load_writes(value["pending_writes"]),
+                )
+
+    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+        """Get a checkpoint tuple from the database.
+
+        This method retrieves a checkpoint tuple from the Postgres database based on the
+        provided config (matching the thread ID in the config).
+
+        Args:
+            config: The config to use for retrieving the checkpoint.
+
+        Returns:
+            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
+
+        Examples:
+
+            Basic:
+            >>> config = {"configurable": {"thread_id": "1"}}
+            >>> checkpoint_tuple = memory.get_tuple(config)
+            >>> print(checkpoint_tuple)
+            CheckpointTuple(...)
+
+            With timestamp:
+
+            >>> config = {
+            ...    "configurable": {
+            ...        "thread_id": "1",
+            ...        "checkpoint_ns": "",
+            ...        "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
+            ...    }
+            ... }
+            >>> checkpoint_tuple = memory.get_tuple(config)
+            >>> print(checkpoint_tuple)
+            CheckpointTuple(...)
+        """  # noqa
+        thread_id = config["configurable"]["thread_id"]
+        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
+        args = (thread_id, checkpoint_ns)
+        where = "WHERE thread_id = %s AND checkpoint_ns = %s"
+
+        with self._cursor() as cur:
+            cur.execute(
+                self.SELECT_SQL + where,
+                args,
+                binary=True,
+            )
+
+            for value in cur:
+                checkpoint: Checkpoint = {
+                    **value["checkpoint"],
+                    "channel_values": self._load_blobs(value["channel_values"]),
+                    "pending_sends": [
+                        self.serde.loads_typed((t.decode(), v))
+                        for t, v in value["pending_sends"]
+                    ]
+                    if value["pending_sends"]
+                    else [],
+                }
+                return CheckpointTuple(
+                    config={
+                        "configurable": {
+                            "thread_id": thread_id,
+                            "checkpoint_ns": checkpoint_ns,
+                            "checkpoint_id": checkpoint["id"],
+                        }
+                    },
+                    checkpoint=checkpoint,
+                    metadata=value["metadata"],
+                    pending_writes=self._load_writes(value["pending_writes"]),
+                )
+
+    def put(
+        self,
+        config: RunnableConfig,
+        checkpoint: Checkpoint,
+        metadata: CheckpointMetadata,
+        new_versions: ChannelVersions,
+    ) -> RunnableConfig:
+        """Save a checkpoint to the database.
+
+        This method saves a checkpoint to the Postgres database. The checkpoint is associated
+        with the provided config. For ShallowPostgresSaver, this method saves ONLY the most recent
+        checkpoint and overwrites a previous checkpoint, if it exists.
+
+        Args:
+            config: The config to associate with the checkpoint.
+            checkpoint: The checkpoint to save.
+            metadata: Additional metadata to save with the checkpoint.
+            new_versions: New channel versions as of this write.
+
+        Returns:
+            RunnableConfig: Updated configuration after storing the checkpoint.
+
+        Examples:
+
+            >>> from langgraph.checkpoint.postgres import ShallowPostgresSaver
+            >>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
+            >>> with ShallowPostgresSaver.from_conn_string(DB_URI) as memory:
+            >>>     config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
+            >>>     checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
+            >>>     saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
+            >>> print(saved_config)
+            {'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}
+        """
+        configurable = config["configurable"].copy()
+        thread_id = configurable.pop("thread_id")
+        checkpoint_ns = configurable.pop("checkpoint_ns")
+
+        copy = checkpoint.copy()
+        next_config = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": checkpoint_ns,
+                "checkpoint_id": checkpoint["id"],
+            }
+        }
+
+        with self._cursor(pipeline=True) as cur:
+            cur.execute(
+                """DELETE FROM checkpoint_writes
+                WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id NOT IN (%s, %s)""",
+                (
+                    thread_id,
+                    checkpoint_ns,
+                    checkpoint["id"],
+                    configurable.get("checkpoint_id", ""),
+                ),
+            )
+            cur.executemany(
+                self.UPSERT_CHECKPOINT_BLOBS_SQL,
+                _dump_blobs(
+                    self.serde,
+                    thread_id,
+                    checkpoint_ns,
+                    copy.pop("channel_values"),  # type: ignore[misc]
+                    new_versions,
+                ),
+            )
+            cur.execute(
+                self.UPSERT_CHECKPOINTS_SQL,
+                (
+                    thread_id,
+                    checkpoint_ns,
+                    Jsonb(copy),
+                    Jsonb(get_checkpoint_metadata(config, metadata)),
+                ),
+            )
+        return next_config
+
+    def put_writes(
+        self,
+        config: RunnableConfig,
+        writes: Sequence[tuple[str, Any]],
+        task_id: str,
+        task_path: str = "",
+    ) -> None:
+        """Store intermediate writes linked to a checkpoint.
+
+        This method saves intermediate writes associated with a checkpoint to the Postgres database.
+
+        Args:
+            config: Configuration of the related checkpoint.
+            writes: List of writes to store.
+            task_id: Identifier for the task creating the writes.
+        """
+        query = (
+            self.UPSERT_CHECKPOINT_WRITES_SQL
+            if all(w[0] in WRITES_IDX_MAP for w in writes)
+            else self.INSERT_CHECKPOINT_WRITES_SQL
+        )
+        with self._cursor(pipeline=True) as cur:
+            cur.executemany(
+                query,
+                self._dump_writes(
+                    config["configurable"]["thread_id"],
+                    config["configurable"]["checkpoint_ns"],
+                    config["configurable"]["checkpoint_id"],
+                    task_id,
+                    task_path,
+                    writes,
+                ),
+            )
+
+    @contextmanager
+    def _cursor(self, *, pipeline: bool = False) -> Iterator[Cursor[DictRow]]:
+        """Create a database cursor as a context manager.
+
+        Args:
+            pipeline: whether to use pipeline for the DB operations inside the context manager.
+                Will be applied regardless of whether the ShallowPostgresSaver instance was initialized with a pipeline.
+                If pipeline mode is not supported, will fall back to using transaction context manager.
+        """
+        with _internal.get_connection(self.conn) as conn:
+            if self.pipe:
+                # a connection in pipeline mode can be used concurrently
+                # in multiple threads/coroutines, but only one cursor can be
+                # used at a time
+                try:
+                    with conn.cursor(binary=True, row_factory=dict_row) as cur:
+                        yield cur
+                finally:
+                    if pipeline:
+                        self.pipe.sync()
+            elif pipeline:
+                # a connection not in pipeline mode can only be used by one
+                # thread/coroutine at a time, so we acquire a lock
+                if self.supports_pipeline:
+                    with (
+                        self.lock,
+                        conn.pipeline(),
+                        conn.cursor(binary=True, row_factory=dict_row) as cur,
+                    ):
+                        yield cur
+                else:
+                    # Use connection's transaction context manager when pipeline mode not supported
+                    with (
+                        self.lock,
+                        conn.transaction(),
+                        conn.cursor(binary=True, row_factory=dict_row) as cur,
+                    ):
+                        yield cur
+            else:
+                with self.lock, conn.cursor(binary=True, row_factory=dict_row) as cur:
+                    yield cur
+
+
+class AsyncShallowPostgresSaver(BasePostgresSaver):
+    """A checkpoint saver that uses Postgres to store checkpoints asynchronously.
+
+    This checkpointer ONLY stores the most recent checkpoint and does NOT retain any history.
+    It is meant to be a light-weight drop-in replacement for the AsyncPostgresSaver that
+    supports most of the LangGraph persistence functionality with the exception of time travel.
+    """
+
+    SELECT_SQL = SELECT_SQL
+    MIGRATIONS = MIGRATIONS
+    UPSERT_CHECKPOINT_BLOBS_SQL = UPSERT_CHECKPOINT_BLOBS_SQL
+    UPSERT_CHECKPOINTS_SQL = UPSERT_CHECKPOINTS_SQL
+    UPSERT_CHECKPOINT_WRITES_SQL = UPSERT_CHECKPOINT_WRITES_SQL
+    INSERT_CHECKPOINT_WRITES_SQL = INSERT_CHECKPOINT_WRITES_SQL
+    lock: asyncio.Lock
+
+    def __init__(
+        self,
+        conn: _ainternal.Conn,
+        pipe: Optional[AsyncPipeline] = None,
+        serde: Optional[SerializerProtocol] = None,
+    ) -> None:
+        warnings.warn(
+            "AsyncShallowPostgresSaver is deprecated as of version 2.0.20 and will be removed in 3.0.0. "
+            "Use AsyncPostgresSaver instead, and invoke the graph with `await graph.ainvoke(..., checkpoint_during=False)`.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        super().__init__(serde=serde)
+        if isinstance(conn, AsyncConnectionPool) and pipe is not None:
+            raise ValueError(
+                "Pipeline should be used only with a single AsyncConnection, not AsyncConnectionPool."
+            )
+
+        self.conn = conn
+        self.pipe = pipe
+        self.lock = asyncio.Lock()
+        self.loop = asyncio.get_running_loop()
+        self.supports_pipeline = Capabilities().has_pipeline()
+
+    @classmethod
+    @asynccontextmanager
+    async def from_conn_string(
+        cls,
+        conn_string: str,
+        *,
+        pipeline: bool = False,
+        serde: Optional[SerializerProtocol] = None,
+    ) -> AsyncIterator["AsyncShallowPostgresSaver"]:
+        """Create a new AsyncShallowPostgresSaver instance from a connection string.
+
+        Args:
+            conn_string: The Postgres connection info string.
+            pipeline: whether to use AsyncPipeline
+
+        Returns:
+            AsyncShallowPostgresSaver: A new AsyncShallowPostgresSaver instance.
+        """
+        async with await AsyncConnection.connect(
+            conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
+        ) as conn:
+            if pipeline:
+                async with conn.pipeline() as pipe:
+                    yield cls(conn=conn, pipe=pipe, serde=serde)
+            else:
+                yield cls(conn=conn, serde=serde)
+
+    async def setup(self) -> None:
+        """Set up the checkpoint database asynchronously.
+
+        This method creates the necessary tables in the Postgres database if they don't
+        already exist and runs database migrations. It MUST be called directly by the user
+        the first time checkpointer is used.
+        """
+        async with self._cursor() as cur:
+            await cur.execute(self.MIGRATIONS[0])
+            results = await cur.execute(
+                "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
+            )
+            row = await results.fetchone()
+            if row is None:
+                version = -1
+            else:
+                version = row["v"]
+            for v, migration in zip(
+                range(version + 1, len(self.MIGRATIONS)),
+                self.MIGRATIONS[version + 1 :],
+            ):
+                await cur.execute(migration)
+                await cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
+        if self.pipe:
+            await self.pipe.sync()
+
+    async def alist(
+        self,
+        config: Optional[RunnableConfig],
+        *,
+        filter: Optional[dict[str, Any]] = None,
+        before: Optional[RunnableConfig] = None,
+        limit: Optional[int] = None,
+    ) -> AsyncIterator[CheckpointTuple]:
+        """List checkpoints from the database asynchronously.
+
+        This method retrieves a list of checkpoint tuples from the Postgres database based
+        on the provided config. For ShallowPostgresSaver, this method returns a list with
+        ONLY the most recent checkpoint.
+        """
+        where, args = self._search_where(config, filter, before)
+        query = self.SELECT_SQL + where
+        if limit:
+            query += f" LIMIT {limit}"
+        async with self._cursor() as cur:
+            await cur.execute(self.SELECT_SQL + where, args, binary=True)
+            async for value in cur:
+                checkpoint: Checkpoint = {
+                    **value["checkpoint"],
+                    "channel_values": self._load_blobs(value["channel_values"]),
+                    "pending_sends": [
+                        self.serde.loads_typed((t.decode(), v))
+                        for t, v in value["pending_sends"]
+                    ]
+                    if value["pending_sends"]
+                    else [],
+                }
+                yield CheckpointTuple(
+                    config={
+                        "configurable": {
+                            "thread_id": value["thread_id"],
+                            "checkpoint_ns": value["checkpoint_ns"],
+                            "checkpoint_id": checkpoint["id"],
+                        }
+                    },
+                    checkpoint=checkpoint,
+                    metadata=value["metadata"],
+                    pending_writes=await asyncio.to_thread(
+                        self._load_writes, value["pending_writes"]
+                    ),
+                )
+
+    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+        """Get a checkpoint tuple from the database asynchronously.
+
+        This method retrieves a checkpoint tuple from the Postgres database based on the
+        provided config (matching the thread ID in the config).
+
+        Args:
+            config: The config to use for retrieving the checkpoint.
+
+        Returns:
+            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
+        """
+        thread_id = config["configurable"]["thread_id"]
+        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
+        args = (thread_id, checkpoint_ns)
+        where = "WHERE thread_id = %s AND checkpoint_ns = %s"
+
+        async with self._cursor() as cur:
+            await cur.execute(
+                self.SELECT_SQL + where,
+                args,
+                binary=True,
+            )
+
+            async for value in cur:
+                checkpoint: Checkpoint = {
+                    **value["checkpoint"],
+                    "channel_values": self._load_blobs(value["channel_values"]),
+                    "pending_sends": [
+                        self.serde.loads_typed((t.decode(), v))
+                        for t, v in value["pending_sends"]
+                    ]
+                    if value["pending_sends"]
+                    else [],
+                }
+                return CheckpointTuple(
+                    config={
+                        "configurable": {
+                            "thread_id": thread_id,
+                            "checkpoint_ns": checkpoint_ns,
+                            "checkpoint_id": checkpoint["id"],
+                        }
+                    },
+                    checkpoint=checkpoint,
+                    metadata=value["metadata"],
+                    pending_writes=await asyncio.to_thread(
+                        self._load_writes, value["pending_writes"]
+                    ),
+                )
+
+    async def aput(
+        self,
+        config: RunnableConfig,
+        checkpoint: Checkpoint,
+        metadata: CheckpointMetadata,
+        new_versions: ChannelVersions,
+    ) -> RunnableConfig:
+        """Save a checkpoint to the database asynchronously.
+
+        This method saves a checkpoint to the Postgres database. The checkpoint is associated
+        with the provided config. For AsyncShallowPostgresSaver, this method saves ONLY the most recent
+        checkpoint and overwrites a previous checkpoint, if it exists.
+
+        Args:
+            config: The config to associate with the checkpoint.
+            checkpoint: The checkpoint to save.
+            metadata: Additional metadata to save with the checkpoint.
+            new_versions: New channel versions as of this write.
+
+        Returns:
+            RunnableConfig: Updated configuration after storing the checkpoint.
+        """
+        configurable = config["configurable"].copy()
+        thread_id = configurable.pop("thread_id")
+        checkpoint_ns = configurable.pop("checkpoint_ns")
+
+        copy = checkpoint.copy()
+        next_config = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": checkpoint_ns,
+                "checkpoint_id": checkpoint["id"],
+            }
+        }
+
+        async with self._cursor(pipeline=True) as cur:
+            await cur.execute(
+                """DELETE FROM checkpoint_writes
+                WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id NOT IN (%s, %s)""",
+                (
+                    thread_id,
+                    checkpoint_ns,
+                    checkpoint["id"],
+                    configurable.get("checkpoint_id", ""),
+                ),
+            )
+            await cur.executemany(
+                self.UPSERT_CHECKPOINT_BLOBS_SQL,
+                _dump_blobs(
+                    self.serde,
+                    thread_id,
+                    checkpoint_ns,
+                    copy.pop("channel_values"),  # type: ignore[misc]
+                    new_versions,
+                ),
+            )
+            await cur.execute(
+                self.UPSERT_CHECKPOINTS_SQL,
+                (
+                    thread_id,
+                    checkpoint_ns,
+                    Jsonb(copy),
+                    Jsonb(get_checkpoint_metadata(config, metadata)),
+                ),
+            )
+        return next_config
+
+    async def aput_writes(
+        self,
+        config: RunnableConfig,
+        writes: Sequence[tuple[str, Any]],
+        task_id: str,
+        task_path: str = "",
+    ) -> None:
+        """Store intermediate writes linked to a checkpoint asynchronously.
+
+        This method saves intermediate writes associated with a checkpoint to the database.
+
+        Args:
+            config: Configuration of the related checkpoint.
+            writes: List of writes to store, each as (channel, value) pair.
+            task_id: Identifier for the task creating the writes.
+        """
+        query = (
+            self.UPSERT_CHECKPOINT_WRITES_SQL
+            if all(w[0] in WRITES_IDX_MAP for w in writes)
+            else self.INSERT_CHECKPOINT_WRITES_SQL
+        )
+        params = await asyncio.to_thread(
+            self._dump_writes,
+            config["configurable"]["thread_id"],
+            config["configurable"]["checkpoint_ns"],
+            config["configurable"]["checkpoint_id"],
+            task_id,
+            task_path,
+            writes,
+        )
+        async with self._cursor(pipeline=True) as cur:
+            await cur.executemany(query, params)
+
+    @asynccontextmanager
+    async def _cursor(
+        self, *, pipeline: bool = False
+    ) -> AsyncIterator[AsyncCursor[DictRow]]:
+        """Create a database cursor as a context manager.
+
+        Args:
+            pipeline: whether to use pipeline for the DB operations inside the context manager.
+                Will be applied regardless of whether the AsyncShallowPostgresSaver instance was initialized with a pipeline.
+                If pipeline mode is not supported, will fall back to using transaction context manager.
+        """
+        async with _ainternal.get_connection(self.conn) as conn:
+            if self.pipe:
+                # a connection in pipeline mode can be used concurrently
+                # in multiple threads/coroutines, but only one cursor can be
+                # used at a time
+                try:
+                    async with conn.cursor(binary=True, row_factory=dict_row) as cur:
+                        yield cur
+                finally:
+                    if pipeline:
+                        await self.pipe.sync()
+            elif pipeline:
+                # a connection not in pipeline mode can only be used by one
+                # thread/coroutine at a time, so we acquire a lock
+                if self.supports_pipeline:
+                    async with (
+                        self.lock,
+                        conn.pipeline(),
+                        conn.cursor(binary=True, row_factory=dict_row) as cur,
+                    ):
+                        yield cur
+                else:
+                    # Use connection's transaction context manager when pipeline mode not supported
+                    async with (
+                        self.lock,
+                        conn.transaction(),
+                        conn.cursor(binary=True, row_factory=dict_row) as cur,
+                    ):
+                        yield cur
+            else:
+                async with (
+                    self.lock,
+                    conn.cursor(binary=True, row_factory=dict_row) as cur,
+                ):
+                    yield cur
+
+    def list(
+        self,
+        config: Optional[RunnableConfig],
+        *,
+        filter: Optional[dict[str, Any]] = None,
+        before: Optional[RunnableConfig] = None,
+        limit: Optional[int] = None,
+    ) -> Iterator[CheckpointTuple]:
+        """List checkpoints from the database.
+
+        This method retrieves a list of checkpoint tuples from the Postgres database based
+        on the provided config. For ShallowPostgresSaver, this method returns a list with
+        ONLY the most recent checkpoint.
+        """
+        aiter_ = self.alist(config, filter=filter, before=before, limit=limit)
+        while True:
+            try:
+                yield asyncio.run_coroutine_threadsafe(
+                    anext(aiter_),  # noqa: F821
+                    self.loop,
+                ).result()
+            except StopAsyncIteration:
+                break
+
+    def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+        """Get a checkpoint tuple from the database.
+
+        This method retrieves a checkpoint tuple from the Postgres database based on the
+        provided config (matching the thread ID in the config).
+
+        Args:
+            config: The config to use for retrieving the checkpoint.
+
+        Returns:
+            Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
+        """
+        try:
+            # check if we are in the main thread, only bg threads can block
+            # we don't check in other methods to avoid the overhead
+            if asyncio.get_running_loop() is self.loop:
+                raise asyncio.InvalidStateError(
+                    "Synchronous calls to AsyncShallowPostgresSaver are only allowed from a "
+                    "different thread. From the main thread, use the async interface."
+                    "For example, use `await checkpointer.aget_tuple(...)` or `await "
+                    "graph.ainvoke(...)`."
+                )
+        except RuntimeError:
+            pass
+        return asyncio.run_coroutine_threadsafe(
+            self.aget_tuple(config), self.loop
+        ).result()
+
+    def put(
+        self,
+        config: RunnableConfig,
+        checkpoint: Checkpoint,
+        metadata: CheckpointMetadata,
+        new_versions: ChannelVersions,
+    ) -> RunnableConfig:
+        """Save a checkpoint to the database.
+
+        This method saves a checkpoint to the Postgres database. The checkpoint is associated
+        with the provided config. For AsyncShallowPostgresSaver, this method saves ONLY the most recent
+        checkpoint and overwrites a previous checkpoint, if it exists.
+
+        Args:
+            config: The config to associate with the checkpoint.
+            checkpoint: The checkpoint to save.
+            metadata: Additional metadata to save with the checkpoint.
+            new_versions: New channel versions as of this write.
+
+        Returns:
+            RunnableConfig: Updated configuration after storing the checkpoint.
+        """
+        return asyncio.run_coroutine_threadsafe(
+            self.aput(config, checkpoint, metadata, new_versions), self.loop
+        ).result()
+
+    def put_writes(
+        self,
+        config: RunnableConfig,
+        writes: Sequence[tuple[str, Any]],
+        task_id: str,
+        task_path: str = "",
+    ) -> None:
+        """Store intermediate writes linked to a checkpoint.
+
+        This method saves intermediate writes associated with a checkpoint to the database.
+
+        Args:
+            config: Configuration of the related checkpoint.
+            writes: List of writes to store, each as (channel, value) pair.
+            task_id: Identifier for the task creating the writes.
+            task_path: Path of the task creating the writes.
+        """
+        return asyncio.run_coroutine_threadsafe(
+            self.aput_writes(config, writes, task_id, task_path), self.loop
+        ).result()

libs/checkpoint-postgres/tests/checkpoint_utils.py

@@ -1,53 +0,0 @@
-from __future__ import annotations
-
-from collections.abc import Mapping
-from datetime import datetime, timezone
-from typing import Any, Protocol
-
-from langgraph.checkpoint.base import Checkpoint, EmptyChannelError
-from langgraph.checkpoint.base.id import uuid6
-
-
-class ChannelProtocol(Protocol):
-    def checkpoint(self) -> Any | None: ...
-
-
-def empty_checkpoint() -> Checkpoint:
-    return Checkpoint(
-        v=1,
-        id=str(uuid6(clock_seq=-2)),
-        ts=datetime.now(timezone.utc).isoformat(),
-        channel_values={},
-        channel_versions={},
-        versions_seen={},
-    )
-
-
-def create_checkpoint(
-    checkpoint: Checkpoint,
-    channels: Mapping[str, ChannelProtocol] | None,
-    step: int,
-    *,
-    id: str | None = None,
-) -> Checkpoint:
-    """Create a checkpoint for the given channels."""
-    ts = datetime.now(timezone.utc).isoformat()
-    if channels is None:
-        values = checkpoint["channel_values"]
-    else:
-        values = {}
-        for k, v in channels.items():
-            if k not in checkpoint["channel_versions"]:
-                continue
-            try:
-                values[k] = v.checkpoint()
-            except EmptyChannelError:
-                pass
-    return Checkpoint(
-        v=1,
-        ts=ts,
-        id=id or str(uuid6(clock_seq=step)),
-        channel_values=values,
-        channel_versions=checkpoint["channel_versions"],
-        versions_seen=checkpoint["versions_seen"],
-    )

libs/checkpoint-postgres/tests/test_async.py

@@ -14,10 +14,14 @@ from langgraph.checkpoint.base import (
     EXCLUDED_METADATA_KEYS,
     Checkpoint,
     CheckpointMetadata,
+    create_checkpoint,
+    empty_checkpoint,
+)
+from langgraph.checkpoint.postgres.aio import (
+    AsyncPostgresSaver,
+    AsyncShallowPostgresSaver,
 )
-from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
 from langgraph.checkpoint.serde.types import TASKS
-from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
 from tests.conftest import DEFAULT_POSTGRES_URI
 
 
@@ -108,11 +112,41 @@ async def _base_saver():
             await conn.execute(f"DROP DATABASE {database}")
 
 
+@asynccontextmanager
+async def _shallow_saver():
+    """Fixture for shallow connection mode testing."""
+    database = f"test_{uuid4().hex[:16]}"
+    # create unique db
+    async with await AsyncConnection.connect(
+        DEFAULT_POSTGRES_URI, autocommit=True
+    ) as conn:
+        await conn.execute(f"CREATE DATABASE {database}")
+    try:
+        async with await AsyncConnection.connect(
+            DEFAULT_POSTGRES_URI + database,
+            autocommit=True,
+            prepare_threshold=0,
+            row_factory=dict_row,
+        ) as conn:
+            checkpointer = AsyncShallowPostgresSaver(conn)
+            await checkpointer.setup()
+            yield checkpointer
+    finally:
+        # drop unique db
+        async with await AsyncConnection.connect(
+            DEFAULT_POSTGRES_URI, autocommit=True
+        ) as conn:
+            await conn.execute(f"DROP DATABASE {database}")
+
+
 @asynccontextmanager
 async def _saver(name: str):
     if name == "base":
         async with _base_saver() as saver:
             yield saver
+    elif name == "shallow":
+        async with _shallow_saver() as saver:
+            yield saver
     elif name == "pool":
         async with _pool_saver() as saver:
             yield saver
@@ -172,7 +206,7 @@ def test_data():
     }
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
 async def test_combined_metadata(saver_name: str, test_data) -> None:
     async with _saver(saver_name) as saver:
         config = {
@@ -194,12 +228,11 @@ async def test_combined_metadata(saver_name: str, test_data) -> None:
         checkpoint = await saver.aget_tuple(config)
         assert checkpoint.metadata == {
             **metadata,
-            "thread_id": "thread-2",
             "run_id": "my_run_id",
         }
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
 async def test_asearch(saver_name: str, test_data) -> None:
     async with _saver(saver_name) as saver:
         configs = test_data["configs"]
@@ -250,7 +283,7 @@ async def test_asearch(saver_name: str, test_data) -> None:
         } == {"", "inner"}
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
 async def test_null_chars(saver_name: str, test_data) -> None:
     async with _saver(saver_name) as saver:
         config = await saver.aput(

libs/checkpoint-postgres/tests/test_sync.py

@@ -15,10 +15,11 @@ from langgraph.checkpoint.base import (
     EXCLUDED_METADATA_KEYS,
     Checkpoint,
     CheckpointMetadata,
+    create_checkpoint,
+    empty_checkpoint,
 )
-from langgraph.checkpoint.postgres import PostgresSaver
+from langgraph.checkpoint.postgres import PostgresSaver, ShallowPostgresSaver
 from langgraph.checkpoint.serde.types import TASKS
-from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
 from tests.conftest import DEFAULT_POSTGRES_URI
 
 
@@ -97,11 +98,37 @@ def _base_saver():
             conn.execute(f"DROP DATABASE {database}")
 
 
+@contextmanager
+def _shallow_saver():
+    """Fixture for regular connection mode testing with a shallow checkpointer."""
+    database = f"test_{uuid4().hex[:16]}"
+    # create unique db
+    with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
+        conn.execute(f"CREATE DATABASE {database}")
+    try:
+        with Connection.connect(
+            DEFAULT_POSTGRES_URI + database,
+            autocommit=True,
+            prepare_threshold=0,
+            row_factory=dict_row,
+        ) as conn:
+            checkpointer = ShallowPostgresSaver(conn)
+            checkpointer.setup()
+            yield checkpointer
+    finally:
+        # drop unique db
+        with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
+            conn.execute(f"DROP DATABASE {database}")
+
+
 @contextmanager
 def _saver(name: str):
     if name == "base":
         with _base_saver() as saver:
             yield saver
+    elif name == "shallow":
+        with _shallow_saver() as saver:
+            yield saver
     elif name == "pool":
         with _pool_saver() as saver:
             yield saver
@@ -161,7 +188,7 @@ def test_data():
     }
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
 def test_combined_metadata(saver_name: str, test_data) -> None:
     with _saver(saver_name) as saver:
         config = {
@@ -183,12 +210,11 @@ def test_combined_metadata(saver_name: str, test_data) -> None:
         checkpoint = saver.get_tuple(config)
         assert checkpoint.metadata == {
             **metadata,
-            "thread_id": "thread-2",
             "run_id": "my_run_id",
         }
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
 def test_search(saver_name: str, test_data) -> None:
     with _saver(saver_name) as saver:
         configs = test_data["configs"]
@@ -237,7 +263,7 @@ def test_search(saver_name: str, test_data) -> None:
         } == {"", "inner"}
 
 
-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
 def test_null_chars(saver_name: str, test_data) -> None:
     with _saver(saver_name) as saver:
         config = saver.put(

libs/checkpoint-postgres/uv.lock

@@ -308,7 +308,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "2.0.26"
+version = "2.1.0"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -327,6 +327,8 @@ dev = [
     { name = "dataclasses-json" },
     { name = "mypy" },
     { name = "numpy" },
+    { name = "pandas" },
+    { name = "pandas-stubs", specifier = ">=2.2.2.240807" },
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-mock" },

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

@@ -536,7 +536,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
         """
         raise NotImplementedError(_AIO_ERROR_MSG)
 
-    def get_next_version(self, current: str | None) -> str:
+    def get_next_version(self, current: str | None, channel: None) -> str:
         """Generate the next version ID for a channel.
 
         This method creates a new version identifier for a channel based on its current version.

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

@@ -591,7 +591,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
             )
             await self.conn.commit()
 
-    def get_next_version(self, current: str | None) -> str:
+    def get_next_version(self, current: str | None, channel: None) -> str:
         """Generate the next version ID for a channel.
 
         This method creates a new version identifier for a channel based on its current version.

libs/checkpoint-sqlite/tests/checkpoint_utils.py

@@ -1,53 +0,0 @@
-from __future__ import annotations
-
-from collections.abc import Mapping
-from datetime import datetime, timezone
-from typing import Any, Protocol
-
-from langgraph.checkpoint.base import Checkpoint, EmptyChannelError
-from langgraph.checkpoint.base.id import uuid6
-
-
-class ChannelProtocol(Protocol):
-    def checkpoint(self) -> Any | None: ...
-
-
-def empty_checkpoint() -> Checkpoint:
-    return Checkpoint(
-        v=1,
-        id=str(uuid6(clock_seq=-2)),
-        ts=datetime.now(timezone.utc).isoformat(),
-        channel_values={},
-        channel_versions={},
-        versions_seen={},
-    )
-
-
-def create_checkpoint(
-    checkpoint: Checkpoint,
-    channels: Mapping[str, ChannelProtocol] | None,
-    step: int,
-    *,
-    id: str | None = None,
-) -> Checkpoint:
-    """Create a checkpoint for the given channels."""
-    ts = datetime.now(timezone.utc).isoformat()
-    if channels is None:
-        values = checkpoint["channel_values"]
-    else:
-        values = {}
-        for k, v in channels.items():
-            if k not in checkpoint["channel_versions"]:
-                continue
-            try:
-                values[k] = v.checkpoint()
-            except EmptyChannelError:
-                pass
-    return Checkpoint(
-        v=1,
-        ts=ts,
-        id=id or str(uuid6(clock_seq=step)),
-        channel_values=values,
-        channel_versions=checkpoint["channel_versions"],
-        versions_seen=checkpoint["versions_seen"],
-    )

libs/checkpoint-sqlite/tests/test_aiosqlite.py

@@ -6,9 +6,10 @@ from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
     Checkpoint,
     CheckpointMetadata,
+    create_checkpoint,
+    empty_checkpoint,
 )
 from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
-from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
 
 
 class TestAsyncSqliteSaver:
@@ -70,7 +71,6 @@ class TestAsyncSqliteSaver:
             checkpoint = await saver.aget_tuple(config)
             assert checkpoint is not None and checkpoint.metadata == {
                 **self.metadata_2,
-                "thread_id": "thread-2",
                 "run_id": "my_run_id",
             }
 
@@ -91,18 +91,11 @@ class TestAsyncSqliteSaver:
 
             search_results_1 = [c async for c in saver.alist(None, filter=query_1)]
             assert len(search_results_1) == 1
-            assert search_results_1[0].metadata == {
-                "thread_id": "thread-1",
-                "thread_ts": "1",
-                **self.metadata_1,
-            }
+            assert search_results_1[0].metadata == self.metadata_1
 
             search_results_2 = [c async for c in saver.alist(None, filter=query_2)]
             assert len(search_results_2) == 1
-            assert search_results_2[0].metadata == {
-                "thread_id": "thread-2",
-                **self.metadata_2,
-            }
+            assert search_results_2[0].metadata == self.metadata_2
 
             search_results_3 = [c async for c in saver.alist(None, filter=query_3)]
             assert len(search_results_3) == 3

libs/checkpoint-sqlite/tests/test_sqlite.py

@@ -6,10 +6,11 @@ from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
     Checkpoint,
     CheckpointMetadata,
+    create_checkpoint,
+    empty_checkpoint,
 )
 from langgraph.checkpoint.sqlite import SqliteSaver
 from langgraph.checkpoint.sqlite.utils import _metadata_predicate, search_where
-from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
 
 
 class TestSqliteSaver:
@@ -71,7 +72,6 @@ class TestSqliteSaver:
             checkpoint = saver.get_tuple(config)
             assert checkpoint is not None and checkpoint.metadata == {
                 **self.metadata_2,
-                "thread_id": "thread-2",
                 "run_id": "my_run_id",
             }
 
@@ -94,18 +94,11 @@ class TestSqliteSaver:
 
             search_results_1 = list(saver.list(None, filter=query_1))
             assert len(search_results_1) == 1
-            assert search_results_1[0].metadata == {
-                "thread_id": "thread-1",
-                "thread_ts": "1",
-                **self.metadata_1,
-            }
+            assert search_results_1[0].metadata == self.metadata_1
 
             search_results_2 = list(saver.list(None, filter=query_2))
             assert len(search_results_2) == 1
-            assert search_results_2[0].metadata == {
-                "thread_id": "thread-2",
-                **self.metadata_2,
-            }
+            assert search_results_2[0].metadata == self.metadata_2
 
             search_results_3 = list(saver.list(None, filter=query_3))
             assert len(search_results_3) == 3

libs/checkpoint-sqlite/uv.lock

@@ -320,7 +320,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "2.0.26"
+version = "2.1.0"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -339,6 +339,8 @@ dev = [
     { name = "dataclasses-json" },
     { name = "mypy" },
     { name = "numpy" },
+    { name = "pandas" },
+    { name = "pandas-stubs", specifier = ">=2.2.2.240807" },
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-mock" },

libs/checkpoint/langgraph/checkpoint/base/__init__.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import AsyncIterator, Iterator, Sequence
+from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
 from typing import (  # noqa: UP035
     Any,
     Generic,
@@ -13,6 +13,7 @@ from typing import (  # noqa: UP035
 
 from langchain_core.runnables import RunnableConfig
 
+from langgraph.checkpoint.base.id import uuid6
 from langgraph.checkpoint.serde.base import SerializerProtocol, maybe_add_typed_methods
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
 from langgraph.checkpoint.serde.types import (
@@ -20,6 +21,7 @@ from langgraph.checkpoint.serde.types import (
     INTERRUPT,
     RESUME,
     SCHEDULED,
+    ChannelProtocol,
 )
 
 V = TypeVar("V", int, float, str)
@@ -89,6 +91,7 @@ def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
         channel_values=checkpoint["channel_values"].copy(),
         channel_versions=checkpoint["channel_versions"].copy(),
         versions_seen={k: v.copy() for k, v in checkpoint["versions_seen"].items()},
+        pending_sends=checkpoint.get("pending_sends", []).copy(),
     )
 
 
@@ -125,6 +128,15 @@ class BaseCheckpointSaver(Generic[V]):
     ) -> None:
         self.serde = maybe_add_typed_methods(serde or self.serde)
 
+    @property
+    def config_specs(self) -> list:
+        """Define the configuration options for the checkpoint saver.
+
+        Returns:
+            list: List of configuration field specs.
+        """
+        return []
+
     def get(self, config: RunnableConfig) -> Checkpoint | None:
         """Fetch a checkpoint using the given configuration.
 
@@ -334,7 +346,7 @@ class BaseCheckpointSaver(Generic[V]):
         """
         raise NotImplementedError
 
-    def get_next_version(self, current: V | None) -> V:
+    def get_next_version(self, current: V | None, channel: None) -> V:
         """Generate the next version ID for a channel.
 
         Default is to use integer versions, incrementing by 1. If you override, you can use str/int/float versions,
@@ -342,6 +354,7 @@ class BaseCheckpointSaver(Generic[V]):
 
         Args:
             current: The current version identifier (int, float, or str).
+            channel: Deprecated argument, kept for backwards compatibility.
 
         Returns:
             V: The next version identifier, which must be increasing.
@@ -379,11 +392,10 @@ def get_checkpoint_metadata(
     for obj in (config.get("metadata"), config.get("configurable")):
         if not obj:
             continue
-        for key in obj:
+        for key, v in obj.items():
             if key in metadata or key in EXCLUDED_METADATA_KEYS or key.startswith("__"):
                 continue
-            v = obj[key]
-            if isinstance(v, str):
+            elif isinstance(v, str):
                 metadata[key] = v.replace("\u0000", "")
             elif isinstance(v, (int, bool, float)):
                 metadata[key] = v
@@ -400,7 +412,65 @@ Each Checkpointer implementation should use this mapping in put_writes.
 WRITES_IDX_MAP = {ERROR: -1, SCHEDULED: -2, INTERRUPT: -3, RESUME: -4}
 
 EXCLUDED_METADATA_KEYS = {
+    "thread_id",
+    "thread_ts",
     "checkpoint_id",
     "checkpoint_ns",
     "checkpoint_map",
+    "langgraph_step",
+    "langgraph_node",
+    "langgraph_triggers",
+    "langgraph_path",
+    "langgraph_checkpoint_ns",
 }
+
+# --- below are deprecated utilities used by past versions of LangGraph ---
+
+LATEST_VERSION = 2
+
+
+def empty_checkpoint() -> Checkpoint:
+    from datetime import datetime, timezone
+
+    return Checkpoint(
+        v=LATEST_VERSION,
+        id=str(uuid6(clock_seq=-2)),
+        ts=datetime.now(timezone.utc).isoformat(),
+        channel_values={},
+        channel_versions={},
+        versions_seen={},
+        pending_sends=[],
+    )
+
+
+def create_checkpoint(
+    checkpoint: Checkpoint,
+    channels: Mapping[str, ChannelProtocol] | None,
+    step: int,
+    *,
+    id: str | None = None,
+) -> Checkpoint:
+    """Create a checkpoint for the given channels."""
+    from datetime import datetime, timezone
+
+    ts = datetime.now(timezone.utc).isoformat()
+    if channels is None:
+        values = checkpoint["channel_values"]
+    else:
+        values = {}
+        for k, v in channels.items():
+            if k not in checkpoint["channel_versions"]:
+                continue
+            try:
+                values[k] = v.checkpoint()
+            except EmptyChannelError:
+                pass
+    return Checkpoint(
+        v=LATEST_VERSION,
+        ts=ts,
+        id=id or str(uuid6(clock_seq=step)),
+        channel_values=values,
+        channel_versions=checkpoint["channel_versions"],
+        versions_seen=checkpoint["versions_seen"],
+        pending_sends=checkpoint.get("pending_sends", []),
+    )

libs/checkpoint/langgraph/checkpoint/memory/__init__.py

@@ -512,7 +512,7 @@ class InMemorySaver(
         """
         return self.delete_thread(thread_id)
 
-    def get_next_version(self, current: str | None) -> str:
+    def get_next_version(self, current: str | None, channel: None) -> str:
         if current is None:
             current_v = 0
         elif isinstance(current, int):

libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py

@@ -475,6 +475,7 @@ def _msgpack_default(obj: Any) -> str | ormsgpack.Ext:
             buf = obj.tobytes(order="A")
             meta = (obj.dtype.str, obj.shape, order, buf)
             return ormsgpack.Ext(EXT_NUMPY_ARRAY, _msgpack_enc(meta))
+
     elif isinstance(obj, BaseException):
         return repr(obj)
     else:

libs/checkpoint/langgraph/checkpoint/serde/types.py

@@ -1,4 +1,13 @@
-from typing import Any, Protocol, TypeVar, runtime_checkable
+from collections.abc import Sequence
+from typing import (
+    Any,
+    Optional,
+    Protocol,
+    TypeVar,
+    runtime_checkable,
+)
+
+from typing_extensions import Self
 
 ERROR = "__error__"
 SCHEDULED = "__scheduled__"
@@ -11,6 +20,25 @@ Update = TypeVar("Update", contravariant=True)
 C = TypeVar("C")
 
 
+class ChannelProtocol(Protocol[Value, Update, C]):
+    # Mirrors langgraph.channels.base.BaseChannel
+    @property
+    def ValueType(self) -> Any: ...
+
+    @property
+    def UpdateType(self) -> Any: ...
+
+    def checkpoint(self) -> Optional[C]: ...
+
+    def from_checkpoint(self, checkpoint: Optional[C]) -> Self: ...
+
+    def update(self, values: Sequence[Update]) -> bool: ...
+
+    def get(self) -> Value: ...
+
+    def consume(self) -> bool: ...
+
+
 @runtime_checkable
 class SendProtocol(Protocol):
     # Mirrors langgraph.constants.Send

libs/checkpoint/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-checkpoint"
-version = "2.0.26"
+version = "2.1.0"
 description = "Library with base interfaces for LangGraph checkpoint savers."
 authors = []
 requires-python = ">=3.9"
@@ -30,6 +30,8 @@ dev = [
   "mypy",
   "dataclasses-json",
   "numpy",
+  "pandas",
+  "pandas-stubs>=2.2.2.240807",
 ]
 
 [tool.hatch.build.targets.wheel]

libs/checkpoint/tests/checkpoint_utils.py

@@ -1,53 +0,0 @@
-from __future__ import annotations
-
-from collections.abc import Mapping
-from datetime import datetime, timezone
-from typing import Any, Protocol
-
-from langgraph.checkpoint.base import Checkpoint, EmptyChannelError
-from langgraph.checkpoint.base.id import uuid6
-
-
-class ChannelProtocol(Protocol):
-    def checkpoint(self) -> Any | None: ...
-
-
-def empty_checkpoint() -> Checkpoint:
-    return Checkpoint(
-        v=1,
-        id=str(uuid6(clock_seq=-2)),
-        ts=datetime.now(timezone.utc).isoformat(),
-        channel_values={},
-        channel_versions={},
-        versions_seen={},
-    )
-
-
-def create_checkpoint(
-    checkpoint: Checkpoint,
-    channels: Mapping[str, ChannelProtocol] | None,
-    step: int,
-    *,
-    id: str | None = None,
-) -> Checkpoint:
-    """Create a checkpoint for the given channels."""
-    ts = datetime.now(timezone.utc).isoformat()
-    if channels is None:
-        values = checkpoint["channel_values"]
-    else:
-        values = {}
-        for k, v in channels.items():
-            if k not in checkpoint["channel_versions"]:
-                continue
-            try:
-                values[k] = v.checkpoint()
-            except EmptyChannelError:
-                pass
-    return Checkpoint(
-        v=1,
-        ts=ts,
-        id=id or str(uuid6(clock_seq=step)),
-        channel_values=values,
-        channel_versions=checkpoint["channel_versions"],
-        versions_seen=checkpoint["versions_seen"],
-    )

libs/checkpoint/tests/test_jsonplus.py

@@ -12,6 +12,7 @@ from zoneinfo import ZoneInfo
 
 import dataclasses_json
 import numpy as np
+import pandas as pd
 import pytest
 from pydantic import BaseModel, SecretStr
 from pydantic.v1 import BaseModel as BaseModelV1
@@ -332,19 +333,139 @@ def test_serde_jsonplus_numpy_array_json_hook(arr: np.ndarray) -> None:
     assert result == arr.tolist()
 
 
-def test_loads_cannot_find() -> None:
-    serde = JsonPlusSerializer()
+@pytest.mark.parametrize(
+    "df",
+    [
+        pd.DataFrame(),
+        pd.DataFrame({"int_col": [1, 2, 3]}),
+        pd.DataFrame({"float_col": [1.1, 2.2, 3.3]}),
+        pd.DataFrame({"str_col": ["a", "b", "c"]}),
+        pd.DataFrame({"bool_col": [True, False, True]}),
+        pd.DataFrame(
+            {
+                "datetime_col": [
+                    datetime(2024, 1, 1),
+                    datetime(2024, 1, 2),
+                    datetime(2024, 1, 3),
+                ]
+            }
+        ),
+        pd.DataFrame(
+            {
+                "int_col": [1, 2, 3],
+                "float_col": [1.1, 2.2, 3.3],
+                "str_col": ["a", "b", "c"],
+            }
+        ),
+        pd.DataFrame(
+            {
+                "int_col": [1, 2, None],
+                "float_col": [1.1, None, 3.3],
+                "str_col": ["a", None, "c"],
+            }
+        ),
+        pd.DataFrame({"cat_col": pd.Categorical(["a", "b", "a", "c"])}),
+        pd.DataFrame(
+            {
+                "int8": pd.array([1, 2, 3], dtype="int8"),
+                "int16": pd.array([10, 20, 30], dtype="int16"),
+                "int32": pd.array([100, 200, 300], dtype="int32"),
+                "int64": pd.array([1000, 2000, 3000], dtype="int64"),
+                "float32": pd.array([1.1, 2.2, 3.3], dtype="float32"),
+                "float64": pd.array([10.1, 20.2, 30.3], dtype="float64"),
+            }
+        ),
+        pd.DataFrame({"value": [1, 2, 3]}, index=["x", "y", "z"]),
+        pd.DataFrame(
+            [[1, 2, 3, 4]],
+            columns=pd.MultiIndex.from_tuples(
+                [("A", "X"), ("A", "Y"), ("B", "X"), ("B", "Y")]
+            ),
+        ),
+        pd.DataFrame(
+            {"value": [1, 2, 3]}, index=pd.date_range("2024-01-01", periods=3, freq="D")
+        ),
+        pd.DataFrame(
+            {
+                "col1": range(1000),
+                "col2": [f"str_{i}" for i in range(1000)],
+                "col3": np.random.rand(1000),
+            }
+        ),
+        pd.DataFrame(
+            {"tz_datetime": pd.date_range("2024-01-01", periods=3, freq="D", tz="UTC")}
+        ),
+        pd.DataFrame({"timedelta": pd.to_timedelta([1, 2, 3], unit="D")}),
+        pd.DataFrame({"period": pd.period_range("2024-01", periods=3, freq="M")}),
+        pd.DataFrame({"interval": pd.interval_range(start=0, end=3, periods=3)}),
+        pd.DataFrame({"unicode": ["Hello 🌍", "Python 🐍", "Data 📊"]}),
+        pd.DataFrame({"mixed": [1, "string", [1, 2, 3], {"key": "value"}]}),
+        pd.DataFrame({"a": [1], "b": ["test"], "c": [3.14]}),
+        pd.DataFrame({"single": [42]}),
+        pd.DataFrame(
+            {
+                "small": [sys.float_info.min, 0, sys.float_info.max],
+                "large_int": [-(2**63), 0, 2**63 - 1],
+            }
+        ),
+        pd.DataFrame({"special_strings": ["", "null", "None", "NaN", "inf", "-inf"]}),
+        pd.DataFrame({"bytes_col": [b"hello", b"world", b"\x00\x01\x02"]}),
+    ],
+)
+def test_serde_jsonplus_pandas_dataframe(df: pd.DataFrame) -> None:
+    serde = JsonPlusSerializer(pickle_fallback=True)
 
-    dumped = (
-        "json",
-        b'{"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyPydanticccc"], "method": null, "args": [], "kwargs": {"foo": "foo", "bar": 1}}',
-    )
+    dumped = serde.dumps_typed(df)
+    assert dumped[0] == "pickle"
+    result = serde.loads_typed(dumped)
+    assert result.equals(df)
 
-    assert serde.loads_typed(dumped) is None, "Should return None if cannot find class"
 
-    dumped = (
-        "json",
-        b'{"lc": 2, "type": "constructor", "id": ["tests", "test_jsonpluss", "MyPydantic"], "method": null, "args": [], "kwargs": {"foo": "foo", "bar": 1}}',
-    )
+@pytest.mark.parametrize(
+    "series",
+    [
+        pd.Series([]),
+        pd.Series([1, 2, 3]),
+        pd.Series([1.1, 2.2, 3.3]),
+        pd.Series(["a", "b", "c"]),
+        pd.Series([True, False, True]),
+        pd.Series([datetime(2024, 1, 1), datetime(2024, 1, 2), datetime(2024, 1, 3)]),
+        pd.Series([1, 2, None]),
+        pd.Series([1.1, None, 3.3]),
+        pd.Series(["a", None, "c"]),
+        pd.Series(pd.Categorical(["a", "b", "a", "c"])),
+        pd.Series([1, 2, 3], dtype="int8"),
+        pd.Series([10, 20, 30], dtype="int16"),
+        pd.Series([100, 200, 300], dtype="int32"),
+        pd.Series([1000, 2000, 3000], dtype="int64"),
+        pd.Series([1.1, 2.2, 3.3], dtype="float32"),
+        pd.Series([10.1, 20.2, 30.3], dtype="float64"),
+        pd.Series([1, 2, 3], index=["x", "y", "z"]),
+        pd.Series([1, 2, 3], index=pd.date_range("2024-01-01", periods=3, freq="D")),
+        pd.Series(range(1000)),
+        pd.Series(pd.date_range("2024-01-01", periods=3, freq="D", tz="UTC")),
+        pd.Series(pd.to_timedelta([1, 2, 3], unit="D")),
+        pd.Series(pd.period_range("2024-01", periods=3, freq="M")),
+        pd.Series(pd.interval_range(start=0, end=3, periods=3)),
+        pd.Series(["Hello 🌍", "Python 🐍", "Data 📊"]),
+        pd.Series([1, "string", [1, 2, 3], {"key": "value"}]),
+        pd.Series([42], name="single"),
+        pd.Series([sys.float_info.min, 0, sys.float_info.max]),
+        pd.Series([-(2**63), 0, 2**63 - 1]),
+        pd.Series(["", "null", "None", "NaN", "inf", "-inf"]),
+        pd.Series([b"hello", b"world", b"\x00\x01\x02"]),
+        pd.Series([1, 2, 3], name="named_series"),
+        pd.Series(
+            [10, 20],
+            index=pd.MultiIndex.from_tuples([("a", 1), ("b", 2)], names=["x", "y"]),
+        ),
+    ],
+)
+def test_serde_jsonplus_pandas_series(series: pd.Series) -> None:
+    serde = JsonPlusSerializer(pickle_fallback=True)
+    dumped = serde.dumps_typed(series)
 
-    assert serde.loads_typed(dumped) is None, "Should return None if cannot find module"
+    assert dumped[0] == "pickle"
+    result = serde.loads_typed(dumped)
+
+    assert result.equals(series)

libs/checkpoint/tests/test_memory.py

@@ -6,12 +6,10 @@ from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
     Checkpoint,
     CheckpointMetadata,
-)
-from langgraph.checkpoint.memory import InMemorySaver
-from tests.checkpoint_utils import (
     create_checkpoint,
     empty_checkpoint,
 )
+from langgraph.checkpoint.memory import InMemorySaver
 
 
 class TestMemorySaver:
@@ -77,7 +75,6 @@ class TestMemorySaver:
         assert checkpoint is not None
         assert checkpoint.metadata == {
             **self.metadata_2,
-            "thread_id": "thread-2",
             "run_id": "my_run_id",
         }
 
@@ -114,18 +111,11 @@ class TestMemorySaver:
 
         search_results_1 = list(self.memory_saver.list(None, filter=query_1))
         assert len(search_results_1) == 1
-        assert search_results_1[0].metadata == {
-            "thread_id": "thread-1",
-            "thread_ts": "1",
-            **self.metadata_1,
-        }
+        assert search_results_1[0].metadata == self.metadata_1
 
         search_results_2 = list(self.memory_saver.list(None, filter=query_2))
         assert len(search_results_2) == 1
-        assert search_results_2[0].metadata == {
-            "thread_id": "thread-2",
-            **self.metadata_2,
-        }
+        assert search_results_2[0].metadata == self.metadata_2
 
         search_results_3 = list(self.memory_saver.list(None, filter=query_3))
         assert len(search_results_3) == 3
@@ -180,20 +170,13 @@ class TestMemorySaver:
             c async for c in self.memory_saver.alist(None, filter=query_1)
         ]
         assert len(search_results_1) == 1
-        assert search_results_1[0].metadata == {
-            "thread_id": "thread-1",
-            "thread_ts": "1",
-            **self.metadata_1,
-        }
+        assert search_results_1[0].metadata == self.metadata_1
 
         search_results_2 = [
             c async for c in self.memory_saver.alist(None, filter=query_2)
         ]
         assert len(search_results_2) == 1
-        assert search_results_2[0].metadata == {
-            "thread_id": "thread-2",
-            **self.metadata_2,
-        }
+        assert search_results_2[0].metadata == self.metadata_2
 
         search_results_3 = [
             c async for c in self.memory_saver.alist(None, filter=query_3)

libs/checkpoint/uv.lock

@@ -3,7 +3,8 @@ revision = 2
 requires-python = ">=3.9"
 resolution-markers = [
     "python_full_version >= '3.12.4'",
-    "python_full_version >= '3.11' and python_full_version < '3.12.4'",
+    "python_full_version >= '3.12' and python_full_version < '3.12.4'",
+    "python_full_version == '3.11.*'",
     "python_full_version == '3.10.*'",
     "python_full_version < '3.10'",
 ]
@@ -323,7 +324,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "2.0.26"
+version = "2.1.0"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -338,6 +339,9 @@ dev = [
     { name = "numpy", version = "2.0.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" },
     { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.10.*'" },
     { name = "numpy", version = "2.3.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
+    { name = "pandas" },
+    { name = "pandas-stubs", version = "2.2.2.240807", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" },
+    { name = "pandas-stubs", version = "2.2.3.250527", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" },
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-mock" },
@@ -357,6 +361,8 @@ dev = [
     { name = "dataclasses-json" },
     { name = "mypy" },
     { name = "numpy" },
+    { name = "pandas" },
+    { name = "pandas-stubs", specifier = ">=2.2.2.240807" },
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-mock" },
@@ -573,7 +579,8 @@ version = "2.3.0"
 source = { registry = "https://pypi.org/simple" }
 resolution-markers = [
     "python_full_version >= '3.12.4'",
-    "python_full_version >= '3.11' and python_full_version < '3.12.4'",
+    "python_full_version >= '3.12' and python_full_version < '3.12.4'",
+    "python_full_version == '3.11.*'",
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/f3/db/8e12381333aea300890829a0a36bfa738cac95475d88982d538725143fd9/numpy-2.3.0.tar.gz", hash = "sha256:581f87f9e9e9db2cba2141400e160e9dd644ee248788d6f90636eeb8fd9260a6", size = 20382813, upload-time = "2025-06-07T14:54:32.608Z" }
 wheels = [
@@ -765,6 +772,99 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/88/ef/eb23f262cca3c0c4eb7ab1933c3b1f03d021f2c48f54763065b6f0e321be/packaging-24.2-py3-none-any.whl", hash = "sha256:09abb1bccd265c01f4a3aa3f7a7db064b36514d2cba19a2f694fe6150451a759", size = 65451, upload-time = "2024-11-08T09:47:44.722Z" },
 ]
 
+[[package]]
+name = "pandas"
+version = "2.3.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "numpy", version = "2.0.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" },
+    { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.10.*'" },
+    { name = "numpy", version = "2.3.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
+    { name = "python-dateutil" },
+    { name = "pytz" },
+    { name = "tzdata" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/72/51/48f713c4c728d7c55ef7444ba5ea027c26998d96d1a40953b346438602fc/pandas-2.3.0.tar.gz", hash = "sha256:34600ab34ebf1131a7613a260a61dbe8b62c188ec0ea4c296da7c9a06b004133", size = 4484490, upload-time = "2025-06-05T03:27:54.133Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/e2/2d/df6b98c736ba51b8eaa71229e8fcd91233a831ec00ab520e1e23090cc072/pandas-2.3.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:625466edd01d43b75b1883a64d859168e4556261a5035b32f9d743b67ef44634", size = 11527531, upload-time = "2025-06-05T03:25:48.648Z" },
+    { url = "https://files.pythonhosted.org/packages/77/1c/3f8c331d223f86ba1d0ed7d3ed7fcf1501c6f250882489cc820d2567ddbf/pandas-2.3.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:a6872d695c896f00df46b71648eea332279ef4077a409e2fe94220208b6bb675", size = 10774764, upload-time = "2025-06-05T03:25:53.228Z" },
+    { url = "https://files.pythonhosted.org/packages/1b/45/d2599400fad7fe06b849bd40b52c65684bc88fbe5f0a474d0513d057a377/pandas-2.3.0-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:f4dd97c19bd06bc557ad787a15b6489d2614ddaab5d104a0310eb314c724b2d2", size = 11711963, upload-time = "2025-06-05T03:25:56.855Z" },
+    { url = "https://files.pythonhosted.org/packages/66/f8/5508bc45e994e698dbc93607ee6b9b6eb67df978dc10ee2b09df80103d9e/pandas-2.3.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:034abd6f3db8b9880aaee98f4f5d4dbec7c4829938463ec046517220b2f8574e", size = 12349446, upload-time = "2025-06-05T03:26:01.292Z" },
+    { url = "https://files.pythonhosted.org/packages/f7/fc/17851e1b1ea0c8456ba90a2f514c35134dd56d981cf30ccdc501a0adeac4/pandas-2.3.0-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:23c2b2dc5213810208ca0b80b8666670eb4660bbfd9d45f58592cc4ddcfd62e1", size = 12920002, upload-time = "2025-06-06T00:00:07.925Z" },
+    { url = "https://files.pythonhosted.org/packages/a1/9b/8743be105989c81fa33f8e2a4e9822ac0ad4aaf812c00fee6bb09fc814f9/pandas-2.3.0-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:39ff73ec07be5e90330cc6ff5705c651ace83374189dcdcb46e6ff54b4a72cd6", size = 13651218, upload-time = "2025-06-05T03:26:09.731Z" },
+    { url = "https://files.pythonhosted.org/packages/26/fa/8eeb2353f6d40974a6a9fd4081ad1700e2386cf4264a8f28542fd10b3e38/pandas-2.3.0-cp310-cp310-win_amd64.whl", hash = "sha256:40cecc4ea5abd2921682b57532baea5588cc5f80f0231c624056b146887274d2", size = 11082485, upload-time = "2025-06-05T03:26:17.572Z" },
+    { url = "https://files.pythonhosted.org/packages/96/1e/ba313812a699fe37bf62e6194265a4621be11833f5fce46d9eae22acb5d7/pandas-2.3.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:8adff9f138fc614347ff33812046787f7d43b3cef7c0f0171b3340cae333f6ca", size = 11551836, upload-time = "2025-06-05T03:26:22.784Z" },
+    { url = "https://files.pythonhosted.org/packages/1b/cc/0af9c07f8d714ea563b12383a7e5bde9479cf32413ee2f346a9c5a801f22/pandas-2.3.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:e5f08eb9a445d07720776df6e641975665c9ea12c9d8a331e0f6890f2dcd76ef", size = 10807977, upload-time = "2025-06-05T16:50:11.109Z" },
+    { url = "https://files.pythonhosted.org/packages/ee/3e/8c0fb7e2cf4a55198466ced1ca6a9054ae3b7e7630df7757031df10001fd/pandas-2.3.0-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:fa35c266c8cd1a67d75971a1912b185b492d257092bdd2709bbdebe574ed228d", size = 11788230, upload-time = "2025-06-05T03:26:27.417Z" },
+    { url = "https://files.pythonhosted.org/packages/14/22/b493ec614582307faf3f94989be0f7f0a71932ed6f56c9a80c0bb4a3b51e/pandas-2.3.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:14a0cc77b0f089d2d2ffe3007db58f170dae9b9f54e569b299db871a3ab5bf46", size = 12370423, upload-time = "2025-06-05T03:26:34.142Z" },
+    { url = "https://files.pythonhosted.org/packages/9f/74/b012addb34cda5ce855218a37b258c4e056a0b9b334d116e518d72638737/pandas-2.3.0-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:c06f6f144ad0a1bf84699aeea7eff6068ca5c63ceb404798198af7eb86082e33", size = 12990594, upload-time = "2025-06-06T00:00:13.934Z" },
+    { url = "https://files.pythonhosted.org/packages/95/81/b310e60d033ab64b08e66c635b94076488f0b6ce6a674379dd5b224fc51c/pandas-2.3.0-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:ed16339bc354a73e0a609df36d256672c7d296f3f767ac07257801aa064ff73c", size = 13745952, upload-time = "2025-06-05T03:26:39.475Z" },
+    { url = "https://files.pythonhosted.org/packages/25/ac/f6ee5250a8881b55bd3aecde9b8cfddea2f2b43e3588bca68a4e9aaf46c8/pandas-2.3.0-cp311-cp311-win_amd64.whl", hash = "sha256:fa07e138b3f6c04addfeaf56cc7fdb96c3b68a3fe5e5401251f231fce40a0d7a", size = 11094534, upload-time = "2025-06-05T03:26:43.23Z" },
+    { url = "https://files.pythonhosted.org/packages/94/46/24192607058dd607dbfacdd060a2370f6afb19c2ccb617406469b9aeb8e7/pandas-2.3.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:2eb4728a18dcd2908c7fccf74a982e241b467d178724545a48d0caf534b38ebf", size = 11573865, upload-time = "2025-06-05T03:26:46.774Z" },
+    { url = "https://files.pythonhosted.org/packages/9f/cc/ae8ea3b800757a70c9fdccc68b67dc0280a6e814efcf74e4211fd5dea1ca/pandas-2.3.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:b9d8c3187be7479ea5c3d30c32a5d73d62a621166675063b2edd21bc47614027", size = 10702154, upload-time = "2025-06-05T16:50:14.439Z" },
+    { url = "https://files.pythonhosted.org/packages/d8/ba/a7883d7aab3d24c6540a2768f679e7414582cc389876d469b40ec749d78b/pandas-2.3.0-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:9ff730713d4c4f2f1c860e36c005c7cefc1c7c80c21c0688fd605aa43c9fcf09", size = 11262180, upload-time = "2025-06-05T16:50:17.453Z" },
+    { url = "https://files.pythonhosted.org/packages/01/a5/931fc3ad333d9d87b10107d948d757d67ebcfc33b1988d5faccc39c6845c/pandas-2.3.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ba24af48643b12ffe49b27065d3babd52702d95ab70f50e1b34f71ca703e2c0d", size = 11991493, upload-time = "2025-06-05T03:26:51.813Z" },
+    { url = "https://files.pythonhosted.org/packages/d7/bf/0213986830a92d44d55153c1d69b509431a972eb73f204242988c4e66e86/pandas-2.3.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:404d681c698e3c8a40a61d0cd9412cc7364ab9a9cc6e144ae2992e11a2e77a20", size = 12470733, upload-time = "2025-06-06T00:00:18.651Z" },
+    { url = "https://files.pythonhosted.org/packages/a4/0e/21eb48a3a34a7d4bac982afc2c4eb5ab09f2d988bdf29d92ba9ae8e90a79/pandas-2.3.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:6021910b086b3ca756755e86ddc64e0ddafd5e58e076c72cb1585162e5ad259b", size = 13212406, upload-time = "2025-06-05T03:26:55.992Z" },
+    { url = "https://files.pythonhosted.org/packages/1f/d9/74017c4eec7a28892d8d6e31ae9de3baef71f5a5286e74e6b7aad7f8c837/pandas-2.3.0-cp312-cp312-win_amd64.whl", hash = "sha256:094e271a15b579650ebf4c5155c05dcd2a14fd4fdd72cf4854b2f7ad31ea30be", size = 10976199, upload-time = "2025-06-05T03:26:59.594Z" },
+    { url = "https://files.pythonhosted.org/packages/d3/57/5cb75a56a4842bbd0511c3d1c79186d8315b82dac802118322b2de1194fe/pandas-2.3.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:2c7e2fc25f89a49a11599ec1e76821322439d90820108309bf42130d2f36c983", size = 11518913, upload-time = "2025-06-05T03:27:02.757Z" },
+    { url = "https://files.pythonhosted.org/packages/05/01/0c8785610e465e4948a01a059562176e4c8088aa257e2e074db868f86d4e/pandas-2.3.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:c6da97aeb6a6d233fb6b17986234cc723b396b50a3c6804776351994f2a658fd", size = 10655249, upload-time = "2025-06-05T16:50:20.17Z" },
+    { url = "https://files.pythonhosted.org/packages/e8/6a/47fd7517cd8abe72a58706aab2b99e9438360d36dcdb052cf917b7bf3bdc/pandas-2.3.0-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:bb32dc743b52467d488e7a7c8039b821da2826a9ba4f85b89ea95274f863280f", size = 11328359, upload-time = "2025-06-05T03:27:06.431Z" },
+    { url = "https://files.pythonhosted.org/packages/2a/b3/463bfe819ed60fb7e7ddffb4ae2ee04b887b3444feee6c19437b8f834837/pandas-2.3.0-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:213cd63c43263dbb522c1f8a7c9d072e25900f6975596f883f4bebd77295d4f3", size = 12024789, upload-time = "2025-06-05T03:27:09.875Z" },
+    { url = "https://files.pythonhosted.org/packages/04/0c/e0704ccdb0ac40aeb3434d1c641c43d05f75c92e67525df39575ace35468/pandas-2.3.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:1d2b33e68d0ce64e26a4acc2e72d747292084f4e8db4c847c6f5f6cbe56ed6d8", size = 12480734, upload-time = "2025-06-06T00:00:22.246Z" },
+    { url = "https://files.pythonhosted.org/packages/e9/df/815d6583967001153bb27f5cf075653d69d51ad887ebbf4cfe1173a1ac58/pandas-2.3.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:430a63bae10b5086995db1b02694996336e5a8ac9a96b4200572b413dfdfccb9", size = 13223381, upload-time = "2025-06-05T03:27:15.641Z" },
+    { url = "https://files.pythonhosted.org/packages/79/88/ca5973ed07b7f484c493e941dbff990861ca55291ff7ac67c815ce347395/pandas-2.3.0-cp313-cp313-win_amd64.whl", hash = "sha256:4930255e28ff5545e2ca404637bcc56f031893142773b3468dc021c6c32a1390", size = 10970135, upload-time = "2025-06-05T03:27:24.131Z" },
+    { url = "https://files.pythonhosted.org/packages/24/fb/0994c14d1f7909ce83f0b1fb27958135513c4f3f2528bde216180aa73bfc/pandas-2.3.0-cp313-cp313t-macosx_10_13_x86_64.whl", hash = "sha256:f925f1ef673b4bd0271b1809b72b3270384f2b7d9d14a189b12b7fc02574d575", size = 12141356, upload-time = "2025-06-05T03:27:34.547Z" },
+    { url = "https://files.pythonhosted.org/packages/9d/a2/9b903e5962134497ac4f8a96f862ee3081cb2506f69f8e4778ce3d9c9d82/pandas-2.3.0-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:e78ad363ddb873a631e92a3c063ade1ecfb34cae71e9a2be6ad100f875ac1042", size = 11474674, upload-time = "2025-06-05T03:27:39.448Z" },
+    { url = "https://files.pythonhosted.org/packages/81/3a/3806d041bce032f8de44380f866059437fb79e36d6b22c82c187e65f765b/pandas-2.3.0-cp313-cp313t-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:951805d146922aed8357e4cc5671b8b0b9be1027f0619cea132a9f3f65f2f09c", size = 11439876, upload-time = "2025-06-05T03:27:43.652Z" },
+    { url = "https://files.pythonhosted.org/packages/15/aa/3fc3181d12b95da71f5c2537c3e3b3af6ab3a8c392ab41ebb766e0929bc6/pandas-2.3.0-cp313-cp313t-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:1a881bc1309f3fce34696d07b00f13335c41f5f5a8770a33b09ebe23261cfc67", size = 11966182, upload-time = "2025-06-05T03:27:47.652Z" },
+    { url = "https://files.pythonhosted.org/packages/37/e7/e12f2d9b0a2c4a2cc86e2aabff7ccfd24f03e597d770abfa2acd313ee46b/pandas-2.3.0-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:e1991bbb96f4050b09b5f811253c4f3cf05ee89a589379aa36cd623f21a31d6f", size = 12547686, upload-time = "2025-06-06T00:00:26.142Z" },
+    { url = "https://files.pythonhosted.org/packages/39/c2/646d2e93e0af70f4e5359d870a63584dacbc324b54d73e6b3267920ff117/pandas-2.3.0-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:bb3be958022198531eb7ec2008cfc78c5b1eed51af8600c6c5d9160d89d8d249", size = 13231847, upload-time = "2025-06-05T03:27:51.465Z" },
+    { url = "https://files.pythonhosted.org/packages/38/86/d786690bd1d666d3369355a173b32a4ab7a83053cbb2d6a24ceeedb31262/pandas-2.3.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:9efc0acbbffb5236fbdf0409c04edce96bec4bdaa649d49985427bd1ec73e085", size = 11552206, upload-time = "2025-06-06T00:00:29.501Z" },
+    { url = "https://files.pythonhosted.org/packages/9c/2f/99f581c1c5b013fcfcbf00a48f5464fb0105da99ea5839af955e045ae3ab/pandas-2.3.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:75651c14fde635e680496148a8526b328e09fe0572d9ae9b638648c46a544ba3", size = 10796831, upload-time = "2025-06-06T00:00:49.502Z" },
+    { url = "https://files.pythonhosted.org/packages/5c/be/3ee7f424367e0f9e2daee93a3145a18b703fbf733ba56e1cf914af4b40d1/pandas-2.3.0-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:bf5be867a0541a9fb47a4be0c5790a4bccd5b77b92f0a59eeec9375fafc2aa14", size = 11736943, upload-time = "2025-06-06T00:01:15.992Z" },
+    { url = "https://files.pythonhosted.org/packages/83/95/81c7bb8f1aefecd948f80464177a7d9a1c5e205c5a1e279984fdacbac9de/pandas-2.3.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:84141f722d45d0c2a89544dd29d35b3abfc13d2250ed7e68394eda7564bd6324", size = 12366679, upload-time = "2025-06-06T00:01:36.162Z" },
+    { url = "https://files.pythonhosted.org/packages/d5/7a/54cf52fb454408317136d683a736bb597864db74977efee05e63af0a7d38/pandas-2.3.0-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:f95a2aef32614ed86216d3c450ab12a4e82084e8102e355707a1d96e33d51c34", size = 12924072, upload-time = "2025-06-06T00:01:44.243Z" },
+    { url = "https://files.pythonhosted.org/packages/0a/bf/25018e431257f8a42c173080f9da7c592508269def54af4a76ccd1c14420/pandas-2.3.0-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:e0f51973ba93a9f97185049326d75b942b9aeb472bec616a129806facb129ebb", size = 13696374, upload-time = "2025-06-06T00:02:14.346Z" },
+    { url = "https://files.pythonhosted.org/packages/db/84/5ffd2c447c02db56326f5c19a235a747fae727e4842cc20e1ddd28f990f6/pandas-2.3.0-cp39-cp39-win_amd64.whl", hash = "sha256:b198687ca9c8529662213538a9bb1e60fa0bf0f6af89292eb68fea28743fcd5a", size = 11104735, upload-time = "2025-06-06T00:02:21.088Z" },
+]
+
+[[package]]
+name = "pandas-stubs"
+version = "2.2.2.240807"
+source = { registry = "https://pypi.org/simple" }
+resolution-markers = [
+    "python_full_version < '3.10'",
+]
+dependencies = [
+    { name = "numpy", version = "2.0.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" },
+    { name = "types-pytz", marker = "python_full_version < '3.10'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/1f/df/0da95bc75c76f1e012e0bc0b76da31faaf4254e94b9870f25e6311145e98/pandas_stubs-2.2.2.240807.tar.gz", hash = "sha256:64a559725a57a449f46225fbafc422520b7410bff9252b661a225b5559192a93", size = 103095, upload-time = "2024-08-07T12:30:54.538Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/0a/f9/22c91632ea1b4c6165952f677bf9ad95f9ac36ffd7ef3e6450144e6d8b1a/pandas_stubs-2.2.2.240807-py3-none-any.whl", hash = "sha256:893919ad82be4275f0d07bb47a95d08bae580d3fdea308a7acfcb3f02e76186e", size = 157069, upload-time = "2024-08-07T12:30:51.868Z" },
+]
+
+[[package]]
+name = "pandas-stubs"
+version = "2.2.3.250527"
+source = { registry = "https://pypi.org/simple" }
+resolution-markers = [
+    "python_full_version >= '3.12.4'",
+    "python_full_version >= '3.12' and python_full_version < '3.12.4'",
+    "python_full_version == '3.11.*'",
+    "python_full_version == '3.10.*'",
+]
+dependencies = [
+    { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.10.*'" },
+    { name = "numpy", version = "2.3.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
+    { name = "types-pytz", marker = "python_full_version >= '3.10'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/5f/0d/5fe7f7f3596eb1c2526fea151e9470f86b379183d8b9debe44b2098651ca/pandas_stubs-2.2.3.250527.tar.gz", hash = "sha256:e2d694c4e72106055295ad143664e5c99e5815b07190d1ff85b73b13ff019e63", size = 106312, upload-time = "2025-05-27T15:24:29.716Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ec/f8/46141ba8c9d7064dc5008bfb4a6ae5bd3c30e4c61c28b5c5ed485bf358ba/pandas_stubs-2.2.3.250527-py3-none-any.whl", hash = "sha256:cd0a49a95b8c5f944e605be711042a4dd8550e2c559b43d70ba2c4b524b66163", size = 159683, upload-time = "2025-05-27T15:24:28.4Z" },
+]
+
 [[package]]
 name = "pluggy"
 version = "1.6.0"
@@ -962,6 +1062,27 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/5b/3a/c44a76c6bb5e9e896d9707fb1c704a31a0136950dec9514373ced0684d56/pytest_watcher-0.4.3-py3-none-any.whl", hash = "sha256:d59b1e1396f33a65ea4949b713d6884637755d641646960056a90b267c3460f9", size = 11852, upload-time = "2024-08-28T17:37:45.731Z" },
 ]
 
+[[package]]
+name = "python-dateutil"
+version = "2.9.0.post0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "six" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/66/c0/0c8b6ad9f17a802ee498c46e004a0eb49bc148f2fd230864601a86dcf6db/python-dateutil-2.9.0.post0.tar.gz", hash = "sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3", size = 342432, upload-time = "2024-03-01T18:36:20.211Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" },
+]
+
+[[package]]
+name = "pytz"
+version = "2025.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/f8/bf/abbd3cdfb8fbc7fb3d4d38d320f2441b1e7cbe29be4f23797b4a2b5d8aac/pytz-2025.2.tar.gz", hash = "sha256:360b9e3dbb49a209c21ad61809c7fb453643e048b38924c765813546746e81c3", size = 320884, upload-time = "2025-03-25T02:25:00.538Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/81/c4/34e93fe5f5429d7570ec1fa436f1986fb1f00c3e0f43a589fe2bbcd22c3f/pytz-2025.2-py2.py3-none-any.whl", hash = "sha256:5ddf76296dd8c44c26eb8f4b6f35488f3ccbf6fbbd7adee0b7262d43f0ec2f00", size = 509225, upload-time = "2025-03-25T02:24:58.468Z" },
+]
+
 [[package]]
 name = "pyyaml"
 version = "6.0.2"
@@ -1067,6 +1188,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/95/3a/2e8704d19f376c799748ff9cb041225c1d59f3e7711bc5596c8cfdc24925/ruff-0.11.10-py3-none-win_arm64.whl", hash = "sha256:ef69637b35fb8b210743926778d0e45e1bffa850a7c61e428c6b971549b5f5d1", size = 10765278, upload-time = "2025-05-15T14:08:54.56Z" },
 ]
 
+[[package]]
+name = "six"
+version = "1.17.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/94/e7/b2c673351809dca68a0e064b6af791aa332cf192da575fd474ed7d6f16a2/six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81", size = 34031, upload-time = "2024-12-04T17:35:28.174Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" },
+]
+
 [[package]]
 name = "sniffio"
 version = "1.3.1"
@@ -1124,6 +1254,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/6e/c2/61d3e0f47e2b74ef40a68b9e6ad5984f6241a942f7cd3bbfbdbd03861ea9/tomli-2.2.1-py3-none-any.whl", hash = "sha256:cb55c73c5f4408779d0cf3eef9f762b9c9f147a77de7b258bef0a5628adc85cc", size = 14257, upload-time = "2024-11-27T22:38:35.385Z" },
 ]
 
+[[package]]
+name = "types-pytz"
+version = "2025.2.0.20250516"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/bd/72/b0e711fd90409f5a76c75349055d3eb19992c110f0d2d6aabbd6cfbc14bf/types_pytz-2025.2.0.20250516.tar.gz", hash = "sha256:e1216306f8c0d5da6dafd6492e72eb080c9a166171fa80dd7a1990fd8be7a7b3", size = 10940, upload-time = "2025-05-16T03:07:01.91Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c1/ba/e205cd11c1c7183b23c97e4bcd1de7bc0633e2e867601c32ecfc6ad42675/types_pytz-2025.2.0.20250516-py3-none-any.whl", hash = "sha256:e0e0c8a57e2791c19f718ed99ab2ba623856b11620cb6b637e5f62ce285a7451", size = 10136, upload-time = "2025-05-16T03:07:01.075Z" },
+]
+
 [[package]]
 name = "typing-extensions"
 version = "4.13.2"
@@ -1158,6 +1297,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/31/08/aa4fdfb71f7de5176385bd9e90852eaf6b5d622735020ad600f2bab54385/typing_inspection-0.4.0-py3-none-any.whl", hash = "sha256:50e72559fcd2a6367a19f7a7e610e6afcb9fac940c650290eed893d61386832f", size = 14125, upload-time = "2025-02-25T17:27:57.754Z" },
 ]
 
+[[package]]
+name = "tzdata"
+version = "2025.2"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/95/32/1a225d6164441be760d75c2c42e2780dc0873fe382da3e98a2e1e48361e5/tzdata-2025.2.tar.gz", hash = "sha256:b60a638fcc0daffadf82fe0f57e53d06bdec2f36c4df66280ae79bce6bd6f2b9", size = 196380, upload-time = "2025-03-23T13:54:43.652Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/5c/23/c7abc0ca0a1526a0774eca151daeb8de62ec457e77262b66b359c3c7679e/tzdata-2025.2-py2.py3-none-any.whl", hash = "sha256:1a403fada01ff9221ca8044d701868fa132215d84beb92242d9acd2147f667a8", size = 347839, upload-time = "2025-03-23T13:54:41.845Z" },
+]
+
 [[package]]
 name = "urllib3"
 version = "2.4.0"

libs/cli/langgraph_cli/config.py

@@ -330,6 +330,11 @@ class HttpConfig(TypedDict, total=False):
     disable_store: bool
     """Optional. If True, /store routes are removed, disabling direct store interactions via HTTP.
     
+    Default is False.
+    """
+    disable_mcp: bool
+    """Optional. If True, /mcp routes are removed, disabling the MCP server.
+    
     Default is False.
     """
     disable_meta: bool
@@ -383,6 +388,14 @@ class Config(TypedDict, total=False):
     Only relevant if Python dependencies are installed via pip. If omitted, default pip settings are used.
     """
 
+    pip_installer: Optional[str]
+    """Optional. Python package installer to use ('auto', 'pip', 'uv').
+    
+    - 'auto' (default): Use uv for supported base images, otherwise pip
+    - 'pip': Force use of pip regardless of base image support
+    - 'uv': Force use of uv (will fail if base image doesn't support it)
+    """
+
     dockerfile_lines: list[str]
     """Optional. Additional Docker instructions that will be appended to your base Dockerfile.
     
@@ -536,6 +549,7 @@ def validate_config(config: Config) -> Config:
         "node_version": node_version,
         "python_version": python_version,
         "pip_config_file": config.get("pip_config_file"),
+        "pip_installer": config.get("pip_installer", "auto"),
         "_INTERNAL_docker_tag": config.get("_INTERNAL_docker_tag"),
         "base_image": config.get("base_image"),
         "image_distro": image_distro,
@@ -600,6 +614,13 @@ def validate_config(config: Config) -> Config:
                 "Must be either 'debian' or 'wolfi'."
             )
 
+    if pip_installer := config.get("pip_installer"):
+        if pip_installer not in ["auto", "pip", "uv"]:
+            raise click.UsageError(
+                f"Invalid pip_installer: '{pip_installer}'. "
+                "Must be 'auto', 'pip', or 'uv'."
+            )
+
     # Validate auth config
     if auth_conf := config.get("auth"):
         if "path" in auth_conf:
@@ -1114,12 +1135,21 @@ def python_config_to_docker(
     base_image: str,
 ) -> tuple[str, dict[str, str]]:
     """Generate a Dockerfile from the configuration."""
-    if _image_supports_uv(base_image):
+    pip_installer = config.get("pip_installer", "auto")
+
+    if pip_installer == "uv":
         install_cmd = "uv pip install --system"
         uv_removal = "RUN uv pip uninstall --system pip setuptools wheel && rm /usr/bin/uv /usr/bin/uvx"
-    else:
+    elif pip_installer == "pip":
         install_cmd = "pip install"
         uv_removal = ""
+    else:
+        if _image_supports_uv(base_image):
+            install_cmd = "uv pip install --system"
+            uv_removal = "RUN uv pip uninstall --system pip setuptools wheel && rm /usr/bin/uv /usr/bin/uvx"
+        else:
+            install_cmd = "pip install"
+            uv_removal = ""
 
     # configure pip
     pip_install = f"PYTHONDONTWRITEBYTECODE=1 {install_cmd} --no-cache-dir -c /api/constraints.txt"

libs/cli/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-cli"
-version = "0.3.1"
+version = "0.3.3"
 description = "CLI for interacting with LangGraph API"
 authors = []
 requires-python = ">=3.9"

libs/cli/schemas/schema.json

@@ -134,6 +134,17 @@
               ],
               "description": "Optional. Linux distribution for the base image.\n\nMust be either 'debian' or 'wolfi'. If omitted, defaults to 'debian'.\n"
             },
+            "pip_installer": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Python package installer to use ('auto', 'pip', 'uv').\n\n"
+            },
             "store": {
               "anyOf": [
                 {
@@ -287,6 +298,17 @@
               ],
               "description": "Optional. Linux distribution for the base image.\n\nMust be either 'debian' or 'wolfi'. If omitted, defaults to 'debian'.\n"
             },
+            "pip_installer": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Python package installer to use ('auto', 'pip', 'uv').\n\n"
+            },
             "store": {
               "anyOf": [
                 {
@@ -477,6 +499,10 @@
           "type": "boolean",
           "description": "Optional. If True, /assistants routes are removed from the server.\n\nDefault is False (meaning /assistants is enabled).\n"
         },
+        "disable_mcp": {
+          "type": "boolean",
+          "description": "Optional. If True, /mcp routes are removed, disabling the MCP server.\n\nDefault is False.\n"
+        },
         "disable_meta": {
           "type": "boolean",
           "description": "Optional. If True, all meta endpoints (/ok, /info, /metrics, /docs) are disabled.\n\nDefault is False.\n"

libs/cli/schemas/schema.v0.json

@@ -134,6 +134,17 @@
               ],
               "description": "Optional. Linux distribution for the base image.\n\nMust be either 'debian' or 'wolfi'. If omitted, defaults to 'debian'.\n"
             },
+            "pip_installer": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Python package installer to use ('auto', 'pip', 'uv').\n\n"
+            },
             "store": {
               "anyOf": [
                 {
@@ -287,6 +298,17 @@
               ],
               "description": "Optional. Linux distribution for the base image.\n\nMust be either 'debian' or 'wolfi'. If omitted, defaults to 'debian'.\n"
             },
+            "pip_installer": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Python package installer to use ('auto', 'pip', 'uv').\n\n"
+            },
             "store": {
               "anyOf": [
                 {
@@ -477,6 +499,10 @@
           "type": "boolean",
           "description": "Optional. If True, /assistants routes are removed from the server.\n\nDefault is False (meaning /assistants is enabled).\n"
         },
+        "disable_mcp": {
+          "type": "boolean",
+          "description": "Optional. If True, /mcp routes are removed, disabling the MCP server.\n\nDefault is False.\n"
+        },
         "disable_meta": {
           "type": "boolean",
           "description": "Optional. If True, all meta endpoints (/ok, /info, /metrics, /docs) are disabled.\n\nDefault is False.\n"

libs/cli/tests/unit_tests/test_config.py

@@ -1,3 +1,4 @@
+import copy
 import json
 import os
 import pathlib
@@ -40,6 +41,7 @@ def test_validate_config():
         "python_version": "3.11",
         "node_version": None,
         "pip_config_file": None,
+        "pip_installer": "auto",
         "image_distro": "debian",
         "dockerfile_lines": [],
         "env": {},
@@ -61,6 +63,7 @@ def test_validate_config():
         "python_version": "3.12",
         "node_version": None,
         "pip_config_file": "pipconfig.txt",
+        "pip_installer": "auto",
         "image_distro": "debian",
         "dockerfile_lines": ["ARG meow"],
         "dependencies": [".", "langchain"],
@@ -216,6 +219,74 @@ def test_validate_config_image_distro():
     assert config["image_distro"] == "debian"
 
 
+def test_validate_config_pip_installer():
+    """Test validation of pip_installer field."""
+    # Valid pip_installer values should work
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "pip_installer": "auto",
+        }
+    )
+    assert config["pip_installer"] == "auto"
+
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "pip_installer": "pip",
+        }
+    )
+    assert config["pip_installer"] == "pip"
+
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "pip_installer": "uv",
+        }
+    )
+    assert config["pip_installer"] == "uv"
+
+    # Missing pip_installer should default to "auto"
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+        }
+    )
+    assert config["pip_installer"] == "auto"
+
+    # Invalid pip_installer values should raise error
+    with pytest.raises(click.UsageError) as exc_info:
+        validate_config(
+            {
+                "python_version": "3.11",
+                "dependencies": ["."],
+                "graphs": {"agent": "./agent.py:graph"},
+                "pip_installer": "conda",
+            }
+        )
+    assert "Invalid pip_installer: 'conda'" in str(exc_info.value)
+    assert "Must be 'auto', 'pip', or 'uv'" in str(exc_info.value)
+
+    with pytest.raises(click.UsageError) as exc_info:
+        validate_config(
+            {
+                "python_version": "3.11",
+                "dependencies": ["."],
+                "graphs": {"agent": "./agent.py:graph"},
+                "pip_installer": "invalid",
+            }
+        )
+    assert "Invalid pip_installer: 'invalid'" in str(exc_info.value)
+
+
 def test_validate_config_file():
     with tempfile.TemporaryDirectory() as tmpdir:
         tmpdir_path = pathlib.Path(tmpdir)
@@ -799,6 +870,61 @@ WORKDIR /deps/__outer_unit_tests/unit_tests"""
     assert additional_contexts == {}
 
 
+def test_config_to_docker_pip_installer():
+    """Test that pip_installer setting affects the generated Dockerfile."""
+    graphs = {"agent": "./graphs/agent.py:graph"}
+    base_config = {
+        "python_version": "3.11",
+        "dependencies": ["."],
+        "graphs": graphs,
+    }
+
+    # Test default (auto) behavior with UV-supporting image
+    config_auto = validate_config(
+        {**copy.deepcopy(base_config), "pip_installer": "auto"}
+    )
+    docker_auto, _ = config_to_docker(
+        PATH_TO_CONFIG, config_auto, "langchain/langgraph-api:0.2.47"
+    )
+    assert "uv pip install --system" in docker_auto
+    assert "rm /usr/bin/uv /usr/bin/uvx" in docker_auto
+
+    # Test explicit pip setting
+    config_pip = validate_config({**copy.deepcopy(base_config), "pip_installer": "pip"})
+    docker_pip, _ = config_to_docker(
+        PATH_TO_CONFIG, config_pip, "langchain/langgraph-api:0.2.47"
+    )
+    assert "uv pip install --system" not in docker_pip
+    assert "pip install" in docker_pip
+    assert "rm /usr/bin/uv" not in docker_pip
+
+    # Test explicit uv setting
+    config_uv = validate_config({**copy.deepcopy(base_config), "pip_installer": "uv"})
+    docker_uv, _ = config_to_docker(
+        PATH_TO_CONFIG, config_uv, "langchain/langgraph-api:0.2.47"
+    )
+    assert "uv pip install --system" in docker_uv
+    assert "rm /usr/bin/uv /usr/bin/uvx" in docker_uv
+
+    # Test auto behavior with older image (should use pip)
+    config_auto_old = validate_config(
+        {**copy.deepcopy(base_config), "pip_installer": "auto"}
+    )
+    docker_auto_old, _ = config_to_docker(
+        PATH_TO_CONFIG, config_auto_old, "langchain/langgraph-api:0.2.46"
+    )
+    assert "uv pip install --system" not in docker_auto_old
+    assert "pip install" in docker_auto_old
+    assert "rm /usr/bin/uv" not in docker_auto_old
+
+    # Test that missing pip_installer defaults to auto behavior
+    config_default = validate_config(copy.deepcopy(base_config))
+    docker_default, _ = config_to_docker(
+        PATH_TO_CONFIG, config_default, "langchain/langgraph-api:0.2.47"
+    )
+    assert "uv pip install --system" in docker_default
+
+
 # config_to_compose
 def test_config_to_compose_simple_config():
     graphs = {"agent": "./agent.py:graph"}

libs/cli/uv.lock

@@ -501,7 +501,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-cli"
-version = "0.3.1"
+version = "0.3.3"
 source = { editable = "." }
 dependencies = [
     { name = "click" },

libs/langgraph/langgraph/channels/__init__.py

@@ -1,14 +1,15 @@
 from langgraph.channels.any_value import AnyValue
 from langgraph.channels.binop import BinaryOperatorAggregate
 from langgraph.channels.ephemeral_value import EphemeralValue
-from langgraph.channels.last_value import LastValue, LastValueAfterFinish
+from langgraph.channels.last_value import LastValue
 from langgraph.channels.topic import Topic
+from langgraph.channels.untracked_value import UntrackedValue
 
 __all__ = [
     "LastValue",
-    "LastValueAfterFinish",
     "Topic",
     "BinaryOperatorAggregate",
+    "UntrackedValue",
     "EphemeralValue",
     "AnyValue",
 ]

libs/langgraph/langgraph/channels/untracked_value.py

@@ -0,0 +1,66 @@
+from collections.abc import Sequence
+from typing import Generic
+
+from typing_extensions import Self
+
+from langgraph.channels.base import BaseChannel, Value
+from langgraph.constants import MISSING
+from langgraph.errors import EmptyChannelError, InvalidUpdateError
+
+
+class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]):
+    """Stores the last value received, never checkpointed."""
+
+    __slots__ = ("value", "guard")
+
+    def __init__(self, typ: type[Value], guard: bool = True) -> None:
+        super().__init__(typ)
+        self.guard = guard
+        self.value = MISSING
+
+    def __eq__(self, value: object) -> bool:
+        return isinstance(value, UntrackedValue) and value.guard == self.guard
+
+    @property
+    def ValueType(self) -> type[Value]:
+        """The type of the value stored in the channel."""
+        return self.typ
+
+    @property
+    def UpdateType(self) -> type[Value]:
+        """The type of the update received by the channel."""
+        return self.typ
+
+    def copy(self) -> Self:
+        """Return a copy of the channel."""
+        empty = self.__class__(self.typ, self.guard)
+        empty.key = self.key
+        empty.value = self.value
+        return empty
+
+    def checkpoint(self) -> Value:
+        return MISSING
+
+    def from_checkpoint(self, checkpoint: Value) -> Self:
+        empty = self.__class__(self.typ, self.guard)
+        empty.key = self.key
+        return empty
+
+    def update(self, values: Sequence[Value]) -> bool:
+        if len(values) == 0:
+            return False
+        if len(values) != 1 and self.guard:
+            raise InvalidUpdateError(
+                f"At key '{self.key}': UntrackedValue(guard=True) can receive only one value per step. Use guard=False if you want to store any one of multiple values."
+            )
+
+        self.value = values[-1]
+        return True
+
+    def get(self) -> Value:
+        if self.value is MISSING:
+            raise EmptyChannelError()
+        return self.value
+
+    def is_available(self) -> bool:
+        return self.value is not MISSING

libs/langgraph/langgraph/constants.py

@@ -39,8 +39,6 @@ ERROR = sys.intern("__error__")
 # for errors raised by nodes
 NO_WRITES = sys.intern("__no_writes__")
 # marker to signal node didn't write anything
-SCHEDULED = sys.intern("__scheduled__")
-# marker to signal node was scheduled (in distributed mode)
 TASKS = sys.intern("__pregel_tasks")
 # for Send objects returned by nodes/edges, corresponds to PUSH below
 RETURN = sys.intern("__return__")
@@ -71,13 +69,6 @@ CONFIG_KEY_RESUMING = sys.intern("__pregel_resuming")
 # holds a boolean indicating if subgraphs should resume from a previous checkpoint
 CONFIG_KEY_TASK_ID = sys.intern("__pregel_task_id")
 # holds the task ID for the current task
-CONFIG_KEY_DEDUPE_TASKS = sys.intern("__pregel_dedupe_tasks")
-# holds a boolean indicating if tasks should be deduplicated (for distributed mode)
-CONFIG_KEY_ENSURE_LATEST = sys.intern("__pregel_ensure_latest")
-# holds a boolean indicating whether to assert the requested checkpoint is the latest
-# (for distributed mode)
-CONFIG_KEY_DELEGATE = sys.intern("__pregel_delegate")
-# holds a boolean indicating whether to delegate subgraphs (for distributed mode)
 CONFIG_KEY_THREAD_ID = sys.intern("thread_id")
 # holds the thread ID for the current invocation
 CONFIG_KEY_CHECKPOINT_MAP = sys.intern("checkpoint_map")
@@ -121,7 +112,6 @@ RESERVED = {
     RESUME,
     ERROR,
     NO_WRITES,
-    SCHEDULED,
     # reserved config.configurable keys
     CONFIG_KEY_SEND,
     CONFIG_KEY_READ,
@@ -132,9 +122,6 @@ RESERVED = {
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_RESUMING,
     CONFIG_KEY_TASK_ID,
-    CONFIG_KEY_DEDUPE_TASKS,
-    CONFIG_KEY_ENSURE_LATEST,
-    CONFIG_KEY_DELEGATE,
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_CHECKPOINT_ID,
     CONFIG_KEY_CHECKPOINT_NS,

libs/langgraph/langgraph/errors.py

@@ -78,13 +78,6 @@ class NodeInterrupt(GraphInterrupt):
         super().__init__([Interrupt(value=value)])
 
 
-class GraphDelegate(GraphBubbleUp):
-    """Raised when a graph is delegated (for distributed mode)."""
-
-    def __init__(self, *args: dict[str, Any]) -> None:
-        super().__init__(*args)
-
-
 class ParentCommand(GraphBubbleUp):
     args: tuple[Command]
 
@@ -102,9 +95,3 @@ class TaskNotFound(Exception):
     """Raised when the executor is unable to find a task (for distributed mode)."""
 
     pass
-
-
-class CheckpointNotLatest(Exception):
-    """Raised when the checkpoint is not the latest version (for distributed mode)."""
-
-    pass

libs/langgraph/langgraph/func/__init__.py

@@ -38,7 +38,7 @@ from langgraph.pregel.read import PregelNode
 from langgraph.pregel.write import ChannelWrite, ChannelWriteEntry
 from langgraph.store.base import BaseStore
 from langgraph.types import _DC_KWARGS, CachePolicy, RetryPolicy, StreamMode
-from langgraph.warnings import LangGraphDeprecatedSinceV10
+from langgraph.warnings import LangGraphDeprecatedSinceV05
 
 
 class TaskFunction(Generic[P, T]):
@@ -179,7 +179,7 @@ def task(
     if (retry := kwargs.get("retry", UNSET)) is not UNSET:
         warnings.warn(
             "`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
-            category=LangGraphDeprecatedSinceV10,
+            category=LangGraphDeprecatedSinceV05,
         )
         if retry_policy is None:
             retry_policy = retry  # type: ignore[assignment]
@@ -383,7 +383,7 @@ class entrypoint:
         if (retry := kwargs.get("retry", UNSET)) is not UNSET:
             warnings.warn(
                 "`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
-                category=LangGraphDeprecatedSinceV10,
+                category=LangGraphDeprecatedSinceV05,
             )
             if retry_policy is None:
                 retry_policy = retry  # type: ignore[assignment]

libs/langgraph/langgraph/graph/__init__.py

@@ -1,11 +1,12 @@
 from langgraph.constants import END, START
-from langgraph.graph.message import MessagesState, add_messages
+from langgraph.graph.message import MessageGraph, MessagesState, add_messages
 from langgraph.graph.state import StateGraph
 
 __all__ = [
     "END",
     "START",
     "StateGraph",
+    "MessageGraph",
     "add_messages",
     "MessagesState",
 ]

libs/langgraph/langgraph/graph/message.py

@@ -25,6 +25,7 @@ from langchain_core.messages import (
 from typing_extensions import TypedDict
 
 from langgraph.constants import CONF, CONFIG_KEY_SEND
+from langgraph.graph.state import StateGraph
 
 Messages = Union[list[MessageLikeRepresentation], MessageLikeRepresentation]
 
@@ -226,6 +227,57 @@ def add_messages(
     return merged
 
 
+class MessageGraph(StateGraph):
+    """A StateGraph where every node receives a list of messages as input and returns one or more messages as output.
+
+    MessageGraph is a subclass of StateGraph whose entire state is a single, append-only* list of messages.
+    Each node in a MessageGraph takes a list of messages as input and returns zero or more
+    messages as output. The `add_messages` function is used to merge the output messages from each node
+    into the existing list of messages in the graph's state.
+
+    Examples:
+        ```pycon
+        >>> from langgraph.graph.message import MessageGraph
+        ...
+        >>> builder = MessageGraph()
+        >>> builder.add_node("chatbot", lambda state: [("assistant", "Hello!")])
+        >>> builder.set_entry_point("chatbot")
+        >>> builder.set_finish_point("chatbot")
+        >>> builder.compile().invoke([("user", "Hi there.")])
+        [HumanMessage(content="Hi there.", id='...'), AIMessage(content="Hello!", id='...')]
+        ```
+
+        ```pycon
+        >>> from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+        >>> from langgraph.graph.message import MessageGraph
+        ...
+        >>> builder = MessageGraph()
+        >>> builder.add_node(
+        ...     "chatbot",
+        ...     lambda state: [
+        ...         AIMessage(
+        ...             content="Hello!",
+        ...             tool_calls=[{"name": "search", "id": "123", "args": {"query": "X"}}],
+        ...         )
+        ...     ],
+        ... )
+        >>> builder.add_node(
+        ...     "search", lambda state: [ToolMessage(content="Searching...", tool_call_id="123")]
+        ... )
+        >>> builder.set_entry_point("chatbot")
+        >>> builder.add_edge("chatbot", "search")
+        >>> builder.set_finish_point("search")
+        >>> builder.compile().invoke([HumanMessage(content="Hi there. Can you search for X?")])
+        {'messages': [HumanMessage(content="Hi there. Can you search for X?", id='b8b7d8f4-7f4d-4f4d-9c1d-f8b8d8f4d9c1'),
+                     AIMessage(content="Hello!", id='f4d9c1d8-8d8f-4d9c-b8b7-d8f4f4d9c1d8'),
+                     ToolMessage(content="Searching...", id='d8f4f4d9-c1d8-4f4d-b8b7-d8f4f4d9c1d8', tool_call_id="123")]}
+        ```
+    """
+
+    def __init__(self) -> None:
+        super().__init__(Annotated[list[AnyMessage], add_messages])  # type: ignore[arg-type]
+
+
 class MessagesState(TypedDict):
     messages: Annotated[list[AnyMessage], add_messages]
 

libs/langgraph/langgraph/graph/state.py

@@ -86,7 +86,7 @@ from langgraph.utils.fields import (
 )
 from langgraph.utils.pydantic import create_model
 from langgraph.utils.runnable import coerce_to_runnable
-from langgraph.warnings import LangGraphDeprecatedSinceV10
+from langgraph.warnings import LangGraphDeprecatedSinceV05
 
 logger = logging.getLogger(__name__)
 
@@ -148,6 +148,15 @@ class _NodeWithConfigWriterStore(Protocol[StateT_contra]):
     ) -> Any: ...
 
 
+class _Invokable(Protocol[StateT_contra]):
+    def invoke(
+        self,
+        input: StateT_contra,
+        config: RunnableConfig | None = None,
+        **kwargs: Any,
+    ) -> Any: ...
+
+
 # TODO: we probably don't want to explicitly support the config / store signatures once
 # we move to adding a context arg. Maybe what we do is we add support for kwargs with param spec
 # this is purely for typing purposes though, so can easily change in the coming weeks.
@@ -160,6 +169,7 @@ StateNode: TypeAlias = Union[
     _NodeWithConfigWriter[StateT_contra],
     _NodeWithConfigStore[StateT_contra],
     _NodeWithConfigWriterStore[StateT_contra],
+    _Invokable[StateT_contra],
 ]
 
 
@@ -261,7 +271,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]):
         if (input_ := kwargs.get("input", UNSET)) is not UNSET:
             warnings.warn(
                 "`input` is deprecated and will be removed. Please use `input_schema` instead.",
-                category=LangGraphDeprecatedSinceV10,
+                category=LangGraphDeprecatedSinceV05,
                 stacklevel=2,
             )
             if input_schema is None:
@@ -270,7 +280,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]):
         if (output := kwargs.get("output", UNSET)) is not UNSET:
             warnings.warn(
                 "`output` is deprecated and will be removed. Please use `output_schema` instead.",
-                category=LangGraphDeprecatedSinceV10,
+                category=LangGraphDeprecatedSinceV05,
                 stacklevel=2,
             )
             if output_schema is None:
@@ -436,7 +446,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]):
         if (retry := kwargs.get("retry", UNSET)) is not UNSET:
             warnings.warn(
                 "`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
-                category=LangGraphDeprecatedSinceV10,
+                category=LangGraphDeprecatedSinceV05,
             )
             if retry_policy is None:
                 retry_policy = retry  # type: ignore[assignment]
@@ -444,7 +454,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]):
         if (input_ := kwargs.get("input", UNSET)) is not UNSET:
             warnings.warn(
                 "`input` is deprecated and will be removed. Please use `input_schema` instead.",
-                category=LangGraphDeprecatedSinceV10,
+                category=LangGraphDeprecatedSinceV05,
             )
             if input_schema is None:
                 input_schema = cast(Union[type[InputT], None], input_)
@@ -535,7 +545,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]):
         if input_schema is not None:
             self._add_schema(input_schema)
         self.nodes[node] = StateNodeSpec(
-            coerce_to_runnable(action, name=node, trace=False),  # type: ignore
+            coerce_to_runnable(action, name=node, trace=False),  # type: ignore[arg-type]
             metadata,
             input=input_schema or self.state_schema,
             retry_policy=retry_policy,
@@ -849,13 +859,6 @@ class StateGraph(Generic[StateT, InputT, OutputT]):
             builder=self,
             schema_to_mapper={},
             config_type=self.config_schema,
-            input_model=(
-                self.input_schema
-                if len(self.channels) > 1
-                and isclass(self.input_schema)
-                and issubclass(self.input_schema, BaseModel)
-                else None
-            ),
             nodes={},
             channels={
                 **self.channels,
@@ -1006,10 +1009,7 @@ class CompiledStateGraph(
             if input_schema in self.schema_to_mapper:
                 mapper = self.schema_to_mapper[input_schema]
             else:
-                mapper = _pick_mapper(
-                    input_channels,
-                    input_schema,
-                )
+                mapper = _pick_mapper(input_channels, input_schema)
                 self.schema_to_mapper[input_schema] = mapper
 
             branch_channel = CHANNEL_BRANCH_TO.format(key)
@@ -1111,6 +1111,7 @@ class CompiledStateGraph(
 
     def _migrate_checkpoint(self, checkpoint: Checkpoint) -> None:
         """Migrate a checkpoint to new channel layout."""
+        super()._migrate_checkpoint(checkpoint)
 
         values = checkpoint["channel_values"]
         versions = checkpoint["channel_versions"]

libs/langgraph/langgraph/pregel/__init__.py

@@ -32,7 +32,6 @@ from langgraph.checkpoint.base import (
     BaseCheckpointSaver,
     Checkpoint,
     CheckpointTuple,
-    copy_checkpoint,
 )
 from langgraph.config import get_config
 from langgraph.constants import (
@@ -60,7 +59,6 @@ from langgraph.constants import (
     NS_SEP,
     NULL_TASK_ID,
     PUSH,
-    SCHEDULED,
     TASKS,
 )
 from langgraph.errors import (
@@ -80,6 +78,7 @@ from langgraph.pregel.algo import (
 from langgraph.pregel.call import identifier
 from langgraph.pregel.checkpoint import (
     channels_from_checkpoint,
+    copy_checkpoint,
     create_checkpoint,
     empty_checkpoint,
 )
@@ -590,8 +589,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
 
     config_type: type[Any] | None = None
 
-    input_model: type[BaseModel] | None = None
-
     config: RunnableConfig | None = None
 
     name: str = "LangGraph"
@@ -619,7 +616,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
         retry_policy: RetryPolicy | Sequence[RetryPolicy] = (),
         cache_policy: CachePolicy | None = None,
         config_type: type[Any] | None = None,
-        input_model: type[BaseModel] | None = None,
         config: RunnableConfig | None = None,
         trigger_to_nodes: Mapping[str, Sequence[str]] | None = None,
         name: str = "LangGraph",
@@ -651,7 +647,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
         )
         self.cache_policy = cache_policy
         self.config_type = config_type
-        self.input_model = input_model
         self.config = config
         self.trigger_to_nodes = trigger_to_nodes or {}
         self.name = name
@@ -750,6 +745,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
         validate_graph(
             self.nodes,
             {k: v for k, v in self.channels.items() if isinstance(v, BaseChannel)},
+            {k: v for k, v in self.channels.items() if not isinstance(v, BaseChannel)},
             self.input_channels,
             self.output_channels,
             self.stream_channels,
@@ -788,8 +784,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 return channel.UpdateType
 
     def get_input_schema(self, config: RunnableConfig | None = None) -> type[BaseModel]:
-        if self.input_model is not None:
-            return self.input_model
         config = merge_configs(self.config, config)
         if isinstance(self.input_channels, str):
             return super().get_input_schema(config)
@@ -914,7 +908,12 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
 
     def _migrate_checkpoint(self, checkpoint: Checkpoint) -> None:
         """Migrate a saved checkpoint to new channel layout."""
-        pass
+        if checkpoint["v"] < 4 and checkpoint.get("pending_sends"):
+            pending_sends: list[Send] = checkpoint.pop("pending_sends")
+            checkpoint["channel_values"][TASKS] = pending_sends
+            checkpoint["channel_versions"][TASKS] = max(
+                checkpoint["channel_versions"].values()
+            )
 
     def _prepare_state_snapshot(
         self,
@@ -1008,7 +1007,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
             )
         if apply_pending_writes and saved.pending_writes:
             for tid, k, v in saved.pending_writes:
-                if k in (ERROR, INTERRUPT, SCHEDULED):
+                if k in (ERROR, INTERRUPT):
                     continue
                 if tid not in next_tasks:
                     continue
@@ -1127,7 +1126,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
             )
         if apply_pending_writes and saved.pending_writes:
             for tid, k, v in saved.pending_writes:
-                if k in (ERROR, INTERRUPT, SCHEDULED):
+                if k in (ERROR, INTERRUPT):
                     continue
                 if tid not in next_tasks:
                     continue
@@ -1416,10 +1415,8 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                     )
                 },
             )
-            checkpoint_metadata = config["metadata"]
             if saved:
                 checkpoint_config = patch_configurable(config, saved.config[CONF])
-                checkpoint_metadata = {**saved.metadata, **checkpoint_metadata}
             channels, managed = channels_from_checkpoint(
                 self.channels,
                 checkpoint,
@@ -1466,7 +1463,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                         )
                     # apply writes from tasks that already ran
                     for tid, k, v in saved.pending_writes or []:
-                        if k in (ERROR, INTERRUPT, SCHEDULED):
+                        if k in (ERROR, INTERRUPT):
                             continue
                         if tid not in next_tasks:
                             continue
@@ -1484,7 +1481,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                     checkpoint_config,
                     create_checkpoint(checkpoint, None, step),
                     {
-                        **checkpoint_metadata,
                         "source": "update",
                         "step": step + 1,
                         "parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1507,7 +1503,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                     checkpoint_config,
                     next_checkpoint,
                     {
-                        **checkpoint_metadata,
                         "source": "update",
                         "step": step + 1,
                         "parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1544,9 +1539,11 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                         checkpoint_config,
                         create_checkpoint(checkpoint, channels, next_step),
                         {
-                            **checkpoint_metadata,
                             "source": "input",
                             "step": next_step,
+                            "parents": saved.metadata.get("parents", {})
+                            if saved
+                            else {},
                         },
                         get_new_channel_versions(
                             checkpoint_previous_versions,
@@ -1582,7 +1579,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                     saved.parent_config or saved.config if saved else checkpoint_config,
                     next_checkpoint,
                     {
-                        **checkpoint_metadata,
                         "source": "fork",
                         "step": step + 1,
                         "parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1630,7 +1626,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                     )
                 # apply writes
                 for tid, k, v in saved.pending_writes:
-                    if k in (ERROR, INTERRUPT, SCHEDULED):
+                    if k in (ERROR, INTERRUPT):
                         continue
                     if tid not in next_tasks:
                         continue
@@ -1744,7 +1740,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 checkpoint_config,
                 checkpoint,
                 {
-                    **checkpoint_metadata,
                     "source": "update",
                     "step": step + 1,
                     "parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1838,10 +1833,8 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                     )
                 },
             )
-            checkpoint_metadata = config["metadata"]
             if saved:
                 checkpoint_config = patch_configurable(config, saved.config[CONF])
-                checkpoint_metadata = {**saved.metadata, **checkpoint_metadata}
             channels, managed = channels_from_checkpoint(
                 self.channels,
                 checkpoint,
@@ -1886,7 +1879,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                         )
                     # apply writes from tasks that already ran
                     for tid, k, v in saved.pending_writes or []:
-                        if k in (ERROR, INTERRUPT, SCHEDULED):
+                        if k in (ERROR, INTERRUPT):
                             continue
                         if tid not in next_tasks:
                             continue
@@ -1904,7 +1897,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                     checkpoint_config,
                     create_checkpoint(checkpoint, None, step),
                     {
-                        **checkpoint_metadata,
                         "source": "update",
                         "step": step + 1,
                         "parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1927,7 +1919,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                     checkpoint_config,
                     next_checkpoint,
                     {
-                        **checkpoint_metadata,
                         "source": "update",
                         "step": step + 1,
                         "parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1964,9 +1955,11 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                         checkpoint_config,
                         create_checkpoint(checkpoint, channels, next_step),
                         {
-                            **checkpoint_metadata,
                             "source": "input",
                             "step": next_step,
+                            "parents": saved.metadata.get("parents", {})
+                            if saved
+                            else {},
                         },
                         get_new_channel_versions(
                             checkpoint_previous_versions,
@@ -2002,7 +1995,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                     saved.parent_config or saved.config if saved else checkpoint_config,
                     next_checkpoint,
                     {
-                        **checkpoint_metadata,
                         "source": "fork",
                         "step": step + 1,
                         "parents": saved.metadata.get("parents", {}) if saved else {},
@@ -2049,7 +2041,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                         self.trigger_to_nodes,
                     )
                 for tid, k, v in saved.pending_writes:
-                    if k in (ERROR, INTERRUPT, SCHEDULED):
+                    if k in (ERROR, INTERRUPT):
                         continue
                     if tid not in next_tasks:
                         continue
@@ -2162,7 +2154,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 checkpoint_config,
                 checkpoint,
                 {
-                    **checkpoint_metadata,
                     "source": "update",
                     "step": step + 1,
                     "parents": saved.metadata.get("parents", {}) if saved else {},
@@ -2304,7 +2295,8 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 - `"custom"`: Emit custom data from inside nodes or tasks using `StreamWriter`.
                 - `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks.
                     Will be emitted as 2-tuples `(LLM token, metadata)`.
-                - `"debug"`: Emit debug events with as much information as possible for each step.
+                - `"checkpoints"`: Emit an event when a checkpoint is created, in the same format as returned by get_state().
+                - `"tasks"`: Emit events when tasks start and finish, including their results and errors.
 
                 You can pass a list as the `stream_mode` parameter to stream multiple modes at once.
                 The streamed outputs will be tuples of `(mode, data)`.
@@ -2404,7 +2396,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 config[CONF][CONFIG_KEY_CHECKPOINT_DURING] = checkpoint_during
             with SyncPregelLoop(
                 input,
-                input_model=self.input_model,
                 stream=StreamProtocol(stream.put, stream_modes),
                 config=config,
                 store=store,
@@ -2413,6 +2404,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 nodes=self.nodes,
                 specs=self.channels,
                 output_keys=output_keys,
+                input_keys=self.input_channels,
                 stream_keys=self.stream_channels_asis,
                 interrupt_before=interrupt_before_,
                 interrupt_after=interrupt_after_,
@@ -2420,7 +2412,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 debug=debug,
                 checkpoint_during=checkpoint_during
                 if checkpoint_during is not None
-                else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, False),
+                else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, True),
                 trigger_to_nodes=self.trigger_to_nodes,
                 migrate_checkpoint=self._migrate_checkpoint,
                 retry_policy=self.retry_policy,
@@ -2467,7 +2459,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 # Channel updates from step N are only visible in step N+1
                 # channels are guaranteed to be immutable for the duration of the step,
                 # with channel updates applied only at the transition between steps.
-                while loop.tick(input_keys=self.input_channels):
+                while loop.tick():
                     for task in loop.match_cached_writes():
                         loop.output_writes(task.id, task.writes, cached=True)
                     for _ in runner.tick(
@@ -2478,6 +2470,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                     ):
                         # emit output
                         yield from output()
+                    loop.after_tick()
             # emit output
             yield from output()
             # handle exit
@@ -2647,7 +2640,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 config[CONF][CONFIG_KEY_CHECKPOINT_DURING] = checkpoint_during
             async with AsyncPregelLoop(
                 input,
-                input_model=self.input_model,
                 stream=StreamProtocol(stream.put_nowait, stream_modes),
                 config=config,
                 store=store,
@@ -2656,6 +2648,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 nodes=self.nodes,
                 specs=self.channels,
                 output_keys=output_keys,
+                input_keys=self.input_channels,
                 stream_keys=self.stream_channels_asis,
                 interrupt_before=interrupt_before_,
                 interrupt_after=interrupt_after_,
@@ -2663,7 +2656,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 debug=debug,
                 checkpoint_during=checkpoint_during
                 if checkpoint_during is not None
-                else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, False),
+                else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, True),
                 trigger_to_nodes=self.trigger_to_nodes,
                 migrate_checkpoint=self._migrate_checkpoint,
                 retry_policy=self.retry_policy,
@@ -2701,7 +2694,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 # channel updates from step N are only visible in step N+1
                 # channels are guaranteed to be immutable for the duration of the step,
                 # with channel updates applied only at the transition between steps
-                while loop.tick(input_keys=self.input_channels):
+                while loop.tick():
                     for task in await loop.amatch_cached_writes():
                         loop.output_writes(task.id, task.writes, cached=True)
                     async for _ in runner.atick(
@@ -2713,6 +2706,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                         # emit output
                         for o in output():
                             yield o
+                    loop.after_tick()
             # emit output
             for o in output():
                 yield o

libs/langgraph/langgraph/pregel/algo.py

@@ -83,7 +83,7 @@ from langgraph.types import (
 )
 from langgraph.utils.config import merge_configs, patch_config
 
-GetNextVersion = Callable[[Optional[V]], V]
+GetNextVersion = Callable[[Optional[V], None], V]
 SUPPORTS_EXC_NOTES = sys.version_info >= (3, 11)
 
 
@@ -214,7 +214,7 @@ def local_read(
     return values
 
 
-def increment(current: int | None) -> int:
+def increment(current: int | None, channel: None) -> int:
     """Default channel versioning function, increments the current int version."""
     return current + 1 if current is not None else 1
 
@@ -265,7 +265,8 @@ def apply_writes(
         next_version = get_next_version(
             max(checkpoint["channel_versions"].values())
             if checkpoint["channel_versions"]
-            else None
+            else None,
+            None,
         )
 
     # Consume all channels that were read
@@ -1034,7 +1035,7 @@ def _proc_input(
             else:
                 return MISSING
         else:
-            val = managed[proc.channels].get(scratchpad)
+            return MISSING
     else:
         raise RuntimeError(
             f"Invalid channels type, expected list or dict, got {proc.channels}"

libs/langgraph/langgraph/pregel/checkpoint.py

@@ -71,3 +71,14 @@ def channels_from_checkpoint(
         },
         managed_specs,
     )
+
+
+def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
+    return Checkpoint(
+        v=checkpoint["v"],
+        ts=checkpoint["ts"],
+        id=checkpoint["id"],
+        channel_values=checkpoint["channel_values"].copy(),
+        channel_versions=checkpoint["channel_versions"].copy(),
+        versions_seen={k: v.copy() for k, v in checkpoint["versions_seen"].items()},
+    )

libs/langgraph/langgraph/pregel/debug.py

@@ -3,13 +3,8 @@ from __future__ import annotations
 from collections import defaultdict
 from collections.abc import Iterable, Iterator, Mapping, Sequence
 from dataclasses import asdict
-from datetime import datetime, timezone
 from pprint import pformat
-from typing import (
-    Any,
-    Literal,
-    Union,
-)
+from typing import Any
 from uuid import UUID
 
 from langchain_core.runnables.config import RunnableConfig
@@ -17,7 +12,7 @@ from langchain_core.utils.input import get_bolded_text, get_colored_text
 from typing_extensions import TypedDict
 
 from langgraph.channels.base import BaseChannel
-from langgraph.checkpoint.base import Checkpoint, CheckpointMetadata, PendingWrite
+from langgraph.checkpoint.base import CheckpointMetadata, PendingWrite
 from langgraph.constants import (
     CONF,
     CONFIG_KEY_CHECKPOINT_NS,
@@ -66,82 +61,43 @@ class CheckpointPayload(TypedDict):
     tasks: list[CheckpointTask]
 
 
-class DebugOutputBase(TypedDict):
-    timestamp: str
-    step: int
-
-
-class DebugOutputTask(DebugOutputBase):
-    type: Literal["task"]
-    payload: TaskPayload
-
-
-class DebugOutputTaskResult(DebugOutputBase):
-    type: Literal["task_result"]
-    payload: TaskResultPayload
-
-
-class DebugOutputCheckpoint(DebugOutputBase):
-    type: Literal["checkpoint"]
-    payload: CheckpointPayload
-
-
-DebugOutput = Union[DebugOutputTask, DebugOutputTaskResult, DebugOutputCheckpoint]
-
-
 TASK_NAMESPACE = UUID("6ba7b831-9dad-11d1-80b4-00c04fd430c8")
 
 
-def map_debug_tasks(
-    step: int, tasks: Iterable[PregelExecutableTask]
-) -> Iterator[DebugOutputTask]:
+def map_debug_tasks(tasks: Iterable[PregelExecutableTask]) -> Iterator[TaskPayload]:
     """Produce "task" events for stream_mode=debug."""
-    ts = datetime.now(timezone.utc).isoformat()
     for task in tasks:
         if task.config is not None and TAG_HIDDEN in task.config.get("tags", []):
             continue
 
         yield {
-            "type": "task",
-            "timestamp": ts,
-            "step": step,
-            "payload": {
-                "id": task.id,
-                "name": task.name,
-                "input": task.input,
-                "triggers": task.triggers,
-            },
+            "id": task.id,
+            "name": task.name,
+            "input": task.input,
+            "triggers": task.triggers,
         }
 
 
 def map_debug_task_results(
-    step: int,
     task_tup: tuple[PregelExecutableTask, Sequence[tuple[str, Any]]],
     stream_keys: str | Sequence[str],
-) -> Iterator[DebugOutputTaskResult]:
+) -> Iterator[TaskResultPayload]:
     """Produce "task_result" events for stream_mode=debug."""
     stream_channels_list = (
         [stream_keys] if isinstance(stream_keys, str) else stream_keys
     )
     task, writes = task_tup
     yield {
-        "type": "task_result",
-        "timestamp": datetime.now(timezone.utc).isoformat(),
-        "step": step,
-        "payload": {
-            "id": task.id,
-            "name": task.name,
-            "error": next((w[1] for w in writes if w[0] == ERROR), None),
-            "result": [
-                w for w in writes if w[0] in stream_channels_list or w[0] == RETURN
-            ],
-            "interrupts": [
-                asdict(v)
-                for w in writes
-                if w[0] == INTERRUPT
-                for v in (w[1] if isinstance(w[1], Sequence) else [w[1]])
-            ],
-        },
+        "id": task.id,
+        "name": task.name,
+        "error": next((w[1] for w in writes if w[0] == ERROR), None),
+        "result": [w for w in writes if w[0] in stream_channels_list or w[0] == RETURN],
+        "interrupts": [
+            asdict(v)
+            for w in writes
+            if w[0] == INTERRUPT
+            for v in (w[1] if isinstance(w[1], Sequence) else [w[1]])
+        ],
     }
 
 
@@ -159,17 +115,15 @@ def rm_pregel_keys(config: RunnableConfig | None) -> RunnableConfig | None:
 
 
 def map_debug_checkpoint(
-    step: int,
     config: RunnableConfig,
     channels: Mapping[str, BaseChannel],
     stream_channels: str | Sequence[str],
     metadata: CheckpointMetadata,
-    checkpoint: Checkpoint,
     tasks: Iterable[PregelExecutableTask],
     pending_writes: list[PendingWrite],
     parent_config: RunnableConfig | None,
     output_keys: str | Sequence[str],
-) -> Iterator[DebugOutputCheckpoint]:
+) -> Iterator[CheckpointPayload]:
     """Produce "checkpoint" events for stream_mode=debug."""
 
     parent_ns = config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, "")
@@ -193,42 +147,35 @@ def map_debug_checkpoint(
         }
 
     yield {
-        "type": "checkpoint",
-        "timestamp": checkpoint["ts"],
-        "step": step,
-        "payload": {
-            "config": rm_pregel_keys(patch_checkpoint_map(config, metadata)),
-            "parent_config": rm_pregel_keys(
-                patch_checkpoint_map(parent_config, metadata)
-            ),
-            "values": read_channels(channels, stream_channels),
-            "metadata": metadata,
-            "next": [t.name for t in tasks],
-            "tasks": [
-                {
-                    "id": t.id,
-                    "name": t.name,
-                    "error": t.error,
-                    "state": t.state,
-                }
-                if t.error
-                else {
-                    "id": t.id,
-                    "name": t.name,
-                    "result": t.result,
-                    "interrupts": tuple(asdict(i) for i in t.interrupts),
-                    "state": t.state,
-                }
-                if t.result
-                else {
-                    "id": t.id,
-                    "name": t.name,
-                    "interrupts": tuple(asdict(i) for i in t.interrupts),
-                    "state": t.state,
-                }
-                for t in tasks_w_writes(tasks, pending_writes, task_states, output_keys)
-            ],
-        },
+        "config": rm_pregel_keys(patch_checkpoint_map(config, metadata)),
+        "parent_config": rm_pregel_keys(patch_checkpoint_map(parent_config, metadata)),
+        "values": read_channels(channels, stream_channels),
+        "metadata": metadata,
+        "next": [t.name for t in tasks],
+        "tasks": [
+            {
+                "id": t.id,
+                "name": t.name,
+                "error": t.error,
+                "state": t.state,
+            }
+            if t.error
+            else {
+                "id": t.id,
+                "name": t.name,
+                "result": t.result,
+                "interrupts": tuple(asdict(i) for i in t.interrupts),
+                "state": t.state,
+            }
+            if t.result
+            else {
+                "id": t.id,
+                "name": t.name,
+                "interrupts": tuple(asdict(i) for i in t.interrupts),
+                "state": t.state,
+            }
+            for t in tasks_w_writes(tasks, pending_writes, task_states, output_keys)
+        ],
     }
 
 

libs/langgraph/langgraph/pregel/loop.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import asyncio
 import binascii
 import concurrent.futures
-import dataclasses
 from collections import defaultdict, deque
 from collections.abc import Iterator, Mapping, Sequence
 from contextlib import (
@@ -12,6 +11,7 @@ from contextlib import (
     AsyncExitStack,
     ExitStack,
 )
+from datetime import datetime, timezone
 from inspect import signature
 from types import TracebackType
 from typing import (
@@ -25,13 +25,11 @@ from typing import (
 
 from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager
 from langchain_core.runnables import RunnableConfig
-from pydantic import BaseModel
 from typing_extensions import ParamSpec, Self
 
 from langgraph.cache.base import BaseCache
 from langgraph.channels.base import BaseChannel
 from langgraph.checkpoint.base import (
-    EXCLUDED_METADATA_KEYS,
     WRITES_IDX_MAP,
     BaseCheckpointSaver,
     ChannelVersions,
@@ -39,16 +37,12 @@ from langgraph.checkpoint.base import (
     CheckpointMetadata,
     CheckpointTuple,
     PendingWrite,
-    copy_checkpoint,
 )
 from langgraph.constants import (
     CONF,
     CONFIG_KEY_CHECKPOINT_ID,
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_CHECKPOINT_NS,
-    CONFIG_KEY_DEDUPE_TASKS,
-    CONFIG_KEY_DELEGATE,
-    CONFIG_KEY_ENSURE_LATEST,
     CONFIG_KEY_RESUME_MAP,
     CONFIG_KEY_RESUMING,
     CONFIG_KEY_SCRATCHPAD,
@@ -60,17 +54,15 @@ from langgraph.constants import (
     INPUT,
     INTERRUPT,
     MISSING,
+    NS_END,
     NS_SEP,
     NULL_TASK_ID,
     PUSH,
     RESUME,
-    SCHEDULED,
     TAG_HIDDEN,
 )
 from langgraph.errors import (
-    CheckpointNotLatest,
     EmptyInputError,
-    GraphDelegate,
     GraphInterrupt,
 )
 from langgraph.managed.base import (
@@ -91,6 +83,7 @@ from langgraph.pregel.algo import (
 )
 from langgraph.pregel.checkpoint import (
     channels_from_checkpoint,
+    copy_checkpoint,
     create_checkpoint,
     empty_checkpoint,
 )
@@ -125,6 +118,7 @@ from langgraph.types import (
     PregelScratchpad,
     RetryPolicy,
     StreamChunk,
+    StreamMode,
     StreamProtocol,
 )
 from langgraph.utils.config import patch_configurable
@@ -132,9 +126,7 @@ from langgraph.utils.config import patch_configurable
 V = TypeVar("V")
 P = ParamSpec("P")
 
-INPUT_DONE = object()
-INPUT_RESUMING = object()
-INPUT_SHOULD_VALIDATE = object()
+
 WritesT = Sequence[tuple[str, Any]]
 
 
@@ -155,11 +147,11 @@ class PregelLoop:
     stop: int
 
     input: Any | None
-    input_model: type[BaseModel] | None
     cache: BaseCache[WritesT] | None
     checkpointer: BaseCheckpointSaver | None
     nodes: Mapping[str, PregelNode]
     specs: Mapping[str, BaseChannel | ManagedValueSpec]
+    input_keys: str | Sequence[str]
     output_keys: str | Sequence[str]
     stream_keys: str | Sequence[str]
     skip_done_tasks: bool
@@ -202,11 +194,16 @@ class PregelLoop:
     prev_checkpoint_config: RunnableConfig | None
 
     status: Literal[
-        "pending", "done", "interrupt_before", "interrupt_after", "out_of_steps"
+        "input",
+        "pending",
+        "done",
+        "interrupt_before",
+        "interrupt_after",
+        "out_of_steps",
     ]
     tasks: dict[str, PregelExecutableTask]
-    to_interrupt: list[PregelExecutableTask]
     output: None | dict[str, Any] | Any = None
+    updated_channels: set[str] | None = None
 
     # public
 
@@ -221,13 +218,13 @@ class PregelLoop:
         checkpointer: BaseCheckpointSaver | None,
         nodes: Mapping[str, PregelNode],
         specs: Mapping[str, BaseChannel | ManagedValueSpec],
+        input_keys: str | Sequence[str],
         output_keys: str | Sequence[str],
         stream_keys: str | Sequence[str],
         trigger_to_nodes: Mapping[str, Sequence[str]],
         interrupt_after: All | Sequence[str] = EMPTY_SEQ,
         interrupt_before: All | Sequence[str] = EMPTY_SEQ,
         manager: None | AsyncParentRunManager | ParentRunManager = None,
-        input_model: type[BaseModel] | None = None,
         debug: bool = False,
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
@@ -240,21 +237,18 @@ class PregelLoop:
         self.step = 0
         self.stop = 0
         self.input = input
-        self.input_model = input_model
         self.checkpointer = checkpointer
         self.cache = cache
         self.nodes = nodes
         self.specs = specs
+        self.input_keys = input_keys
         self.output_keys = output_keys
         self.stream_keys = stream_keys
         self.interrupt_after = interrupt_after
         self.interrupt_before = interrupt_before
         self.manager = manager
         self.is_nested = CONFIG_KEY_TASK_ID in self.config.get(CONF, {})
-        self.skip_done_tasks = (
-            CONFIG_KEY_CHECKPOINT_ID not in config[CONF]
-            or CONFIG_KEY_DEDUPE_TASKS in config[CONF]
-        )
+        self.skip_done_tasks = CONFIG_KEY_CHECKPOINT_ID not in config[CONF]
         self._migrate_checkpoint = migrate_checkpoint
         self.trigger_to_nodes = trigger_to_nodes
         self.retry_policy = retry_policy
@@ -264,9 +258,7 @@ class PregelLoop:
         if self.stream is not None and CONFIG_KEY_STREAM in config[CONF]:
             self.stream = DuplexStream(self.stream, config[CONF][CONFIG_KEY_STREAM])
         scratchpad: PregelScratchpad | None = config[CONF].get(CONFIG_KEY_SCRATCHPAD)
-        if not self.config[CONF].get(CONFIG_KEY_DELEGATE) and isinstance(
-            scratchpad, PregelScratchpad
-        ):
+        if isinstance(scratchpad, PregelScratchpad):
             # if count is > 0, append to checkpoint_ns
             # if count is 0, leave as is
             if cnt := scratchpad.subgraph_counter():
@@ -321,10 +313,22 @@ class PregelLoop:
         # deduplicate writes to special channels, last write wins
         if all(w[0] in WRITES_IDX_MAP for w in writes):
             writes = list({w[0]: w for w in writes}.values())
-        # remove existing writes for this task
-        self.checkpoint_pending_writes = [
-            w for w in self.checkpoint_pending_writes if w[0] != task_id
-        ]
+        if task_id == NULL_TASK_ID:
+            # writes for the null task are accumulated
+            self.checkpoint_pending_writes = [
+                w
+                for w in self.checkpoint_pending_writes
+                if w[0] != task_id or w[1] not in WRITES_IDX_MAP
+            ]
+            writes_to_save: WritesT = [
+                w[1:] for w in self.checkpoint_pending_writes if w[0] == task_id
+            ] + list(writes)
+        else:
+            # remove existing writes for this task
+            self.checkpoint_pending_writes = [
+                w for w in self.checkpoint_pending_writes if w[0] != task_id
+            ]
+            writes_to_save = writes
         # save writes
         self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes)
         if self.checkpoint_during and self.checkpointer_put_writes is not None:
@@ -345,7 +349,7 @@ class PregelLoop:
                 self.submit(
                     self.checkpointer_put_writes,
                     config,
-                    writes,
+                    writes_to_save,
                     task_id,
                     task_path_str(task.path) if task else "",
                 )
@@ -353,7 +357,7 @@ class PregelLoop:
                 self.submit(
                     self.checkpointer_put_writes,
                     config,
-                    writes,
+                    writes_to_save,
                     task_id,
                 )
         # output writes
@@ -404,12 +408,6 @@ class PregelLoop:
         self, task: PregelExecutableTask, write_idx: int, call: Call | None = None
     ) -> PregelExecutableTask | None:
         """Accept a PUSH from a task, potentially returning a new task to start."""
-        # don't start if we should interrupt *after* the original task
-        if self.interrupt_after and should_interrupt(
-            self.checkpoint, self.interrupt_after, [task]
-        ):
-            self.to_interrupt.append(task)
-            return
         checkpoint_id_bytes = binascii.unhexlify(self.checkpoint["id"].replace("-", ""))
         null_version = checkpoint_null_version(self.checkpoint)
         if pushed := cast(
@@ -435,14 +433,8 @@ class PregelLoop:
                 cache_policy=self.cache_policy,
             ),
         ):
-            # don't start if we should interrupt *before* the new task
-            if self.interrupt_before and should_interrupt(
-                self.checkpoint, self.interrupt_before, [pushed]
-            ):
-                self.to_interrupt.append(pushed)
-                return
             # produce debug output
-            self._emit("debug", map_debug_tasks, self.step, [pushed])
+            self._emit("tasks", map_debug_tasks, [pushed])
             # debug flag
             if self.debug:
                 print_step_tasks(self.step, [pushed])
@@ -454,11 +446,7 @@ class PregelLoop:
             # return the new task, to be started if not run before
             return pushed
 
-    def tick(
-        self,
-        *,
-        input_keys: str | Sequence[str],
-    ) -> bool:
+    def tick(self) -> bool:
         """Execute a single iteration of the Pregel loop.
 
         Args:
@@ -467,72 +455,6 @@ class PregelLoop:
         Returns:
             True if more iterations are needed.
         """
-        if self.status != "pending":
-            raise RuntimeError("Cannot tick when status is no longer 'pending'")
-
-        updated_channels: set[str] | None = None
-
-        if self.input not in (INPUT_DONE, INPUT_RESUMING, INPUT_SHOULD_VALIDATE):
-            updated_channels = self._first(input_keys=input_keys)
-        elif self.to_interrupt:
-            # if we need to interrupt, do so
-            self.status = "interrupt_before"
-            raise GraphInterrupt()
-        elif all(task.writes for task in self.tasks.values()):
-            # finish superstep
-            writes = [w for t in self.tasks.values() for w in t.writes]
-            # debug flag
-            if self.debug:
-                print_step_writes(
-                    self.step,
-                    writes,
-                    (
-                        [self.stream_keys]
-                        if isinstance(self.stream_keys, str)
-                        else self.stream_keys
-                    ),
-                )
-            # all tasks have finished
-            updated_channels = apply_writes(
-                self.checkpoint,
-                self.channels,
-                self.tasks.values(),
-                self.checkpointer_get_next_version,
-                self.trigger_to_nodes,
-            )
-            # validate input if requested
-            if self.input is INPUT_SHOULD_VALIDATE:
-                self.input = INPUT_DONE
-                # validate
-                cast(type[BaseModel], self.input_model)(
-                    **read_channels(self.channels, self.stream_keys)
-                )
-            # produce values output
-            if not updated_channels.isdisjoint(
-                (self.output_keys,)
-                if isinstance(self.output_keys, str)
-                else self.output_keys
-            ):
-                self._emit(
-                    "values", map_output_values, self.output_keys, writes, self.channels
-                )
-            # clear pending writes
-            self.checkpoint_pending_writes.clear()
-            # "not skip_done_tasks" only applies to first tick after resuming
-            self.skip_done_tasks = True
-            # save checkpoint
-            self._put_checkpoint({"source": "loop"})
-            # after execution, check if we should interrupt
-            if self.interrupt_after and should_interrupt(
-                self.checkpoint, self.interrupt_after, self.tasks.values()
-            ):
-                self.status = "interrupt_after"
-                raise GraphInterrupt()
-
-            # unset resuming flag
-            self.config[CONF].pop(CONFIG_KEY_RESUMING, None)
-        else:
-            return False
 
         # check if iteration limit is reached
         if self.step > self.stop:
@@ -554,18 +476,16 @@ class PregelLoop:
             store=self.store,
             checkpointer=self.checkpointer,
             trigger_to_nodes=self.trigger_to_nodes,
-            updated_channels=updated_channels,
+            updated_channels=self.updated_channels,
             retry_policy=self.retry_policy,
             cache_policy=self.cache_policy,
         )
-        self.to_interrupt = []
 
         # produce debug output
         if self._checkpointer_put_after_previous is not None:
             self._emit(
-                "debug",
+                "checkpoints",
                 map_debug_checkpoint,
-                self.step - 1,  # printing checkpoint for previous step
                 {
                     **self.checkpoint_config,
                     CONF: {
@@ -576,7 +496,6 @@ class PregelLoop:
                 self.channels,
                 self.stream_keys,
                 self.checkpoint_metadata,
-                self.checkpoint,
                 self.tasks.values(),
                 self.checkpoint_pending_writes,
                 self.prev_checkpoint_config,
@@ -588,26 +507,10 @@ class PregelLoop:
             self.status = "done"
             return False
 
-        # check if we should delegate (used by subgraphs in distributed mode)
-        if self.config[CONF].get(CONFIG_KEY_DELEGATE):
-            assert self.input is INPUT_RESUMING
-            raise GraphDelegate(
-                {
-                    "config": patch_configurable(
-                        self.config, {CONFIG_KEY_DELEGATE: False}
-                    ),
-                    "input": None,
-                }
-            )
-
         # if there are pending writes from a previous loop, apply them
         if self.skip_done_tasks and self.checkpoint_pending_writes:
             self._match_writes(self.tasks)
 
-        # if all tasks have finished, re-tick
-        if all(task.writes for task in self.tasks.values()):
-            return self.tick(input_keys=input_keys)
-
         # before execution, check if we should interrupt
         if self.interrupt_before and should_interrupt(
             self.checkpoint, self.interrupt_before, self.tasks.values()
@@ -616,7 +519,7 @@ class PregelLoop:
             raise GraphInterrupt()
 
         # produce debug output
-        self._emit("debug", map_debug_tasks, self.step, self.tasks.values())
+        self._emit("tasks", map_debug_tasks, self.tasks.values())
 
         # debug flag
         if self.debug:
@@ -629,6 +532,52 @@ class PregelLoop:
 
         return True
 
+    def after_tick(self) -> None:
+        # finish superstep
+        writes = [w for t in self.tasks.values() for w in t.writes]
+        # debug flag
+        if self.debug:
+            print_step_writes(
+                self.step,
+                writes,
+                (
+                    [self.stream_keys]
+                    if isinstance(self.stream_keys, str)
+                    else self.stream_keys
+                ),
+            )
+        # all tasks have finished
+        self.updated_channels = apply_writes(
+            self.checkpoint,
+            self.channels,
+            self.tasks.values(),
+            self.checkpointer_get_next_version,
+            self.trigger_to_nodes,
+        )
+        # produce values output
+        if not self.updated_channels.isdisjoint(
+            (self.output_keys,)
+            if isinstance(self.output_keys, str)
+            else self.output_keys
+        ):
+            self._emit(
+                "values", map_output_values, self.output_keys, writes, self.channels
+            )
+        # clear pending writes
+        self.checkpoint_pending_writes.clear()
+        # "not skip_done_tasks" only applies to first tick after resuming
+        self.skip_done_tasks = True
+        # save checkpoint
+        self._put_checkpoint({"source": "loop"})
+        # after execution, check if we should interrupt
+        if self.interrupt_after and should_interrupt(
+            self.checkpoint, self.interrupt_after, self.tasks.values()
+        ):
+            self.status = "interrupt_after"
+            raise GraphInterrupt()
+        # unset resuming flag
+        self.config[CONF].pop(CONFIG_KEY_RESUMING, None)
+
     def match_cached_writes(self) -> Sequence[PregelExecutableTask]:
         raise NotImplementedError
 
@@ -642,14 +591,7 @@ class PregelLoop:
             if k in (ERROR, INTERRUPT, RESUME):
                 continue
             if task := tasks.get(tid):
-                if k == SCHEDULED:
-                    if v == max(
-                        self.checkpoint["versions_seen"].get(INTERRUPT, {}).values(),
-                        default=None,
-                    ):
-                        self.tasks[tid] = dataclasses.replace(task, scheduled=True)
-                else:
-                    task.writes.append((k, v))
+                task.writes.append((k, v))
 
     def _first(self, *, input_keys: str | Sequence[str]) -> set[str] | None:
         # resuming from previous checkpoint requires
@@ -715,21 +657,8 @@ class PregelLoop:
             self._emit(
                 "values", map_output_values, self.output_keys, True, self.channels
             )
-            # set flag
-            self.input = INPUT_RESUMING
         # map inputs to channel updates
         elif input_writes := deque(map_input(input_keys, self.input)):
-            # TODO shouldn't these writes be passed to put_writes too?
-            # check if we should delegate (used by subgraphs in distributed mode)
-            if self.config[CONF].get(CONFIG_KEY_DELEGATE):
-                raise GraphDelegate(
-                    {
-                        "config": patch_configurable(
-                            self.config, {CONFIG_KEY_DELEGATE: False}
-                        ),
-                        "input": self.input,
-                    }
-                )
             # discard any unfinished tasks from previous checkpoint
             discard_tasks = prepare_next_tasks(
                 self.checkpoint,
@@ -758,24 +687,15 @@ class PregelLoop:
             )
             # save input checkpoint
             self._put_checkpoint({"source": "input"})
-            # set flag
-            if (
-                self.input_model is not None
-                and not isinstance(self.input, self.input_model)
-                and not isinstance(self.stream_keys, str)
-            ):
-                self.input = INPUT_SHOULD_VALIDATE
-            else:
-                self.input = INPUT_DONE
         elif CONFIG_KEY_RESUMING not in configurable:
             raise EmptyInputError(f"Received no input for {input_keys}")
-        else:
-            self.input = INPUT_DONE
         # update config
         if not self.is_nested:
             self.config = patch_configurable(
                 self.config, {CONFIG_KEY_RESUMING: is_resuming}
             )
+        # set flag
+        self.status = "pending"
         return updated_channels
 
     def _put_checkpoint(self, metadata: CheckpointMetadata) -> None:
@@ -812,11 +732,6 @@ class PregelLoop:
         )
         # bail if no checkpointer
         if do_checkpoint and self._checkpointer_put_after_previous is not None:
-            for k, v in self.config["metadata"].items():
-                if k in EXCLUDED_METADATA_KEYS:
-                    continue
-                metadata.setdefault(k, v)  # type: ignore
-
             self.prev_checkpoint_config = (
                 self.checkpoint_config
                 if CONFIG_KEY_CHECKPOINT_ID in self.checkpoint_config[CONF]
@@ -868,7 +783,14 @@ class PregelLoop:
         traceback: TracebackType | None,
     ) -> bool | None:
         # persist current checkpoint and writes
-        if not self.checkpoint_during:
+        if not self.checkpoint_during and (
+            # if it's a top graph
+            not self.is_nested
+            # or a nested graph with error or interrupt
+            or exc_value is not None
+            # or a nested graph with checkpointer=True
+            or all(NS_END not in part for part in self.checkpoint_ns)
+        ):
             self._put_checkpoint(self.checkpoint_metadata)
             self._put_pending_writes()
         # suppress interrupt
@@ -917,17 +839,39 @@ class PregelLoop:
 
     def _emit(
         self,
-        mode: str,
+        mode: StreamMode,
         values: Callable[P, Iterator[Any]],
         *args: P.args,
         **kwargs: P.kwargs,
     ) -> None:
         if self.stream is None:
             return
-        if mode not in self.stream.modes:
+        debug_remap = mode in ("checkpoints", "tasks") and "debug" in self.stream.modes
+        if mode not in self.stream.modes and not debug_remap:
             return
         for v in values(*args, **kwargs):
-            self.stream((self.checkpoint_ns, mode, v))
+            if mode in self.stream.modes:
+                self.stream((self.checkpoint_ns, mode, v))
+            # "debug" mode is "checkpoints" or "tasks" with a wrapper dict
+            if debug_remap:
+                self.stream(
+                    (
+                        self.checkpoint_ns,
+                        "debug",
+                        {
+                            "step": self.step - 1
+                            if mode == "checkpoints"
+                            else self.step,
+                            "timestamp": datetime.now(timezone.utc).isoformat(),
+                            "type": "checkpoint"
+                            if mode == "checkpoints"
+                            else "task_result"
+                            if "result" in v
+                            else "task",
+                            "payload": v,
+                        },
+                    )
+                )
 
     def output_writes(
         self, task_id: str, writes: WritesT, *, cached: bool = False
@@ -968,9 +912,8 @@ class PregelLoop:
                 )
             if not cached:
                 self._emit(
-                    "debug",
+                    "tasks",
                     map_debug_task_results,
-                    self.step,
                     (task, writes),
                     self.stream_keys,
                 )
@@ -992,9 +935,9 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         manager: None | AsyncParentRunManager | ParentRunManager = None,
         interrupt_after: All | Sequence[str] = EMPTY_SEQ,
         interrupt_before: All | Sequence[str] = EMPTY_SEQ,
+        input_keys: str | Sequence[str] = EMPTY_SEQ,
         output_keys: str | Sequence[str] = EMPTY_SEQ,
         stream_keys: str | Sequence[str] = EMPTY_SEQ,
-        input_model: type[BaseModel] | None = None,
         debug: bool = False,
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
@@ -1003,7 +946,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
     ) -> None:
         super().__init__(
             input,
-            input_model=input_model,
             stream=stream,
             config=config,
             checkpointer=checkpointer,
@@ -1011,6 +953,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             store=store,
             nodes=nodes,
             specs=specs,
+            input_keys=input_keys,
             output_keys=output_keys,
             stream_keys=stream_keys,
             interrupt_after=interrupt_after,
@@ -1097,25 +1040,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
     # context manager
 
     def __enter__(self) -> Self:
-        if self.config.get(CONF, {}).get(
-            CONFIG_KEY_ENSURE_LATEST
-        ) and self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
-            if self.checkpointer is None:
-                raise RuntimeError(
-                    "Cannot ensure latest checkpoint without checkpointer"
-                )
-            saved = self.checkpointer.get_tuple(
-                patch_configurable(
-                    self.checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
-                )
-            )
-            if (
-                saved is None
-                or saved.checkpoint["id"]
-                != self.checkpoint_config[CONF][CONFIG_KEY_CHECKPOINT_ID]
-            ):
-                raise CheckpointNotLatest
-        elif self.checkpointer:
+        if self.checkpointer:
             saved = self.checkpointer.get_tuple(self.checkpoint_config)
         else:
             saved = None
@@ -1149,10 +1074,11 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             self.specs, self.checkpoint
         )
         self.stack.push(self._suppress_interrupt)
-        self.status = "pending"
+        self.status = "input"
         self.step = self.checkpoint_metadata["step"] + 1
         self.stop = self.step + self.config["recursion_limit"] + 1
         self.checkpoint_previous_versions = self.checkpoint["channel_versions"].copy()
+        self.updated_channels = self._first(input_keys=self.input_keys)
 
         return self
 
@@ -1182,9 +1108,9 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         interrupt_after: All | Sequence[str] = EMPTY_SEQ,
         interrupt_before: All | Sequence[str] = EMPTY_SEQ,
         manager: None | AsyncParentRunManager | ParentRunManager = None,
+        input_keys: str | Sequence[str] = EMPTY_SEQ,
         output_keys: str | Sequence[str] = EMPTY_SEQ,
         stream_keys: str | Sequence[str] = EMPTY_SEQ,
-        input_model: type[BaseModel] | None = None,
         debug: bool = False,
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
@@ -1193,7 +1119,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
     ) -> None:
         super().__init__(
             input,
-            input_model=input_model,
             stream=stream,
             config=config,
             checkpointer=checkpointer,
@@ -1201,6 +1126,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             store=store,
             nodes=nodes,
             specs=specs,
+            input_keys=input_keys,
             output_keys=output_keys,
             stream_keys=stream_keys,
             interrupt_after=interrupt_after,
@@ -1290,25 +1216,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
     # context manager
 
     async def __aenter__(self) -> Self:
-        if self.config.get(CONF, {}).get(
-            CONFIG_KEY_ENSURE_LATEST
-        ) and self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
-            if self.checkpointer is None:
-                raise RuntimeError(
-                    "Cannot ensure latest checkpoint without checkpointer"
-                )
-            saved = await self.checkpointer.aget_tuple(
-                patch_configurable(
-                    self.checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
-                )
-            )
-            if (
-                saved is None
-                or saved.checkpoint["id"]
-                != self.checkpoint_config[CONF][CONFIG_KEY_CHECKPOINT_ID]
-            ):
-                raise CheckpointNotLatest
-        elif self.checkpointer:
+        if self.checkpointer:
             saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
         else:
             saved = None
@@ -1344,11 +1252,11 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             self.specs, self.checkpoint
         )
         self.stack.push(self._suppress_interrupt)
-        self.status = "pending"
+        self.status = "input"
         self.step = self.checkpoint_metadata["step"] + 1
         self.stop = self.step + self.config["recursion_limit"] + 1
-
         self.checkpoint_previous_versions = self.checkpoint["channel_versions"].copy()
+        self.updated_channels = self._first(input_keys=self.input_keys)
 
         return self
 

libs/langgraph/langgraph/pregel/read.py

@@ -12,12 +12,11 @@ from langchain_core.runnables import Runnable, RunnableConfig
 
 from langgraph.constants import CONF, CONFIG_KEY_READ
 from langgraph.pregel.protocol import PregelProtocol
-from langgraph.pregel.retry import RetryPolicy
 from langgraph.pregel.utils import find_subgraph_pregel
 from langgraph.pregel.write import ChannelWrite
-from langgraph.types import CachePolicy
+from langgraph.types import CachePolicy, RetryPolicy
 from langgraph.utils.config import merge_configs
-from langgraph.utils.runnable import RunnableCallable, RunnableSeq, coerce_to_runnable
+from langgraph.utils.runnable import RunnableCallable, RunnableSeq
 
 READ_TYPE = Callable[[Union[str, Sequence[str]], bool], Union[Any, dict[str, Any]]]
 INPUT_CACHE_KEY_TYPE = tuple[Callable[..., Any], tuple[str, ...]]
@@ -96,7 +95,7 @@ class ChannelRead(RunnableCallable):
 DEFAULT_BOUND = RunnableCallable(lambda input: input)
 
 
-class PregelNode(Runnable):
+class PregelNode:
     """A node in a Pregel graph. This won't be invoked as a runnable by the graph
     itself, but instead acts as a container for the components necessary to make
     a PregelExecutableTask for a node."""
@@ -227,38 +226,6 @@ class PregelNode(Runnable):
             else (self.channels,),
         )
 
-    def __or__(
-        self,
-        other: Runnable[Any, Any]
-        | Callable[[Any], Any]
-        | Mapping[str, Runnable[Any, Any] | Callable[[Any], Any]],
-    ) -> PregelNode:
-        if isinstance(other, Runnable) and ChannelWrite.is_writer(other):
-            return self.copy(update=dict(writers=[*self.writers, other]))
-        elif self.bound is DEFAULT_BOUND:
-            return self.copy(
-                update=dict(bound=coerce_to_runnable(other, name=None, trace=True))
-            )
-        else:
-            return self.copy(update=dict(bound=RunnableSeq(self.bound, other)))
-
-    def pipe(
-        self,
-        *others: Runnable[Any, Any] | Callable[[Any], Any],
-        name: str | None = None,
-    ) -> PregelNode:
-        for other in others:
-            self = self | other
-        return self
-
-    def __ror__(
-        self,
-        other: Runnable[Any, Any]
-        | Callable[[Any], Any]
-        | Mapping[str, Runnable[Any, Any] | Callable[[Any], Any]],
-    ) -> PregelNode:
-        raise NotImplementedError()
-
     def invoke(
         self,
         input: Any,

libs/langgraph/langgraph/pregel/runner.py

@@ -104,7 +104,8 @@ class FuturesDict(Generic[F, E], dict[F, Optional[PregelExecutableTask]]):
         fut: F,
     ) -> None:
         try:
-            self.callback()(task, _exception(fut))  # type: ignore[misc]
+            if cb := self.callback():
+                cb(task, _exception(fut))
         finally:
             with self.lock:
                 self.done.add(fut)

libs/langgraph/langgraph/pregel/validate.py

@@ -5,6 +5,7 @@ from typing import Any
 
 from langgraph.channels.base import BaseChannel
 from langgraph.constants import RESERVED
+from langgraph.managed.base import ManagedValueMapping
 from langgraph.pregel.read import PregelNode
 from langgraph.types import All
 
@@ -12,6 +13,7 @@ from langgraph.types import All
 def validate_graph(
     nodes: Mapping[str, PregelNode],
     channels: dict[str, BaseChannel],
+    managed: ManagedValueMapping,
     input_channels: str | Sequence[str],
     output_channels: str | Sequence[str],
     stream_channels: str | Sequence[str] | None,
@@ -20,14 +22,30 @@ def validate_graph(
 ) -> None:
     for chan in channels:
         if chan in RESERVED:
-            raise ValueError(f"Channel names {chan} are reserved")
+            raise ValueError(f"Channel name '{chan}' is reserved")
+    for name in managed:
+        if name in RESERVED:
+            raise ValueError(f"Managed name '{name}' is reserved")
 
     subscribed_channels = set[str]()
     for name, node in nodes.items():
         if name in RESERVED:
-            raise ValueError(f"Node names {RESERVED} are reserved")
+            raise ValueError(f"Node name '{name}' is reserved")
         if isinstance(node, PregelNode):
             subscribed_channels.update(node.triggers)
+            if isinstance(node.channels, str):
+                if node.channels not in channels:
+                    raise ValueError(
+                        f"Node {name} reads channel '{node.channels}' "
+                        f"not in known channels: '{repr(sorted(channels))[:100]}'"
+                    )
+            else:
+                for chan in node.channels:
+                    if chan not in channels and chan not in managed:
+                        raise ValueError(
+                            f"Node {name} reads channel '{chan}' "
+                            f"not in known channels: '{repr(sorted(channels))[:100]}'"
+                        )
         else:
             raise TypeError(
                 f"Invalid node type {type(node)}, expected PregelNode or NodeBuilder"

libs/langgraph/langgraph/types.py

@@ -46,7 +46,9 @@ Checkpointer = Union[None, bool, BaseCheckpointSaver]
 - False disables checkpointing, even if the parent graph has a checkpointer.
 - None inherits checkpointer from the parent graph."""
 
-StreamMode = Literal["values", "updates", "debug", "messages", "custom"]
+StreamMode = Literal[
+    "values", "updates", "checkpoints", "tasks", "debug", "messages", "custom"
+]
 """How the stream method should emit outputs.
 
 - `"values"`: Emit all values in the state after each step, including interrupts.
@@ -55,7 +57,9 @@ StreamMode = Literal["values", "updates", "debug", "messages", "custom"]
     If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately.
 - `"custom"`: Emit custom data using from inside nodes or tasks using `StreamWriter`.
 - `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks.
-- `"debug"`: Emit debug events with as much information as possible for each step.
+- `"checkpoints"`: Emit an event when a checkpoint is created, in the same format as returned by get_state().
+- `"tasks"`: Emit events when tasks start and finish, including their results and errors.
+- `"debug"`: Emit "checlkpoints" and "tasks" events, for debugging purposes.
 """
 
 StreamWriter = Callable[[Any], None]
@@ -203,7 +207,6 @@ class PregelExecutableTask:
     cache_key: CacheKey | None
     id: str
     path: tuple[str | int | tuple, ...]
-    scheduled: bool = False
     writers: Sequence[Runnable] = ()
     subgraphs: Sequence[PregelProtocol] = ()
 

libs/langgraph/langgraph/typing.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from typing import Union
-
 from typing_extensions import TypeVar
 
 from langgraph._typing import StateLike
@@ -19,12 +17,8 @@ InputT = TypeVar("InputT", bound=StateLike, default=StateT)
 Defaults to `StateT`.
 """
 
-ResolvedInputT = TypeVar("ResolvedInputT", bound=StateLike)
-"""Type variable used to represent the resolved input to a state graph.
+OutputT = TypeVar("OutputT", bound=StateLike, default=StateT)
+"""Type variable used to represent the output of a state graph.
 
-No default.
+Defaults to `StateT`.
 """
-
-
-OutputT = TypeVar("OutputT", bound=Union[StateLike, None], default=StateT)
-"""Type variable used to represent the output of a state graph."""

libs/langgraph/langgraph/warnings.py

@@ -41,8 +41,8 @@ class LangGraphDeprecationWarning(DeprecationWarning):
         return message
 
 
-class LangGraphDeprecatedSinceV10(LangGraphDeprecationWarning):
-    """A specific `LangGraphDeprecationWarning` subclass defining functionality deprecated since LangGraph v1.0.0"""
+class LangGraphDeprecatedSinceV05(LangGraphDeprecationWarning):
+    """A specific `LangGraphDeprecationWarning` subclass defining functionality deprecated since LangGraph v0.5.0"""
 
     def __init__(self, message: str, *args: object) -> None:
-        super().__init__(message, *args, since=(1, 0), expected_removal=(2, 0))
+        super().__init__(message, *args, since=(0, 5), expected_removal=(2, 0))

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "0.4.7"
+version = "0.5.0rc1"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.9"
@@ -13,9 +13,9 @@ license = "MIT"
 license-files = ['LICENSE']
 dependencies = [
     "langchain-core>=0.1",
-    "langgraph-checkpoint>=2.0.26",
+    "langgraph-checkpoint>=2.1.0",
     "langgraph-sdk>=0.1.42",
-    "langgraph-prebuilt>=0.2.0",
+    "langgraph-prebuilt>=0.5.0rc0",
     "xxhash>=3.5.0",
     "pydantic>=2.7.4",
 ]

libs/langgraph/tests/conftest.py

@@ -12,6 +12,7 @@ from langgraph.checkpoint.base import BaseCheckpointSaver
 from langgraph.store.base import BaseStore
 from tests.conftest_checkpointer import (
     _checkpointer_memory,
+    _checkpointer_memory_migrate_sends,
     _checkpointer_postgres,
     _checkpointer_postgres_aio,
     _checkpointer_postgres_aio_pipe,
@@ -125,6 +126,7 @@ async def async_store(request: pytest.FixtureRequest) -> AsyncIterator[BaseStore
     if NO_DOCKER
     else [
         "memory",
+        "memory_migrate_sends",
         "sqlite",
         "sqlite_aes",
         "postgres",
@@ -139,6 +141,9 @@ def sync_checkpointer(
     if checkpointer_name == "memory":
         with _checkpointer_memory() as checkpointer:
             yield checkpointer
+    elif checkpointer_name == "memory_migrate_sends":
+        with _checkpointer_memory_migrate_sends() as checkpointer:
+            yield checkpointer
     elif checkpointer_name == "sqlite":
         with _checkpointer_sqlite() as checkpointer:
             yield checkpointer

libs/langgraph/tests/conftest_checkpointer.py

@@ -14,7 +14,10 @@ from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
 
 pytest.register_assert_rewrite("tests.memory_assert")
 
-from tests.memory_assert import MemorySaverAssertImmutable  # noqa: E402
+from tests.memory_assert import (  # noqa: E402
+    MemorySaverAssertImmutable,
+    MemorySaverNeedsPendingSendsMigration,
+)
 
 DEFAULT_POSTGRES_URI = "postgres://postgres:postgres@localhost:5442/"
 
@@ -24,6 +27,11 @@ def _checkpointer_memory():
     yield MemorySaverAssertImmutable()
 
 
+@contextmanager
+def _checkpointer_memory_migrate_sends():
+    yield MemorySaverNeedsPendingSendsMigration()
+
+
 @contextmanager
 def _checkpointer_sqlite():
     with SqliteSaver.from_conn_string(":memory:") as checkpointer:
@@ -187,6 +195,7 @@ async def _checkpointer_postgres_aio_pool():
 
 __all__ = [
     "_checkpointer_memory",
+    "_checkpointer_memory_migrate_sends",
     "_checkpointer_sqlite",
     "_checkpointer_sqlite_aes",
     "_checkpointer_postgres",

libs/langgraph/tests/memory_assert.py

@@ -7,6 +7,7 @@ from typing import Any, Optional
 from langchain_core.runnables import RunnableConfig
 
 from langgraph.checkpoint.base import (
+    BaseCheckpointSaver,
     ChannelVersions,
     Checkpoint,
     CheckpointMetadata,
@@ -14,6 +15,7 @@ from langgraph.checkpoint.base import (
     SerializerProtocol,
 )
 from langgraph.checkpoint.memory import InMemorySaver, PersistentDict
+from langgraph.constants import TASKS
 
 
 class NoopSerializer(SerializerProtocol):
@@ -24,6 +26,28 @@ class NoopSerializer(SerializerProtocol):
         return "type", obj
 
 
+class MemorySaverNeedsPendingSendsMigration(BaseCheckpointSaver):
+    def __init__(self) -> None:
+        self.saver = InMemorySaver()
+
+    def __getattribute__(self, name):
+        if name in ("saver", "__class__", "get_tuple"):
+            return object.__getattribute__(self, name)
+        return getattr(self.saver, name)
+
+    def get_tuple(self, config):
+        if tup := self.saver.get_tuple(config):
+            if tup.checkpoint["v"] == 4 and tup.checkpoint["channel_values"].get(TASKS):
+                tup.checkpoint["v"] = 3
+                tup.checkpoint["pending_sends"] = tup.checkpoint["channel_values"].pop(
+                    TASKS
+                )
+                tup.checkpoint["channel_versions"].pop(TASKS)
+                for seen in tup.checkpoint["versions_seen"].values():
+                    seen.pop(TASKS, None)
+        return tup
+
+
 class MemorySaverAssertImmutable(InMemorySaver):
     storage_for_copies: defaultdict[str, dict[str, dict[str, Checkpoint]]]
 

libs/langgraph/tests/test_checkpoint_migration.py

@@ -7,12 +7,9 @@ from typing import Annotated, Literal, Optional, Union
 import pytest
 from typing_extensions import TypedDict
 
-from langgraph.checkpoint.base import (
-    BaseCheckpointSaver,
-    CheckpointTuple,
-    copy_checkpoint,
-)
+from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointTuple
 from langgraph.graph.state import StateGraph
+from langgraph.pregel.checkpoint import copy_checkpoint
 from langgraph.types import Command, Interrupt, PregelTask, StateSnapshot, interrupt
 from langgraph.utils.config import patch_configurable
 from tests.any_int import AnyInt
@@ -46,7 +43,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
                 "source": "loop",
                 "step": 4,
                 "parents": {},
-                "thread_id": "1",
             },
             created_at=AnyStr(),
             parent_config={
@@ -76,7 +72,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
                 "source": "loop",
                 "step": 3,
                 "parents": {},
-                "thread_id": "1",
             },
             created_at=AnyStr(),
             parent_config={
@@ -134,7 +129,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
                 "source": "loop",
                 "step": 2,
                 "parents": {},
-                "thread_id": "1",
             },
             created_at=AnyStr(),
             parent_config={
@@ -171,7 +165,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
                 "source": "loop",
                 "step": 1,
                 "parents": {},
-                "thread_id": "1",
             },
             created_at=AnyStr(),
             parent_config={
@@ -221,7 +214,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
                 "source": "loop",
                 "step": 0,
                 "parents": {},
-                "thread_id": "1",
             },
             created_at=AnyStr(),
             parent_config={
@@ -260,7 +252,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
                 "source": "input",
                 "step": -1,
                 "parents": {},
-                "thread_id": "1",
             },
             created_at=AnyStr(),
             parent_config=None,
@@ -346,7 +337,6 @@ SAVED_CHECKPOINTS = {
                 "source": "loop",
                 "step": 4,
                 "parents": {},
-                "thread_id": "1",
             },
             parent_config={
                 "configurable": {
@@ -407,7 +397,6 @@ SAVED_CHECKPOINTS = {
                 "source": "loop",
                 "step": 3,
                 "parents": {},
-                "thread_id": "1",
             },
             parent_config={
                 "configurable": {
@@ -483,7 +472,6 @@ SAVED_CHECKPOINTS = {
                 "source": "loop",
                 "step": 2,
                 "parents": {},
-                "thread_id": "1",
             },
             parent_config={
                 "configurable": {
@@ -535,7 +523,6 @@ SAVED_CHECKPOINTS = {
                 "source": "loop",
                 "step": 1,
                 "parents": {},
-                "thread_id": "1",
             },
             parent_config={
                 "configurable": {
@@ -590,7 +577,6 @@ SAVED_CHECKPOINTS = {
                 "source": "loop",
                 "step": 0,
                 "parents": {},
-                "thread_id": "1",
             },
             parent_config={
                 "configurable": {
@@ -639,7 +625,6 @@ SAVED_CHECKPOINTS = {
                 "source": "input",
                 "step": -1,
                 "parents": {},
-                "thread_id": "1",
             },
             parent_config=None,
             pending_writes=[
@@ -723,7 +708,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 4,
                 "parents": {},
             },
@@ -785,7 +769,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 3,
                 "parents": {},
             },
@@ -864,7 +847,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 2,
                 "parents": {},
             },
@@ -920,7 +902,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 1,
                 "parents": {},
             },
@@ -980,7 +961,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 0,
                 "parents": {},
             },
@@ -1029,7 +1009,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "input",
-                "thread_id": "1",
                 "step": -1,
                 "parents": {},
             },
@@ -1115,7 +1094,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 4,
                 "parents": {},
             },
@@ -1177,7 +1155,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 3,
                 "parents": {},
             },
@@ -1256,7 +1233,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 2,
                 "parents": {},
             },
@@ -1312,7 +1288,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 1,
                 "parents": {},
             },
@@ -1372,7 +1347,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 0,
                 "parents": {},
             },
@@ -1421,7 +1395,6 @@ SAVED_CHECKPOINTS = {
             },
             metadata={
                 "source": "input",
-                "thread_id": "1",
                 "step": -1,
                 "parents": {},
             },

libs/langgraph/tests/test_deprecation.py

@@ -4,7 +4,7 @@ from typing_extensions import TypedDict
 from langgraph.func import entrypoint, task
 from langgraph.graph import StateGraph
 from langgraph.types import RetryPolicy
-from langgraph.warnings import LangGraphDeprecatedSinceV10
+from langgraph.warnings import LangGraphDeprecatedSinceV05
 
 
 class PlainState(TypedDict): ...
@@ -14,7 +14,7 @@ def test_add_node_retry_arg() -> None:
     builder = StateGraph(PlainState)
 
     with pytest.warns(
-        LangGraphDeprecatedSinceV10,
+        LangGraphDeprecatedSinceV05,
         match="`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
     ):
         builder.add_node("test_node", lambda state: state, retry=RetryPolicy())  # type: ignore[arg-type]
@@ -22,7 +22,7 @@ def test_add_node_retry_arg() -> None:
 
 def test_task_retry_arg() -> None:
     with pytest.warns(
-        LangGraphDeprecatedSinceV10,
+        LangGraphDeprecatedSinceV05,
         match="`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
     ):
 
@@ -33,7 +33,7 @@ def test_task_retry_arg() -> None:
 
 def test_entrypoint_retry_arg() -> None:
     with pytest.warns(
-        LangGraphDeprecatedSinceV10,
+        LangGraphDeprecatedSinceV05,
         match="`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
     ):
 
@@ -44,7 +44,7 @@ def test_entrypoint_retry_arg() -> None:
 
 def test_state_graph_input_schema() -> None:
     with pytest.warns(
-        LangGraphDeprecatedSinceV10,
+        LangGraphDeprecatedSinceV05,
         match="`input` is deprecated and will be removed. Please use `input_schema` instead.",
     ):
         StateGraph(PlainState, input=PlainState)  # type: ignore[arg-type]
@@ -52,7 +52,7 @@ def test_state_graph_input_schema() -> None:
 
 def test_state_graph_output_schema() -> None:
     with pytest.warns(
-        LangGraphDeprecatedSinceV10,
+        LangGraphDeprecatedSinceV05,
         match="`output` is deprecated and will be removed. Please use `output_schema` instead.",
     ):
         StateGraph(PlainState, output=PlainState)  # type: ignore[arg-type]
@@ -62,7 +62,7 @@ def test_add_node_input_schema() -> None:
     builder = StateGraph(PlainState)
 
     with pytest.warns(
-        LangGraphDeprecatedSinceV10,
+        LangGraphDeprecatedSinceV05,
         match="`input` is deprecated and will be removed. Please use `input_schema` instead.",
     ):
         builder.add_node("test_node", lambda state: state, input=PlainState)  # type: ignore[arg-type]

libs/langgraph/tests/test_pregel.py

@@ -46,7 +46,7 @@ from langgraph.constants import CONFIG_KEY_NODE_FINISHED, ERROR, PULL, START
 from langgraph.errors import InvalidUpdateError, ParentCommand
 from langgraph.func import entrypoint, task
 from langgraph.graph import END, StateGraph
-from langgraph.graph.message import MessagesState, add_messages
+from langgraph.graph.message import MessageGraph, MessagesState, add_messages
 from langgraph.prebuilt.tool_node import ToolNode
 from langgraph.pregel import (
     GraphRecursionError,
@@ -159,7 +159,7 @@ def test_checkpoint_errors() -> None:
             raise ValueError("Faulty put_writes")
 
     class FaultyVersionCheckpointer(InMemorySaver):
-        def get_next_version(self, current: Optional[int]) -> int:
+        def get_next_version(self, current: Optional[int], channel: None) -> int:
             raise ValueError("Faulty get_next_version")
 
     def logic(inp: str) -> str:
@@ -904,7 +904,6 @@ def test_pending_writes_resume(
         "parents": {},
         "source": "loop",
         "step": 0,
-        "thread_id": "1",
     }
     # get_state with checkpoint_id should not apply any pending writes
     state = graph.get_state(state.config)
@@ -994,7 +993,6 @@ def test_pending_writes_resume(
             "parents": {},
             "step": 1,
             "source": "loop",
-            "thread_id": "1",
         },
         parent_config={
             "configurable": {
@@ -1042,7 +1040,6 @@ def test_pending_writes_resume(
             "parents": {},
             "step": 0,
             "source": "loop",
-            "thread_id": "1",
         },
         parent_config={
             "configurable": {
@@ -1094,7 +1091,6 @@ def test_pending_writes_resume(
             "parents": {},
             "step": -1,
             "source": "input",
-            "thread_id": "1",
         },
         parent_config=None,
         pending_writes=UnsortedSequence(
@@ -2058,7 +2054,6 @@ def test_in_one_fan_out_state_graph_waiting_edge(
             "parents": {},
             "source": "update",
             "step": 4,
-            "thread_id": "2",
         },
         parent_config=expected_parent_config,
         interrupts=(),
@@ -2328,7 +2323,6 @@ def test_in_one_fan_out_state_graph_defer_node(
             "parents": {},
             "source": "update",
             "step": 4,
-            "thread_id": "2",
         },
         parent_config=expected_parent_config,
         interrupts=(),
@@ -3276,6 +3270,57 @@ def test_subgraph_checkpoint_true(
         ),
     ]
 
+    checkpoints = list(app.get_state_history(config))
+    if checkpoint_during:
+        assert len(checkpoints) == 4
+    else:
+        assert len(checkpoints) == 1
+
+
+def test_subgraph_checkpoint_during_false_inherited() -> None:
+    sync_checkpointer = InMemorySaver()
+
+    class InnerState(TypedDict):
+        my_key: Annotated[str, operator.add]
+        my_other_key: str
+
+    def inner_1(state: InnerState):
+        return {"my_key": " got here", "my_other_key": state["my_key"]}
+
+    def inner_2(state: InnerState):
+        return {"my_key": " and there"}
+
+    inner = StateGraph(InnerState)
+    inner.add_node("inner_1", inner_1)
+    inner.add_node("inner_2", inner_2)
+    inner.add_edge("inner_1", "inner_2")
+    inner.set_entry_point("inner_1")
+    inner.set_finish_point("inner_2")
+
+    class State(TypedDict):
+        my_key: str
+
+    inner_app = inner.compile(checkpointer=sync_checkpointer)
+    graph = StateGraph(State)
+    graph.add_node("inner", inner_app)
+    graph.add_edge(START, "inner")
+    graph.add_conditional_edges(
+        "inner", lambda s: "inner" if s["my_key"].count("there") < 2 else END
+    )
+    app = graph.compile(checkpointer=sync_checkpointer)
+    for checkpoint_during in [True, False]:
+        thread_id = str(uuid.uuid4())
+        config = {"configurable": {"thread_id": thread_id}}
+        app.invoke(
+            {"my_key": ""}, config, subgraphs=True, checkpoint_during=checkpoint_during
+        )
+        if checkpoint_during:
+            checkpoints = list(sync_checkpointer.list(config))
+            assert len(checkpoints) == 12
+        else:
+            checkpoints = list(sync_checkpointer.list(config))
+            assert len(checkpoints) == 1
+
 
 def test_subgraph_checkpoint_true_interrupt(
     sync_checkpointer: BaseCheckpointSaver, checkpoint_during: bool
@@ -3879,7 +3924,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
 
     # assert that checkpoint metadata contains the run's configurable fields
     chkpnt_metadata_1 = sync_checkpointer.get_tuple(config).metadata
-    assert chkpnt_metadata_1["thread_id"] == "1"
     assert chkpnt_metadata_1["test_config_1"] == "foo"
     assert chkpnt_metadata_1["test_config_2"] == "bar"
 
@@ -3888,7 +3932,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
     # on how the graph is constructed.
     chkpnt_tuples_1 = sync_checkpointer.list(config)
     for chkpnt_tuple in chkpnt_tuples_1:
-        assert chkpnt_tuple.metadata["thread_id"] == "1"
         assert chkpnt_tuple.metadata["test_config_1"] == "foo"
         assert chkpnt_tuple.metadata["test_config_2"] == "bar"
 
@@ -3908,7 +3951,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
 
     # assert that checkpoint metadata contains the run's configurable fields
     chkpnt_metadata_2 = sync_checkpointer.get_tuple(config).metadata
-    assert chkpnt_metadata_2["thread_id"] == "2"
     assert chkpnt_metadata_2["test_config_3"] == "foo"
     assert chkpnt_metadata_2["test_config_4"] == "bar"
 
@@ -3926,7 +3968,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
 
     # assert that checkpoint metadata contains the run's configurable fields
     chkpnt_metadata_3 = sync_checkpointer.get_tuple(config).metadata
-    assert chkpnt_metadata_3["thread_id"] == "2"
     assert chkpnt_metadata_3["test_config_3"] == "foo"
     assert chkpnt_metadata_3["test_config_4"] == "bar"
 
@@ -3935,7 +3976,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
     # on how the graph is constructed.
     chkpnt_tuples_2 = sync_checkpointer.list(config)
     for chkpnt_tuple in chkpnt_tuples_2:
-        assert chkpnt_tuple.metadata["thread_id"] == "2"
         assert chkpnt_tuple.metadata["test_config_3"] == "foo"
         assert chkpnt_tuple.metadata["test_config_4"] == "bar"
 
@@ -3943,14 +3983,9 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
 def test_remove_message_via_state_update(
     sync_checkpointer: BaseCheckpointSaver,
 ) -> None:
-    from langchain_core.messages import (
-        AIMessage,
-        AnyMessage,
-        HumanMessage,
-        RemoveMessage,
-    )
+    from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
 
-    workflow = StateGraph(Annotated[list[AnyMessage], add_messages])
+    workflow = MessageGraph()
     workflow.add_node(
         "chatbot",
         lambda state: [
@@ -3981,14 +4016,9 @@ def test_remove_message_via_state_update(
 
 
 def test_remove_message_from_node():
-    from langchain_core.messages import (
-        AIMessage,
-        AnyMessage,
-        HumanMessage,
-        RemoveMessage,
-    )
+    from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
 
-    workflow = StateGraph(Annotated[list[AnyMessage], add_messages])
+    workflow = MessageGraph()
     workflow.add_node(
         "chatbot",
         lambda state: [
@@ -4300,7 +4330,7 @@ def test_store_injected(
     builder = StateGraph(State)
     builder.add_node("node", Node())
     builder.add_edge("__start__", "node")
-    N = 500
+    N = 50
     M = 1
 
     for i in range(N):
@@ -4575,11 +4605,14 @@ def test_debug_nested_subgraphs(
 
         return clean_config
 
-    for checkpoint_events, checkpoint_history in zip(
-        stream_ns.values(), history_ns.values()
+    for checkpoint_events, checkpoint_history, ns in zip(
+        stream_ns.values(), history_ns.values(), stream_ns.keys()
     ):
         if not checkpoint_during:
             checkpoint_events = checkpoint_events[-1:]
+            if ns:  # Save no checkpoints for subgraphs when checkpoint_during=False
+                assert not checkpoint_history
+                continue
         assert len(checkpoint_events) == len(checkpoint_history)
         for stream, history in zip(checkpoint_events, checkpoint_history):
             assert stream["values"] == history.values
@@ -4778,7 +4811,9 @@ def test_parent_command(
 
     config = {"configurable": {"thread_id": "1"}}
 
-    assert graph.invoke({"messages": [("user", "get user name")]}, config) == {
+    assert graph.invoke(
+        {"messages": [("user", "get user name")]}, config, checkpoint_during=False
+    ) == {
         "messages": [
             _AnyIdHumanMessage(
                 content="get user name", additional_kwargs={}, response_metadata={}
@@ -4805,7 +4840,6 @@ def test_parent_command(
         },
         metadata={
             "source": "loop",
-            "thread_id": "1",
             "step": 1,
             "parents": {},
         },
@@ -4877,7 +4911,7 @@ def test_interrupt_multiple(sync_checkpointer: BaseCheckpointSaver):
     assert [
         event
         for event in graph.stream(
-            Command(resume="answer 1", update={"my_key": "foofoo"}), thread1
+            Command(resume="answer 1", update={"my_key": " foofoo "}), thread1
         )
     ] == [
         {
@@ -4892,8 +4926,14 @@ def test_interrupt_multiple(sync_checkpointer: BaseCheckpointSaver):
         }
     ]
 
-    assert [event for event in graph.stream(Command(resume="answer 2"), thread1)] == [
-        {"node": {"my_key": "answer 1 answer 2"}},
+    assert [
+        event
+        for event in graph.stream(
+            Command(resume="answer 2"), thread1, stream_mode="values"
+        )
+    ] == [
+        {"my_key": "DE foofoo "},
+        {"my_key": "DE foofoo answer 1 answer 2"},
     ]
 
 
@@ -5501,7 +5541,10 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver):
 
     configurable = {"configurable": {"thread_id": uuid.uuid4()}}
     assert [
-        chunk for chunk in graph.stream({"a": 5}, configurable, stream_mode="debug")
+        chunk
+        for chunk in graph.stream(
+            {"a": 5}, configurable, stream_mode="debug", checkpoint_during=False
+        )
     ] == [
         {
             "payload": {
@@ -5603,7 +5646,12 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver):
     ]
     assert [
         c
-        for c in graph.stream(Command(resume="123"), configurable, stream_mode="debug")
+        for c in graph.stream(
+            Command(resume="123"),
+            configurable,
+            stream_mode="debug",
+            checkpoint_during=False,
+        )
     ] == [
         {
             "payload": {
@@ -5618,7 +5666,6 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver):
                     "parents": {},
                     "source": "input",
                     "step": -1,
-                    "thread_id": AnyStr(),
                 },
                 "next": [
                     "graph",

libs/langgraph/tests/test_pregel_async.py

@@ -103,7 +103,7 @@ async def test_checkpoint_errors() -> None:
             raise ValueError("Faulty put_writes")
 
     class FaultyVersionCheckpointer(InMemorySaver):
-        def get_next_version(self, current: Optional[int]) -> int:
+        def get_next_version(self, current: Optional[int], channel: None) -> int:
             raise ValueError("Faulty get_next_version")
 
     def logic(inp: str) -> str:
@@ -274,7 +274,9 @@ async def test_checkpoint_put_after_cancellation() -> None:
     thread1 = {"configurable": {"thread_id": "1"}}
 
     # start the task
-    t = asyncio.create_task(graph.ainvoke({"hello": "world"}, thread1))
+    t = asyncio.create_task(
+        graph.ainvoke({"hello": "world"}, thread1, checkpoint_during=False)
+    )
     # cancel after 0.2 seconds
     await asyncio.sleep(0.2)
     t.cancel()
@@ -340,7 +342,7 @@ async def test_checkpoint_put_after_cancellation_stream_anext() -> None:
     thread1 = {"configurable": {"thread_id": "1"}}
 
     # start the task
-    s = graph.astream({"hello": "world"}, thread1)
+    s = graph.astream({"hello": "world"}, thread1, checkpoint_during=False)
     t = asyncio.create_task(s.__anext__())
     # cancel after 0.2 seconds
     await asyncio.sleep(0.2)
@@ -408,7 +410,11 @@ async def test_checkpoint_put_after_cancellation_stream_events_anext() -> None:
 
     # start the task
     s = graph.astream_events(
-        {"hello": "world"}, thread1, version="v2", include_names=["LangGraph"]
+        {"hello": "world"},
+        thread1,
+        version="v2",
+        include_names=["LangGraph"],
+        checkpoint_during=False,
     )
     # skip first event (happens right away)
     await s.__anext__()
@@ -595,7 +601,9 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non
     # stop when about to enter node
     assert [
         c
-        async for c in tool_two.astream({"my_key": "value ⛰️", "market": "DE"}, thread1)
+        async for c in tool_two.astream(
+            {"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False
+        )
     ] == [
         {
             "__interrupt__": (
@@ -612,7 +620,6 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non
             "parents": {},
             "source": "loop",
             "step": 0,
-            "thread_id": "1",
         },
     ]
     tup = await tool_two.checkpointer.aget_tuple(thread1)
@@ -639,7 +646,6 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non
             "parents": {},
             "source": "loop",
             "step": 0,
-            "thread_id": "1",
         },
         parent_config=None,
         interrupts=(
@@ -665,7 +671,6 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non
             "parents": {},
             "source": "update",
             "step": 1,
-            "thread_id": "1",
         },
         parent_config=(
             [c async for c in tool_two.checkpointer.alist(thread1, limit=2)][-1].config
@@ -768,7 +773,9 @@ async def test_dynamic_interrupt_subgraph(
     # stop when about to enter node
     assert [
         c
-        async for c in tool_two.astream({"my_key": "value ⛰️", "market": "DE"}, thread1)
+        async for c in tool_two.astream(
+            {"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False
+        )
     ] == [
         {
             "__interrupt__": (
@@ -785,7 +792,6 @@ async def test_dynamic_interrupt_subgraph(
             "parents": {},
             "source": "loop",
             "step": 0,
-            "thread_id": "1",
         },
     ]
     tup = await tool_two.checkpointer.aget_tuple(thread1)
@@ -818,7 +824,6 @@ async def test_dynamic_interrupt_subgraph(
             "parents": {},
             "source": "loop",
             "step": 0,
-            "thread_id": "1",
         },
         parent_config=None,
         interrupts=(
@@ -844,7 +849,6 @@ async def test_dynamic_interrupt_subgraph(
             "parents": {},
             "source": "update",
             "step": 1,
-            "thread_id": "1",
         },
         parent_config=(
             [c async for c in tool_two.checkpointer.alist(thread1root, limit=2)][
@@ -946,7 +950,9 @@ async def test_copy_checkpoint(async_checkpointer: BaseCheckpointSaver) -> None:
     # flow: interrupt -> clear tasks
     thread1 = {"configurable": {"thread_id": "1"}}
     # stop when about to enter node
-    assert await tool_two.ainvoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == {
+    assert await tool_two.ainvoke(
+        {"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False
+    ) == {
         "my_key": "value ⛰️ one",
         "market": "DE",
         "__interrupt__": [
@@ -963,7 +969,6 @@ async def test_copy_checkpoint(async_checkpointer: BaseCheckpointSaver) -> None:
             "parents": {},
             "source": "loop",
             "step": 0,
-            "thread_id": "1",
         },
     ]
 
@@ -1000,7 +1005,6 @@ async def test_copy_checkpoint(async_checkpointer: BaseCheckpointSaver) -> None:
             "parents": {},
             "source": "loop",
             "step": 0,
-            "thread_id": "1",
         },
         parent_config=None,
         interrupts=(
@@ -1039,7 +1043,6 @@ async def test_copy_checkpoint(async_checkpointer: BaseCheckpointSaver) -> None:
             "parents": {},
             "source": "fork",
             "step": 1,
-            "thread_id": "1",
         },
         parent_config=(
             [c async for c in tool_two.checkpointer.alist(thread1, limit=2)][-1].config
@@ -1217,7 +1220,6 @@ async def test_cancel_graph_astream(async_checkpointer: BaseCheckpointSaver) ->
         "parents": {},
         "source": "loop",
         "step": 0,
-        "thread_id": "1",
     }
 
 
@@ -1292,7 +1294,6 @@ async def test_cancel_graph_astream_events_v2(
         "parents": {},
         "source": "loop",
         "step": 1,
-        "thread_id": "2",
     }
 
 
@@ -1338,7 +1339,7 @@ async def test_node_schemas_custom_output() -> None:
             "now": 123,
         }
 
-    builder = StateGraph(State, output=Output)
+    builder = StateGraph(State, output_schema=Output)
     builder.add_node("a", node_a)
     builder.add_node("b", node_b)
     builder.add_node("c", node_c)
@@ -1353,7 +1354,7 @@ async def test_node_schemas_custom_output() -> None:
         "messages": [_AnyIdHumanMessage(content="hello")],
     }
 
-    builder = StateGraph(State, output=Output)
+    builder = StateGraph(State, output_schema=Output)
     builder.add_node("a", node_a)
     builder.add_node("b", node_b)
     builder.add_node("c", node_c)
@@ -1815,7 +1816,6 @@ async def test_pending_writes_resume(
         "parents": {},
         "source": "loop",
         "step": 0,
-        "thread_id": "1",
     }
     # get_state with checkpoint_id should not apply any pending writes
     state = await graph.aget_state(state.config)
@@ -1905,7 +1905,6 @@ async def test_pending_writes_resume(
             "parents": {},
             "step": 1,
             "source": "loop",
-            "thread_id": "1",
         },
         parent_config={
             "configurable": {
@@ -1953,7 +1952,6 @@ async def test_pending_writes_resume(
             "parents": {},
             "step": 0,
             "source": "loop",
-            "thread_id": "1",
         },
         parent_config={
             "configurable": {
@@ -2001,7 +1999,6 @@ async def test_pending_writes_resume(
             "parents": {},
             "step": -1,
             "source": "input",
-            "thread_id": "1",
         },
         parent_config=None,
         pending_writes=UnsortedSequence(
@@ -2601,7 +2598,6 @@ async def test_send_dedupe_on_resume(
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 4,
                 "parents": {},
             },
@@ -2637,7 +2633,6 @@ async def test_send_dedupe_on_resume(
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 3,
                 "parents": {},
             },
@@ -2680,7 +2675,6 @@ async def test_send_dedupe_on_resume(
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 2,
                 "parents": {},
             },
@@ -2735,7 +2729,6 @@ async def test_send_dedupe_on_resume(
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 1,
                 "parents": {},
             },
@@ -2790,7 +2783,6 @@ async def test_send_dedupe_on_resume(
             },
             metadata={
                 "source": "loop",
-                "thread_id": "1",
                 "step": 0,
                 "parents": {},
             },
@@ -2827,7 +2819,6 @@ async def test_send_dedupe_on_resume(
             },
             metadata={
                 "source": "input",
-                "thread_id": "1",
                 "step": -1,
                 "parents": {},
             },
@@ -2957,7 +2948,9 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
     foo_called = 0
     graph = builder.compile(checkpointer=async_checkpointer, interrupt_before=["foo"])
     thread1 = {"configurable": {"thread_id": "2"}}
-    assert await graph.ainvoke({"messages": [HumanMessage("hello")]}, thread1) == {
+    assert await graph.ainvoke(
+        {"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False
+    ) == {
         "messages": [
             _AnyIdHumanMessage(content="hello"),
             _AnyIdAIMessage(
@@ -3006,7 +2999,6 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
             "step": 1,
             "source": "loop",
             "parents": {},
-            "thread_id": "2",
         },
         created_at=AnyStr(),
         parent_config=None,
@@ -3052,7 +3044,6 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
             "step": 2,
             "source": "update",
             "parents": {},
-            "thread_id": "2",
         },
         created_at=AnyStr(),
         parent_config=(
@@ -3081,7 +3072,9 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
     foo_called = 0
     graph = builder.compile(checkpointer=async_checkpointer, interrupt_before=["foo"])
     thread1 = {"configurable": {"thread_id": "3"}}
-    assert await graph.ainvoke({"messages": [HumanMessage("hello")]}, thread1) == {
+    assert await graph.ainvoke(
+        {"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False
+    ) == {
         "messages": [
             _AnyIdHumanMessage(content="hello"),
             _AnyIdAIMessage(
@@ -3130,7 +3123,6 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
             "step": 1,
             "source": "loop",
             "parents": {},
-            "thread_id": "3",
         },
         created_at=AnyStr(),
         parent_config=None,
@@ -3197,7 +3189,6 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
             "step": 2,
             "source": "update",
             "parents": {},
-            "thread_id": "3",
         },
         created_at=AnyStr(),
         parent_config=(
@@ -3346,7 +3337,9 @@ async def test_send_react_interrupt_control(
     foo_called = 0
     graph = builder.compile(checkpointer=async_checkpointer, interrupt_before=["foo"])
     thread1 = {"configurable": {"thread_id": "2"}}
-    assert await graph.ainvoke({"messages": [HumanMessage("hello")]}, thread1) == {
+    assert await graph.ainvoke(
+        {"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False
+    ) == {
         "messages": [
             _AnyIdHumanMessage(content="hello"),
             _AnyIdAIMessage(
@@ -3395,7 +3388,6 @@ async def test_send_react_interrupt_control(
             "step": 1,
             "source": "loop",
             "parents": {},
-            "thread_id": "2",
         },
         created_at=AnyStr(),
         parent_config=None,
@@ -3441,7 +3433,6 @@ async def test_send_react_interrupt_control(
             "step": 2,
             "source": "update",
             "parents": {},
-            "thread_id": "2",
         },
         created_at=AnyStr(),
         parent_config=(
@@ -4387,7 +4378,6 @@ async def test_in_one_fan_out_state_graph_waiting_edge_custom_state_class(
             "parents": {},
             "source": "loop",
             "step": 4,
-            "thread_id": "1",
         },
         created_at=AnyStr(),
         parent_config=(
@@ -5029,6 +5019,51 @@ async def test_subgraph_checkpoint_true(
     ]
 
 
+async def test_subgraph_checkpoint_during_false_inherited() -> None:
+    async_checkpointer = InMemorySaver()
+
+    class InnerState(TypedDict):
+        my_key: Annotated[str, operator.add]
+        my_other_key: str
+
+    def inner_1(state: InnerState):
+        return {"my_key": " got here", "my_other_key": state["my_key"]}
+
+    def inner_2(state: InnerState):
+        return {"my_key": " and there"}
+
+    inner = StateGraph(InnerState)
+    inner.add_node("inner_1", inner_1)
+    inner.add_node("inner_2", inner_2)
+    inner.add_edge("inner_1", "inner_2")
+    inner.set_entry_point("inner_1")
+    inner.set_finish_point("inner_2")
+
+    class State(TypedDict):
+        my_key: str
+
+    inner_app = inner.compile(checkpointer=async_checkpointer)
+    graph = StateGraph(State)
+    graph.add_node("inner", inner_app)
+    graph.add_edge(START, "inner")
+    graph.add_conditional_edges(
+        "inner", lambda s: "inner" if s["my_key"].count("there") < 2 else END
+    )
+    app = graph.compile(checkpointer=async_checkpointer)
+    for checkpoint_during in [True, False]:
+        thread_id = str(uuid.uuid4())
+        config = {"configurable": {"thread_id": thread_id}}
+        await app.ainvoke(
+            {"my_key": ""}, config, subgraphs=True, checkpoint_during=checkpoint_during
+        )
+        if checkpoint_during:
+            checkpoints = list(async_checkpointer.list(config))
+            assert len(checkpoints) == 12
+        else:
+            checkpoints = list(async_checkpointer.list(config))
+            assert len(checkpoints) == 1
+
+
 @NEEDS_CONTEXTVARS
 async def test_subgraph_checkpoint_true_interrupt(
     async_checkpointer: BaseCheckpointSaver, checkpoint_during: bool
@@ -5601,7 +5636,6 @@ async def test_checkpoint_metadata(async_checkpointer: BaseCheckpointSaver) -> N
 
     # assert that checkpoint metadata contains the run's configurable fields
     chkpnt_metadata_1 = (await async_checkpointer.aget_tuple(config)).metadata
-    assert chkpnt_metadata_1["thread_id"] == "1"
     assert chkpnt_metadata_1["test_config_1"] == "foo"
     assert chkpnt_metadata_1["test_config_2"] == "bar"
 
@@ -5610,7 +5644,6 @@ async def test_checkpoint_metadata(async_checkpointer: BaseCheckpointSaver) -> N
     # on how the graph is constructed.
     chkpnt_tuples_1 = async_checkpointer.alist(config)
     async for chkpnt_tuple in chkpnt_tuples_1:
-        assert chkpnt_tuple.metadata["thread_id"] == "1"
         assert chkpnt_tuple.metadata["test_config_1"] == "foo"
         assert chkpnt_tuple.metadata["test_config_2"] == "bar"
 
@@ -5630,7 +5663,6 @@ async def test_checkpoint_metadata(async_checkpointer: BaseCheckpointSaver) -> N
 
     # assert that checkpoint metadata contains the run's configurable fields
     chkpnt_metadata_2 = (await async_checkpointer.aget_tuple(config)).metadata
-    assert chkpnt_metadata_2["thread_id"] == "2"
     assert chkpnt_metadata_2["test_config_3"] == "foo"
     assert chkpnt_metadata_2["test_config_4"] == "bar"
 
@@ -5648,7 +5680,6 @@ async def test_checkpoint_metadata(async_checkpointer: BaseCheckpointSaver) -> N
 
     # assert that checkpoint metadata contains the run's configurable fields
     chkpnt_metadata_3 = (await async_checkpointer.aget_tuple(config)).metadata
-    assert chkpnt_metadata_3["thread_id"] == "2"
     assert chkpnt_metadata_3["test_config_3"] == "foo"
     assert chkpnt_metadata_3["test_config_4"] == "bar"
 
@@ -5657,7 +5688,6 @@ async def test_checkpoint_metadata(async_checkpointer: BaseCheckpointSaver) -> N
     # on how the graph is constructed.
     chkpnt_tuples_2 = async_checkpointer.alist(config)
     async for chkpnt_tuple in chkpnt_tuples_2:
-        assert chkpnt_tuple.metadata["thread_id"] == "2"
         assert chkpnt_tuple.metadata["test_config_3"] == "foo"
         assert chkpnt_tuple.metadata["test_config_4"] == "bar"
 
@@ -5736,7 +5766,7 @@ async def test_store_injected_async(
     builder.add_edge("__start__", "node")
     builder.add_edge("node", "other_node")
 
-    N = 500
+    N = 50
     M = 1
 
     for i in range(N):
@@ -6007,11 +6037,14 @@ async def test_debug_nested_subgraphs(
 
         return clean_config
 
-    for checkpoint_events, checkpoint_history in zip(
-        stream_ns.values(), history_ns.values()
+    for checkpoint_events, checkpoint_history, ns in zip(
+        stream_ns.values(), history_ns.values(), stream_ns.keys()
     ):
         if not checkpoint_during:
             checkpoint_events = checkpoint_events[-1:]
+            if ns:  # Save no checkpoints for subgraphs when checkpoint_during=False
+                assert not checkpoint_history
+                continue
         assert len(checkpoint_events) == len(checkpoint_history)
         for stream, history in zip(checkpoint_events, checkpoint_history):
             assert stream["values"] == history.values
@@ -6062,7 +6095,9 @@ async def test_parent_command(
 
     config = {"configurable": {"thread_id": "1"}}
 
-    assert await graph.ainvoke({"messages": [("user", "get user name")]}, config) == {
+    assert await graph.ainvoke(
+        {"messages": [("user", "get user name")]}, config, checkpoint_during=False
+    ) == {
         "messages": [
             _AnyIdHumanMessage(
                 content="get user name", additional_kwargs={}, response_metadata={}
@@ -6091,7 +6126,6 @@ async def test_parent_command(
         },
         metadata={
             "source": "loop",
-            "thread_id": "1",
             "step": 1,
             "parents": {},
         },
@@ -6982,14 +7016,17 @@ async def test_multiple_subgraphs(async_checkpointer: BaseCheckpointSaver) -> No
         return {"result": state["a"] + state["b"]}
 
     add_subgraph = (
-        StateGraph(State, output=Output).add_node(add).add_edge(START, "add").compile()
+        StateGraph(State, output_schema=Output)
+        .add_node(add)
+        .add_edge(START, "add")
+        .compile()
     )
 
     async def multiply(state):
         return {"result": state["a"] * state["b"]}
 
     multiply_subgraph = (
-        StateGraph(State, output=Output)
+        StateGraph(State, output_schema=Output)
         .add_node(multiply)
         .add_edge(START, "multiply")
         .compile()
@@ -7002,7 +7039,7 @@ async def test_multiple_subgraphs(async_checkpointer: BaseCheckpointSaver) -> No
         return another_result
 
     parent_call_same_subgraph = (
-        StateGraph(State, output=Output)
+        StateGraph(State, output_schema=Output)
         .add_node(call_same_subgraph)
         .add_edge(START, "call_same_subgraph")
         .compile(checkpointer=async_checkpointer)
@@ -7026,7 +7063,7 @@ async def test_multiple_subgraphs(async_checkpointer: BaseCheckpointSaver) -> No
         }
 
     parent_call_multiple_subgraphs = (
-        StateGraph(State, output=Output)
+        StateGraph(State, output_schema=Output)
         .add_node(call_multiple_subgraphs)
         .add_edge(START, "call_multiple_subgraphs")
         .compile(checkpointer=async_checkpointer)
@@ -7104,14 +7141,17 @@ async def test_multiple_subgraphs_mixed_entrypoint(
         return {"result": state["a"] + state["b"]}
 
     add_subgraph = (
-        StateGraph(State, output=Output).add_node(add).add_edge(START, "add").compile()
+        StateGraph(State, output_schema=Output)
+        .add_node(add)
+        .add_edge(START, "add")
+        .compile()
     )
 
     async def multiply(state):
         return {"result": state["a"] * state["b"]}
 
     multiply_subgraph = (
-        StateGraph(State, output=Output)
+        StateGraph(State, output_schema=Output)
         .add_node(multiply)
         .add_edge(START, "multiply")
         .compile()
@@ -7181,7 +7221,7 @@ async def test_multiple_subgraphs_mixed_state_graph(
         return {"result": another_result}
 
     parent_call_same_subgraph = (
-        StateGraph(State, output=Output)
+        StateGraph(State, output_schema=Output)
         .add_node(call_same_subgraph)
         .add_edge(START, "call_same_subgraph")
         .compile(checkpointer=async_checkpointer)
@@ -7205,7 +7245,7 @@ async def test_multiple_subgraphs_mixed_state_graph(
         }
 
     parent_call_multiple_subgraphs = (
-        StateGraph(State, output=Output)
+        StateGraph(State, output_schema=Output)
         .add_node(call_multiple_subgraphs)
         .add_edge(START, "call_multiple_subgraphs")
         .compile(checkpointer=async_checkpointer)

libs/langgraph/tests/test_state.py

@@ -92,7 +92,7 @@ def test_state_schema_with_type_hint():
             assert state.pop("foo") == "bar"
             return {"input_state": state}
 
-    graph = StateGraph(InputState, output=OutputState)
+    graph = StateGraph(InputState, output_schema=OutputState)
     actions = [
         complete_hint,
         miss_first_hint,

libs/langgraph/tests/test_type_checking.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from operator import add
-from typing import Annotated, Any
+from typing import Annotated, Any, Union
 
 from langchain_core.runnables import RunnableConfig
 from pydantic import BaseModel
@@ -103,3 +103,21 @@ def test_input_state_specified() -> None:
 
     new_graph.invoke({"something": 1})
     new_graph.invoke({"something": 2, "info": ["hello", "world"]})  # type: ignore[arg-type]
+
+
+def test_invokeable_node_signature() -> None:
+    class State(TypedDict):
+        info: Annotated[list[str], add]
+
+    graph_builder = StateGraph(State)
+
+    class RunnableIsh:
+        def invoke(
+            self,
+            input: State,
+            config: Union[RunnableConfig, None] = None,
+            **kwargs: Any,
+        ) -> dict[str, str]:
+            return {}
+
+    graph_builder.add_node("runnable", RunnableIsh())

libs/langgraph/uv.lock

@@ -1201,7 +1201,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "0.4.7"
+version = "0.5.0rc1"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },
@@ -1310,7 +1310,7 @@ wheels = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "2.0.26"
+version = "2.1.0"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -1329,6 +1329,8 @@ dev = [
     { name = "dataclasses-json" },
     { name = "mypy" },
     { name = "numpy" },
+    { name = "pandas" },
+    { name = "pandas-stubs", specifier = ">=2.2.2.240807" },
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-mock" },
@@ -1421,7 +1423,7 @@ inmem = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "0.2.2"
+version = "0.5.0rc0"
 source = { editable = "../prebuilt" }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py

@@ -591,7 +591,7 @@ def create_react_agent(
         workflow = StateGraph(state_schema, config_schema=config_schema)
         workflow.add_node(
             "agent",
-            RunnableCallable(call_model, acall_model),  # type: ignore[call-overload]
+            RunnableCallable(call_model, acall_model),
             input_schema=input_schema,
         )
         if pre_model_hook is not None:
@@ -610,7 +610,7 @@ def create_react_agent(
         if response_format is not None:
             workflow.add_node(
                 "generate_structured_response",
-                RunnableCallable(  # type: ignore[call-overload]
+                RunnableCallable(
                     generate_structured_response,
                     agenerate_structured_response,
                 ),
@@ -660,10 +660,10 @@ def create_react_agent(
     # Define the two nodes we will cycle between
     workflow.add_node(
         "agent",
-        RunnableCallable(call_model, acall_model),  # type: ignore[call-overload]
+        RunnableCallable(call_model, acall_model),
         input_schema=input_schema,
     )
-    workflow.add_node("tools", tool_node)  # type: ignore[call-overload]
+    workflow.add_node("tools", tool_node)
 
     # Optionally add a pre-model hook node that will be called
     # every time before the "agent" (LLM-calling node)
@@ -693,7 +693,7 @@ def create_react_agent(
     if response_format is not None:
         workflow.add_node(
             "generate_structured_response",
-            RunnableCallable(  # type: ignore[call-overload]
+            RunnableCallable(
                 generate_structured_response,
                 agenerate_structured_response,
             ),

libs/prebuilt/langgraph/prebuilt/tool_node.py

@@ -629,7 +629,7 @@ def tools_condition(
 
     Args:
         state: The state to check for
-            tool calls. Must have a list of messages or have the
+            tool calls. Must have a list of messages (MessageGraph) or have the
             "messages" key (StateGraph).
 
     Returns:

libs/prebuilt/langgraph/prebuilt/tool_validator.py

@@ -2,7 +2,7 @@
 in a langchain graph. It applies a pydantic schema to tool_calls in the models' outputs,
 and returns a ToolMessage with the validated content. If the schema is not valid, it
 returns a ToolMessage with the error message. The ValidationNode can be used in a
-StateGraph with a "messages" key. If multiple tool calls are
+StateGraph with a "messages" key or in a MessageGraph. If multiple tool calls are
 requested, they will be run in parallel.
 """
 
@@ -49,7 +49,7 @@ def _default_format_error(
 class ValidationNode(RunnableCallable):
     """A node that validates all tools requests from the last AIMessage.
 
-    It can be used in StateGraph with a "messages" key.
+    It can be used either in StateGraph with a "messages" key or in MessageGraph.
 
     !!! note
 

libs/prebuilt/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph-prebuilt"
-version = "0.2.2"
+version = "0.5.0rc0"
 description = "Library with high-level APIs for creating and executing LangGraph agents and tools."
 authors = []
 requires-python = ">=3.9"
@@ -12,7 +12,7 @@ readme = "README.md"
 license = "MIT"
 license-files = ['LICENSE']
 dependencies = [
-    "langgraph-checkpoint>=2.0.10",
+    "langgraph-checkpoint>=2.1.0",
     "langchain-core>=0.3.22",
 ]
 

libs/prebuilt/tests/memory_assert.py

@@ -13,9 +13,9 @@ from langgraph.checkpoint.base import (
     CheckpointMetadata,
     CheckpointTuple,
     SerializerProtocol,
-    copy_checkpoint,
 )
 from langgraph.checkpoint.memory import InMemorySaver, PersistentDict
+from langgraph.pregel.checkpoint import copy_checkpoint
 
 
 class NoopSerializer(SerializerProtocol):

libs/prebuilt/tests/test_react_agent.py

@@ -91,7 +91,6 @@ def test_no_prompt(sync_checkpointer: BaseCheckpointSaver, version: str) -> None
         "parents": {},
         "source": "loop",
         "step": 1,
-        "thread_id": "123",
     }
     assert saved.pending_writes == []
 
@@ -118,7 +117,6 @@ async def test_no_prompt_async(async_checkpointer: BaseCheckpointSaver) -> None:
         "parents": {},
         "source": "loop",
         "step": 1,
-        "thread_id": "123",
     }
     assert saved.pending_writes == []
 
@@ -1089,14 +1087,17 @@ def test_react_with_subgraph_tools(
         return {"result": state["a"] + state["b"]}
 
     add_subgraph = (
-        StateGraph(State, output=Output).add_node(add).add_edge(START, "add").compile()
+        StateGraph(State, output_schema=Output)
+        .add_node(add)
+        .add_edge(START, "add")
+        .compile()
     )
 
     def multiply(state):
         return {"result": state["a"] * state["b"]}
 
     multiply_subgraph = (
-        StateGraph(State, output=Output)
+        StateGraph(State, output_schema=Output)
         .add_node(multiply)
         .add_edge(START, "multiply")
         .compile()

libs/prebuilt/uv.lock

@@ -320,7 +320,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "0.4.7"
+version = "0.5.0rc1"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },
@@ -371,7 +371,7 @@ dev = [
 
 [[package]]
 name = "langgraph-checkpoint"
-version = "2.0.26"
+version = "2.1.0"
 source = { editable = "../checkpoint" }
 dependencies = [
     { name = "langchain-core" },
@@ -390,6 +390,8 @@ dev = [
     { name = "dataclasses-json" },
     { name = "mypy" },
     { name = "numpy" },
+    { name = "pandas" },
+    { name = "pandas-stubs", specifier = ">=2.2.2.240807" },
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-mock" },
@@ -462,7 +464,7 @@ dev = [
 
 [[package]]
 name = "langgraph-prebuilt"
-version = "0.2.2"
+version = "0.5.0rc0"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-js/src/react/stream.tsx

@@ -894,7 +894,13 @@ export function useStream<
         if (event === "events") options.onLangChainEvent?.(data);
         if (event === "debug") options.onDebugEvent?.(data);
 
-        if (event === "values") setStreamValues(data);
+        if (event === "values") {
+          if ("__interrupt__" in data) {
+            // don't update values on interrupt values event
+            continue;
+          }
+          setStreamValues(data);
+        }
         if (event === "messages") {
           const [serialized] = data;
 

libs/sdk-py/langgraph_sdk/schema.py

@@ -36,7 +36,15 @@ Represents the status of a thread:
 """
 
 StreamMode = Literal[
-    "values", "messages", "updates", "events", "debug", "custom", "messages-tuple"
+    "values",
+    "messages",
+    "updates",
+    "events",
+    "tasks",
+    "checkpoints",
+    "debug",
+    "custom",
+    "messages-tuple",
 ]
 """
 Defines the mode of streaming:
@@ -44,6 +52,8 @@ Defines the mode of streaming:
 - "messages": Stream complete messages.
 - "updates": Stream updates to the state.
 - "events": Stream events occurring during execution.
+- "checkpoints": Stream checkpoints as they are created.
+- "tasks": Stream task start and finish events.
 - "debug": Stream detailed debug information.
 - "custom": Stream custom events.
 """