cli 0.2.12

🐛 [CLI] Generate one --build-context argument for each dependency in the docker build command. (#4962 )
* Generate one `--build-context` for each dependency in the `docker build` command. * Try and fix test --------- Co-authored-by: Nuno Campos <nuno@langchain.dev>
2026-08-18 05:35:43 +02:00 · 2025-06-04 15:50:07 -07:00 · 2025-06-04 22:48:48 +00:00 · 2025-06-04 13:27:48 -04:00 · 2025-06-04 17:20:19 +00:00 · 2025-06-04 17:12:13 +00:00
180 changed files with 10930 additions and 28416 deletions
@@ -0,0 +1,11 @@
+# Please see the documentation for all configuration options:
+# https://docs.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
+# and
+# https://docs.github.com/code-security/dependabot/dependabot-version-updates/configuration-options-for-the-dependabot.yml-file
+
+version: 2
+updates:
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "weekly"
@@ -49,7 +49,7 @@ jobs:

      - name: Get .mypy_cache to speed up mypy
        if: steps.changed-files.outputs.all
-        uses: actions/cache@v3
+        uses: actions/cache@v4
        env:
          SEGMENT_DOWNLOAD_TIMEOUT_MIN: "2"
        with:
@@ -75,7 +75,7 @@ jobs:

      - name: Get .mypy_cache_test to speed up mypy
        if: steps.changed-files.outputs.all
-        uses: actions/cache@v3
+        uses: actions/cache@v4
        env:
          SEGMENT_DOWNLOAD_TIMEOUT_MIN: "2"
        with:
@@ -1,52 +0,0 @@
-name: test
-
-on:
-  workflow_call:
-
-jobs:
-  build:
-    runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        python-version:
-          - "3.11"
-          - "3.12"
-
-    defaults:
-      run:
-        working-directory: libs/scheduler-kafka
-    name: "test #${{ matrix.python-version }}"
-    steps:
-      - uses: actions/checkout@v4
-      - name: Set up Python ${{ matrix.python-version }}
-        uses: astral-sh/setup-uv@v6
-        with:
-          python-version: ${{ matrix.python-version }}
-          enable-cache: true
-          cache-suffix: "test-scheduler-kafka"
-      - name: Login to Docker Hub
-        uses: docker/login-action@v3
-        if: ${{ !github.event.pull_request.head.repo.fork }}
-        with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_RO_TOKEN }}
-
-      - name: Install dependencies
-        shell: bash
-        run: uv sync --frozen --group dev
-
-      - name: Run tests
-        shell: bash
-        run: make test
-
-      - name: Ensure the tests did not create any additional files
-        shell: bash
-        run: |
-          set -eu
-
-          STATUS="$(git status)"
-          echo "$STATUS"
-
-          # grep will exit non-zero if the target message isn't found,
-          # and `set -e` above will cause the step to fail.
-          echo "$STATUS" | grep 'nothing to commit, working tree clean'
@@ -35,7 +35,6 @@ jobs:
              - 'libs/checkpoint/**'
              - 'libs/checkpoint-sqlite/**'
              - 'libs/checkpoint-postgres/**'
-              - 'libs/scheduler-kafka/**'
              - 'libs/prebuilt/**'
            sdk-js:
              - 'libs/sdk-js/**'
@@ -53,7 +52,7 @@ jobs:
            "libs/checkpoint",
            "libs/checkpoint-sqlite",
            "libs/checkpoint-postgres",
-            "libs/scheduler-kafka",
+
            "libs/prebuilt",
          ]
    if: needs.changes.outputs.python == 'true'
@@ -89,14 +88,6 @@ jobs:
    uses: ./.github/workflows/_test_langgraph.yml
    secrets: inherit

-  # NOTE: we're testing scheduler-kafka separately because it requires a different matrix
-  test-scheduler-kafka:
-    needs: changes
-    if: needs.changes.outputs.python == 'true'
-    name: "cd libs/scheduler-kafka"
-    uses: ./.github/workflows/_test_scheduler_kafka.yml
-    secrets: inherit
-
  check-sdk-methods:
    needs: changes
    if: needs.changes.outputs.python == 'true'
@@ -166,9 +157,9 @@ jobs:
      run:
        working-directory: ${{ matrix.working-directory }}
    steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
      - name: Setup Node.js (LTS)
-        uses: actions/setup-node@v3
+        uses: actions/setup-node@v4
        with:
          node-version: "20"
          cache: "yarn"
@@ -192,9 +183,9 @@ jobs:
      run:
        working-directory: ${{ matrix.working-directory }}
    steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
      - name: Setup Node.js (LTS)
-        uses: actions/setup-node@v3
+        uses: actions/setup-node@v4
        with:
          node-version: "20"
          cache: "yarn"
@@ -212,7 +203,6 @@ jobs:
        lint-js,
        test,
        test-langgraph,
-        test-scheduler-kafka,
        check-sdk-methods,
        check-schema,
        integration-test,
@@ -4,9 +4,11 @@ on:
  push:
    branches:
      - main
+      - v0
  pull_request:
    branches:
      - main
+      - v0
  workflow_dispatch:

 permissions:
@@ -82,9 +84,9 @@ jobs:
        run: make llms-text
      - name: Build site
        run: |
-          # If this is main branch, then we want to download stats. we do this 
+          # If this is v0 branch, then we want to download stats. we do this 
          # with the env variable DOWNLOAD_STATS=true
-          if [ "${{ github.ref }}" == "refs/heads/main" ]; then
+          if [ "${{ github.ref }}" == "refs/heads/v0" ]; then
            DOWNLOAD_STATS=true make build-docs
          else
            make build-docs
@@ -144,8 +146,8 @@ jobs:
          fi

      - name: Configure GitHub Pages
-        if: github.ref == 'refs/heads/main'
-        uses: actions/configure-pages@v4
+        if: github.ref == 'refs/heads/v0'
+        uses: actions/configure-pages@v5

      - name: Upload Pages Artifact
        # if: github.ref == 'refs/heads/main'
@@ -154,6 +156,6 @@ jobs:
          path: ./docs/site/

      - name: Deploy to GitHub Pages
-        if: github.ref == 'refs/heads/main'
+        if: github.ref == 'refs/heads/v0'
        id: deployment
        uses: actions/deploy-pages@v4
@@ -22,7 +22,7 @@ jobs:
      - uses: actions/checkout@v4
      # JS Build
      - name: Use Node.js
-        uses: actions/setup-node@v3
+        uses: actions/setup-node@v4
        with:
          node-version: "20"
          cache: "yarn"
@@ -181,3 +181,4 @@ Chinook.db
 .vercel
 .turbo
 .editorconfig
+.scratch
@@ -109,7 +109,7 @@ Here are some high-level tips on writing a good how-to guide:
 LangGraph's conceptual guides fall under the **Explanation** quadrant of Diataxis. They should cover LangChain terms and concepts
 in a more abstract way than how-to guides or tutorials, and should be geared towards curious users interested in
 gaining a deeper understanding of the framework. Try to avoid excessively large code examples. The goal here is to
-impart perspective to the user rather than to finish a practical project. These guides should cover **why** things work they way they do.
+impart perspective to the user rather than to finish a practical project. These guides should cover **why** things work the way they do.


 To quote the Diataxis website:
@@ -1,9 +1,16 @@
+"""mkdocs hooks for adding custom logic to documentation pipeline.
+
+Lifecycle events: https://www.mkdocs.org/dev-guide/plugins/#events
+"""
+
 import logging
 import os
 import posixpath
 import re
 from typing import Any, Dict

+from bs4 import BeautifulSoup
+from mkdocs.config.defaults import MkDocsConfig
 from mkdocs.structure.files import Files, File
 from mkdocs.structure.pages import Page

@@ -101,8 +108,7 @@ REDIRECT_MAP = {
    "how-tos/deploy-self-hosted.md": "cloud/deployment/self_hosted_data_plane.md",
    "concepts/self_hosted.md": "concepts/langgraph_self_hosted_data_plane.md",
    # assistant redirects
-    "cloud/how-tos/assistant_versioning.md": "cloud/how-tos/configuration_cloud.md"
-
+    "cloud/how-tos/assistant_versioning.md": "cloud/how-tos/configuration_cloud.md",
 }


@@ -292,7 +298,7 @@ Redirecting...
 """


-def write_html(site_dir, old_path, new_path):
+def _write_html(site_dir, old_path, new_path):
    """Write an HTML file in the site_dir with a meta redirect to the new page"""
    # Determine all relevant paths
    old_path_abs = os.path.join(site_dir, old_path)
@@ -308,6 +314,52 @@ def write_html(site_dir, old_path, new_path):
        f.write(content)


+def _inject_gtm(html: str) -> str:
+    """Inject Google Tag Manager code into the HTML.
+
+    Code to inject Google Tag Manager noscript tag immediately after <body>.
+
+    This is done via hooks rather than via a template because the MkDocs material
+    theme does not seem to allow placing the code immediately after the <body> tag
+    without modifying the template files directly.
+
+    Args:
+        html: The HTML content to modify.
+
+    Returns:
+        The modified HTML content with GTM code injected.
+    """
+    # Code was copied from Google Tag Manager setup instructions.
+    gtm_code = """
+<!-- Google Tag Manager (noscript) -->
+<noscript><iframe src="https://www.googletagmanager.com/ns.html?id=GTM-T35S4S46"
+height="0" width="0" style="display:none;visibility:hidden"></iframe></noscript>
+<!-- End Google Tag Manager (noscript) -->
+"""
+    soup = BeautifulSoup(html, "html.parser")
+    body = soup.body
+    if body:
+        # Insert the GTM code as raw HTML at the top of <body>
+        body.insert(0, BeautifulSoup(gtm_code, "html.parser"))
+        return str(soup)
+    else:
+        return html  # fallback if no <body> found
+
+
+def on_post_page(output: str, page: Page, config: MkDocsConfig) -> str:
+    """Inject Google Tag Manager noscript tag immediately after <body>.
+
+    Args:
+        output: The HTML output of the page.
+        page: The page instance.
+        config: The MkDocs configuration object.
+
+    Returns:
+        modified HTML output with GTM code injected.
+    """
+    return _inject_gtm(output)
+
+
 # Create HTML files for redirects after site dir has been built
 def on_post_build(config):
    use_directory_urls = config.get("use_directory_urls")
@@ -324,4 +376,4 @@ def on_post_build(config):
            + hash
            + suffix
        )
-        write_html(config["site_dir"], old_html_path, new_html_path)
+        _write_html(config["site_dir"], old_html_path, new_html_path)
@@ -38,7 +38,7 @@ client = MultiServerMCPClient(
            "transport": "stdio",
        },
        "weather": {
-            # Ensure your start your weather server on port 8000
+            # Ensure you start your weather server on port 8000
            "url": "http://localhost:8000/mcp",
            "transport": "streamable_http",
        }
@@ -88,7 +88,7 @@ ny_response = agent.invoke(

 When the agent is invoked the second time with the same `thread_id`, the original message history from the first conversation is automatically included, allowing the agent to infer that the user is asking specifically about the **weather** in New York.

-!!! Note "LangGraph Platform providers a production-ready checkpointer"
+!!! Note "LangGraph Platform provides a production-ready checkpointer"

    If you're using [LangGraph Platform](./deployment.md), during deployment your checkpointer will be automatically configured to use a production-ready database.

@@ -9,7 +9,7 @@ hide:

 # Multi-agent

-A single agent might struggle if it needs to specialize in multiple domains or manage many tools. To tackle this, you can break your agent into smaller, independent agents and composing them into a [multi-agent system](../concepts/multi_agent.md).
+A single agent might struggle if it needs to specialize in multiple domains or manage many tools. To tackle this, you can break your agent into smaller, independent agents and compose them into a [multi-agent system](../concepts/multi_agent.md).

 In multi-agent systems, agents need to communicate between each other. They do so via [handoffs](#handoffs) — a primitive that describes which agent to hand control to and the payload to send to that agent.

@@ -16,4 +16,4 @@ Users can add an array of additional lines to add to the Dockerfile following th
 }
 ```

-This would install the system packages required to use Pillow if we were working with `jpeq` or `png` image formats. 
+This would install the system packages required to use Pillow if we were working with `jpeg` or `png` image formats. 
@@ -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, MessageGraph
+from langgraph.graph import END, START, StateGraph, MessagesState

 model = ChatOpenAI(temperature=0)

-graph_workflow = MessageGraph()
+graph_workflow = StateGraph(MessagesState)

 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, MessageGraph
+from langgraph.graph import END, START
 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)
@@ -41,7 +41,8 @@ Before deploying, review the [conceptual guide for the Self-Hosted Control Plane
          pullPolicy: IfNotPresent
          tag: "aa9dff4"

-1. In your `values.yaml` file, enable the `langgraphPlatform` option. Note that you must also have a valid ingress setup:
+1. In your `langsmith_config.yaml` file, enable the `langgraphPlatform` option. Note that you must also have a valid ingress setup:
+
        config:
          langgraphPlatform:
            enabled: true
@@ -95,7 +95,7 @@ my-app/

 ## Define Graphs

-Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledGraph][langgraph.graph.graph.CompiledGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).
+Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledStateGraph][langgraph.graph.state.CompiledStateGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).

 Example `agent.py` file, which shows how to import from other modules you define (code for the modules is not shown here, please see [this repository](https://github.com/langchain-ai/langgraph-example) to see their implementation):

@@ -108,7 +108,7 @@ my-app/

 ## Define Graphs

-Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledGraph][langgraph.graph.graph.CompiledGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).
+Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledStateGraph][langgraph.graph.state.CompiledStateGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).

 Example `agent.py` file, which shows how to import from other modules you define (code for the modules is not shown here, please see [this repository](https://github.com/langchain-ai/langgraph-example-pyproject) to see their implementation):

@@ -1,8 +1,8 @@
-# How to integrate LangGraph into your React application
+How to integrate LangGraph into your React application# How to integrate LangGraph into your React application

-!!! info "Prerequisites" 
+!!! info "Prerequisites"

-    - [LangGraph Platform](../../concepts/langgraph_platform.md) 
+    - [LangGraph Platform](../../concepts/langgraph_platform.md)
    - [LangGraph Server](../../concepts/langgraph_server.md)

 The `useStream()` React hook provides a seamless way to integrate LangGraph into your React applications. It handles all the complexities of streaming, state management, and branching logic, letting you focus on building great chat experiences.
@@ -113,6 +113,115 @@ export default function App() {
 }
 ```

+### Resume a stream after page refresh
+
+The `useStream()` hook can automatically resume an ongoing run upon mounting by setting `reconnectOnMount: true`. This is useful for continuing a stream after a page refresh, ensuring no messages and events generated during the downtime are lost.
+
+```tsx
+const thread = useStream<{ messages: Message[] }>({
+  apiUrl: "http://localhost:2024",
+  assistantId: "agent",
+  reconnectOnMount: true,
+});
+```
+
+By default the ID of the created run is stored in `window.sessionStorage`, which can be swapped by passing a custom storage in `reconnectOnMount` instead. The storage is used to persist the in-flight run ID for a thread (under `lg:stream:${threadId}` key).
+
+```tsx
+const thread = useStream<{ messages: Message[] }>({
+  apiUrl: "http://localhost:2024",
+  assistantId: "agent",
+  reconnectOnMount: () => window.localStorage,
+});
+```
+
+You can also manually manage the resuming process by using the run callbacks to persist the run metadata and the `joinStream` function to resume the stream. Make sure to pass `streamResumable: true` when creating the run; otherwise some events might be lost.
+
+````tsx
+import type { Message } from "@langchain/langgraph-sdk";
+import { useStream } from "@langchain/langgraph-sdk/react";
+import { useCallback, useState, useEffect, useRef } from "react";
+
+export default function App() {
+  const [threadId, onThreadId] = useSearchParam("threadId");
+
+  const thread = useStream<{ messages: Message[] }>({
+    apiUrl: "http://localhost:2024",
+    assistantId: "agent",
+
+    threadId,
+    onThreadId,
+
+    onCreated: (run) => {
+      window.sessionStorage.setItem(`resume:${run.thread_id}`, run.run_id);
+    },
+    onFinish: (_, run) => {
+      window.sessionStorage.removeItem(`resume:${run?.thread_id}`);
+    },
+  });
+
+  // Ensure that we only join the stream once per thread.
+  const joinedThreadId = useRef<string | null>(null);
+  useEffect(() => {
+    if (!threadId) return;
+
+    const resume = window.sessionStorage.getItem(`resume:${threadId}`);
+    if (resume && joinedThreadId.current !== threadId) {
+      thread.joinStream(resume);
+      joinedThreadId.current = threadId;
+    }
+  }, [threadId]);
+
+  return (
+    <form
+      onSubmit={(e) => {
+        e.preventDefault();
+        const form = e.target as HTMLFormElement;
+        const message = new FormData(form).get("message") as string;
+        thread.submit(
+          { messages: [{ type: "human", content: message }] },
+          { streamResumable: true }
+        );
+      }}
+    >
+      <div>
+        {thread.messages.map((message) => (
+          <div key={message.id}>{message.content as string}</div>
+        ))}
+      </div>
+      <input type="text" name="message" />
+      <button type="submit">Send</button>
+    </form>
+  );
+}
+
+// Utility method to retrieve and persist data in URL as search param
+function useSearchParam(key: string) {
+  const [value, setValue] = useState<string | null>(() => {
+    const params = new URLSearchParams(window.location.search);
+    return params.get(key) ?? null;
+  });
+
+  const update = useCallback(
+    (value: string | null) => {
+      setValue(value);
+
+      const url = new URL(window.location.href);
+      if (value == null) {
+        url.searchParams.delete(key);
+      } else {
+        url.searchParams.set(key, value);
+      }
+
+      window.history.pushState({}, "", url.toString());
+    },
+    [key]
+  );
+
+  return [value, update] as const;
+}
+```
+
 ### Thread Management

 Keep track of conversations with built-in thread management. You can access the current thread ID and get notified when new threads are created:
@@ -127,7 +236,7 @@ const thread = useStream<{ messages: Message[] }>({
  threadId: threadId,
  onThreadId: setThreadId,
 });
-```
+````

 We recommend storing the `threadId` in your URL's query parameters to let users resume conversations after page refreshes.

@@ -32,7 +32,7 @@ Below are examples of directory structures for Python and JavaScript application
    │   ├── utils # utilities for your graph
    │   │   ├── __init__.py
    │   │   ├── tools.py # tools for your graph
-    │   │   ├── nodes.py # node functions for you graph
+    │   │   ├── nodes.py # node functions for your graph
    │   │   └── state.py # state definition of your graph
    │   ├── __init__.py
    │   └── agent.py # code for constructing your graph
@@ -26,4 +26,4 @@ Once you've created an assistant, subsequent edits to that assistant will create

 ## Learn more

-* The LangGraph Cloud API provides several endpoints for creating and managing assistants their versions. See the [API reference](../cloud/reference/api/api_ref.html#tag/assistants) for more details.
+* The LangGraph Cloud API provides several endpoints for creating and managing assistants and their versions. See the [API reference](../cloud/reference/api/api_ref.html#tag/assistants) for more details.
@@ -59,7 +59,7 @@ For more information, please see:
 !!! info "Important"
    The Self-Hosted Control Plane deployment option is currently in beta stage and requires an [Enterprise](../concepts/plans.md) plan.

-The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option give you full control and responsibility of the control plane and data plane infrastructure.
+The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option gives you full control and responsibility of the control plane and data plane infrastructure.

 Build a Docker image using the [LangGraph CLI](./langgraph_cli.md) and deploy your LangGraph Server from the [control plane UI](./langgraph_control_plane.md#control-plane-ui).

@@ -59,8 +59,8 @@ Yes! You can use LangGraph with any LLMs. The main reason we use LLMs that suppo

 Yes! LangGraph is totally ambivalent to what LLMs are used under the hood. The main reason we use closed LLMs in most of the tutorials is that they seamlessly support tool calling, while OSS LLMs often don't. But tool calling is not necessary (see [this section](#does-langgraph-work-with-llms-that-dont-support-tool-calling)) so you can totally use LangGraph with OSS LLMs.

-## Can I use LangGraph Studio without logging to LangSmith
+## Can I use LangGraph Studio without logging in to LangSmith

 Yes! You can use the [development version of LangGraph Server](../tutorials/langgraph-platform/local-server.md) to run the backend locally.
 This will connect to the studio frontend hosted as part of LangSmith.
-If you set an environment variable of `LANGSMITH_TRACING=false` then no traces will be sent to LangSmith.
+If you set an environment variable of `LANGSMITH_TRACING=false`, then no traces will be sent to LangSmith.
@@ -186,7 +186,7 @@ When declaring an `entrypoint`, you can request access to additional parameters

 | Parameter    | Description                                                                                                                                                        |
 |--------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| **previous** | Access the the state associated with the previous `checkpoint` for the given thread. See [short-term-memory](#short-term-memory).                                  |
+| **previous** | Access the state associated with the previous `checkpoint` for the given thread. See [short-term-memory](#short-term-memory).                                  |
 | **store**    | An instance of [BaseStore][langgraph.store.base.BaseStore]. Useful for [long-term memory](../how-tos/use-functional-api.md#long-term-memory).                      |
 | **writer**   | Use to access the StreamWriter when working with Async Python < 3.11. See [streaming with functional API for details](../how-tos/use-functional-api.md#streaming). |
 | **config**   | For accessing run time configuration. See [RunnableConfig](https://python.langchain.com/docs/concepts/runnables/#runnableconfig) for information.                  |
@@ -9,7 +9,7 @@ The term "data plane" is used broadly to refer to [LangGraph Servers](./langgrap

 ## Server Infrastructure

-In addition to the [LangGraph Server](./langgraph_server.md) itself, the following infrastructure for each server are also included in the broad definition of "data plane":
+In addition to the [LangGraph Server](./langgraph_server.md) itself, the following infrastructure components for each server are also included in the broad definition of "data plane":

 - Postgres
 - Redis
@@ -44,7 +44,7 @@ All runs in a LangGraph Server are executed by a pool of background workers that

 ### Ephemeral metadata

-Runs in a LangGraph Server may be retried for specific failures (currently only for transient Postgres errors encountered during the run). In order to limit the number of retries (currently limited to 3 attempts per run) we record the attempt number in a Redis string when is picked up. This contains no run-specific info other than its ID, and expires after a short delay.
+Runs in a LangGraph Server may be retried for specific failures (currently only for transient Postgres errors encountered during the run). In order to limit the number of retries (currently limited to 3 attempts per run) we record the attempt number in a Redis string when it is picked up. This contains no run-specific info other than its ID, and expires after a short delay.

 ## Data Plane Features

@@ -62,7 +62,7 @@ For CPU utilization, the autoscaler targets 75% utilization. This means the auto

 For number of pending runs, the autoscaler targets 10 pending runs. For example, if the current number of containers is 1, but the number of pending runs in 20, the autoscaler will scale up the deployment to 2 containers (20 pending runs / 2 containers = 10 pending runs per container).

-Each metric is computed independently and the autoscaler will determine the scaling action based on the metric that results in the most number of containers.
+Each metric is computed independently and the autoscaler will determine the scaling action based on the metric that results in the largest number of containers.

 Scale down actions are delayed for 30 minutes before any action is taken. In other words, if the autoscaler decides to scale down a deployment, it will first wait for 30 minutes before scaling down. After 30 minutes, the metrics are recomputed and the deployment will scale down if the recomputed metrics result in a lower number of containers than the current number. Otherwise, the deployment remains scaled up. This "cool down" period ensures that deployments do not scale up and down too frequently.

@@ -9,7 +9,7 @@ Develop, deploy, scale, and manage agents with **LangGraph Platform** — the pu

 !!! tip "Get started with LangGraph Platform"

-    Check out the [LangGraph Platform quickstart](../tutorials/langgraph-platform/local-server.md) for instructions on how to use LangGraph Platform run a LangGraph application locally.
+    Check out the [LangGraph Platform quickstart](../tutorials/langgraph-platform/local-server.md) for instructions on how to use LangGraph Platform to run a LangGraph application locally.

 ## Why use LangGraph Platform?

@@ -33,4 +33,4 @@ LangGraph Platform makes it easy to get your agent running in production —  wh

 - **[LangGraph Studio](./langgraph_studio.md)**: Enables visualization, interaction, and debugging of agentic systems that implement the LangGraph Server API protocol. Studio also integrates with LangSmith to enable tracing, evaluation, and prompt engineering.

- **[Deployment](./deployment_options.md)**: There are four ways to deploy on LangGraph Platform: [Cloud Saas](../concepts/langgraph_cloud.md), [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).
+- **[Deployment](./deployment_options.md)**: There are four ways to deploy on LangGraph Platform: [Cloud SaaS](../concepts/langgraph_cloud.md), [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).
@@ -9,10 +9,12 @@ There are two versions of the self-hosted deployment: [Self-Hosted Data Plane](.

 - You use `langgraph-cli` and/or [LangGraph Studio](./langgraph_studio.md) app to test graph locally.
 - You use `langgraph build` command to build image.
+- You have a Self-Hosted LangSmith instance deployed.
+- You are using Ingress for your LangSmith instance. All agents will be deployed as Kubernetes services behind this ingress.

 ## Self-Hosted Control Plane

-The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option give you full control and responsibility of the control plane and data plane infrastructure.
+The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option gives you full control and responsibility of the control plane and data plane infrastructure.

 |                   | [Control plane](../concepts/langgraph_control_plane.md) | [Data plane](../concepts/langgraph_data_plane.md) |
 |-------------------|-------------------|------------|
@@ -29,4 +31,4 @@ The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deploy
 - **Kubernetes**: The Self-Hosted Control Plane deployment option supports deploying control plane and data plane infrastructure to any Kubernetes cluster.

 !!! tip
-    If you would like to deploy to Kubernetes, you can use this [Helm chart](https://github.com/langchain-ai/helm/blob/main/charts/langgraph-cloud/README.md).
+    If you would like to enable this on your LangSmith instance, please follow the [Self-Hosted Control Plane deployment guide](../cloud/deployment/self_hosted_control_plane.md).
@@ -37,4 +37,4 @@ For information on how to deploy a [LangGraph Server](../concepts/langgraph_serv
 - **Amazon ECS**: Coming soon!

 !!! tip
-    If you would like to deploy to Kubernetes, you can use this [Helm chart](https://github.com/langchain-ai/helm/blob/main/charts/langgraph-cloud/README.md).
+    If you would like to deploy to Kubernetes, you can follow the [Self-Hosted Data Plane deployment guide](../cloud/deployment/self_hosted_data_plane.md).
@@ -26,7 +26,7 @@ Feature Differences:
 |-------|------------|------------|
 | [Cron Jobs](../cloud/concepts/cron_jobs.md) |❌|✅|
 | [Custom Authentication](../concepts/auth.md) |❌|✅|
-| [Deployment options](../concepts/deployment_options.md) | Standalone container | Cloud Saas, Self-Hosted Data Plane, Self-Hosted Control Plane, Standalone container
+| [Deployment options](../concepts/deployment_options.md) | Standalone container | Cloud SaaS, Self-Hosted Data Plane, Self-Hosted Control Plane, Standalone container

 ## Application structure

@@ -21,7 +21,7 @@ Key features of LangGraph Studio:

 - Visualize your graph architecture
 - [Run and interact with your agent](../cloud/how-tos/invoke_studio.md)
- [Manage assistants](../cloud/how-tos/studio/manage_assistants.md.md)
+- [Manage assistants](../cloud/how-tos/studio/manage_assistants.md)
 - [Manage threads](../cloud/how-tos/threads_studio.md)
 - [Iterate on prompts](../cloud/how-tos/iterate_graph_studio.md)
 - Manage [long term memory](memory.md)
@@ -33,7 +33,7 @@ Studio supports two modes:

 ### Graph mode

-Graph mode exposes the full feature-set of Studio and is useful when you would like as many details about the execution of your agent, including the nodes traversed, intermediate states, and LangSmith integrations (such as adding to datasets an playground).
+Graph mode exposes the full feature-set of Studio and is useful when you would like as many details about the execution of your agent, including the nodes traversed, intermediate states, and LangSmith integrations (such as adding to datasets and playground).

 ### Chat mode

@@ -105,7 +105,7 @@ graph.invoke({"user_input":"My"})

 There are two subtle and important points to note here:

-1. We pass `state: InputState` as the input schema to `node_1`. But, we write out to `foo`, a channel in `OverallState`. How can we write out to a state channel that is not included in the input schema? This is because a node _can write to any state channel in the graph state._ The graph state is the union of of the state channels defined at initialization, which includes `OverallState` and the filters `InputState` and `OutputState`.
+1. We pass `state: InputState` as the input schema to `node_1`. But, we write out to `foo`, a channel in `OverallState`. How can we write out to a state channel that is not included in the input schema? This is because a node _can write to any state channel in the graph state._ The graph state is the union of the state channels defined at initialization, which includes `OverallState` and the filters `InputState` and `OutputState`.

 2. We initialize the graph with `StateGraph(OverallState,input=InputState,output=OutputState)`. So, how can we write to `PrivateState` in `node_2`? How does the graph gain access to this schema if it was not passed in the `StateGraph` initialization? We can do this because _nodes can also declare additional state channels_ as long as the state schema definition exists. In this case, the `PrivateState` schema is defined, so we can add `bar` as a new state channel in the graph and write to it.

@@ -167,7 +167,7 @@ In addition to keeping track of message IDs, the `add_messages` function will al
 {"messages": [{"type": "human", "content": "message"}]}
 ```

-Since the state updates are always deserialized into LangChain `Messages` when using `add_messages`, you should use dot notation to access message attributes, like `state["messages"][-1].content`. Below is an example of a graph that uses `add_messages` as it's reducer function.
+Since the state updates are always deserialized into LangChain `Messages` when using `add_messages`, you should use dot notation to access message attributes, like `state["messages"][-1].content`. Below is an example of a graph that uses `add_messages` as its reducer function.

 ```python
 from langchain_core.messages import AnyMessage
@@ -383,7 +383,7 @@ def update_memory(state: MessagesState, config: RunnableConfig, *, store: BaseSt

 ```

-As we showed above, we can also access the store in any node and use the `store.search` method to get memories. Recall the the memories are returned as a list of objects that can be converted to a dictionary.
+As we showed above, we can also access the store in any node and use the `store.search` method to get memories. Recall the memories are returned as a list of objects that can be converted to a dictionary.

 ```python
 memories[-1].dict()
@@ -25,7 +25,7 @@ Each step consists of three phases:

 Repeat until no **actors** are selected for execution, or a maximum number of steps is reached.

-## Actors 
+## Actors

 An **actor** is a `PregelNode`. It subscribes to channels, reads data from them, and writes data to them. It can be thought of as an **actor** in the Pregel algorithm. `PregelNodes` implement LangChain's Runnable interface.

@@ -39,7 +39,7 @@ Channels are used to communicate between actors (PregelNodes). Each channel has

 ## Examples

-While most users will interact with Pregel through the [StateGraph][langgraph.graph.StateGraph] API or 
+While most users will interact with Pregel through the [StateGraph][langgraph.graph.StateGraph] API or
 the [entrypoint][langgraph.func.entrypoint] decorator, it is possible to interact with Pregel directly.

 Below are a few different examples to give you a sense of the Pregel API.
@@ -49,12 +49,12 @@ Below are a few different examples to give you a sense of the Pregel API.
    ```python

    from langgraph.channels import EphemeralValue
-    from langgraph.pregel import Pregel, Channel 
+    from langgraph.pregel import Pregel, NodeBuilder

    node1 = (
-        Channel.subscribe_to("a")
-        | (lambda x: x + x)
-        | Channel.write_to("b")
+        NodeBuilder().subscribe_only("a")
+        .do(lambda x: x + x)
+        .write_to("b")
    )

    app = Pregel(
@@ -78,18 +78,18 @@ Below are a few different examples to give you a sense of the Pregel API.

    ```python
    from langgraph.channels import LastValue, EphemeralValue
-    from langgraph.pregel import Pregel, Channel 
+    from langgraph.pregel import Pregel, NodeBuilder

    node1 = (
-        Channel.subscribe_to("a")
-        | (lambda x: x + x)
-        | Channel.write_to("b")
+        NodeBuilder().subscribe_only("a")
+        .do(lambda x: x + x)
+        .write_to("b")
    )

    node2 = (
-        Channel.subscribe_to("b")
-        | (lambda x: x + x)
-        | Channel.write_to("c")
+        NodeBuilder().subscribe_only("b")
+        .do(lambda x: x + x)
+        .write_to("c")
    )


@@ -115,23 +115,18 @@ Below are a few different examples to give you a sense of the Pregel API.

    ```python
    from langgraph.channels import EphemeralValue, Topic
-    from langgraph.pregel import Pregel, Channel 
+    from langgraph.pregel import Pregel, NodeBuilder

    node1 = (
-        Channel.subscribe_to("a")
-        | (lambda x: x + x)
-        | {
-            "b": Channel.write_to("b"),
-            "c": Channel.write_to("c")
-        }
+        NodeBuilder().subscribe_only("a")
+        .do(lambda x: x + x)
+        .write_to("b", "c")
    )

    node2 = (
-        Channel.subscribe_to("b")
-        | (lambda x: x + x)
-        | {
-            "c": Channel.write_to("c"),
-        }
+        NodeBuilder().subscribe_to("b")
+        .do(lambda x: x["b"] + x["b"])
+        .write_to("c")
    )

    app = Pregel(
@@ -158,24 +153,19 @@ Below are a few different examples to give you a sense of the Pregel API.

    ```python
    from langgraph.channels import EphemeralValue, BinaryOperatorAggregate
-    from langgraph.pregel import Pregel, Channel
+    from langgraph.pregel import Pregel, NodeBuilder


    node1 = (
-        Channel.subscribe_to("a")
-        | (lambda x: x + x)
-        | {
-            "b": Channel.write_to("b"),
-            "c": Channel.write_to("c")
-        }
+        NodeBuilder().subscribe_only("a")
+        .do(lambda x: x + x)
+        .write_to("b", "c")
    )

    node2 = (
-        Channel.subscribe_to("b")
-        | (lambda x: x + x)
-        | {
-            "c": Channel.write_to("c"),
-        }
+        NodeBuilder().subscribe_only("b")
+        .do(lambda x: x + x)
+        .write_to("c")
    )

    def reducer(current, update):
@@ -197,8 +187,7 @@ Below are a few different examples to give you a sense of the Pregel API.

    app.invoke({"a": "foo"})
    ```
-    
-    
+
 === "Cycle"

    This example demonstrates how to introduce a cycle in the graph, by having
@@ -207,12 +196,12 @@ Below are a few different examples to give you a sense of the Pregel API.

    ```python
    from langgraph.channels import EphemeralValue
-    from langgraph.pregel import Pregel, Channel, ChannelWrite, ChannelWriteEntry
+    from langgraph.pregel import Pregel, NodeBuilder, ChannelWriteEntry

    example_node = (
-        Channel.subscribe_to("value")
-        | (lambda x: x + x if len(x) < 10 else None)
-        | ChannelWrite(writes=[ChannelWriteEntry(channel="value", skip_none=True)])
+        NodeBuilder().subscribe_only("value")
+        .do(lambda x: x + x if len(x) < 10 else None)
+        .write_to(ChannelWriteEntry("value", skip_none=True))
    )

    app = Pregel(
@@ -235,7 +224,6 @@ Below are a few different examples to give you a sense of the Pregel API.

 LangGraph provides two high-level APIs for creating a Pregel application: the [StateGraph (Graph API)](./low_level.md) and the [Functional API](functional_api.md).

-
 === "StateGraph (Graph API)"

    The [StateGraph (Graph API)][langgraph.graph.StateGraph] is a higher-level abstraction that simplifies the creation of Pregel applications. It allows you to define a graph of nodes and edges. When you compile the graph, the StateGraph API automatically creates the Pregel application for you.
@@ -266,7 +254,7 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S
    builder.add_node(score_essay)
    builder.add_edge(START, "write_essay")

-    # Compile the graph. 
+    # Compile the graph.
    # This will return a Pregel instance.
    graph = builder.compile()
    ```
@@ -279,7 +267,7 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S

    You will see something like this:

-    ```pycon 
+    ```pycon
    {'__start__': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1810>,
     'write_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba14d0>,
     'score_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1710>}
@@ -310,7 +298,7 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S
 === "Functional API"

    In the [Functional API](functional_api.md), you can use an [`entrypoint`][langgraph.func.entrypoint] to create
-    a Pregel application. The `entrypoint` decorator allows you to define a function that takes input and returns output. 
+    a Pregel application. The `entrypoint` decorator allows you to define a function that takes input and returns output.

    ```python
    from typing import TypedDict, Optional
@@ -339,8 +327,8 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S
    ```

    ```pycon
-    Nodes: 
+    Nodes:
    {'write_essay': <langgraph.pregel.read.PregelNode object at 0x7d05e2f9aad0>}
-    Channels: 
+    Channels:
    {'__start__': <langgraph.channels.ephemeral_value.EphemeralValue object at 0x7d05e2c906c0>, '__end__': <langgraph.channels.last_value.LastValue object at 0x7d05e2c90c40>, '__previous__': <langgraph.channels.last_value.LastValue object at 0x7d05e1007280>}
    ```
@@ -25,7 +25,7 @@ When a graceful shutdown request is received (SIGINT) an instance enters shutdow
 - gives any in-progress runs a limited number of seconds to finish (if not finished it will be put back in the queue)
 - stops the instance from picking up more runs from the queue

-If a hard shutdown occurs due to a server crash or an infrastructure failure, any runs that were in progress will be picked up by a internal sweeper task that looks for in-progress runs that have breached their heartbeat window. The sweeper runs every 2 minutes and will put the runs back in the queue for another instance to pick them up.
+If a hard shutdown occurs due to a server crash or an infrastructure failure, any runs that were in progress will be picked up by an internal sweeper task that looks for in-progress runs that have breached their heartbeat window. The sweeper runs every 2 minutes and will put the runs back in the queue for another instance to pick them up.

 ## Postgres resilience

@@ -74,7 +74,7 @@ In your `langgraph.json`, add the path to your auth file:

 ## 3. Connect from the client

-Once you've set up authentication in your server, requests must include the the required authorization information based on your chosen scheme.
+Once you've set up authentication in your server, requests must include the required authorization information based on your chosen scheme.
 Assuming you are using JWT token authentication, you could access your deployments using any of the following methods:

 === "Python Client"
@@ -1198,7 +1198,7 @@
    "\n",
    "There are many use cases where you may wish for your node to have a custom retry policy, for example if you are calling an API, querying a database, or calling an LLM, etc. LangGraph lets you add retry policies to nodes.\n",
    "\n",
-    "To configure a retry policy, pass the `retry` parameter to the [add_node](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph.add_node). The `retry` parameter takes in a `RetryPolicy` named tuple object. Below we instantiate a `RetryPolicy` object with the default parameters and associate it with a node:\n",
+    "To configure a retry policy, pass the `retry_policy` parameter to the [add_node](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph.add_node). The `retry_policy` parameter takes in a `RetryPolicy` named tuple object. Below we instantiate a `RetryPolicy` object with the default parameters and associate it with a node:\n",
    "\n",
    "```python\n",
    "from langgraph.pregel import RetryPolicy\n",
@@ -1206,7 +1206,7 @@
    "builder.add_node(\n",
    "    \"node_name\",\n",
    "    node_function,\n",
-    "    retry=RetryPolicy(),\n",
+    "    retry_policy=RetryPolicy(),\n",
    ")\n",
    "```"
   ]
@@ -1241,7 +1241,7 @@
  },
  {
   "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": null,
   "id": "ad92598c-b688-42fa-aae0-9de36273d584",
   "metadata": {},
   "outputs": [],
@@ -1276,9 +1276,9 @@
    "builder.add_node(\n",
    "    \"query_database\",\n",
    "    query_database,\n",
-    "    retry=RetryPolicy(retry_on=sqlite3.OperationalError),\n",
+    "    retry_policy=RetryPolicy(retry_on=sqlite3.OperationalError),\n",
    ")\n",
-    "builder.add_node(\"model\", call_model, retry=RetryPolicy(max_attempts=5))\n",
+    "builder.add_node(\"model\", call_model, retry_policy=RetryPolicy(max_attempts=5))\n",
    "builder.add_edge(START, \"model\")\n",
    "builder.add_edge(\"model\", \"query_database\")\n",
    "builder.add_edge(\"query_database\", END)\n",
@@ -3416,7 +3416,7 @@
 ],
 "metadata": {
  "kernelspec": {
-   "display_name": "Python 3 (ipykernel)",
+   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
@@ -3430,7 +3430,7 @@
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
-   "version": "3.10.4"
+   "version": "3.12.9"
  }
 },
 "nbformat": 4,
@@ -12,9 +12,9 @@
    "\n",
    "\n",
    "1. **Run the graph** with initial inputs using `invoke` or `stream` APIs.\n",
-    "2. **Identify a checkpoint in an existing thread**: Use the [`get_state_history()`][langgraph.graph.graph.CompiledGraph.get_state_history] method to retrieve the execution history for a specific `thread_id` and locate the desired `checkpoint_id`.  \n",
+    "2. **Identify a checkpoint in an existing thread**: Use the [`get_state_history()`][langgraph.graph.state.CompiledStateGraph.get_state_history] method to retrieve the execution history for a specific `thread_id` and locate the desired `checkpoint_id`.  \n",
    "   Alternatively, set a [breakpoint](../../../concepts/breakpoints/) before the node(s) where you want execution to pause. You can then find the most recent checkpoint recorded up to that breakpoint.\n",
-    "3. **(Optional) modify the graph state**: Use the [`update_state`][langgraph.graph.graph.CompiledGraph.update_state] method to modify the graph’s state at the checkpoint and resume execution from alternative state.\n",
+    "3. **(Optional) modify the graph state**: Use the [`update_state`][langgraph.graph.state.CompiledStateGraph.update_state] method to modify the graph’s state at the checkpoint and resume execution from alternative state.\n",
    "4. **Resume execution from the checkpoint**: Use the `invoke` or `stream` APIs with an input of `None` and a configuration containing the appropriate `thread_id` and `checkpoint_id`.\n",
    "\n",
    "## Example\n",
@@ -405,7 +405,7 @@
  },
  {
   "cell_type": "code",
-   "execution_count": 46,
+   "execution_count": null,
   "id": "1954a5f1-91e4-4b32-9be9-c8bc1cc43cb5",
   "metadata": {},
   "outputs": [],
@@ -465,7 +465,9 @@
    "\n",
    "graph_builder = StateGraph(State)\n",
    "graph_builder.add_node(\"agent\", agent)\n",
-    "graph_builder.add_node(\"select_tools\", select_tools, retry=RetryPolicy(max_attempts=3))\n",
+    "graph_builder.add_node(\n",
+    "    \"select_tools\", select_tools, retry_policy=RetryPolicy(max_attempts=3)\n",
+    ")\n",
    "\n",
    "tool_node = ToolNode(tools=tools)\n",
    "graph_builder.add_node(\"tools\", tool_node)\n",
@@ -207,7 +207,7 @@
   "id": "213d661e-6ba4-42b9-bc7f-6c8c423e3419",
   "metadata": {},
   "source": [
-    "Let's now create our agents using the the prebuilt [`create_react_agent`][langgraph.prebuilt.chat_agent_executor.create_react_agent] and our multi-agent workflow. Note that will be calling [`interrupt`][langgraph.types.interrupt] every time after we get the final response from each of the agents."
+    "Let's now create our agents using the prebuilt [`create_react_agent`][langgraph.prebuilt.chat_agent_executor.create_react_agent] and our multi-agent workflow. Note that will be calling [`interrupt`][langgraph.types.interrupt] every time after we get the final response from each of the agents."
   ]
  },
  {
@@ -739,7 +739,6 @@
    "                'id': '1f029ca3-1f5b-6704-8004-820c16b69a5a',\n",
    "                'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000006.0.3205149138784782', 'branch:to:call_model': '00000000000000000000000000000006.0.14611156755133758'}, 'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}},\n",
    "                'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today?), HumanMessage(content=\"what's my name?\"), AIMessage(content='Your name is Bob.')]},\n",
-    "                'pending_sends': []\n",
    "            },\n",
    "            metadata={\n",
    "                'source': 'loop',\n",
@@ -856,7 +855,7 @@
    "                    'id': '1f029ca3-1f5b-6704-8004-820c16b69a5a', \n",
    "                    'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000006.0.3205149138784782', 'branch:to:call_model': '00000000000000000000000000000006.0.14611156755133758'}, \n",
    "                    'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}},\n",
-    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\"), AIMessage(content='Your name is Bob.')]}, 'pending_sends': []\n",
+    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\"), AIMessage(content='Your name is Bob.')]},\n",
    "                },\n",
    "                metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}}, 'step': 4, 'parents': {}, 'thread_id': '1'}, \n",
    "                parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}}, \n",
@@ -870,8 +869,7 @@
    "                    'id': '1f029ca3-1790-6b0a-8003-baf965b6a38f', \n",
    "                    'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000005.0.7935064215293443', 'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}, \n",
    "                    'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, \n",
-    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\")], 'branch:to:call_model': None}, \n",
-    "                    'pending_sends': []\n",
+    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\")], 'branch:to:call_model': None}\n",
    "                }, \n",
    "                metadata={'source': 'loop', 'writes': None, 'step': 3, 'parents': {}, 'thread_id': '1'}, \n",
    "                parent_config={...}, \n",
@@ -885,8 +883,7 @@
    "                    'id': '1f029ca3-1790-616e-8002-9e021694a0cd', \n",
    "                    'channel_versions': {'__start__': '00000000000000000000000000000004.0.5736472536395331', 'messages': '00000000000000000000000000000003.0.7056767754077798', 'branch:to:call_model': '00000000000000000000000000000003.0.22059023329132854'}, \n",
    "                    'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, \n",
-    "                    'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"what's my name?\"}]}, 'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}, \n",
-    "                    'pending_sends': []\n",
+    "                    'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"what's my name?\"}]}, 'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}\n",
    "                }, \n",
    "                metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': \"what's my name?\"}]}}, 'step': 2, 'parents': {}, 'thread_id': '1'}, \n",
    "                parent_config={...}, \n",
@@ -900,8 +897,7 @@
    "                    'id': '1f029ca3-178d-6f54-8001-d7b180db0c89', \n",
    "                    'channel_versions': {'__start__': '00000000000000000000000000000002.0.18673090920108737', 'messages': '00000000000000000000000000000003.0.7056767754077798', 'branch:to:call_model': '00000000000000000000000000000003.0.22059023329132854'}, \n",
    "                    'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, \n",
-    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}, \n",
-    "                    'pending_sends': []\n",
+    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}\n",
    "                }, \n",
    "                metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}}, 'step': 1, 'parents': {}, 'thread_id': '1'}, \n",
    "                parent_config={...}, \n",
@@ -915,8 +911,7 @@
    "                    'id': '1f029ca3-0874-6612-8000-339f2abc83b1', \n",
    "                    'channel_versions': {'__start__': '00000000000000000000000000000002.0.18673090920108737', 'messages': '00000000000000000000000000000002.0.30296526818059655', 'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}, \n",
    "                    'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}}, \n",
-    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\")], 'branch:to:call_model': None}, \n",
-    "                    'pending_sends': []\n",
+    "                    'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\")], 'branch:to:call_model': None}\n",
    "                }, \n",
    "                metadata={'source': 'loop', 'writes': None, 'step': 0, 'parents': {}, 'thread_id': '1'}, \n",
    "                parent_config={...}, \n",
@@ -930,8 +925,7 @@
    "                    'id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565', \n",
    "                    'channel_versions': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, \n",
    "                    'versions_seen': {'__input__': {}}, \n",
-    "                    'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"hi! I'm bob\"}]}}, \n",
-    "                    'pending_sends': []\n",
+    "                    'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"hi! I'm bob\"}]}}\n",
    "                }, \n",
    "                metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': \"hi! I'm bob\"}]}}, 'step': -1, 'parents': {}, 'thread_id': '1'}, \n",
    "                parent_config=None, \n",
@@ -321,7 +321,7 @@ attempts = 0
 # The default RetryPolicy is optimized for retrying specific network errors.
 retry_policy = RetryPolicy(retry_on=ValueError)

-@task(retry=retry_policy) 
+@task(retry_policy=retry_policy) 
 def get_info():
    global attempts
    attempts += 1
@@ -36,41 +36,6 @@
        - aget_subgraphs
        - with_config

-::: langgraph.graph.graph.Graph
-    options:
-      show_if_no_docstring: true
-      show_root_heading: true
-      show_root_full_path: false
-      members:
-        - add_node
-        - add_edge
-        - add_conditional_edges
-        - compile
-
-::: langgraph.graph.graph.CompiledGraph
-    options:
-      show_if_no_docstring: true
-      show_root_heading: true
-      show_root_full_path: false
-      members:
-        - stream
-        - astream
-        - invoke
-        - ainvoke
-        - get_state
-        - aget_state
-        - get_state_history
-        - aget_state_history
-        - update_state
-        - aupdate_state
-        - bulk_update_state
-        - abulk_update_state
-        - get_graph
-        - aget_graph
-        - get_subgraphs
-        - aget_subgraphs
-        - with_config
-
 ::: langgraph.graph.message
    options:
      members:
@@ -1,5 +1,21 @@
 # Pregel

+::: langgraph.pregel.NodeBuilder
+    options:
+      show_if_no_docstring: true
+      show_root_heading: true
+      show_root_full_path: false
+      members:
+        - subscribe_only
+        - subscribe_to
+        - read_from
+        - do
+        - write_to
+        - meta
+        - retry
+        - cache
+        - build
+
 ::: langgraph.pregel.Pregel
    options:
      show_if_no_docstring: true
@@ -833,7 +833,7 @@
    "@tool\n",
    "def book_excursion(recommendation_id: int) -> str:\n",
    "    \"\"\"\n",
-    "    Book a excursion by its recommendation ID.\n",
+    "    Book an excursion by its recommendation ID.\n",
    "\n",
    "    Args:\n",
    "        recommendation_id (int): The ID of the trip recommendation to book.\n",
@@ -89,7 +89,7 @@
  },
  {
   "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": null,
   "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 MessageGraph\n",
-    "    # or a StateGraph with 'messages' as the key.\n",
+    "    # These functions let the step be used in a\n",
+    "    # 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",
@@ -1,6 +1,6 @@
 # Add tools

-To handle queries you chatbot can't answer "from memory", integrate a web search tool. The chatbot can use this tool to find relevant information and provide better responses.
+To handle queries that your chatbot can't answer "from memory", integrate a web search tool. The chatbot can use this tool to find relevant information and provide better responses.

 !!! note

@@ -516,11 +516,13 @@
  },
  {
   "cell_type": "code",
-   "execution_count": 12,
+   "execution_count": null,
   "id": "b76b5ec3-0720-443d-85b1-c0e79659ca0a",
   "metadata": {},
   "outputs": [],
   "source": [
+    "from pprint import pprint\n",
+    "\n",
    "from langchain.schema import Document\n",
    "\n",
    "\n",
@@ -796,7 +798,7 @@
  },
  {
   "cell_type": "code",
-   "execution_count": 14,
+   "execution_count": null,
   "id": "29acc541-d726-4b75-84d1-a215845fe88a",
   "metadata": {},
   "outputs": [
@@ -823,8 +825,6 @@
    }
   ],
   "source": [
-    "from pprint import pprint\n",
-    "\n",
    "# Run\n",
    "inputs = {\n",
    "    \"question\": \"What player at the Bears expected to draft first in the 2024 NFL draft?\"\n",
@@ -389,7 +389,7 @@
     "text": [
      "{'generate': {'messages': [AIMessage(content='Title: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nAntoine de Saint-Exupéry\\'s \"The Little Prince\" is a classic novella that has captured the hearts of millions since its publication in 1943. While it might be easy to dismiss this work as a children\\'s story, its profound themes and timeless message make it a relevant and topical piece in modern life. This essay will explore the allegorical nature of \"The Little Prince\" and discuss how its message can be applied to the complexities of the modern world.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\n\"The Little Prince\" is an allegorical tale that explores various aspects of the human condition through its whimsical characters and situations. The Little Prince himself represents innocence, curiosity, and the importance of human connection. As the story unfolds, readers encounter different characters that symbolize various aspects of adult life, such as vanity, materialism, and authority. These representations allow the story to transcend age and culture, making it relatable to a wide range of readers, even in the modern context.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe Little Prince\\'s message is centered around the importance of looking beyond superficial appearances and forming meaningful connections with others. In a world increasingly dominated by technology and social media, where surface-level interactions are commonplace, this message is more relevant than ever. The Little Prince encourages readers to cherish and nurture genuine relationships, reminding us that true happiness and fulfillment come from understanding and empathizing with others.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\n\"The Little Prince\" also offers a critique of modern society, highlighting the dangers of materialism, consumerism, and the pursuit of power. These themes resonate strongly in today\\'s world, where wealth inequality and environmental degradation are pressing issues. The story serves as a reminder that the pursuit of material possessions and status often comes at the expense of our own happiness and the well-being of our planet.\\n\\nConclusion:\\nIn conclusion, \"The Little Prince\" remains a topical and relevant work in modern life due to its allegorical nature, timeless message, and critique of modern society. Its exploration of human connections, materialism, and the pursuit of power offers valuable insights for readers of all ages. By embracing the story\\'s wisdom, we can better navigate the complexities of the modern world and foster a more compassionate, sustainable, and interconnected society.', response_metadata={'token_usage': {'prompt_tokens': 72, 'total_tokens': 632, 'completion_tokens': 560}, 'model_name': 'accounts/fireworks/models/mixtral-8x7b-instruct', 'system_fingerprint': '', 'finish_reason': 'stop', 'logprobs': None}, id='run-b39a25ab-24f6-42d0-96c2-0f74c3ecc8f7-0', usage_metadata={'input_tokens': 72, 'output_tokens': 560, 'total_tokens': 632})]}}\n",
      "---\n",
-      "{'reflect': {'messages': [HumanMessage(content='Essay Critique and Recommendations:\\n\\nTitle: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nThe introduction effectively sets the stage for the essay by providing background information on \"The Little Prince\" and its relevance in modern life. However, consider adding a hook to engage the reader\\'s attention and create a stronger first impression.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\nThis paragraph provides a clear explanation of the allegorical nature of \"The Little Prince.\" To enhance this section, consider offering specific examples from the text to illustrate how the characters and situations symbolize various aspects of adult life. This will strengthen your analysis and make it more engaging for the reader.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe relevance of the Little Prince\\'s message is well-articulated in this paragraph. To further strengthen your argument, consider discussing the consequences of ignoring this message in the context of modern society. This will help emphasize the importance of the Little Prince\\'s wisdom and its relevance to contemporary issues.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\nThis paragraph effectively highlights the story\\'s critique of modern society. To deepen your analysis, explore how the themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince\\'s message that could help address these issues.\\n\\nConclusion:\\nThe conclusion effectively summarizes the main points of the essay and emphasizes the relevance of \"The Little Prince\" in modern life. To further enhance this section, consider incorporating a thought-provoking question or statement that encourages readers to reflect on the story\\'s message and its implications for their own lives.\\n\\nRecommendations:\\n1. Expand the essay to approximately 1,200-1,500 words to allow for a more in-depth analysis.\\n2. Incorporate specific examples and quotes from \"The Little Prince\" to support your arguments and engage the reader.\\n3. Ensure that each body paragraph contains a clear thesis statement, supporting evidence, and analysis.\\n4. Consider discussing counterarguments or potential criticisms of the Little Prince\\'s message to add depth and complexity to your essay.\\n5. Revise and edit the essay for clarity, coherence, and grammar.')]}}\n",
+      "{'reflect': {'messages': [HumanMessage(content='Essay Critique and Recommendations:\\n\\nTitle: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nThe introduction effectively sets the stage for the essay by providing background information on \"The Little Prince\" and its relevance in modern life. However, consider adding a hook to engage the reader\\'s attention and create a stronger first impression.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\nThis paragraph provides a clear explanation of the allegorical nature of \"The Little Prince.\" To enhance this section, consider offering specific examples from the text to illustrate how the characters and situations symbolize various aspects of adult life. This will strengthen your analysis and make it more engaging for the reader.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe relevance of the Little Prince\\'s message is well-articulated in this paragraph. To further strengthen your argument, consider discussing the consequences of ignoring this message in the context of modern society. This will help emphasize the importance of the Little Prince\\'s wisdom and its relevance to contemporary issues.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\nThis paragraph effectively highlights the story\\'s critique of modern society. To deepen your analysis, explore how themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince\\'s message that could help address these issues.\\n\\nConclusion:\\nThe conclusion effectively summarizes the main points of the essay and emphasizes the relevance of \"The Little Prince\" in modern life. To further enhance this section, consider incorporating a thought-provoking question or statement that encourages readers to reflect on the story\\'s message and its implications for their own lives.\\n\\nRecommendations:\\n1. Expand the essay to approximately 1,200-1,500 words to allow for a more in-depth analysis.\\n2. Incorporate specific examples and quotes from \"The Little Prince\" to support your arguments and engage the reader.\\n3. Ensure that each body paragraph contains a clear thesis statement, supporting evidence, and analysis.\\n4. Consider discussing counterarguments or potential criticisms of the Little Prince\\'s message to add depth and complexity to your essay.\\n5. Revise and edit the essay for clarity, coherence, and grammar.')]}}\n",
      "---\n",
      "{'generate': {'messages': [AIMessage(content='Title: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nIn Antoine de Saint-Exupéry\\'s classic novella \"The Little Prince,\" a young boy embarks on a journey through the universe, meeting various characters that symbolize different aspects of adult life. This timeless tale, published in 1943, remains incredibly relevant in today\\'s modern world. Its allegorical nature, thought-provoking message, and critique of modern society offer invaluable insights for readers of all ages. This essay will explore the allegory of \"The Little Prince,\" analyze the relevance of its message, and discuss its critique of modern society, demonstrating its topicality in contemporary life.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\n\"The Little Prince\" is an allegorical tale that uses whimsical characters and situations to explore various aspects of the human condition. For instance, the king represents authority without substance, while the businessman embodies the futility of materialism. The fox, conversely, symbolizes the importance of forming genuine connections and nurturing meaningful relationships. These allegorical representations allow the story to transcend age and culture, making it relatable to a wide range of readers, even in the modern context.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe Little Prince\\'s message is centered around the importance of looking beyond superficial appearances and forming meaningful connections with others. In a world increasingly dominated by technology and social media, where surface-level interactions are commonplace, this message is more relevant than ever. Neglecting this message can lead to feelings of isolation, loneliness, and dissatisfaction. By embracing the story\\'s wisdom, we can prioritize genuine relationships, fostering a more compassionate and interconnected society.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\n\"The Little Prince\" offers a critique of modern society, highlighting the dangers of materialism, consumerism, and the pursuit of power. These themes resonate strongly in today\\'s world, where wealth inequality and environmental degradation are pressing issues. The story serves as a reminder that the pursuit of material possessions and status often comes at the expense of our own happiness and the well-being of our planet. To address these challenges, we must reevaluate our priorities, focusing on sustainability, empathy, and the cultivation of meaningful relationships.\\n\\nConclusion:\\nIn conclusion, \"The Little Prince\" remains a topical and relevant work in modern life due to its allegorical nature, timeless message, and critique of modern society. Its exploration of human connections, materialism, and the pursuit of power offers valuable insights for readers of all ages. By embracing the story\\'s wisdom, we can better navigate the complexities of the modern world and foster a more compassionate, sustainable, and interconnected society. As the Little Prince so eloquently states, \"What is essential is invisible to the eye,\" reminding us that true happiness and fulfillment come from understanding and empathizing with others.\\n\\nExpanded Essay Recommendations:\\n\\n1. Expand the essay to approximately 1,200-1,500 words to allow for a more in-depth analysis.\\n2. Incorporate specific examples and quotes from \"The Little Prince\" to support your arguments and engage the reader. For instance, use quotes like, \"You become responsible, forever, for what you have tamed,\" to emphasize the importance of forming genuine connections.\\n3. Ensure that each body paragraph contains a clear thesis statement, supporting evidence, and analysis.\\n4. Consider discussing counterarguments or potential criticisms of the Little Prince\\'s message to add depth and complexity to your essay. For example, explore the idea that the pursuit of material possessions can provide a sense of security and comfort.\\n5. Revise and edit the essay for clarity, coherence, and grammar. Ensure that transitions between paragraphs are smooth and that your arguments flow logically.', response_metadata={'token_usage': {'prompt_tokens': 1168, 'total_tokens': 2044, 'completion_tokens': 876}, 'model_name': 'accounts/fireworks/models/mixtral-8x7b-instruct', 'system_fingerprint': '', 'finish_reason': 'stop', 'logprobs': None}, id='run-9bfc9ff2-3186-43f5-8b75-498d532d8d1a-0', usage_metadata={'input_tokens': 1168, 'output_tokens': 876, 'total_tokens': 2044})]}}\n",
      "---\n",
@@ -478,7 +478,7 @@
      "The relevance of the Little Prince's message is well-articulated in this paragraph. To further strengthen your argument, consider discussing the consequences of ignoring this message in the context of modern society. This will help emphasize the importance of the Little Prince's wisdom and its relevance to contemporary issues.\n",
      "\n",
      "Body Paragraph 3 - The Critique of Modern Society:\n",
-      "This paragraph effectively highlights the story's critique of modern society. To deepen your analysis, explore how the themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince's message that could help address these issues.\n",
+      "This paragraph effectively highlights the story's critique of modern society. To deepen your analysis, explore how themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince's message that could help address these issues.\n",
      "\n",
      "Conclusion:\n",
      "The conclusion effectively summarizes the main points of the essay and emphasizes the relevance of \"The Little Prince\" in modern life. To further enhance this section, consider incorporating a thought-provoking question or statement that encourages readers to reflect on the story's message and its implications for their own lives.\n",
@@ -1,5 +1,16 @@
 {% extends "base.html" %}

+{% block analytics %}
+<!-- Google Tag Manager -->
+<script>(function(w,d,s,l,i){w[l]=w[l]||[];w[l].push({'gtm.start':
+new Date().getTime(),event:'gtm.js'});var f=d.getElementsByTagName(s)[0],
+j=d.createElement(s),dl=l!='dataLayer'?'&l='+l:'';j.async=true;j.src=
+'https://www.googletagmanager.com/gtm.js?id='+i+dl;f.parentNode.insertBefore(j,f);
+})(window,document,'script','dataLayer','GTM-T35S4S46');</script>
+<!-- End Google Tag Manager -->
+{% endblock %}
+
+
 {% block extrahead %}
  <meta name="algolia-site-verification" content="165B7E7C89E49946" />
  <style>
@@ -185,7 +196,6 @@
  </style>
 {% endblock %}

-
 {% block content %}
 <div class="notebook-links">
  {% if page.nb_url %}
@@ -209,7 +219,6 @@
  {% endif %}
 {% endblock %}

-
 {% block announce %}
 <strong>We are growing and hiring for multiple roles for LangChain, LangGraph and LangSmith. <a href="https://www.langchain.com/careers" target="_blank" rel="noopener noreferrer"> Join our team!</a></strong>
 {% endblock %}
@@ -25,7 +25,7 @@ with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # call .setup() the first time you're using the checkpointer
    checkpointer.setup()
    checkpoint = {
-        "v": 2,
+        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
@@ -47,7 +47,6 @@ with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
            "start:node": 2
            }
        },
-        "pending_sends": [],
    }

    # store checkpoint
@@ -67,7 +66,7 @@ from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

 async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpoint = {
-        "v": 2,
+        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
@@ -89,7 +88,6 @@ async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
            "start:node": 2
            }
        },
-        "pending_sends": [],
    }

    # store checkpoint
@@ -1,4 +1,5 @@
 import threading
+from collections import defaultdict
 from collections.abc import Iterator, Sequence
 from contextlib import contextmanager
 from typing import Any, Optional
@@ -20,7 +21,6 @@ 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
@@ -143,8 +143,36 @@ class PostgresSaver(BasePostgresSaver):
            query += f" LIMIT {limit}"
        # if we change this to use .stream() we need to make sure to close the cursor
        with self._cursor() as cur:
-            cur.execute(query, args, binary=True)
-            for value in cur:
+            cur.execute(query, args)
+            values = cur.fetchall()
+            if not values:
+                return
+            # migrate pending sends if necessary
+            if to_migrate := [
+                v
+                for v in values
+                if v["checkpoint"]["v"] < 4 and v["parent_checkpoint_id"]
+            ]:
+                cur.execute(
+                    self.SELECT_PENDING_SENDS_SQL,
+                    (
+                        values[0]["thread_id"],
+                        [v["parent_checkpoint_id"] for v in to_migrate],
+                    ),
+                )
+                grouped_by_parent = defaultdict(list)
+                for value in to_migrate:
+                    grouped_by_parent[value["parent_checkpoint_id"]].append(value)
+                for sends in cur:
+                    for value in grouped_by_parent[sends["checkpoint_id"]]:
+                        if value["channel_values"] is None:
+                            value["channel_values"] = []
+                        self._migrate_pending_sends(
+                            sends["sends"],
+                            value["checkpoint"],
+                            value["channel_values"],
+                        )
+            for value in values:
                yield CheckpointTuple(
                    {
                        "configurable": {
@@ -153,12 +181,11 @@ class PostgresSaver(BasePostgresSaver):
                            "checkpoint_id": value["checkpoint_id"],
                        }
                    },
-                    self._load_checkpoint(
-                        value["checkpoint"],
-                        value["channel_values"],
-                        value["pending_sends"],
-                    ),
-                    self._load_metadata(value["metadata"]),
+                    {
+                        **value["checkpoint"],
+                        "channel_values": self._load_blobs(value["channel_values"]),
+                    },
+                    value["metadata"],
                    (
                        {
                            "configurable": {
@@ -222,37 +249,52 @@ class PostgresSaver(BasePostgresSaver):
            cur.execute(
                self.SELECT_SQL + where,
                args,
-                binary=True,
            )
+            value = cur.fetchone()
+            if value is None:
+                return None

-            for value in cur:
-                return CheckpointTuple(
+            # migrate pending sends if necessary
+            if value["checkpoint"]["v"] < 4 and value["parent_checkpoint_id"]:
+                cur.execute(
+                    self.SELECT_PENDING_SENDS_SQL,
+                    (thread_id, [value["parent_checkpoint_id"]]),
+                )
+                if sends := cur.fetchone():
+                    if value["channel_values"] is None:
+                        value["channel_values"] = []
+                    self._migrate_pending_sends(
+                        sends["sends"],
+                        value["checkpoint"],
+                        value["channel_values"],
+                    )
+
+            return CheckpointTuple(
+                {
+                    "configurable": {
+                        "thread_id": thread_id,
+                        "checkpoint_ns": checkpoint_ns,
+                        "checkpoint_id": value["checkpoint_id"],
+                    }
+                },
+                {
+                    **value["checkpoint"],
+                    "channel_values": self._load_blobs(value["channel_values"]),
+                },
+                value["metadata"],
+                (
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
-                            "checkpoint_id": value["checkpoint_id"],
+                            "checkpoint_id": value["parent_checkpoint_id"],
                        }
-                    },
-                    self._load_checkpoint(
-                        value["checkpoint"],
-                        value["channel_values"],
-                        value["pending_sends"],
-                    ),
-                    self._load_metadata(value["metadata"]),
-                    (
-                        {
-                            "configurable": {
-                                "thread_id": thread_id,
-                                "checkpoint_ns": checkpoint_ns,
-                                "checkpoint_id": value["parent_checkpoint_id"],
-                            }
-                        }
-                        if value["parent_checkpoint_id"]
-                        else None
-                    ),
-                    self._load_writes(value["pending_writes"]),
-                )
+                    }
+                    if value["parent_checkpoint_id"]
+                    else None
+                ),
+                self._load_writes(value["pending_writes"]),
+            )

    def put(
        self,
@@ -319,8 +361,8 @@ class PostgresSaver(BasePostgresSaver):
                    checkpoint_ns,
                    checkpoint["id"],
                    checkpoint_id,
-                    Jsonb(self._dump_checkpoint(copy)),
-                    self._dump_metadata(get_checkpoint_metadata(config, metadata)),
+                    Jsonb(copy),
+                    Jsonb(get_checkpoint_metadata(config, metadata)),
                ),
            )
        return next_config
@@ -391,7 +433,7 @@ class PostgresSaver(BasePostgresSaver):
                Will be applied regardless of whether the PostgresSaver 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:
+        with self.lock, _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
@@ -407,7 +449,6 @@ class PostgresSaver(BasePostgresSaver):
                # 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,
                    ):
@@ -415,14 +456,13 @@ class PostgresSaver(BasePostgresSaver):
                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:
+                with conn.cursor(binary=True, row_factory=dict_row) as cur:
                    yield cur


-__all__ = ["PostgresSaver", "BasePostgresSaver", "ShallowPostgresSaver", "Conn"]
+__all__ = ["PostgresSaver", "BasePostgresSaver", "Conn"]
@@ -1,4 +1,5 @@
 import asyncio
+from collections import defaultdict
 from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import asynccontextmanager
 from typing import Any, Optional
@@ -20,7 +21,6 @@ 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
@@ -131,7 +131,35 @@ class AsyncPostgresSaver(BasePostgresSaver):
        # if we change this to use .stream() we need to make sure to close the cursor
        async with self._cursor() as cur:
            await cur.execute(query, args, binary=True)
-            async for value in cur:
+            values = await cur.fetchall()
+            if not values:
+                return
+            # migrate pending sends if necessary
+            if to_migrate := [
+                v
+                for v in values
+                if v["checkpoint"]["v"] < 4 and v["parent_checkpoint_id"]
+            ]:
+                await cur.execute(
+                    self.SELECT_PENDING_SENDS_SQL,
+                    (
+                        values[0]["thread_id"],
+                        [v["parent_checkpoint_id"] for v in to_migrate],
+                    ),
+                )
+                grouped_by_parent = defaultdict(list)
+                for value in to_migrate:
+                    grouped_by_parent[value["parent_checkpoint_id"]].append(value)
+                async for sends in cur:
+                    for value in grouped_by_parent[sends["checkpoint_id"]]:
+                        if value["channel_values"] is None:
+                            value["channel_values"] = []
+                        self._migrate_pending_sends(
+                            sends["sends"],
+                            value["checkpoint"],
+                            value["channel_values"],
+                        )
+            for value in values:
                yield CheckpointTuple(
                    {
                        "configurable": {
@@ -140,13 +168,11 @@ class AsyncPostgresSaver(BasePostgresSaver):
                            "checkpoint_id": value["checkpoint_id"],
                        }
                    },
-                    await asyncio.to_thread(
-                        self._load_checkpoint,
-                        value["checkpoint"],
-                        value["channel_values"],
-                        value["pending_sends"],
-                    ),
-                    self._load_metadata(value["metadata"]),
+                    {
+                        **value["checkpoint"],
+                        "channel_values": self._load_blobs(value["channel_values"]),
+                    },
+                    value["metadata"],
                    (
                        {
                            "configurable": {
@@ -191,36 +217,51 @@ class AsyncPostgresSaver(BasePostgresSaver):
                args,
                binary=True,
            )
+            value = await cur.fetchone()
+            if value is None:
+                return None

-            async for value in cur:
-                return CheckpointTuple(
+            # migrate pending sends if necessary
+            if value["checkpoint"]["v"] < 4 and value["parent_checkpoint_id"]:
+                await cur.execute(
+                    self.SELECT_PENDING_SENDS_SQL,
+                    (thread_id, [value["parent_checkpoint_id"]]),
+                )
+                if sends := await cur.fetchone():
+                    if value["channel_values"] is None:
+                        value["channel_values"] = []
+                    self._migrate_pending_sends(
+                        sends["sends"],
+                        value["checkpoint"],
+                        value["channel_values"],
+                    )
+
+            return CheckpointTuple(
+                {
+                    "configurable": {
+                        "thread_id": thread_id,
+                        "checkpoint_ns": checkpoint_ns,
+                        "checkpoint_id": value["checkpoint_id"],
+                    }
+                },
+                {
+                    **value["checkpoint"],
+                    "channel_values": self._load_blobs(value["channel_values"]),
+                },
+                value["metadata"],
+                (
                    {
                        "configurable": {
                            "thread_id": thread_id,
                            "checkpoint_ns": checkpoint_ns,
-                            "checkpoint_id": value["checkpoint_id"],
+                            "checkpoint_id": value["parent_checkpoint_id"],
                        }
-                    },
-                    await asyncio.to_thread(
-                        self._load_checkpoint,
-                        value["checkpoint"],
-                        value["channel_values"],
-                        value["pending_sends"],
-                    ),
-                    self._load_metadata(value["metadata"]),
-                    (
-                        {
-                            "configurable": {
-                                "thread_id": thread_id,
-                                "checkpoint_ns": checkpoint_ns,
-                                "checkpoint_id": value["parent_checkpoint_id"],
-                            }
-                        }
-                        if value["parent_checkpoint_id"]
-                        else None
-                    ),
-                    await asyncio.to_thread(self._load_writes, value["pending_writes"]),
-                )
+                    }
+                    if value["parent_checkpoint_id"]
+                    else None
+                ),
+                await asyncio.to_thread(self._load_writes, value["pending_writes"]),
+            )

    async def aput(
        self,
@@ -277,8 +318,8 @@ class AsyncPostgresSaver(BasePostgresSaver):
                    checkpoint_ns,
                    checkpoint["id"],
                    checkpoint_id,
-                    Jsonb(self._dump_checkpoint(copy)),
-                    self._dump_metadata(get_checkpoint_metadata(config, metadata)),
+                    Jsonb(copy),
+                    Jsonb(get_checkpoint_metadata(config, metadata)),
                ),
            )
        return next_config
@@ -350,7 +391,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
                Will be applied regardless of whether the AsyncPostgresSaver 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:
+        async with self.lock, _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
@@ -366,7 +407,6 @@ class AsyncPostgresSaver(BasePostgresSaver):
                # 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,
                    ):
@@ -374,16 +414,12 @@ class AsyncPostgresSaver(BasePostgresSaver):
                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,
-                ):
+                async with conn.cursor(binary=True, row_factory=dict_row) as cur:
                    yield cur

    def list(
@@ -532,4 +568,4 @@ class AsyncPostgresSaver(BasePostgresSaver):
        ).result()


-__all__ = ["AsyncPostgresSaver", "AsyncShallowPostgresSaver", "Conn"]
+__all__ = ["AsyncPostgresSaver", "Conn"]
@@ -9,12 +9,9 @@ from langgraph.checkpoint.base import (
    WRITES_IDX_MAP,
    BaseCheckpointSaver,
    ChannelVersions,
-    Checkpoint,
-    CheckpointMetadata,
    get_checkpoint_id,
 )
-from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
-from langgraph.checkpoint.serde.types import TASKS, ChannelProtocol
+from langgraph.checkpoint.serde.types import TASKS

 MetadataInput = Optional[dict[str, Any]]

@@ -72,7 +69,7 @@ MIGRATIONS = [
    """ALTER TABLE checkpoint_writes ADD COLUMN task_path TEXT NOT NULL DEFAULT '';""",
 ]

-SELECT_SQL = f"""
+SELECT_SQL = """
 select
    thread_id,
    checkpoint,
@@ -96,17 +93,20 @@ select
        where cw.thread_id = checkpoints.thread_id
            and cw.checkpoint_ns = checkpoints.checkpoint_ns
            and cw.checkpoint_id = checkpoints.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.checkpoint_id = checkpoints.parent_checkpoint_id
-            and cw.channel = '{TASKS}'
-    ) as pending_sends
+    ) as pending_writes
 from checkpoints """

+SELECT_PENDING_SENDS_SQL = f"""
+select
+    checkpoint_id,
+    array_agg(array[type::bytea, blob] order by task_path, task_id, idx) as sends
+from checkpoint_writes
+where thread_id = %s
+    and checkpoint_id = any(%s)
+    and channel = '{TASKS}'
+group by checkpoint_id
+"""
+
 UPSERT_CHECKPOINT_BLOBS_SQL = """
    INSERT INTO checkpoint_blobs (thread_id, checkpoint_ns, channel, version, type, blob)
    VALUES (%s, %s, %s, %s, %s, %s)
@@ -140,31 +140,34 @@ INSERT_CHECKPOINT_WRITES_SQL = """

 class BasePostgresSaver(BaseCheckpointSaver[str]):
    SELECT_SQL = SELECT_SQL
+    SELECT_PENDING_SENDS_SQL = SELECT_PENDING_SENDS_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

-    jsonplus_serde = JsonPlusSerializer()
    supports_pipeline: bool

-    def _load_checkpoint(
+    def _migrate_pending_sends(
        self,
+        pending_sends: list[tuple[bytes, bytes]],
        checkpoint: dict[str, Any],
        channel_values: list[tuple[bytes, bytes, bytes]],
-        pending_sends: list[tuple[bytes, bytes]],
-    ) -> Checkpoint:
-        return {
-            **checkpoint,
-            "pending_sends": [
-                self.serde.loads_typed((c.decode(), b)) for c, b in pending_sends or []
-            ],
-            "channel_values": self._load_blobs(channel_values),
-        }
-
-    def _dump_checkpoint(self, checkpoint: Checkpoint) -> dict[str, Any]:
-        return {**checkpoint, "pending_sends": []}
+    ) -> None:
+        if not pending_sends:
+            return
+        # add to values
+        enc, blob = self.serde.dumps_typed(
+            [self.serde.loads_typed((c.decode(), b)) for c, b in pending_sends],
+        )
+        channel_values.append((TASKS.encode(), enc.encode(), blob))
+        # add to versions
+        checkpoint["channel_versions"][TASKS] = (
+            max(checkpoint["channel_versions"].values())
+            if checkpoint["channel_versions"]
+            else self.get_next_version(None)
+        )

    def _load_blobs(
        self, blob_values: list[tuple[bytes, bytes, bytes]]
@@ -241,15 +244,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
            for idx, (channel, value) in enumerate(writes)
        ]

-    def _load_metadata(self, metadata: dict[str, Any]) -> CheckpointMetadata:
-        return self.jsonplus_serde.loads(self.jsonplus_serde.dumps(metadata))
-
-    def _dump_metadata(self, metadata: CheckpointMetadata) -> str:
-        serialized_metadata = self.jsonplus_serde.dumps(metadata)
-        # NOTE: we're using JSON serializer (not msgpack), so we need to remove null characters before writing
-        return serialized_metadata.decode().replace("\\u0000", "")
-
-    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
+    def get_next_version(self, current: Optional[str]) -> str:
        if current is None:
            current_v = 0
        elif isinstance(current, int):
@@ -1,941 +0,0 @@
-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 = self._load_checkpoint(
-                    value["checkpoint"],
-                    value["channel_values"],
-                    value["pending_sends"],
-                )
-                yield CheckpointTuple(
-                    config={
-                        "configurable": {
-                            "thread_id": value["thread_id"],
-                            "checkpoint_ns": value["checkpoint_ns"],
-                            "checkpoint_id": checkpoint["id"],
-                        }
-                    },
-                    checkpoint=checkpoint,
-                    metadata=self._load_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 = self._load_checkpoint(
-                    value["checkpoint"],
-                    value["channel_values"],
-                    value["pending_sends"],
-                )
-                return CheckpointTuple(
-                    config={
-                        "configurable": {
-                            "thread_id": thread_id,
-                            "checkpoint_ns": checkpoint_ns,
-                            "checkpoint_id": checkpoint["id"],
-                        }
-                    },
-                    checkpoint=checkpoint,
-                    metadata=self._load_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(self._dump_checkpoint(copy)),
-                    self._dump_metadata(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 = await asyncio.to_thread(
-                    self._load_checkpoint,
-                    value["checkpoint"],
-                    value["channel_values"],
-                    value["pending_sends"],
-                )
-                yield CheckpointTuple(
-                    config={
-                        "configurable": {
-                            "thread_id": value["thread_id"],
-                            "checkpoint_ns": value["checkpoint_ns"],
-                            "checkpoint_id": checkpoint["id"],
-                        }
-                    },
-                    checkpoint=checkpoint,
-                    metadata=self._load_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 = await asyncio.to_thread(
-                    self._load_checkpoint,
-                    value["checkpoint"],
-                    value["channel_values"],
-                    value["pending_sends"],
-                )
-                return CheckpointTuple(
-                    config={
-                        "configurable": {
-                            "thread_id": thread_id,
-                            "checkpoint_ns": checkpoint_ns,
-                            "checkpoint_id": checkpoint["id"],
-                        }
-                    },
-                    checkpoint=checkpoint,
-                    metadata=self._load_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(self._dump_checkpoint(copy)),
-                    self._dump_metadata(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()
@@ -0,0 +1,51 @@
+from collections.abc import Mapping
+from datetime import datetime, timezone
+from typing import Any, Optional, Protocol
+
+from langgraph.checkpoint.base import Checkpoint, EmptyChannelError
+from langgraph.checkpoint.base.id import uuid6
+
+
+class ChannelProtocol(Protocol):
+    def checkpoint(self) -> Optional[Any]: ...
+
+
+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: Optional[Mapping[str, ChannelProtocol]],
+    step: int,
+    *,
+    id: Optional[str] = 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"],
+    )
@@ -14,13 +14,10 @@ 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


@@ -111,41 +108,11 @@ 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
@@ -205,7 +172,7 @@ def test_data():
    }


-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
 async def test_combined_metadata(saver_name: str, test_data) -> None:
    async with _saver(saver_name) as saver:
        config = {
@@ -232,7 +199,7 @@ async def test_combined_metadata(saver_name: str, test_data) -> None:
        }


-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
 async def test_asearch(saver_name: str, test_data) -> None:
    async with _saver(saver_name) as saver:
        configs = test_data["configs"]
@@ -283,7 +250,7 @@ async def test_asearch(saver_name: str, test_data) -> None:
        } == {"", "inner"}


-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
 async def test_null_chars(saver_name: str, test_data) -> None:
    async with _saver(saver_name) as saver:
        config = await saver.aput(
@@ -296,3 +263,52 @@ async def test_null_chars(saver_name: str, test_data) -> None:
        assert [c async for c in saver.alist(None, filter={"my_key": "abc"})][
            0
        ].metadata["my_key"] == "abc"
+
+
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+async def test_pending_sends_migration(saver_name: str) -> None:
+    async with _saver(saver_name) as saver:
+        config = {
+            "configurable": {
+                "thread_id": "thread-1",
+                "checkpoint_ns": "",
+            }
+        }
+
+        # create the first checkpoint
+        # and put some pending sends
+        checkpoint_0 = empty_checkpoint()
+        config = await saver.aput(config, checkpoint_0, {}, {})
+        await saver.aput_writes(
+            config, [(TASKS, "send-1"), (TASKS, "send-2")], task_id="task-1"
+        )
+        await saver.aput_writes(config, [(TASKS, "send-3")], task_id="task-2")
+
+        # check that fetching checkpoint_0 doesn't attach pending sends
+        # (they should be attached to the next checkpoint)
+        tuple_0 = await saver.aget_tuple(config)
+        assert tuple_0.checkpoint["channel_values"] == {}
+        assert tuple_0.checkpoint["channel_versions"] == {}
+
+        # create the second checkpoint
+        checkpoint_1 = create_checkpoint(checkpoint_0, {}, 1)
+        config = await saver.aput(config, checkpoint_1, {}, {})
+
+        # check that pending sends are attached to checkpoint_1
+        tuple_1 = await saver.aget_tuple(config)
+        assert tuple_1.checkpoint["channel_values"] == {
+            TASKS: ["send-1", "send-2", "send-3"]
+        }
+        assert TASKS in tuple_1.checkpoint["channel_versions"]
+
+        # check that list also applies the migration
+        search_results = [
+            c async for c in saver.alist({"configurable": {"thread_id": "thread-1"}})
+        ]
+        assert len(search_results) == 2
+        assert search_results[-1].checkpoint["channel_values"] == {}
+        assert search_results[-1].checkpoint["channel_versions"] == {}
+        assert search_results[0].checkpoint["channel_values"] == {
+            TASKS: ["send-1", "send-2", "send-3"]
+        }
+        assert TASKS in search_results[0].checkpoint["channel_versions"]
@@ -15,10 +15,10 @@ from langgraph.checkpoint.base import (
    EXCLUDED_METADATA_KEYS,
    Checkpoint,
    CheckpointMetadata,
-    create_checkpoint,
-    empty_checkpoint,
 )
-from langgraph.checkpoint.postgres import PostgresSaver, ShallowPostgresSaver
+from langgraph.checkpoint.postgres import PostgresSaver
+from langgraph.checkpoint.serde.types import TASKS
+from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
 from tests.conftest import DEFAULT_POSTGRES_URI


@@ -97,37 +97,11 @@ 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
@@ -187,7 +161,7 @@ def test_data():
    }


-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
 def test_combined_metadata(saver_name: str, test_data) -> None:
    with _saver(saver_name) as saver:
        config = {
@@ -214,7 +188,7 @@ def test_combined_metadata(saver_name: str, test_data) -> None:
        }


-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
 def test_search(saver_name: str, test_data) -> None:
    with _saver(saver_name) as saver:
        configs = test_data["configs"]
@@ -263,7 +237,7 @@ def test_search(saver_name: str, test_data) -> None:
        } == {"", "inner"}


-@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
 def test_null_chars(saver_name: str, test_data) -> None:
    with _saver(saver_name) as saver:
        config = saver.put(
@@ -284,3 +258,52 @@ def test_nonnull_migrations() -> None:
    for migration in PostgresSaver.MIGRATIONS:
        statement = _leading_comment_remover.sub("", migration).split()[0]
        assert statement.strip()
+
+
+@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
+def test_pending_sends_migration(saver_name: str) -> None:
+    with _saver(saver_name) as saver:
+        config = {
+            "configurable": {
+                "thread_id": "thread-1",
+                "checkpoint_ns": "",
+            }
+        }
+
+        # create the first checkpoint
+        # and put some pending sends
+        checkpoint_0 = empty_checkpoint()
+        config = saver.put(config, checkpoint_0, {}, {})
+        saver.put_writes(
+            config, [(TASKS, "send-1"), (TASKS, "send-2")], task_id="task-1"
+        )
+        saver.put_writes(config, [(TASKS, "send-3")], task_id="task-2")
+
+        # check that fetching checkpoint_0 doesn't attach pending sends
+        # (they should be attached to the next checkpoint)
+        tuple_0 = saver.get_tuple(config)
+        assert tuple_0.checkpoint["channel_values"] == {}
+        assert tuple_0.checkpoint["channel_versions"] == {}
+
+        # create the second checkpoint
+        checkpoint_1 = create_checkpoint(checkpoint_0, {}, 1)
+        config = saver.put(config, checkpoint_1, {}, {})
+
+        # check that pending sends are attached to checkpoint_1
+        checkpoint_1 = saver.get_tuple(config)
+        assert checkpoint_1.checkpoint["channel_values"] == {
+            TASKS: ["send-1", "send-2", "send-3"]
+        }
+        assert TASKS in checkpoint_1.checkpoint["channel_versions"]
+
+        # check that list also applies the migration
+        search_results = [
+            c for c in saver.list({"configurable": {"thread_id": "thread-1"}})
+        ]
+        assert len(search_results) == 2
+        assert search_results[-1].checkpoint["channel_values"] == {}
+        assert search_results[-1].checkpoint["channel_versions"] == {}
+        assert search_results[0].checkpoint["channel_values"] == {
+            TASKS: ["send-1", "send-2", "send-3"]
+        }
+        assert TASKS in search_results[0].checkpoint["channel_versions"]
@@ -12,7 +12,7 @@ read_config = {"configurable": {"thread_id": "1"}}

 with SqliteSaver.from_conn_string(":memory:") as checkpointer:
    checkpoint = {
-        "v": 2,
+        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
@@ -34,7 +34,6 @@ with SqliteSaver.from_conn_string(":memory:") as checkpointer:
                "start:node": 2
            }
        },
-        "pending_sends": [],
    }

    # store checkpoint
@@ -54,7 +53,7 @@ from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

 async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
    checkpoint = {
-        "v": 2,
+        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
@@ -76,7 +75,6 @@ async with AsyncSqliteSaver.from_conn_string(":memory:") as checkpointer:
                "start:node": 2
            }
        },
-        "pending_sends": [],
    }

    # store checkpoint
@@ -19,7 +19,6 @@ from langgraph.checkpoint.base import (
    get_checkpoint_metadata,
 )
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
-from langgraph.checkpoint.serde.types import ChannelProtocol
 from langgraph.checkpoint.sqlite.utils import search_where

 _AIO_ERROR_MSG = (
@@ -535,14 +534,13 @@ class SqliteSaver(BaseCheckpointSaver[str]):
        """
        raise NotImplementedError(_AIO_ERROR_MSG)

-    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
+    def get_next_version(self, current: Optional[str]) -> str:
        """Generate the next version ID for a channel.

        This method creates a new version identifier for a channel based on its current version.

        Args:
            current (Optional[str]): The current version identifier of the channel.
-            channel (BaseChannel): The channel being versioned.

        Returns:
            str: The next version identifier, which is guaranteed to be monotonically increasing.
@@ -19,7 +19,6 @@ from langgraph.checkpoint.base import (
    get_checkpoint_metadata,
 )
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
-from langgraph.checkpoint.serde.types import ChannelProtocol
 from langgraph.checkpoint.sqlite.utils import search_where

 T = TypeVar("T", bound=Callable)
@@ -590,14 +589,13 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
            )
            await self.conn.commit()

-    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
+    def get_next_version(self, current: Optional[str]) -> str:
        """Generate the next version ID for a channel.

        This method creates a new version identifier for a channel based on its current version.

        Args:
            current (Optional[str]): The current version identifier of the channel.
-            channel (BaseChannel): The channel being versioned.

        Returns:
            str: The next version identifier, which is guaranteed to be monotonically increasing.
@@ -0,0 +1,51 @@
+from collections.abc import Mapping
+from datetime import datetime, timezone
+from typing import Any, Optional, Protocol
+
+from langgraph.checkpoint.base import Checkpoint, EmptyChannelError
+from langgraph.checkpoint.base.id import uuid6
+
+
+class ChannelProtocol(Protocol):
+    def checkpoint(self) -> Optional[Any]: ...
+
+
+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: Optional[Mapping[str, ChannelProtocol]],
+    step: int,
+    *,
+    id: Optional[str] = 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"],
+    )
@@ -6,10 +6,9 @@ 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:
@@ -6,11 +6,10 @@ 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:
@@ -51,7 +51,7 @@ read_config = {"configurable": {"thread_id": "1"}}

 checkpointer = MemorySaver()
 checkpoint = {
-    "v": 2,
+    "v": 4,
    "ts": "2024-07-31T20:14:19.804150+00:00",
    "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
    "channel_values": {
@@ -73,7 +73,6 @@ checkpoint = {
        "start:node": 2
      }
    },
-    "pending_sends": [],
 }

 # store checkpoint
@@ -1,22 +1,18 @@
-from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
-from datetime import datetime, timezone
+from collections.abc import AsyncIterator, Iterator, Sequence
 from typing import (  # noqa: UP035
    Any,
-    Dict,
    Generic,
    List,
    Literal,
    NamedTuple,
    Optional,
-    Tuple,
    TypedDict,
    TypeVar,
    Union,
 )

-from langchain_core.runnables import ConfigurableFieldSpec, RunnableConfig
+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 (
@@ -24,14 +20,10 @@ from langgraph.checkpoint.serde.types import (
    INTERRUPT,
    RESUME,
    SCHEDULED,
-    ChannelProtocol,
-    SendProtocol,
 )

 V = TypeVar("V", int, float, str)
-PendingWrite = Tuple[str, str, Any]
-# Kept for backwards compat, newer versions of LangGraph no longer use this.
-LATEST_VERSION = 2
+PendingWrite = tuple[str, str, Any]


 # Marked as total=False to allow for future expansion.
@@ -53,11 +45,6 @@ class CheckpointMetadata(TypedDict, total=False):
    0 for the first "loop" checkpoint.
    ... for the nth checkpoint afterwards.
    """
-    writes: dict[str, Any]
-    """The writes that were made between the previous checkpoint and this one.
-
-    Mapping from node name to writes emitted by that node.
-    """
    parents: dict[str, str]
    """The IDs of the parent checkpoints.

@@ -65,10 +52,6 @@ class CheckpointMetadata(TypedDict, total=False):
    """


-class TaskInfo(TypedDict):
-    status: Literal["scheduled", "success", "error"]
-
-
 ChannelVersions = dict[str, Union[str, int, float]]


@@ -96,22 +79,6 @@ class Checkpoint(TypedDict):
    This keeps track of the versions of the channels that each node has seen.
    Used to determine which nodes to execute next.
    """
-    pending_sends: List[SendProtocol]
-    """List of inputs pushed to nodes but not yet processed.
-    Cleared by the next checkpoint."""
-
-
-# Kept for backwards compat, newer versions of LangGraph no longer use this.
-def empty_checkpoint() -> Checkpoint:
-    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 copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
@@ -122,39 +89,6 @@ 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(),
-    )
-
-
-# Kept for backwards compat, newer versions of LangGraph no longer use this.
-def create_checkpoint(
-    checkpoint: Checkpoint,
-    channels: Optional[Mapping[str, ChannelProtocol]],
-    step: int,
-    *,
-    id: Optional[str] = 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=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", []),
    )


@@ -168,34 +102,6 @@ class CheckpointTuple(NamedTuple):
    pending_writes: Optional[List[PendingWrite]] = None


-CheckpointThreadId = ConfigurableFieldSpec(
-    id="thread_id",
-    annotation=str,
-    name="Thread ID",
-    description=None,
-    default="",
-    is_shared=True,
-)
-
-CheckpointNS = ConfigurableFieldSpec(
-    id="checkpoint_ns",
-    annotation=str,
-    name="Checkpoint NS",
-    description='Checkpoint namespace. Denotes the path to the subgraph node the checkpoint originates from, separated by `|` character, e.g. `"child|grandchild"`. Defaults to "" (root graph).',
-    default="",
-    is_shared=True,
-)
-
-CheckpointId = ConfigurableFieldSpec(
-    id="checkpoint_id",
-    annotation=Optional[str],
-    name="Checkpoint ID",
-    description="Pass to fetch a past checkpoint. If None, fetches the latest checkpoint.",
-    default=None,
-    is_shared=True,
-)
-
-
 class BaseCheckpointSaver(Generic[V]):
    """Base class for creating a graph checkpointer.

@@ -219,15 +125,6 @@ class BaseCheckpointSaver(Generic[V]):
    ) -> None:
        self.serde = maybe_add_typed_methods(serde or self.serde)

-    @property
-    def config_specs(self) -> list[ConfigurableFieldSpec]:
-        """Define the configuration options for the checkpoint saver.
-
-        Returns:
-            list[ConfigurableFieldSpec]: List of configuration field specs.
-        """
-        return [CheckpointThreadId, CheckpointNS, CheckpointId]
-
    def get(self, config: RunnableConfig) -> Optional[Checkpoint]:
        """Fetch a checkpoint using the given configuration.

@@ -258,7 +155,7 @@ class BaseCheckpointSaver(Generic[V]):
        self,
        config: Optional[RunnableConfig],
        *,
-        filter: Optional[Dict[str, Any]] = None,
+        filter: Optional[dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> Iterator[CheckpointTuple]:
@@ -304,7 +201,7 @@ class BaseCheckpointSaver(Generic[V]):
    def put_writes(
        self,
        config: RunnableConfig,
-        writes: Sequence[Tuple[str, Any]],
+        writes: Sequence[tuple[str, Any]],
        task_id: str,
        task_path: str = "",
    ) -> None:
@@ -362,7 +259,7 @@ class BaseCheckpointSaver(Generic[V]):
        self,
        config: Optional[RunnableConfig],
        *,
-        filter: Optional[Dict[str, Any]] = None,
+        filter: Optional[dict[str, Any]] = None,
        before: Optional[RunnableConfig] = None,
        limit: Optional[int] = None,
    ) -> AsyncIterator[CheckpointTuple]:
@@ -409,7 +306,7 @@ class BaseCheckpointSaver(Generic[V]):
    async def aput_writes(
        self,
        config: RunnableConfig,
-        writes: Sequence[Tuple[str, Any]],
+        writes: Sequence[tuple[str, Any]],
        task_id: str,
        task_path: str = "",
    ) -> None:
@@ -437,7 +334,7 @@ class BaseCheckpointSaver(Generic[V]):
        """
        raise NotImplementedError

-    def get_next_version(self, current: Optional[V], channel: ChannelProtocol) -> V:
+    def get_next_version(self, current: Optional[V]) -> 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,
@@ -445,7 +342,6 @@ class BaseCheckpointSaver(Generic[V]):

        Args:
            current: The current version identifier (int, float, or str).
-            channel: The channel being versioned.

        Returns:
            V: The next version identifier, which must be increasing.
@@ -476,7 +372,10 @@ def get_checkpoint_metadata(
    config: RunnableConfig, metadata: CheckpointMetadata
 ) -> CheckpointMetadata:
    """Get checkpoint metadata in a backwards-compatible manner."""
-    metadata = metadata.copy()
+    metadata = {
+        k: v.replace("\u0000", "") if isinstance(v, str) else v
+        for k, v in metadata.items()
+    }
    for obj in (config.get("metadata"), config.get("configurable")):
        if not obj:
            continue
@@ -484,8 +383,10 @@ def get_checkpoint_metadata(
            if key in metadata or key in EXCLUDED_METADATA_KEYS or key.startswith("__"):
                continue
            v = obj[key]
-            if isinstance(v, (str, int, bool, float)):
-                metadata[key] = v  # type: ignore[literal-required]
+            if isinstance(v, str):
+                metadata[key] = v.replace("\u0000", "")
+            elif isinstance(v, (int, bool, float)):
+                metadata[key] = v
    return metadata


@@ -22,7 +22,6 @@ from langgraph.checkpoint.base import (
    get_checkpoint_id,
    get_checkpoint_metadata,
 )
-from langgraph.checkpoint.serde.types import TASKS, ChannelProtocol

 logger = logging.getLogger(__name__)

@@ -150,19 +149,6 @@ class InMemorySaver(
            if saved := self.storage[thread_id][checkpoint_ns].get(checkpoint_id):
                checkpoint, metadata, parent_checkpoint_id = saved
                writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
-                if parent_checkpoint_id:
-                    sends = sorted(
-                        (
-                            (*w, k[1])
-                            for k, w in self.writes[
-                                (thread_id, checkpoint_ns, parent_checkpoint_id)
-                            ].items()
-                            if w[1] == TASKS
-                        ),
-                        key=lambda w: (w[3], w[0], w[4]),
-                    )
-                else:
-                    sends = []
                checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)
                return CheckpointTuple(
                    config=config,
@@ -171,7 +157,6 @@ class InMemorySaver(
                        "channel_values": self._load_blobs(
                            thread_id, checkpoint_ns, checkpoint_["channel_versions"]
                        ),
-                        "pending_sends": [self.serde.loads_typed(s[2]) for s in sends],
                    },
                    metadata=self.serde.loads_typed(metadata),
                    pending_writes=[
@@ -194,22 +179,7 @@ class InMemorySaver(
                checkpoint_id = max(checkpoints.keys())
                checkpoint, metadata, parent_checkpoint_id = checkpoints[checkpoint_id]
                writes = self.writes[(thread_id, checkpoint_ns, checkpoint_id)].values()
-                if parent_checkpoint_id:
-                    sends = sorted(
-                        (
-                            (*w, k[1])
-                            for k, w in self.writes[
-                                (thread_id, checkpoint_ns, parent_checkpoint_id)
-                            ].items()
-                            if w[1] == TASKS
-                        ),
-                        key=lambda w: (w[3], w[0], w[4]),
-                    )
-                else:
-                    sends = []
-
                checkpoint_ = self.serde.loads_typed(checkpoint)
-
                return CheckpointTuple(
                    config={
                        "configurable": {
@@ -223,7 +193,6 @@ class InMemorySaver(
                        "channel_values": self._load_blobs(
                            thread_id, checkpoint_ns, checkpoint_["channel_versions"]
                        ),
-                        "pending_sends": [self.serde.loads_typed(s[2]) for s in sends],
                    },
                    metadata=self.serde.loads_typed(metadata),
                    pending_writes=[
@@ -316,20 +285,6 @@ class InMemorySaver(
                        (thread_id, checkpoint_ns, checkpoint_id)
                    ].values()

-                    if parent_checkpoint_id:
-                        sends = sorted(
-                            (
-                                (*w, k[1])
-                                for k, w in self.writes[
-                                    (thread_id, checkpoint_ns, parent_checkpoint_id)
-                                ].items()
-                                if w[1] == TASKS
-                            ),
-                            key=lambda w: (w[3], w[0], w[4]),
-                        )
-                    else:
-                        sends = []
-
                    checkpoint_: Checkpoint = self.serde.loads_typed(checkpoint)

                    yield CheckpointTuple(
@@ -347,9 +302,6 @@ class InMemorySaver(
                                checkpoint_ns,
                                checkpoint_["channel_versions"],
                            ),
-                            "pending_sends": [
-                                self.serde.loads_typed(s[2]) for s in sends
-                            ],
                        },
                        metadata=metadata,
                        parent_config=(
@@ -390,7 +342,6 @@ class InMemorySaver(
            RunnableConfig: The updated config containing the saved checkpoint's timestamp.
        """
        c = checkpoint.copy()
-        c.pop("pending_sends")  # type: ignore[misc]
        thread_id = config["configurable"]["thread_id"]
        checkpoint_ns = config["configurable"]["checkpoint_ns"]
        values: dict[str, Any] = c.pop("channel_values")  # type: ignore[misc]
@@ -561,7 +512,7 @@ class InMemorySaver(
        """
        return self.delete_thread(thread_id)

-    def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
+    def get_next_version(self, current: Optional[str]) -> str:
        if current is None:
            current_v = 0
        elif isinstance(current, int):
@@ -1,13 +1,4 @@
-from collections.abc import Sequence
-from typing import (
-    Any,
-    Optional,
-    Protocol,
-    TypeVar,
-    runtime_checkable,
-)
-
-from typing_extensions import Self
+from typing import Any, Protocol, TypeVar, runtime_checkable

 ERROR = "__error__"
 SCHEDULED = "__scheduled__"
@@ -20,25 +11,6 @@ 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
@@ -0,0 +1,51 @@
+from collections.abc import Mapping
+from datetime import datetime, timezone
+from typing import Any, Optional, Protocol
+
+from langgraph.checkpoint.base import Checkpoint, EmptyChannelError
+from langgraph.checkpoint.base.id import uuid6
+
+
+class ChannelProtocol(Protocol):
+    def checkpoint(self) -> Optional[Any]: ...
+
+
+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: Optional[Mapping[str, ChannelProtocol]],
+    step: int,
+    *,
+    id: Optional[str] = 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"],
+    )
@@ -6,10 +6,12 @@ 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:
@@ -165,6 +165,13 @@ def generate_schema():
    if "python_version" in python_schema["properties"]:
        python_schema["properties"]["python_version"]["enum"] = ["3.11", "3.12"]

+    # Add enum constraint for image_distro
+    if "image_distro" in python_schema["properties"]:
+        python_schema["properties"]["image_distro"]["anyOf"] = [
+            {"type": "string", "enum": ["debian", "wolfi"]},
+            {"type": "null"},
+        ]
+
    # Create Node.js schema with node_version
    node_schema = {
        "type": "object",
@@ -184,6 +191,13 @@ def generate_schema():
            {"type": "null"},
        ]

+    # Add enum constraint for image_distro
+    if "image_distro" in node_schema["properties"]:
+        node_schema["properties"]["image_distro"]["anyOf"] = [
+            {"type": "string", "enum": ["debian", "wolfi"]},
+            {"type": "null"},
+        ]
+
    # Replace the Config schema with a oneOf constraint
    config_schema["oneOf"] = [python_schema, node_schema]

@@ -20,6 +20,7 @@ from langgraph_cli.docker import DockerCapabilities
 from langgraph_cli.exec import Runner, subp_exec
 from langgraph_cli.progress import Progress
 from langgraph_cli.templates import TEMPLATE_HELP_STRING, create_new
+from langgraph_cli.util import warn_non_wolfi_distro
 from langgraph_cli.version import __version__

 OPT_DOCKER_COMPOSE = click.option(
@@ -317,10 +318,8 @@ def _build(
    )
    # add additional_contexts
    if additional_contexts:
-        additional_contexts_str = ",".join(
-            f"{k}={v}" for k, v in additional_contexts.items()
-        )
-        args.extend(["--build-context", additional_contexts_str])
+        for k, v in additional_contexts.items():
+            args.extend(["--build-context", f"{k}={v}"])
    # run docker build
    runner.run(
        subp_exec(
@@ -375,6 +374,7 @@ def build(
        if shutil.which("docker") is None:
            raise click.UsageError("Docker not installed") from None
        config_json = langgraph_cli.config.validate_config_file(config)
+        warn_non_wolfi_distro(config_json)
        _build(
            runner, set, config, config_json, base_image, pull, tag, docker_build_args
        )
@@ -466,6 +466,7 @@ def dockerfile(
    save_path = pathlib.Path(save_path).absolute()
    secho(f"🔍 Validating configuration at path: {config}", fg="yellow")
    config_json = langgraph_cli.config.validate_config_file(config)
+    warn_non_wolfi_distro(config_json)
    secho("✅ Configuration validated!", fg="green")

    secho(f"📝 Generating Dockerfile at {save_path}", fg="yellow")
@@ -791,6 +792,7 @@ def prepare(
 ) -> tuple[list[str], str]:
    """Prepare the arguments and stdin for running the LangGraph API server."""
    config_json = langgraph_cli.config.validate_config_file(config_path)
+    warn_non_wolfi_distro(config_json)
    # pull latest images
    if pull:
        runner.run(
@@ -13,6 +13,8 @@ DEFAULT_NODE_VERSION = "20"
 MIN_PYTHON_VERSION = "3.11"
 DEFAULT_PYTHON_VERSION = "3.11"

+DEFAULT_IMAGE_DISTRO = "debian"
+

 class TTLConfig(TypedDict, total=False):
    """Configuration for TTL (time-to-live) behavior in the store."""
@@ -367,6 +369,12 @@ class Config(TypedDict, total=False):
    
    Defaults to langchain/langgraph-api or langchain/langgraphjs-api."""

+    image_distro: Optional[str]
+    """Optional. Linux distribution for the base image.
+    
+    Must be either 'debian' or 'wolfi'. If omitted, defaults to 'debian'.
+    """
+
    pip_config_file: Optional[str]
    """Optional. Path to a pip config file (e.g., "/etc/pip.conf" or "pip.ini") for controlling
    package installation (custom indices, credentials, etc.).
@@ -458,7 +466,10 @@ RUN PYTHONDONTWRITEBYTECODE=1 pip install --no-cache-dir --no-deps -e /api
 # -- Removing pip from the final image ~<:===~~~ --
 RUN pip uninstall -y pip setuptools wheel && \
    rm -rf /usr/local/lib/python*/site-packages/pip* /usr/local/lib/python*/site-packages/setuptools* /usr/local/lib/python*/site-packages/wheel* && \
-    find /usr/local/bin -name "pip*" -delete
+    find /usr/local/bin -name "pip*" -delete || true
+# pip removal for wolfi
+RUN rm -rf /usr/lib/python*/site-packages/pip* /usr/lib/python*/site-packages/setuptools* /usr/lib/python*/site-packages/wheel* && \
+    find /usr/bin -name "pip*" -delete || true
 # -- End of pip removal --"""


@@ -517,12 +528,15 @@ def validate_config(config: Config) -> Config:
        "python_version", DEFAULT_PYTHON_VERSION if some_python else None
    )

+    image_distro = config.get("image_distro", DEFAULT_IMAGE_DISTRO)
+
    config = {
        "node_version": node_version,
        "python_version": python_version,
        "pip_config_file": config.get("pip_config_file"),
        "_INTERNAL_docker_tag": config.get("_INTERNAL_docker_tag"),
        "base_image": config.get("base_image"),
+        "image_distro": image_distro,
        "dependencies": config.get("dependencies", []),
        "dockerfile_lines": config.get("dockerfile_lines", []),
        "graphs": config.get("graphs", {}),
@@ -576,6 +590,14 @@ def validate_config(config: Config) -> Config:
            "Add at least one graph to 'graphs' dictionary."
        )

+    # Validate image_distro config
+    if image_distro := config.get("image_distro"):
+        if image_distro not in ["debian", "wolfi"]:
+            raise click.UsageError(
+                f"Invalid image_distro: '{image_distro}'. "
+                "Must be either 'debian' or 'wolfi'."
+            )
+
    # Validate auth config
    if auth_conf := config.get("auth"):
        if "path" in auth_conf:
@@ -1085,8 +1107,6 @@ def python_config_to_docker(
        else ""
    )

-    docker_tag = config.get("_INTERNAL_docker_tag") or config["python_version"]
-
    # collect dependencies
    pypi_deps = [dep for dep in config["dependencies"] if not dep.startswith(".")]
    local_deps = _assemble_local_deps(config_path, config)
@@ -1205,10 +1225,7 @@ ADD {relpath} /deps/{name}
                "# -- End of JS dependencies install --",
            ]
        )
-    if "/langgraph-server" in base_image:
-        image_str = f"{base_image}-py{docker_tag}"
-    else:
-        image_str = f"{base_image}:{docker_tag}"
+    image_str = docker_tag(config, base_image)
    docker_file_contents = [
        f"FROM {image_str}",
        "",
@@ -1248,7 +1265,7 @@ def node_config_to_docker(
 ) -> tuple[str, dict[str, str]]:
    faux_path = f"/deps/{config_path.parent.name}"
    install_cmd = _get_node_pm_install_cmd(config_path, config)
-    docker_tag = config.get("_INTERNAL_docker_tag") or config["node_version"]
+    image_str = docker_tag(config, base_image)

    env_vars: list[str] = []

@@ -1275,7 +1292,7 @@ def node_config_to_docker(
    env_vars.append(f"ENV LANGSERVE_GRAPHS='{json.dumps(config['graphs'])}'")

    docker_file_contents = [
-        f"FROM {base_image}:{docker_tag}",
+        f"FROM {image_str}",
        "",
        os.linesep.join(config["dockerfile_lines"]),
        "",
@@ -1306,6 +1323,10 @@ def docker_tag(
    base_image: Optional[str] = None,
 ) -> str:
    base_image = base_image or default_base_image(config)
+
+    image_distro = config.get("image_distro")
+    distro_tag = "" if image_distro == DEFAULT_IMAGE_DISTRO else f"-{image_distro}"
+
    if config.get("_INTERNAL_docker_tag"):
        return f"{base_image}:{config['_INTERNAL_docker_tag']}"

@@ -1313,8 +1334,8 @@ def docker_tag(
        return f"{base_image}-py{config['python_version']}"

    if config.get("node_version") and not config.get("python_version"):
-        return f"{base_image}:{config['node_version']}"
-    return f"{base_image}:{config['python_version']}"
+        return f"{base_image}:{config['node_version']}{distro_tag}"
+    return f"{base_image}:{config['python_version']}{distro_tag}"


 def config_to_docker(
@@ -1,2 +1,25 @@
+import click
+
+
 def clean_empty_lines(input_str: str):
    return "\n".join(filter(None, input_str.splitlines()))
+
+
+def warn_non_wolfi_distro(config_json: dict) -> None:
+    """Show warning if image_distro is not set to 'wolfi'."""
+    image_distro = config_json.get("image_distro", "debian")  # Default is debian
+    if image_distro != "wolfi":
+        click.secho(
+            "⚠️  Security Recommendation: Consider switching to Wolfi Linux for enhanced security.",
+            fg="yellow",
+            bold=True,
+        )
+        click.secho(
+            "   Wolfi is a security-oriented, minimal Linux distribution designed for containers.",
+            fg="yellow",
+        )
+        click.secho(
+            '   To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.',
+            fg="yellow",
+        )
+        click.secho("")  # Empty line for better readability
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"

 [project]
 name = "langgraph-cli"
-version = "0.2.10"
+version = "0.2.12"
 description = "CLI for interacting with LangGraph API"
 authors = []
 requires-python = ">=3.9"
@@ -119,6 +119,21 @@
              ],
              "description": "Optional. Configuration for the built-in HTTP server, controlling which custom routes are exposed\nand how cross-origin requests are handled.\n"
            },
+            "image_distro": {
+              "anyOf": [
+                {
+                  "type": "string",
+                  "enum": [
+                    "debian",
+                    "wolfi"
+                  ]
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Linux distribution for the base image.\n\nMust be either 'debian' or 'wolfi'. If omitted, defaults to 'debian'.\n"
+            },
            "store": {
              "anyOf": [
                {
@@ -257,6 +272,21 @@
              ],
              "description": "Optional. Configuration for the built-in HTTP server, controlling which custom routes are exposed\nand how cross-origin requests are handled.\n"
            },
+            "image_distro": {
+              "anyOf": [
+                {
+                  "type": "string",
+                  "enum": [
+                    "debian",
+                    "wolfi"
+                  ]
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Linux distribution for the base image.\n\nMust be either 'debian' or 'wolfi'. If omitted, defaults to 'debian'.\n"
+            },
            "store": {
              "anyOf": [
                {
@@ -119,6 +119,21 @@
              ],
              "description": "Optional. Configuration for the built-in HTTP server, controlling which custom routes are exposed\nand how cross-origin requests are handled.\n"
            },
+            "image_distro": {
+              "anyOf": [
+                {
+                  "type": "string",
+                  "enum": [
+                    "debian",
+                    "wolfi"
+                  ]
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Linux distribution for the base image.\n\nMust be either 'debian' or 'wolfi'. If omitted, defaults to 'debian'.\n"
+            },
            "store": {
              "anyOf": [
                {
@@ -257,6 +272,21 @@
              ],
              "description": "Optional. Configuration for the built-in HTTP server, controlling which custom routes are exposed\nand how cross-origin requests are handled.\n"
            },
+            "image_distro": {
+              "anyOf": [
+                {
+                  "type": "string",
+                  "enum": [
+                    "debian",
+                    "wolfi"
+                  ]
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "description": "Optional. Linux distribution for the base image.\n\nMust be either 'debian' or 'wolfi'. If omitted, defaults to 'debian'.\n"
+            },
            "store": {
              "anyOf": [
                {
@@ -22,12 +22,14 @@ DEFAULT_DOCKER_CAPABILITIES = DockerCapabilities(


@contextmanager
-def temporary_config_folder(config_content: dict):
+def temporary_config_folder(config_content: dict, levels: int = 0):
    # Create a temporary directory
    temp_dir = tempfile.mkdtemp()
    try:
        # Define the path for the config.json file
-        config_path = Path(temp_dir) / "config.json"
+        config_path = Path(temp_dir) / f"{'a/' * levels}config.json"
+        # Ensure the parent directory exists
+        config_path.parent.mkdir(parents=True, exist_ok=True)

        # Write the provided dictionary content to config.json
        with open(config_path, "w", encoding="utf-8") as config_file:
@@ -438,3 +440,132 @@ def test_dockerfile_command_with_bad_config() -> None:
        # Assert command was successful
        assert result.exit_code == 2
        assert "conf.json' does not exist" in result.output
+
+
+def test_dockerfile_command_shows_wolfi_warning() -> None:
+    """Test the 'dockerfile' command shows warning when image_distro is not wolfi."""
+    runner = CliRunner()
+    config_content = {
+        "python_version": "3.11",
+        "graphs": {"agent": "agent.py:graph"},
+        "dependencies": ["."],
+        # No image_distro specified - should default to debian and show warning
+    }
+
+    with temporary_config_folder(config_content) as temp_dir:
+        save_path = temp_dir / "Dockerfile"
+        agent_path = temp_dir / "agent.py"
+        agent_path.touch()
+
+        result = runner.invoke(
+            cli,
+            ["dockerfile", str(save_path), "--config", str(temp_dir / "config.json")],
+        )
+
+        # Assert command was successful
+        assert result.exit_code == 0, result.output
+
+        # Check that warning is shown
+        assert "Security Recommendation" in result.output
+        assert "Wolfi Linux" in result.output
+        assert "image_distro" in result.output
+        assert "wolfi" in result.output
+
+
+def test_dockerfile_command_no_wolfi_warning_when_wolfi_set() -> None:
+    """Test the 'dockerfile' command does NOT show warning when image_distro is wolfi."""
+    runner = CliRunner()
+    config_content = {
+        "python_version": "3.11",
+        "graphs": {"agent": "agent.py:graph"},
+        "dependencies": ["."],
+        "image_distro": "wolfi",  # Explicitly set to wolfi - should not show warning
+    }
+
+    with temporary_config_folder(config_content) as temp_dir:
+        save_path = temp_dir / "Dockerfile"
+        agent_path = temp_dir / "agent.py"
+        agent_path.touch()
+
+        result = runner.invoke(
+            cli,
+            ["dockerfile", str(save_path), "--config", str(temp_dir / "config.json")],
+        )
+
+        # Assert command was successful
+        assert result.exit_code == 0, result.output
+
+        # Check that warning is NOT shown
+        assert "Security Recommendation" not in result.output
+        assert "Wolfi Linux" not in result.output
+
+
+def test_build_command_shows_wolfi_warning() -> None:
+    """Test the 'build' command shows warning when image_distro is not wolfi."""
+    runner = CliRunner()
+    config_content = {
+        "python_version": "3.11",
+        "graphs": {"agent": "agent.py:graph"},
+        "dependencies": ["."],
+        # No image_distro specified - should default to debian and show warning
+    }
+
+    with temporary_config_folder(config_content) as temp_dir:
+        agent_path = temp_dir / "agent.py"
+        agent_path.touch()
+
+        # Mock docker command since we don't want to actually build
+        with runner.isolated_filesystem():
+            result = runner.invoke(
+                cli,
+                [
+                    "build",
+                    "--tag",
+                    "test-image",
+                    "--config",
+                    str(temp_dir / "config.json"),
+                ],
+                catch_exceptions=True,
+            )
+
+        # The command will fail because docker isn't available or we're mocking,
+        # but we should still see the warning before it fails
+        assert "Security Recommendation" in result.output
+        assert "Wolfi Linux" in result.output
+        assert "image_distro" in result.output
+        assert "wolfi" in result.output
+
+
+def test_build_generate_proper_build_context():
+    runner = CliRunner()
+    config_content = {
+        "python_version": "3.11",
+        "graphs": {"agent": "agent.py:graph"},
+        "dependencies": [".", "../../..", "../.."],
+        "image_distro": "wolfi",
+    }
+
+    with temporary_config_folder(config_content, levels=3) as temp_dir:
+        agent_path = temp_dir / "agent.py"
+        agent_path.touch()
+
+        # Mock docker command since we don't want to actually build
+        with runner.isolated_filesystem():
+            result = runner.invoke(
+                cli,
+                [
+                    "build",
+                    "--tag",
+                    "test-image",
+                    "--config",
+                    str(temp_dir / "config.json"),
+                ],
+                catch_exceptions=True,
+            )
+
+        build_context_pattern = re.compile(r"--build-context\s+(\w+)=([^\s]+)")
+
+        build_contexts = re.findall(build_context_pattern, result.output)
+        assert (
+            len(build_contexts) == 2
+        ), f"Expected 2 build contexts, but found {len(build_contexts)}"
@@ -11,6 +11,7 @@ from langgraph_cli.config import (
    PIP_CLEANUP_LINES,
    config_to_compose,
    config_to_docker,
+    docker_tag,
    validate_config,
    validate_config_file,
 )
@@ -34,6 +35,7 @@ def test_validate_config():
        "python_version": "3.11",
        "node_version": None,
        "pip_config_file": None,
+        "image_distro": "debian",
        "dockerfile_lines": [],
        "env": {},
        "store": None,
@@ -54,6 +56,7 @@ def test_validate_config():
        "python_version": "3.12",
        "node_version": None,
        "pip_config_file": "pipconfig.txt",
+        "image_distro": "debian",
        "dockerfile_lines": ["ARG meow"],
        "dependencies": [".", "langchain"],
        "graphs": {
@@ -120,10 +123,7 @@ def test_validate_config():
        }
    )
    assert config["python_version"] == "3.12-slim"
-    with pytest.raises(
-        ValueError,
-        match="Invalid http.app format",
-    ):
+    with pytest.raises(ValueError, match="Invalid http.app format"):
        validate_config(
            {
                "python_version": "3.12",
@@ -134,6 +134,83 @@ def test_validate_config():
        )


+def test_validate_config_image_distro():
+    """Test validation of image_distro field."""
+    # Valid image_distro values should work
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "image_distro": "debian",
+        }
+    )
+    assert config["image_distro"] == "debian"
+
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "image_distro": "wolfi",
+        }
+    )
+    assert config["image_distro"] == "wolfi"
+
+    # Missing image_distro should default to 'debian'
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+        }
+    )
+    assert config["image_distro"] == "debian"
+
+    # Invalid image_distro values should raise error
+    with pytest.raises(click.UsageError) as exc_info:
+        validate_config(
+            {
+                "python_version": "3.11",
+                "dependencies": ["."],
+                "graphs": {"agent": "./agent.py:graph"},
+                "image_distro": "ubuntu",
+            }
+        )
+    assert "Invalid image_distro: 'ubuntu'" in str(exc_info.value)
+    assert "Must be either 'debian' or 'wolfi'" 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"},
+                "image_distro": "alpine",
+            }
+        )
+    assert "Invalid image_distro: 'alpine'" in str(exc_info.value)
+
+    # Test base Node.js config with image distro
+    config = validate_config(
+        {
+            "node_version": "20",
+            "graphs": {"agent": "./agent.js:graph"},
+            "image_distro": "wolfi",
+        }
+    )
+    assert config["image_distro"] == "wolfi"
+
+    # Test Node.js config with no distro specified
+    config = validate_config(
+        {
+            "node_version": "20",
+            "graphs": {"agent": "./agent.js:graph"},
+        }
+    )
+    assert config["image_distro"] == "debian"
+
+
 def test_validate_config_file():
    with tempfile.TemporaryDirectory() as tmpdir:
        tmpdir_path = pathlib.Path(tmpdir)
@@ -891,3 +968,148 @@ def test_config_to_compose_end_to_end():
        watch=True,
    )
    assert clean_empty_lines(actual_compose_stdin) == expected_compose_stdin
+
+
+def test_docker_tag_image_distro():
+    """Test docker_tag function with different image_distro configurations."""
+
+    # Test 1: Default distro (debian) - no suffix
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+        }
+    )
+    tag = docker_tag(config)
+    assert tag == "langchain/langgraph-api:3.11"
+
+    # Test 2: Explicit debian distro - no suffix (same as default)
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "image_distro": "debian",
+        }
+    )
+    tag = docker_tag(config)
+    assert tag == "langchain/langgraph-api:3.11"
+
+    # Test 3: Wolfi distro - should add suffix
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "image_distro": "wolfi",
+        }
+    )
+    tag = docker_tag(config)
+    assert tag == "langchain/langgraph-api:3.11-wolfi"
+
+    # Test 4: Node.js with default distro
+    config = validate_config(
+        {
+            "node_version": "20",
+            "graphs": {"agent": "./agent.js:graph"},
+        }
+    )
+    tag = docker_tag(config)
+    assert tag == "langchain/langgraphjs-api:20"
+
+    # Test 5: Node.js with wolfi distro
+    config = validate_config(
+        {
+            "node_version": "20",
+            "graphs": {"agent": "./agent.js:graph"},
+            "image_distro": "wolfi",
+        }
+    )
+    tag = docker_tag(config)
+    assert tag == "langchain/langgraphjs-api:20-wolfi"
+
+    # Test 6: Custom base image with wolfi
+    config = validate_config(
+        {
+            "python_version": "3.12",
+            "dependencies": ["."],
+            "graphs": {"agent": "./agent.py:graph"},
+            "image_distro": "wolfi",
+            "base_image": "my-registry/custom-image",
+        }
+    )
+    tag = docker_tag(config, base_image="my-registry/custom-image")
+    assert tag == "my-registry/custom-image:3.12-wolfi"
+
+
+def test_docker_tag_multiplatform_with_distro():
+    """Test docker_tag with multiplatform configs and image_distro."""
+
+    # Test 1: Multiplatform (Python + Node) with wolfi
+    config = validate_config(
+        {
+            "python_version": "3.11",
+            "node_version": "20",
+            "dependencies": ["."],
+            "graphs": {"python": "./agent.py:graph", "js": "./agent.js:graph"},
+            "image_distro": "wolfi",
+        }
+    )
+    tag = docker_tag(config)
+    # Should default to Python when both are present
+    assert tag == "langchain/langgraph-api:3.11-wolfi"
+
+    # Test 2: Node-only multiplatform with wolfi
+    config = validate_config(
+        {
+            "node_version": "20",
+            "graphs": {"js": "./agent.js:graph"},
+            "image_distro": "wolfi",
+        }
+    )
+    tag = docker_tag(config)
+    assert tag == "langchain/langgraphjs-api:20-wolfi"
+
+
+def test_docker_tag_different_python_versions_with_distro():
+    """Test docker_tag with different Python versions and distros."""
+
+    versions_and_expected = [
+        ("3.11", "langchain/langgraph-api:3.11-wolfi"),
+        ("3.12", "langchain/langgraph-api:3.12-wolfi"),
+        ("3.13", "langchain/langgraph-api:3.13-wolfi"),
+    ]
+
+    for python_version, expected_tag in versions_and_expected:
+        config = validate_config(
+            {
+                "python_version": python_version,
+                "dependencies": ["."],
+                "graphs": {"agent": "./agent.py:graph"},
+                "image_distro": "wolfi",
+            }
+        )
+        tag = docker_tag(config)
+        assert tag == expected_tag, f"Failed for Python {python_version}"
+
+
+def test_docker_tag_different_node_versions_with_distro():
+    """Test docker_tag with different Node.js versions and distros."""
+
+    versions_and_expected = [
+        ("20", "langchain/langgraphjs-api:20-wolfi"),
+        ("21", "langchain/langgraphjs-api:21-wolfi"),
+        ("22", "langchain/langgraphjs-api:22-wolfi"),
+    ]
+
+    for node_version, expected_tag in versions_and_expected:
+        config = validate_config(
+            {
+                "node_version": node_version,
+                "graphs": {"agent": "./agent.js:graph"},
+                "image_distro": "wolfi",
+            }
+        )
+        tag = docker_tag(config)
+        assert tag == expected_tag, f"Failed for Node.js {node_version}"
@@ -0,0 +1,12 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(rg:*)",
+      "Bash(python:*)",
+      "Bash(grep:*)",
+      "Bash(sed:*)",
+      "Bash(awk:*)"
+    ],
+    "deny": []
+  }
+}
@@ -44,9 +44,8 @@ stop-postgres:
 	docker compose -f tests/compose-postgres.yml down -v

 start-dev-server:
-	uv run langgraph dev --config tests/example_app/langgraph.json --no-browser &
+	LOG_LEVEL=warning uv run langgraph dev --config tests/example_app/langgraph.json --no-browser & echo "$$!" > .devserver.pid
 	@echo "Dev server started."
-	@echo "Dev server PID: $$!" > .devserver.pid

 stop-dev-server:
 	@if [ -f .devserver.pid ]; then \
@@ -57,15 +56,22 @@ stop-dev-server:
 	fi

 TEST ?= .
+NO_DOCKER ?= $(sh command -v docker >/dev/null 2>&1 && echo "false" || echo "true")

 test:
-	make start-postgres &&\
-	make start-dev-server &&\
-	uv run pytest $(TEST); \
-	EXIT_CODE=$$?; \
-	make stop-postgres; \
-	make stop-dev-server; \
-	exit $$EXIT_CODE
+	if [ "$(NO_DOCKER)" = "false" ]; then \
+		make start-postgres &&\
+		make start-dev-server &&\
+		uv run pytest $(TEST); \
+		EXIT_CODE=$$?; \
+		make stop-postgres; \
+		make stop-dev-server; \
+		exit $$EXIT_CODE; \
+	else \
+		NO_DOCKER=true uv run pytest $(TEST) ; \
+		EXIT_CODE=$$?; \
+		exit $$EXIT_CODE; \
+	fi

 test_parallel:
 	make start-postgres &&\
@@ -21,6 +21,8 @@ def fanout_to_subgraph() -> StateGraph:
    class JokeOutput(TypedDict):
        jokes: list[str]

+    class JokeState(JokeInput, JokeOutput): ...
+
    async def bump(state: JokeOutput):
        return {"jokes": [state["jokes"][0] + " a"]}

@@ -35,7 +37,7 @@ def fanout_to_subgraph() -> StateGraph:
        return END if state["jokes"][0].endswith(" a" * 10) else "bump"

    # subgraph
-    subgraph = StateGraph(input=JokeInput, output=JokeOutput)
+    subgraph = StateGraph(JokeState, input=JokeInput, output=JokeOutput)
    subgraph.add_node("edit", edit)
    subgraph.add_node("generate", generate)
    subgraph.add_node("bump", bump)
@@ -69,6 +71,8 @@ def fanout_to_subgraph_sync() -> StateGraph:
    class JokeOutput(TypedDict):
        jokes: list[str]

+    class JokeState(JokeInput, JokeOutput): ...
+
    def bump(state: JokeOutput):
        return {"jokes": [state["jokes"][0] + " a"]}

@@ -83,7 +87,7 @@ def fanout_to_subgraph_sync() -> StateGraph:
        return END if state["jokes"][0].endswith(" a" * 10) else "bump"

    # subgraph
-    subgraph = StateGraph(input=JokeInput, output=JokeOutput)
+    subgraph = StateGraph(JokeState, input=JokeInput, output=JokeOutput)
    subgraph.add_node("edit", edit)
    subgraph.add_node("generate", generate)
    subgraph.add_node("bump", bump)
@@ -1,84 +0,0 @@
-import functools
-import warnings
-from typing import Any, Callable, TypeVar, Union, cast
-
-
-class LangGraphDeprecationWarning(DeprecationWarning):
-    pass
-
-
-F = TypeVar("F", bound=Callable[..., Any])
-C = TypeVar("C", bound=type[Any])
-
-
-def deprecated(
-    since: str, alternative: str, *, removal: str = "", example: str = ""
-) -> Callable[[F], F]:
-    def decorator(obj: Union[F, C]) -> Union[F, C]:
-        removal_str = removal if removal else "a future version"
-        message = (
-            f"{obj.__name__} is deprecated as of version {since} and will be"
-            f" removed in {removal_str}. Use {alternative} instead.{example}"
-        )
-        if isinstance(obj, type):
-            original_init = obj.__init__  # type: ignore[misc]
-
-            @functools.wraps(original_init)
-            def new_init(self, *args: Any, **kwargs: Any) -> None:  # type: ignore[no-untyped-def]
-                warnings.warn(message, LangGraphDeprecationWarning, stacklevel=2)
-                original_init(self, *args, **kwargs)
-
-            obj.__init__ = new_init  # type: ignore[misc]
-
-            docstring = (
-                f"**Deprecated**: This class is deprecated as of version {since}. "
-                f"Use `{alternative}` instead."
-            )
-            if obj.__doc__:
-                docstring = docstring + f"\n\n{obj.__doc__}"
-            obj.__doc__ = docstring
-
-            return cast(C, obj)
-        elif callable(obj):
-
-            @functools.wraps(obj)
-            def wrapper(*args: Any, **kwargs: Any) -> Any:
-                warnings.warn(message, LangGraphDeprecationWarning, stacklevel=2)
-                return obj(*args, **kwargs)
-
-            docstring = (
-                f"**Deprecated**: This function is deprecated as of version {since}. "
-                f"Use `{alternative}` instead."
-            )
-            if obj.__doc__:
-                docstring = docstring + f"\n\n{obj.__doc__}"
-            wrapper.__doc__ = docstring
-
-            return cast(F, wrapper)
-        else:
-            raise TypeError(
-                f"Can only add deprecation decorator to classes or callables, got '{type(obj)}' instead."
-            )
-
-    return decorator
-
-
-def deprecated_parameter(
-    arg_name: str, since: str, alternative: str, *, removal: str
-) -> Callable[[F], F]:
-    def decorator(func: F) -> F:
-        @functools.wraps(func)
-        def wrapper(*args, **kwargs):  # type: ignore[no-untyped-def]
-            if arg_name in kwargs:
-                warnings.warn(
-                    f"Parameter '{arg_name}' in function '{func.__name__}' is "
-                    f"deprecated as of version {since} and will be removed in version {removal}. "
-                    f"Use '{alternative}' parameter instead.",
-                    category=LangGraphDeprecationWarning,
-                    stacklevel=2,
-                )
-            return func(*args, **kwargs)
-
-        return cast(F, wrapper)
-
-    return decorator
@@ -1,17 +1,14 @@
 from langgraph.channels.any_value import AnyValue
 from langgraph.channels.binop import BinaryOperatorAggregate
-from langgraph.channels.context import Context
 from langgraph.channels.ephemeral_value import EphemeralValue
-from langgraph.channels.last_value import LastValue
+from langgraph.channels.last_value import LastValue, LastValueAfterFinish
 from langgraph.channels.topic import Topic
-from langgraph.channels.untracked_value import UntrackedValue

 __all__ = [
    "LastValue",
+    "LastValueAfterFinish",
    "Topic",
-    "Context",
    "BinaryOperatorAggregate",
-    "UntrackedValue",
    "EphemeralValue",
    "AnyValue",
 ]
@@ -1,5 +0,0 @@
-from langgraph.managed.context import Context as ContextManagedValue
-
-Context = ContextManagedValue.of
-
-__all__ = ["Context"]
@@ -1,206 +0,0 @@
-from collections.abc import Sequence, Set
-from typing import Any, Generic, NamedTuple, Optional, Union
-
-from typing_extensions import Self
-
-from langgraph.channels.base import BaseChannel, Value
-from langgraph.constants import MISSING
-from langgraph.errors import EmptyChannelError, InvalidUpdateError
-
-
-class WaitForNames(NamedTuple):
-    names: Set[Any]
-
-
-class DynamicBarrierValue(
-    Generic[Value], BaseChannel[Value, Union[Value, WaitForNames], Set[Value]]
-):
-    """A channel that switches between two states
-
-    - in the "priming" state it can't be read from.
-        - if it receives a WaitForNames update, it switches to the "waiting" state.
-    - in the "waiting" state it collects named values until all are received.
-        - once all named values are received, it can be read once, and it switches
-          back to the "priming" state.
-    """
-
-    __slots__ = ("names", "seen")
-
-    names: Optional[Set[Value]]
-    seen: set[Value]
-
-    def __init__(self, typ: type[Value]) -> None:
-        super().__init__(typ)
-        self.names = None
-        self.seen = set()
-
-    def __eq__(self, value: object) -> bool:
-        return isinstance(value, DynamicBarrierValue) and value.names == self.names
-
-    @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)
-        empty.key = self.key
-        empty.names = self.names
-        empty.seen = self.seen.copy()
-        return empty
-
-    def checkpoint(self) -> tuple[Optional[Set[Value]], set[Value]]:
-        return (self.names, self.seen)
-
-    def from_checkpoint(
-        self, checkpoint: tuple[Optional[Set[Value]], set[Value]]
-    ) -> Self:
-        empty = self.__class__(self.typ)
-        empty.key = self.key
-        if checkpoint is not MISSING:
-            names, seen = checkpoint
-            empty.names = names if names is not None else None
-            empty.seen = seen
-        return empty
-
-    def update(self, values: Sequence[Union[Value, WaitForNames]]) -> bool:
-        if wait_for_names := [v for v in values if isinstance(v, WaitForNames)]:
-            if len(wait_for_names) > 1:
-                raise InvalidUpdateError(
-                    f"At key '{self.key}': Received multiple WaitForNames updates in the same step."
-                )
-            self.names = wait_for_names[0].names
-            return True
-        elif self.names is not None:
-            updated = False
-            for value in values:
-                assert not isinstance(value, WaitForNames)
-                if value in self.names and value not in self.seen:
-                    self.seen.add(value)
-                    updated = True
-            return updated
-
-    def get(self) -> Value:
-        if self.seen != self.names:
-            raise EmptyChannelError()
-        return None
-
-    def is_available(self) -> bool:
-        return self.seen == self.names
-
-    def consume(self) -> bool:
-        if self.seen == self.names:
-            self.seen = set()
-            self.names = None
-            return True
-        return False
-
-
-class DynamicBarrierValueAfterFinish(
-    Generic[Value], BaseChannel[Value, Union[Value, WaitForNames], Set[Value]]
-):
-    """A channel that switches between two states
-
-    - in the "priming" state it can't be read from.
-        - if it receives a WaitForNames update, it switches to the "waiting" state.
-    - in the "waiting" state it collects named values until all are received.
-        - once all named values are received, and the finished flag is set, it can be read once, and it switches
-          back to the "priming" state.
-    """
-
-    __slots__ = ("names", "seen", "finished")
-
-    names: Optional[Set[Value]]
-    seen: set[Value]
-    finished: bool
-
-    def __init__(self, typ: type[Value]) -> None:
-        super().__init__(typ)
-        self.names = None
-        self.seen = set()
-        self.finished = False
-
-    def __eq__(self, value: object) -> bool:
-        return (
-            isinstance(value, DynamicBarrierValueAfterFinish)
-            and value.names == self.names
-        )
-
-    @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)
-        empty.key = self.key
-        empty.names = self.names
-        empty.seen = self.seen.copy()
-        empty.finished = self.finished
-        return empty
-
-    def checkpoint(self) -> tuple[Optional[Set[Value]], set[Value], bool]:
-        return (self.names, self.seen, self.finished)
-
-    def from_checkpoint(
-        self, checkpoint: tuple[Optional[Set[Value]], set[Value], bool]
-    ) -> Self:
-        empty = self.__class__(self.typ)
-        empty.key = self.key
-        if checkpoint is not MISSING:
-            names, seen, finished = checkpoint
-            empty.names = names if names is not None else None
-            empty.seen = seen
-            empty.finished = finished
-        return empty
-
-    def update(self, values: Sequence[Union[Value, WaitForNames]]) -> bool:
-        if wait_for_names := [v for v in values if isinstance(v, WaitForNames)]:
-            if len(wait_for_names) > 1:
-                raise InvalidUpdateError(
-                    f"At key '{self.key}': Received multiple WaitForNames updates in the same step."
-                )
-            self.names = wait_for_names[0].names
-            return True
-        elif self.names is not None:
-            updated = False
-            for value in values:
-                assert not isinstance(value, WaitForNames)
-                if value in self.names and value not in self.seen:
-                    self.seen.add(value)
-                    updated = True
-            return updated
-
-    def get(self) -> Value:
-        if not self.finished and self.seen != self.names:
-            raise EmptyChannelError()
-        return None
-
-    def is_available(self) -> bool:
-        return self.seen == self.names and self.finished
-
-    def consume(self) -> bool:
-        if self.finished and self.seen == self.names:
-            self.seen = set()
-            self.names = None
-            return True
-        return False
-
-    def finish(self) -> bool:
-        if not self.finished and self.seen == self.names:
-            self.finished = True
-            return True
-        else:
-            return False
@@ -71,12 +71,14 @@ class Topic(
        return empty

    def update(self, values: Sequence[Union[Value, list[Value]]]) -> bool:
-        current = list(self.values)
+        updated = False
        if not self.accumulate:
+            updated = bool(self.values)
            self.values = list[Value]()
-        if flat_values := flatten(values):
+        if flat_values := tuple(flatten(values)):
+            updated = True
            self.values.extend(flat_values)
-        return self.values != current
+        return updated

    def get(self) -> Sequence[Value]:
        if self.values:
@@ -1,66 +0,0 @@
-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
@@ -122,7 +122,6 @@ RESERVED = {
    ERROR,
    NO_WRITES,
    SCHEDULED,
-    TASKS,
    # reserved config.configurable keys
    CONFIG_KEY_SEND,
    CONFIG_KEY_READ,
@@ -2,6 +2,7 @@ import asyncio
 import concurrent.futures
 import functools
 import inspect
+import warnings
 from collections.abc import Awaitable, Sequence
 from dataclasses import dataclass
 from typing import (
@@ -16,6 +17,8 @@ from typing import (
    overload,
 )

+from typing_extensions import Unpack
+
 from langgraph.cache.base import BaseCache
 from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue
@@ -34,6 +37,8 @@ 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.typing import DeprecatedKwargs
+from langgraph.warnings import LangGraphDeprecatedSinceV10


 class TaskFunction(Generic[P, T]):
@@ -41,7 +46,7 @@ class TaskFunction(Generic[P, T]):
        self,
        func: Callable[P, T],
        *,
-        retry: Optional[Sequence[RetryPolicy]] = (),
+        retry_policy: Sequence[RetryPolicy],
        cache_policy: Optional[CachePolicy[Callable[P, Union[str, bytes]]]] = None,
        name: Optional[str] = None,
    ) -> None:
@@ -57,13 +62,17 @@ class TaskFunction(Generic[P, T]):
                # handle regular functions / partials / callable classes, etc.
                func.__name__ = name
        self.func = func
-        self.retry = retry
+        self.retry_policy = retry_policy
        self.cache_policy = cache_policy
        functools.update_wrapper(self, func)

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> SyncAsyncFuture[T]:
        return call(
-            self.func, retry=self.retry, cache_policy=self.cache_policy, *args, **kwargs
+            self.func,
+            retry_policy=self.retry_policy,
+            cache_policy=self.cache_policy,
+            *args,
+            **kwargs,
        )

    def clear_cache(self, cache: BaseCache) -> None:
@@ -83,8 +92,9 @@ class TaskFunction(Generic[P, T]):
 def task(
    *,
    name: Optional[str] = None,
-    retry: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]] = None,
+    retry_policy: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]] = None,
    cache_policy: Optional[CachePolicy[Callable[P, Union[str, bytes]]]] = None,
+    **kwargs: Unpack[DeprecatedKwargs],
 ) -> Callable[
    [Union[Callable[P, Awaitable[T]], Callable[P, T]]],
    TaskFunction[P, T],
@@ -101,8 +111,9 @@ def task(
    __func_or_none__: Optional[Union[Callable[P, Awaitable[T]], Callable[P, T]]] = None,
    *,
    name: Optional[str] = None,
-    retry: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]] = None,
+    retry_policy: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]] = None,
    cache_policy: Optional[CachePolicy[Callable[P, Union[str, bytes]]]] = None,
+    **kwargs: Unpack[DeprecatedKwargs],
 ) -> Union[
    Callable[
        [Union[Callable[P, Awaitable[T]], Callable[P, T]]],
@@ -125,7 +136,9 @@ def task(
    - Calling the function produces a future. This makes it easy to parallelize tasks.

    Args:
-        retry: An optional retry policy to use for the task in case of a failure.
+        name: An optional name for the task. If not provided, the function name will be used.
+        retry_policy: An optional retry policy (or list of policies) to use for the task in case of a failure.
+        cache_policy: An optional cache policy to use for the task. This allows caching of the task results.

    Returns:
        A callable function when used as a decorator.
@@ -166,10 +179,21 @@ def task(
        await add_one.ainvoke([1, 2, 3])  # Returns [2, 3, 4]
        ```
    """
-    if isinstance(retry, RetryPolicy):
-        retry_policies: Optional[Sequence[RetryPolicy]] = (retry,)
-    else:
-        retry_policies = retry
+    if (retry := kwargs.get("retry")) is not None:
+        warnings.warn(
+            "`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
+            category=LangGraphDeprecatedSinceV10,
+        )
+        if retry_policy is None:
+            retry_policy = retry  # type: ignore[assignment]
+
+    retry_policies: Sequence[RetryPolicy] = (
+        ()
+        if retry_policy is None
+        else (retry_policy,)
+        if isinstance(retry_policy, RetryPolicy)
+        else retry_policy
+    )

    def decorator(
        func: Union[Callable[P, Awaitable[T]], Callable[P, T]],
@@ -177,7 +201,7 @@ def task(
        Callable[P, concurrent.futures.Future[T]], Callable[P, asyncio.Future[T]]
    ]:
        return TaskFunction(
-            func, retry=retry_policies, cache_policy=cache_policy, name=name
+            func, retry_policy=retry_policies, cache_policy=cache_policy, name=name
        )

    if __func_or_none__ is not None:
@@ -232,8 +256,11 @@ class entrypoint:
            its state across runs.
        store: A generalized key-value store. Some implementations may support
            semantic search capabilities through an optional `index` configuration.
+        cache: A cache to use for caching the results of the workflow.
        config_schema: Specifies the schema for the configuration object that will be
            passed to the workflow.
+        cache_policy: A cache policy to use for caching the results of the workflow.
+        retry_policy: A retry policy (or list of policies) to use for the workflow in case of a failure.

    Example: Using entrypoint and tasks
        ```python
@@ -354,14 +381,23 @@ class entrypoint:
        cache: Optional[BaseCache] = None,
        config_schema: Optional[type[Any]] = None,
        cache_policy: Optional[CachePolicy] = None,
-        retry: Union[RetryPolicy, Sequence[RetryPolicy]] = (),
+        retry_policy: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]] = None,
+        **kwargs: Unpack[DeprecatedKwargs],
    ) -> None:
        """Initialize the entrypoint decorator."""
+        if (retry := kwargs.get("retry")) is not None:
+            warnings.warn(
+                "`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
+                category=LangGraphDeprecatedSinceV10,
+            )
+            if retry_policy is None:
+                retry_policy = retry  # type: ignore[assignment]
+
        self.checkpointer = checkpointer
        self.store = store
        self.cache = cache
        self.cache_policy = cache_policy
-        self.retry = retry
+        self.retry_policy = retry_policy
        self.config_schema = config_schema

    @dataclass(**_DC_KWARGS)
@@ -493,6 +529,6 @@ class entrypoint:
            store=self.store,
            cache=self.cache,
            cache_policy=self.cache_policy,
-            retry_policy=self.retry,
+            retry_policy=self.retry_policy or (),
            config_type=self.config_schema,
        )
@@ -1,13 +1,11 @@
-from langgraph.graph.graph import END, START, Graph
-from langgraph.graph.message import MessageGraph, MessagesState, add_messages
+from langgraph.constants import END, START
+from langgraph.graph.message import MessagesState, add_messages
 from langgraph.graph.state import StateGraph

 __all__ = [
    "END",
    "START",
-    "Graph",
    "StateGraph",
-    "MessageGraph",
    "add_messages",
    "MessagesState",
 ]
@@ -87,7 +87,6 @@ def _get_branch_path_input_schema(
 class Branch(NamedTuple):
    path: Runnable[Any, Union[Hashable, list[Hashable]]]
    ends: Optional[dict[Hashable, str]]
-    then: Optional[str] = None
    input_schema: Optional[type[Any]] = None

    @classmethod
@@ -95,7 +94,6 @@ class Branch(NamedTuple):
        cls,
        path: Runnable[Any, Union[Hashable, list[Hashable]]],
        path_map: Optional[Union[dict[Hashable, str], list[str]]],
-        then: Optional[str] = None,
        infer_schema: bool = False,
    ) -> "Branch":
        # coerce path_map to a dictionary
@@ -123,7 +121,7 @@ class Branch(NamedTuple):
        # infer input schema
        input_schema = _get_branch_path_input_schema(path) if infer_schema else None
        # create branch
-        return cls(path=path, ends=path_map_, then=then, input_schema=input_schema)
+        return cls(path=path, ends=path_map_, input_schema=input_schema)

    def run(
        self,
@@ -1,443 +0,0 @@
-import logging
-from collections import defaultdict
-from collections.abc import Awaitable, Hashable, Sequence
-from typing import (
-    Any,
-    Callable,
-    NamedTuple,
-    Optional,
-    Union,
-    cast,
-    overload,
-)
-
-from langchain_core.runnables import Runnable
-from typing_extensions import Self
-
-from langgraph.cache.base import BaseCache
-from langgraph.channels.ephemeral_value import EphemeralValue
-from langgraph.constants import (
-    EMPTY_SEQ,
-    END,
-    NS_END,
-    NS_SEP,
-    START,
-    TAG_HIDDEN,
-    Send,
-)
-from langgraph.graph.branch import Branch
-from langgraph.pregel import Channel, Pregel
-from langgraph.pregel.read import PregelNode
-from langgraph.pregel.write import ChannelWrite, ChannelWriteEntry
-from langgraph.store.base import BaseStore
-from langgraph.types import All, Checkpointer
-from langgraph.utils.runnable import RunnableLike, coerce_to_runnable
-
-logger = logging.getLogger(__name__)
-
-
-class NodeSpec(NamedTuple):
-    runnable: Runnable
-    metadata: Optional[dict[str, Any]] = None
-    ends: Optional[Union[tuple[str, ...], dict[str, str]]] = EMPTY_SEQ
-
-
-class Graph:
-    def __init__(self) -> None:
-        self.nodes: dict[str, NodeSpec] = {}
-        self.edges = set[tuple[str, str]]()
-        self.branches: defaultdict[str, dict[str, Branch]] = defaultdict(dict)
-        self.support_multiple_edges = False
-        self.compiled = False
-
-    @property
-    def _all_edges(self) -> set[tuple[str, str]]:
-        return self.edges
-
-    @overload
-    def add_node(
-        self,
-        node: RunnableLike,
-        *,
-        metadata: Optional[dict[str, Any]] = None,
-    ) -> Self: ...
-
-    @overload
-    def add_node(
-        self,
-        node: str,
-        action: RunnableLike,
-        *,
-        metadata: Optional[dict[str, Any]] = None,
-    ) -> Self: ...
-
-    def add_node(
-        self,
-        node: Union[str, RunnableLike],
-        action: Optional[RunnableLike] = None,
-        *,
-        metadata: Optional[dict[str, Any]] = None,
-    ) -> Self:
-        """Add a new node to the graph.
-
-        Args:
-            node: The function or runnable this node will run.
-                If a string is provided, it will be used as the node name, and action will be used as the function or runnable.
-            action: The action associated with the node. (default: None)
-                Will be used as the node function or runnable if `node` is a string (node name).
-            metadata: The metadata associated with the node. (default: None)
-        """
-        if isinstance(node, str):
-            for character in (NS_SEP, NS_END):
-                if character in node:
-                    raise ValueError(
-                        f"'{character}' is a reserved character and is not allowed in the node names."
-                    )
-
-        if self.compiled:
-            logger.warning(
-                "Adding a node to a graph that has already been compiled. This will "
-                "not be reflected in the compiled graph."
-            )
-        if not isinstance(node, str):
-            action = node
-            node = getattr(action, "name", getattr(action, "__name__"))
-            if node is None:
-                raise ValueError(
-                    "Node name must be provided if action is not a function"
-                )
-        if action is None:
-            raise RuntimeError(
-                "Expected a function or Runnable action in add_node. Received None."
-            )
-        if node in self.nodes:
-            raise ValueError(f"Node `{node}` already present.")
-        if node == END or node == START:
-            raise ValueError(f"Node `{node}` is reserved.")
-
-        self.nodes[cast(str, node)] = NodeSpec(
-            coerce_to_runnable(action, name=cast(str, node), trace=False), metadata
-        )
-        return self
-
-    def add_edge(self, start_key: str, end_key: str) -> Self:
-        """Add a directed edge from the start node to the end node.
-
-        Args:
-            start_key: The key of the start node of the edge.
-            end_key: The key of the end node of the edge.
-        """
-        if self.compiled:
-            logger.warning(
-                "Adding an edge to a graph that has already been compiled. This will "
-                "not be reflected in the compiled graph."
-            )
-        if start_key == END:
-            raise ValueError("END cannot be a start node")
-        if end_key == START:
-            raise ValueError("START cannot be an end node")
-
-        # run this validation only for non-StateGraph graphs
-        if not hasattr(self, "channels") and start_key in set(
-            start for start, _ in self.edges
-        ):
-            raise ValueError(
-                f"Already found path for node '{start_key}'.\n"
-                "For multiple edges, use StateGraph with an Annotated state key."
-            )
-
-        self.edges.add((start_key, end_key))
-        return self
-
-    def add_conditional_edges(
-        self,
-        source: str,
-        path: Union[
-            Callable[..., Union[Hashable, list[Hashable]]],
-            Callable[..., Awaitable[Union[Hashable, list[Hashable]]]],
-            Runnable[Any, Union[Hashable, list[Hashable]]],
-        ],
-        path_map: Optional[Union[dict[Hashable, str], list[str]]] = None,
-        then: Optional[str] = None,
-    ) -> Self:
-        """Add a conditional edge from the starting node to any number of destination nodes.
-
-        Args:
-            source: The starting node. This conditional edge will run when
-                exiting this node.
-            path: The callable that determines the next
-                node or nodes. If not specifying `path_map` it should return one or
-                more nodes. If it returns END, the graph will stop execution.
-            path_map: Optional mapping of paths to node
-                names. If omitted the paths returned by `path` should be node names.
-            then: The name of a node to execute after the nodes
-                selected by `path`.
-
-        Returns:
-            Self: The instance of the graph, allowing for method chaining.
-
-        Note: Without typehints on the `path` function's return value (e.g., `-> Literal["foo", "__end__"]:`)
-            or a path_map, the graph visualization assumes the edge could transition to any node in the graph.
-
-        """  # noqa: E501
-        if self.compiled:
-            logger.warning(
-                "Adding an edge to a graph that has already been compiled. This will "
-                "not be reflected in the compiled graph."
-            )
-
-        # find a name for the condition
-        path = coerce_to_runnable(path, name=None, trace=True)
-        name = path.name or "condition"
-        # validate the condition
-        if name in self.branches[source]:
-            raise ValueError(
-                f"Branch with name `{path.name}` already exists for node `{source}`"
-            )
-        # save it
-        self.branches[source][name] = Branch.from_path(path, path_map, then, False)
-        return self
-
-    def set_entry_point(self, key: str) -> Self:
-        """Specifies the first node to be called in the graph.
-
-        Equivalent to calling `add_edge(START, key)`.
-
-        Parameters:
-            key (str): The key of the node to set as the entry point.
-
-        Returns:
-            Self: The instance of the graph, allowing for method chaining.
-        """
-        return self.add_edge(START, key)
-
-    def set_conditional_entry_point(
-        self,
-        path: Union[
-            Callable[..., Union[Hashable, list[Hashable]]],
-            Callable[..., Awaitable[Union[Hashable, list[Hashable]]]],
-            Runnable[Any, Union[Hashable, list[Hashable]]],
-        ],
-        path_map: Optional[Union[dict[Hashable, str], list[str]]] = None,
-        then: Optional[str] = None,
-    ) -> Self:
-        """Sets a conditional entry point in the graph.
-
-        Args:
-            path: The callable that determines the next
-                node or nodes. If not specifying `path_map` it should return one or
-                more nodes. If it returns END, the graph will stop execution.
-            path_map: Optional mapping of paths to node
-                names. If omitted the paths returned by `path` should be node names.
-            then: The name of a node to execute after the nodes
-                selected by `path`.
-
-        Returns:
-            Self: The instance of the graph, allowing for method chaining.
-        """
-        return self.add_conditional_edges(START, path, path_map, then)
-
-    def set_finish_point(self, key: str) -> Self:
-        """Marks a node as a finish point of the graph.
-
-        If the graph reaches this node, it will cease execution.
-
-        Parameters:
-            key (str): The key of the node to set as the finish point.
-
-        Returns:
-            Self: The instance of the graph, allowing for method chaining.
-        """
-        return self.add_edge(key, END)
-
-    def validate(self, interrupt: Optional[Sequence[str]] = None) -> Self:
-        # assemble sources
-        all_sources = {src for src, _ in self._all_edges}
-        for start, branches in self.branches.items():
-            all_sources.add(start)
-            for cond, branch in branches.items():
-                if branch.then is not None:
-                    if branch.ends is not None:
-                        for end in branch.ends.values():
-                            if end != END:
-                                all_sources.add(end)
-                    else:
-                        for node in self.nodes:
-                            if node != start and node != branch.then:
-                                all_sources.add(node)
-        for name, spec in self.nodes.items():
-            if spec.ends:
-                all_sources.add(name)
-        # validate sources
-        for source in all_sources:
-            if source not in self.nodes and source != START:
-                raise ValueError(f"Found edge starting at unknown node '{source}'")
-
-        if START not in all_sources:
-            raise ValueError(
-                "Graph must have an entrypoint: add at least one edge from START to another node"
-            )
-
-        # assemble targets
-        all_targets = {end for _, end in self._all_edges}
-        for start, branches in self.branches.items():
-            for cond, branch in branches.items():
-                if branch.then is not None:
-                    all_targets.add(branch.then)
-                if branch.ends is not None:
-                    for end in branch.ends.values():
-                        if end not in self.nodes and end != END:
-                            raise ValueError(
-                                f"At '{start}' node, '{cond}' branch found unknown target '{end}'"
-                            )
-                        all_targets.add(end)
-                else:
-                    all_targets.add(END)
-                    for node in self.nodes:
-                        if node != start and node != branch.then:
-                            all_targets.add(node)
-        for name, spec in self.nodes.items():
-            if spec.ends:
-                all_targets.update(spec.ends)
-        for target in all_targets:
-            if target not in self.nodes and target != END:
-                raise ValueError(f"Found edge ending at unknown node `{target}`")
-        # validate interrupts
-        if interrupt:
-            for node in interrupt:
-                if node not in self.nodes:
-                    raise ValueError(f"Interrupt node `{node}` not found")
-
-        self.compiled = True
-        return self
-
-    def compile(
-        self,
-        checkpointer: Checkpointer = None,
-        interrupt_before: Optional[Union[All, list[str]]] = None,
-        interrupt_after: Optional[Union[All, list[str]]] = None,
-        debug: bool = False,
-        name: Optional[str] = None,
-        *,
-        cache: Optional[BaseCache] = None,
-        store: Optional[BaseStore] = None,
-    ) -> "CompiledGraph":
-        """Compiles the graph into a `CompiledGraph` object.
-
-        The compiled graph implements the `Runnable` interface and can be invoked,
-        streamed, batched, and run asynchronously.
-
-        Args:
-            checkpointer: A checkpoint saver object or flag.
-                If provided, this Checkpointer serves as a fully versioned "short-term memory" for the graph,
-                allowing it to be paused, resumed, and replayed from any point.
-                If None, it may inherit the parent graph's checkpointer when used as a subgraph.
-                If False, it will not use or inherit any checkpointer.
-            interrupt_before: An optional list of node names to interrupt before.
-            interrupt_after: An optional list of node names to interrupt after.
-            debug: A flag indicating whether to enable debug mode.
-            name: The name to use for the compiled graph.
-
-        Returns:
-            CompiledGraph: The compiled graph.
-        """
-        # assign default values
-        interrupt_before = interrupt_before or []
-        interrupt_after = interrupt_after or []
-
-        # validate the graph
-        self.validate(
-            interrupt=(
-                (interrupt_before if interrupt_before != "*" else []) + interrupt_after
-                if interrupt_after != "*"
-                else []
-            )
-        )
-
-        # create empty compiled graph
-        compiled = CompiledGraph(
-            builder=self,
-            nodes={},
-            channels={START: EphemeralValue(Any), END: EphemeralValue(Any)},
-            input_channels=START,
-            output_channels=END,
-            stream_mode="values",
-            stream_channels=[],
-            checkpointer=checkpointer,
-            interrupt_before_nodes=interrupt_before,
-            interrupt_after_nodes=interrupt_after,
-            auto_validate=False,
-            debug=debug,
-            name=name or "LangGraph",
-            cache=cache,
-            store=store,
-        )
-
-        # attach nodes, edges, and branches
-        for key, node in self.nodes.items():
-            compiled.attach_node(key, node)
-
-        for start, end in self.edges:
-            compiled.attach_edge(start, end)
-
-        for start, branches in self.branches.items():
-            for name, branch in branches.items():
-                compiled.attach_branch(start, name, branch)
-
-        # validate the compiled graph
-        return compiled.validate()
-
-
-class CompiledGraph(Pregel):
-    builder: Graph
-
-    def __init__(self, *, builder: Graph, **kwargs: Any) -> None:
-        super().__init__(**kwargs)
-        self.builder = builder
-
-    def attach_node(self, key: str, node: NodeSpec) -> None:
-        self.channels[key] = EphemeralValue(Any)
-        self.nodes[key] = (
-            PregelNode(channels=[], triggers=[], metadata=node.metadata)
-            | node.runnable
-            | ChannelWrite([ChannelWriteEntry(key)])
-        )
-        cast(list[str], self.stream_channels).append(key)
-
-    def attach_edge(self, start: str, end: str) -> None:
-        if end == END:
-            # publish to end channel
-            self.nodes[start].writers.append(ChannelWrite([ChannelWriteEntry(END)]))
-        else:
-            # subscribe to start channel
-            self.nodes[end].triggers.append(start)
-            cast(list[str], self.nodes[end].channels).append(start)
-
-    def attach_branch(self, start: str, name: str, branch: Branch) -> None:
-        def get_writes(
-            packets: Sequence[Union[str, Send]], static: bool = False
-        ) -> Sequence[Union[ChannelWriteEntry, Send]]:
-            return [
-                (
-                    ChannelWriteEntry(f"branch:{start}:{name}:{p}" if p != END else END)
-                    if not isinstance(p, Send)
-                    else p
-                )
-                for p in packets
-            ]
-
-        # add hidden start node
-        if start == START and start not in self.nodes:
-            self.nodes[start] = Channel.subscribe_to(START, tags=[TAG_HIDDEN])
-
-        # attach branch writer
-        self.nodes[start] |= branch.run(get_writes)
-
-        # attach branch readers
-        ends = branch.ends.values() if branch.ends else [node for node in self.nodes]
-        for end in ends:
-            if end != END:
-                channel_name = f"branch:{start}:{name}:{end}"
-                self.channels[channel_name] = EphemeralValue(Any)
-                self.nodes[end].triggers.append(channel_name)
-                cast(list[str], self.nodes[end].channels).append(channel_name)
@@ -24,7 +24,6 @@ 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,57 +225,6 @@ 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]

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import inspect
 import logging
 import typing
@@ -10,9 +12,11 @@ from types import FunctionType
 from typing import (
    Any,
    Callable,
+    Generic,
    Literal,
    NamedTuple,
    Optional,
+    Protocol,
    Union,
    cast,
    get_args,
@@ -23,17 +27,11 @@ from typing import (

 from langchain_core.runnables import Runnable, RunnableConfig
 from pydantic import BaseModel
-from typing_extensions import Self
+from typing_extensions import Self, TypeAlias, Unpack

-from langgraph._api.deprecation import LangGraphDeprecationWarning
 from langgraph.cache.base import BaseCache
 from langgraph.channels.base import BaseChannel
 from langgraph.channels.binop import BinaryOperatorAggregate
-from langgraph.channels.dynamic_barrier_value import (
-    DynamicBarrierValue,
-    DynamicBarrierValueAfterFinish,
-    WaitForNames,
-)
 from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue, LastValueAfterFinish
 from langgraph.channels.named_barrier_value import (
@@ -43,10 +41,12 @@ from langgraph.channels.named_barrier_value import (
 from langgraph.checkpoint.base import Checkpoint
 from langgraph.constants import (
    EMPTY_SEQ,
+    END,
    INTERRUPT,
    MISSING,
    NS_END,
    NS_SEP,
+    START,
    TAG_HIDDEN,
    TASKS,
 )
@@ -57,21 +57,11 @@ from langgraph.errors import (
    create_error_message,
 )
 from langgraph.graph.branch import Branch
-from langgraph.graph.graph import (
-    END,
-    START,
-    CompiledGraph,
-    Graph,
-    Send,
-)
 from langgraph.managed.base import (
-    ChannelKeyPlaceholder,
-    ChannelTypePlaceholder,
-    ConfiguredManagedValue,
    ManagedValueSpec,
    is_managed_value,
-    is_writable_managed_value,
 )
+from langgraph.pregel import Pregel
 from langgraph.pregel.read import ChannelRead, PregelNode
 from langgraph.pregel.write import (
    ChannelWrite,
@@ -79,10 +69,24 @@ from langgraph.pregel.write import (
    ChannelWriteTupleEntry,
 )
 from langgraph.store.base import BaseStore
-from langgraph.types import All, CachePolicy, Checkpointer, Command, RetryPolicy
-from langgraph.utils.fields import get_field_default, get_update_as_tuples
+from langgraph.types import (
+    All,
+    CachePolicy,
+    Checkpointer,
+    Command,
+    RetryPolicy,
+    Send,
+    StreamWriter,
+)
+from langgraph.typing import DeprecatedKwargs, InputT, StateT, StateT_contra, Unset
+from langgraph.utils.fields import (
+    get_cached_annotated_keys,
+    get_field_default,
+    get_update_as_tuples,
+)
 from langgraph.utils.pydantic import create_model
-from langgraph.utils.runnable import RunnableLike, coerce_to_runnable
+from langgraph.utils.runnable import coerce_to_runnable
+from langgraph.warnings import LangGraphDeprecatedSinceV10

 logger = logging.getLogger(__name__)

@@ -99,17 +103,77 @@ def _warn_invalid_state_schema(schema: Union[type[Any], Any]) -> None:
    )


-def _get_node_name(node: RunnableLike) -> str:
-    if isinstance(node, Runnable):
-        return node.get_name()
-    elif callable(node):
+class _StateNode(Protocol[StateT_contra]):
+    def __call__(self, state: StateT_contra) -> Any: ...
+
+
+class _NodeWithConfig(Protocol[StateT_contra]):
+    def __call__(self, state: StateT_contra, config: RunnableConfig) -> Any: ...
+
+
+class _NodeWithWriter(Protocol[StateT_contra]):
+    def __call__(self, state: StateT_contra, *, writer: StreamWriter) -> Any: ...
+
+
+class _NodeWithStore(Protocol[StateT_contra]):
+    def __call__(self, state: StateT_contra, *, store: BaseStore) -> Any: ...
+
+
+class _NodeWithWriterStore(Protocol[StateT_contra]):
+    def __call__(
+        self, state: StateT_contra, *, writer: StreamWriter, store: BaseStore
+    ) -> Any: ...
+
+
+class _NodeWithConfigWriter(Protocol[StateT_contra]):
+    def __call__(
+        self, state: StateT_contra, *, config: RunnableConfig, writer: StreamWriter
+    ) -> Any: ...
+
+
+class _NodeWithConfigStore(Protocol[StateT_contra]):
+    def __call__(
+        self, state: StateT_contra, *, config: RunnableConfig, store: BaseStore
+    ) -> Any: ...
+
+
+class _NodeWithConfigWriterStore(Protocol[StateT_contra]):
+    def __call__(
+        self,
+        state: StateT_contra,
+        *,
+        config: RunnableConfig,
+        writer: StreamWriter,
+        store: BaseStore,
+    ) -> 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.
+StateNode: TypeAlias = Union[
+    _StateNode[StateT_contra],
+    _NodeWithConfig[StateT_contra],
+    _NodeWithWriter[StateT_contra],
+    _NodeWithStore[StateT_contra],
+    _NodeWithWriterStore[StateT_contra],
+    _NodeWithConfigWriter[StateT_contra],
+    _NodeWithConfigStore[StateT_contra],
+    _NodeWithConfigWriterStore[StateT_contra],
+]
+
+
+def _get_node_name(node: StateNode) -> str:
+    try:
        return getattr(node, "__name__", node.__class__.__name__)
-    else:
+    except AttributeError:
        raise TypeError(f"Unsupported node type: {type(node)}")


 class StateNodeSpec(NamedTuple):
-    runnable: Runnable
+    # TODO: rename this callable, also move away from NamedTuple so that we can use
+    # a generic StateNode, so maybe a dataclass
+    runnable: StateNode
    metadata: Optional[dict[str, Any]]
    input: type[Any]
    retry_policy: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]]
@@ -118,7 +182,7 @@ class StateNodeSpec(NamedTuple):
    defer: bool = False


-class StateGraph(Graph):
+class StateGraph(Generic[StateT, InputT]):
    """A graph whose nodes communicate by reading and writing to a shared state.
    The signature of each node is State -> Partial<State>.

@@ -170,39 +234,32 @@ class StateGraph(Graph):
        ```
    """

-    nodes: dict[str, StateNodeSpec]  # type: ignore[assignment]
+    edges: set[tuple[str, str]]
+    nodes: dict[str, StateNodeSpec]
+    branches: defaultdict[str, dict[str, Branch]]
    channels: dict[str, BaseChannel]
    managed: dict[str, ManagedValueSpec]
    schemas: dict[type[Any], dict[str, Union[BaseChannel, ManagedValueSpec]]]

    def __init__(
        self,
-        state_schema: Optional[type[Any]] = None,
-        config_schema: Optional[type[Any]] = None,
+        state_schema: type[StateT],
+        config_schema: type[Any] | None = None,
        *,
-        input: Optional[type[Any]] = None,
-        output: Optional[type[Any]] = None,
+        input: type[InputT] | None = None,
+        output: type[Any] | None = None,
    ) -> None:
-        super().__init__()
-        if state_schema is None:
-            if input is None or output is None:
-                raise ValueError("Must provide state_schema or input and output")
-            state_schema = input
-            warnings.warn(
-                "Initializing StateGraph without state_schema is deprecated. "
-                "Please pass in an explicit state_schema instead of just an input and output schema.",
-                LangGraphDeprecationWarning,
-                stacklevel=2,
-            )
-        else:
-            if input is None:
-                input = state_schema
-            if output is None:
-                output = state_schema
+        input = input or state_schema
+        output = output or state_schema
+
+        self.nodes = {}
+        self.edges = set[tuple[str, str]]()
+        self.branches = defaultdict(dict)
+        self.support_multiple_edges = False
+        self.compiled = False
        self.schemas = {}
        self.channels = {}
        self.managed = {}
-        self.type_hints: dict[type[Any], dict[str, Any]] = {}
        self.schema = state_schema
        self.input = input
        self.output = output
@@ -230,7 +287,6 @@ class StateGraph(Graph):
                    " Managed channels are not permitted in Input/Output schema."
                )
            self.schemas[schema] = {**channels, **managed}
-            self.type_hints[schema] = type_hints
            for key, channel in channels.items():
                if key in self.channels:
                    if self.channels[key] != channel:
@@ -254,14 +310,15 @@ class StateGraph(Graph):
    @overload
    def add_node(
        self,
-        node: RunnableLike,
+        node: StateNode[StateT],
        *,
        defer: bool = False,
        metadata: Optional[dict[str, Any]] = None,
        input: Optional[type[Any]] = None,
-        retry: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]] = None,
+        retry_policy: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]] = None,
        cache_policy: Optional[CachePolicy] = None,
        destinations: Optional[Union[dict[str, str], tuple[str, ...]]] = None,
+        **kwargs: Unpack[DeprecatedKwargs],
    ) -> Self:
        """Add a new node to the state graph.
        Will take the name of the function/runnable as the node name.
@@ -272,29 +329,31 @@ class StateGraph(Graph):
    def add_node(
        self,
        node: str,
-        action: RunnableLike,
+        action: StateNode[StateT],
        *,
        defer: bool = False,
        metadata: Optional[dict[str, Any]] = None,
        input: Optional[type[Any]] = None,
-        retry: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]] = None,
+        retry_policy: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]] = None,
        cache_policy: Optional[CachePolicy] = None,
        destinations: Optional[Union[dict[str, str], tuple[str, ...]]] = None,
+        **kwargs: Unpack[DeprecatedKwargs],
    ) -> Self:
        """Add a new node to the state graph."""
        ...

    def add_node(
        self,
-        node: Union[str, RunnableLike],
-        action: Optional[RunnableLike] = None,
+        node: Union[str, StateNode[StateT]],
+        action: Optional[StateNode[StateT]] = None,
        *,
        defer: bool = False,
        metadata: Optional[dict[str, Any]] = None,
        input: Optional[type[Any]] = None,
-        retry: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]] = None,
+        retry_policy: Optional[Union[RetryPolicy, Sequence[RetryPolicy]]] = None,
        cache_policy: Optional[CachePolicy] = None,
        destinations: Optional[Union[dict[str, str], tuple[str, ...]]] = None,
+        **kwargs: Unpack[DeprecatedKwargs],
    ) -> Self:
        """Add a new node to the state graph.

@@ -306,7 +365,7 @@ class StateGraph(Graph):
            defer: Whether to defer the execution of the node until the run is about to end.
            metadata: The metadata associated with the node. (default: None)
            input: The input schema for the node. (default: the graph's input schema)
-            retry: The policy for retrying the node. (default: None)
+            retry_policy: The retry policy for the node. (default: None)
                If a sequence is provided, the first matching policy will be applied.
            cache_policy: The cache policy for the node. (default: None)
            destinations: Destinations that indicate where a node can route to.
@@ -314,8 +373,6 @@ class StateGraph(Graph):
                If a dict is provided, the keys will be used as the target node names and the values will be used as the labels for the edges.
                If a tuple is provided, the values will be used as the target node names.
                NOTE: this is only used for graph rendering and doesn't have any effect on the graph execution.
-        Raises:
-            ValueError: If the key is already being used as a state key.

        Example:
            ```python
@@ -345,6 +402,14 @@ class StateGraph(Graph):
        Returns:
            Self: The instance of the state graph, allowing for method chaining.
        """
+        if (retry := kwargs.get("retry")) is not None:
+            warnings.warn(
+                "`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
+                category=LangGraphDeprecatedSinceV10,
+            )
+            if retry_policy is None:
+                retry_policy = retry  # type: ignore[assignment]
+
        if not isinstance(node, str):
            action = node
            if isinstance(action, Runnable):
@@ -355,8 +420,6 @@ class StateGraph(Graph):
                raise ValueError(
                    "Node name must be provided if action is not a function"
                )
-        if node in self.channels:
-            raise ValueError(f"'{node}' is already being used as a state key")
        if self.compiled:
            logger.warning(
                "Adding a node to a graph that has already been compiled. This will "
@@ -377,7 +440,7 @@ class StateGraph(Graph):
            raise ValueError(f"Node `{node}` is reserved.")

        for character in (NS_SEP, NS_END):
-            if character in cast(str, node):
+            if character in node:
                raise ValueError(
                    f"'{character}' is a reserved character and is not allowed in the node names."
                )
@@ -432,11 +495,11 @@ class StateGraph(Graph):

        if input is not None:
            self._add_schema(input)
-        self.nodes[cast(str, node)] = StateNodeSpec(
-            coerce_to_runnable(action, name=cast(str, node), trace=False),
+        self.nodes[node] = StateNodeSpec(
+            coerce_to_runnable(action, name=node, trace=False),  # type: ignore
            metadata,
            input=input or self.schema,
-            retry_policy=retry,
+            retry_policy=retry_policy,
            cache_policy=cache_policy,
            ends=ends,
            defer=defer,
@@ -460,14 +523,30 @@ class StateGraph(Graph):
        Returns:
            Self: The instance of the state graph, allowing for method chaining.
        """
-        if isinstance(start_key, str):
-            return super().add_edge(start_key, end_key)
-
        if self.compiled:
            logger.warning(
                "Adding an edge to a graph that has already been compiled. This will "
                "not be reflected in the compiled graph."
            )
+
+        if isinstance(start_key, str):
+            if start_key == END:
+                raise ValueError("END cannot be a start node")
+            if end_key == START:
+                raise ValueError("START cannot be an end node")
+
+            # run this validation only for non-StateGraph graphs
+            if not hasattr(self, "channels") and start_key in set(
+                start for start, _ in self.edges
+            ):
+                raise ValueError(
+                    f"Already found path for node '{start_key}'.\n"
+                    "For multiple edges, use StateGraph with an Annotated state key."
+                )
+
+            self.edges.add((start_key, end_key))
+            return self
+
        for start in start_key:
            if start == END:
                raise ValueError("END cannot be a start node")
@@ -490,7 +569,6 @@ class StateGraph(Graph):
            Runnable[Any, Union[Hashable, list[Hashable]]],
        ],
        path_map: Optional[Union[dict[Hashable, str], list[str]]] = None,
-        then: Optional[str] = None,
    ) -> Self:
        """Add a conditional edge from the starting node to any number of destination nodes.

@@ -502,8 +580,6 @@ class StateGraph(Graph):
                more nodes. If it returns END, the graph will stop execution.
            path_map: Optional mapping of paths to node
                names. If omitted the paths returned by `path` should be node names.
-            then: The name of a node to execute after the nodes
-                selected by `path`.

        Returns:
            Self: The instance of the graph, allowing for method chaining.
@@ -527,19 +603,19 @@ class StateGraph(Graph):
                f"Branch with name `{path.name}` already exists for node `{source}`"
            )
        # save it
-        self.branches[source][name] = Branch.from_path(path, path_map, then, True)
+        self.branches[source][name] = Branch.from_path(path, path_map, True)
        if schema := self.branches[source][name].input_schema:
            self._add_schema(schema)
        return self

    def add_sequence(
        self,
-        nodes: Sequence[Union[RunnableLike, tuple[str, RunnableLike]]],
+        nodes: Sequence[Union[StateNode[StateT], tuple[str, StateNode[StateT]]]],
    ) -> Self:
        """Add a sequence of nodes that will be executed in the provided order.

        Args:
-            nodes: A sequence of RunnableLike objects (e.g. a LangChain Runnable or a callable) or (name, RunnableLike) tuples.
+            nodes: A sequence of StateNodes (callables that accept a state arg) or (name, StateNode) tuples.
                If no names are provided, the name will be inferred from the node object (e.g. a runnable or a callable name).
                Each node will be executed in the order provided.

@@ -574,6 +650,130 @@ class StateGraph(Graph):

        return self

+    def set_entry_point(self, key: str) -> Self:
+        """Specifies the first node to be called in the graph.
+
+        Equivalent to calling `add_edge(START, key)`.
+
+        Parameters:
+            key (str): The key of the node to set as the entry point.
+
+        Returns:
+            Self: The instance of the graph, allowing for method chaining.
+        """
+        return self.add_edge(START, key)
+
+    def set_conditional_entry_point(
+        self,
+        path: Union[
+            Callable[..., Union[Hashable, list[Hashable]]],
+            Callable[..., Awaitable[Union[Hashable, list[Hashable]]]],
+            Runnable[Any, Union[Hashable, list[Hashable]]],
+        ],
+        path_map: Optional[Union[dict[Hashable, str], list[str]]] = None,
+    ) -> Self:
+        """Sets a conditional entry point in the graph.
+
+        Args:
+            path: The callable that determines the next
+                node or nodes. If not specifying `path_map` it should return one or
+                more nodes. If it returns END, the graph will stop execution.
+            path_map: Optional mapping of paths to node
+                names. If omitted the paths returned by `path` should be node names.
+
+        Returns:
+            Self: The instance of the graph, allowing for method chaining.
+        """
+        return self.add_conditional_edges(START, path, path_map)
+
+    def set_finish_point(self, key: str) -> Self:
+        """Marks a node as a finish point of the graph.
+
+        If the graph reaches this node, it will cease execution.
+
+        Parameters:
+            key (str): The key of the node to set as the finish point.
+
+        Returns:
+            Self: The instance of the graph, allowing for method chaining.
+        """
+        return self.add_edge(key, END)
+
+    def validate(self, interrupt: Optional[Sequence[str]] = None) -> Self:
+        # assemble sources
+        all_sources = {src for src, _ in self._all_edges}
+        for start, branches in self.branches.items():
+            all_sources.add(start)
+        for name, spec in self.nodes.items():
+            if spec.ends:
+                all_sources.add(name)
+        # validate sources
+        for source in all_sources:
+            if source not in self.nodes and source != START:
+                raise ValueError(f"Found edge starting at unknown node '{source}'")
+
+        if START not in all_sources:
+            raise ValueError(
+                "Graph must have an entrypoint: add at least one edge from START to another node"
+            )
+
+        # assemble targets
+        all_targets = {end for _, end in self._all_edges}
+        for start, branches in self.branches.items():
+            for cond, branch in branches.items():
+                if branch.ends is not None:
+                    for end in branch.ends.values():
+                        if end not in self.nodes and end != END:
+                            raise ValueError(
+                                f"At '{start}' node, '{cond}' branch found unknown target '{end}'"
+                            )
+                        all_targets.add(end)
+                else:
+                    all_targets.add(END)
+                    for node in self.nodes:
+                        if node != start:
+                            all_targets.add(node)
+        for name, spec in self.nodes.items():
+            if spec.ends:
+                all_targets.update(spec.ends)
+        for target in all_targets:
+            if target not in self.nodes and target != END:
+                raise ValueError(f"Found edge ending at unknown node `{target}`")
+        # validate interrupts
+        if interrupt:
+            for node in interrupt:
+                if node not in self.nodes:
+                    raise ValueError(f"Interrupt node `{node}` not found")
+
+        self.compiled = True
+        return self
+
+    @overload
+    def compile(
+        self: StateGraph[StateT, Unset],
+        checkpointer: Checkpointer = None,
+        *,
+        cache: Optional[BaseCache] = None,
+        store: Optional[BaseStore] = None,
+        interrupt_before: Optional[Union[All, list[str]]] = None,
+        interrupt_after: Optional[Union[All, list[str]]] = None,
+        debug: bool = False,
+        name: Optional[str] = None,
+    ) -> CompiledStateGraph[StateT, StateT]: ...
+
+    @overload
+    def compile(
+        self: StateGraph[StateT, InputT],
+        checkpointer: Checkpointer = None,
+        *,
+        cache: Optional[BaseCache] = None,
+        store: Optional[BaseStore] = None,
+        interrupt_before: Optional[Union[All, list[str]]] = None,
+        interrupt_after: Optional[Union[All, list[str]]] = None,
+        debug: bool = False,
+        name: Optional[str] = None,
+    ) -> CompiledStateGraph[StateT, InputT]: ...
+
    def compile(
        self,
        checkpointer: Checkpointer = None,
@@ -584,7 +784,7 @@ class StateGraph(Graph):
        interrupt_after: Optional[Union[All, list[str]]] = None,
        debug: bool = False,
        name: Optional[str] = None,
-    ) -> "CompiledStateGraph":
+    ) -> Union[CompiledStateGraph[StateT, StateT], CompiledStateGraph[StateT, InputT]]:
        """Compiles the state graph into a `CompiledStateGraph` object.

        The compiled graph implements the `Runnable` interface and can be invoked,
@@ -636,7 +836,8 @@ class StateGraph(Graph):
            ]
        )

-        compiled = CompiledStateGraph(
+        ResolvedInputT: Union[type[InputT], type[StateT]] = self.input or self.schema
+        compiled = CompiledStateGraph[StateT, ResolvedInputT](  # type: ignore[valid-type]
            builder=self,
            schema_to_mapper={},
            config_type=self.config_schema,
@@ -684,17 +885,19 @@ class StateGraph(Graph):
        return compiled.validate()


-class CompiledStateGraph(CompiledGraph):
-    builder: StateGraph
+class CompiledStateGraph(Pregel[InputT], Generic[StateT, InputT]):
+    builder: StateGraph[StateT, InputT]
    schema_to_mapper: dict[type[Any], Optional[Callable[[Any], Any]]]

    def __init__(
        self,
        *,
+        builder: StateGraph[StateT, InputT],
        schema_to_mapper: dict[type[Any], Optional[Callable[[Any], Any]]],
        **kwargs: Any,
    ) -> None:
        super().__init__(**kwargs)
+        self.builder = builder
        self.schema_to_mapper = schema_to_mapper

    def get_input_schema(
@@ -726,9 +929,7 @@ class CompiledStateGraph(CompiledGraph):
            ]
        else:
            output_keys = list(self.builder.channels) + [
-                k
-                for k, v in self.builder.managed.items()
-                if is_writable_managed_value(v)
+                k for k, v in self.builder.managed.items()
            ]

        def _get_updates(
@@ -760,7 +961,7 @@ class CompiledStateGraph(CompiledGraph):
                    else:
                        updates.extend(_get_updates(i) or ())
                return updates
-            elif (t := type(input)) and get_type_hints(t):
+            elif (t := type(input)) and get_cached_annotated_keys(t):
                return get_update_as_tuples(input, output_keys)
            else:
                msg = create_error_message(
@@ -800,7 +1001,6 @@ class CompiledStateGraph(CompiledGraph):
                mapper = _pick_mapper(
                    list(input_values),
                    input_schema,
-                    self.builder.type_hints[input_schema],
                )
                self.schema_to_mapper[input_schema] = mapper

@@ -821,7 +1021,7 @@ class CompiledStateGraph(CompiledGraph):
                metadata=node.metadata,
                retry_policy=node.retry_policy,
                cache_policy=node.cache_policy,
-                bound=node.runnable,
+                bound=node.runnable,  # type: ignore[arg-type]
            )
        else:
            raise RuntimeError
@@ -871,17 +1071,6 @@ class CompiledStateGraph(CompiledGraph):
            ]
            if not writes:
                return []
-            if branch.then and branch.then != END:
-                writes.append(
-                    ChannelWriteEntry(
-                        f"branch:{start}:{name}::then",
-                        WaitForNames(
-                            frozenset(
-                                p.node if isinstance(p, Send) else p for p in packets
-                            )
-                        ),
-                    )
-                )
            return writes

        if with_reader:
@@ -896,7 +1085,7 @@ class CompiledStateGraph(CompiledGraph):
            if schema in self.schema_to_mapper:
                mapper = self.schema_to_mapper[schema]
            else:
-                mapper = _pick_mapper(channels, schema, self.builder.type_hints[schema])
+                mapper = _pick_mapper(channels, schema)
                self.schema_to_mapper[schema] = mapper
            # create reader
            reader: Optional[Callable[[RunnableConfig], Any]] = partial(
@@ -912,25 +1101,6 @@ class CompiledStateGraph(CompiledGraph):
        # attach branch publisher
        self.nodes[start].writers.append(branch.run(get_writes, reader))

-        # attach then subscriber
-        if branch.then and branch.then != END:
-            ends = (
-                branch.ends.values()
-                if branch.ends
-                else [node for node in self.builder.nodes if node != branch.then]
-            )
-            channel_name = f"branch:{start}:{name}::then"
-            if self.builder.nodes[branch.then].defer:
-                self.channels[channel_name] = DynamicBarrierValueAfterFinish(str)
-            else:
-                self.channels[channel_name] = DynamicBarrierValue(str)
-            self.nodes[branch.then].triggers.append(channel_name)
-            for end in ends:
-                if end != END:
-                    self.nodes[end].writers.append(
-                        ChannelWrite((ChannelWriteEntry(channel_name, end),))
-                    )
-
    def _migrate_checkpoint(self, checkpoint: Checkpoint) -> None:
        """Migrate a checkpoint to new channel layout."""

@@ -1037,7 +1207,7 @@ class CompiledStateGraph(CompiledGraph):


 def _pick_mapper(
-    state_keys: Sequence[str], schema: type[Any], type_hints: Optional[dict[str, Any]]
+    state_keys: Sequence[str], schema: type[Any]
 ) -> Optional[Callable[[Any], Any]]:
    if state_keys == ["__root__"]:
        return None
@@ -1207,12 +1377,6 @@ def _is_field_managed_value(name: str, typ: type[Any]) -> Optional[ManagedValueS
        if len(meta) >= 1:
            decoration = get_origin(meta[-1]) or meta[-1]
            if is_managed_value(decoration):
-                if isinstance(decoration, ConfiguredManagedValue):
-                    for k, v in decoration.kwargs.items():
-                        if v is ChannelKeyPlaceholder:
-                            decoration.kwargs[k] = name
-                        if v is ChannelTypePlaceholder:
-                            decoration.kwargs[k] = typ.__origin__
                return decoration

    return None
@@ -1,101 +1,30 @@
 from abc import ABC, abstractmethod
-from collections.abc import AsyncIterator, Iterator, Sequence
-from contextlib import asynccontextmanager, contextmanager
 from inspect import isclass
 from typing import (
    Any,
    Generic,
-    NamedTuple,
    TypeVar,
-    Union,
 )

-from typing_extensions import Self, TypeGuard
+from typing_extensions import TypeGuard

-from langgraph.types import LoopProtocol
+from langgraph.types import PregelScratchpad

 V = TypeVar("V")
 U = TypeVar("U")


 class ManagedValue(ABC, Generic[V]):
-    def __init__(self, loop: LoopProtocol) -> None:
-        self.loop = loop
-
-    @classmethod
-    @contextmanager
-    def enter(cls, loop: LoopProtocol, **kwargs: Any) -> Iterator[Self]:
-        try:
-            value = cls(loop, **kwargs)
-            yield value
-        finally:
-            # because managed value and Pregel have reference to each other
-            # let's make sure to break the reference on exit
-            try:
-                del value
-            except UnboundLocalError:
-                pass
-
-    @classmethod
-    @asynccontextmanager
-    async def aenter(cls, loop: LoopProtocol, **kwargs: Any) -> AsyncIterator[Self]:
-        try:
-            value = cls(loop, **kwargs)
-            yield value
-        finally:
-            # because managed value and Pregel have reference to each other
-            # let's make sure to break the reference on exit
-            try:
-                del value
-            except UnboundLocalError:
-                pass
-
+    @staticmethod
    @abstractmethod
-    def __call__(self) -> V: ...
+    def get(scratchpad: PregelScratchpad) -> V: ...


-class WritableManagedValue(Generic[V, U], ManagedValue[V], ABC):
-    @abstractmethod
-    def update(self, writes: Sequence[U]) -> None: ...
-
-    @abstractmethod
-    async def aupdate(self, writes: Sequence[U]) -> None: ...
-
-
-class ConfiguredManagedValue(NamedTuple):
-    cls: type[ManagedValue]
-    kwargs: dict[str, Any]
-
-
-ManagedValueSpec = Union[type[ManagedValue], ConfiguredManagedValue]
+ManagedValueSpec = type[ManagedValue]


 def is_managed_value(value: Any) -> TypeGuard[ManagedValueSpec]:
-    return (isclass(value) and issubclass(value, ManagedValue)) or isinstance(
-        value, ConfiguredManagedValue
-    )
+    return isclass(value) and issubclass(value, ManagedValue)


-def is_readonly_managed_value(value: Any) -> TypeGuard[type[ManagedValue]]:
-    return (
-        isclass(value)
-        and issubclass(value, ManagedValue)
-        and not issubclass(value, WritableManagedValue)
-    ) or (
-        isinstance(value, ConfiguredManagedValue)
-        and not issubclass(value.cls, WritableManagedValue)
-    )
-
-
-def is_writable_managed_value(value: Any) -> TypeGuard[type[WritableManagedValue]]:
-    return (isclass(value) and issubclass(value, WritableManagedValue)) or (
-        isinstance(value, ConfiguredManagedValue)
-        and issubclass(value.cls, WritableManagedValue)
-    )
-
-
-ChannelKeyPlaceholder = object()
-ChannelTypePlaceholder = object()
-
-
-ManagedValueMapping = dict[str, ManagedValue]
+ManagedValueMapping = dict[str, ManagedValueSpec]
@@ -1,111 +0,0 @@
-from collections.abc import AsyncIterator, Iterator
-from contextlib import (
-    AbstractAsyncContextManager,
-    AbstractContextManager,
-    asynccontextmanager,
-    contextmanager,
-)
-from inspect import signature
-from typing import (
-    Any,
-    Callable,
-    Generic,
-    Optional,
-    Union,
-)
-
-from typing_extensions import Self
-
-from langgraph.managed.base import ConfiguredManagedValue, ManagedValue, V
-from langgraph.types import LoopProtocol
-
-
-class Context(ManagedValue[V], Generic[V]):
-    runtime = True
-
-    value: V
-
-    @staticmethod
-    def of(
-        ctx: Union[
-            None,
-            Callable[..., AbstractContextManager[V]],
-            type[AbstractContextManager[V]],
-            Callable[..., AbstractAsyncContextManager[V]],
-            type[AbstractAsyncContextManager[V]],
-        ] = None,
-        actx: Optional[
-            Union[
-                Callable[..., AbstractAsyncContextManager[V]],
-                type[AbstractAsyncContextManager[V]],
-            ]
-        ] = None,
-    ) -> ConfiguredManagedValue:
-        if ctx is None and actx is None:
-            raise ValueError("Must provide either sync or async context manager.")
-        return ConfiguredManagedValue(Context, {"ctx": ctx, "actx": actx})
-
-    @classmethod
-    @contextmanager
-    def enter(cls, loop: LoopProtocol, **kwargs: Any) -> Iterator[Self]:
-        with super().enter(loop, **kwargs) as self:
-            if self.ctx is None:
-                raise ValueError(
-                    "Synchronous context manager not found. Please initialize Context value with a sync context manager, or invoke your graph asynchronously."
-                )
-            ctx = (
-                self.ctx(loop.config)  # type: ignore[call-arg]
-                if signature(self.ctx).parameters.get("config")
-                else self.ctx()
-            )
-            with ctx as v:  # type: ignore[union-attr]
-                self.value = v
-                yield self
-
-    @classmethod
-    @asynccontextmanager
-    async def aenter(cls, loop: LoopProtocol, **kwargs: Any) -> AsyncIterator[Self]:
-        async with super().aenter(loop, **kwargs) as self:
-            if self.actx is not None:
-                ctx = (
-                    self.actx(loop.config)  # type: ignore[call-arg]
-                    if signature(self.actx).parameters.get("config")
-                    else self.actx()
-                )
-            elif self.ctx is not None:
-                ctx = (
-                    self.ctx(loop.config)  # type: ignore
-                    if signature(self.ctx).parameters.get("config")
-                    else self.ctx()
-                )
-            else:
-                raise ValueError(
-                    "Asynchronous context manager not found. Please initialize Context value with an async context manager, or invoke your graph synchronously."
-                )
-            if hasattr(ctx, "__aenter__"):
-                async with ctx as v:
-                    self.value = v
-                    yield self
-            elif hasattr(ctx, "__enter__") and hasattr(ctx, "__exit__"):
-                with ctx as v:
-                    self.value = v
-                    yield self
-            else:
-                raise ValueError(
-                    "Context manager must have either __enter__ or __aenter__ method."
-                )
-
-    def __init__(
-        self,
-        loop: LoopProtocol,
-        *,
-        ctx: Union[
-            None, type[AbstractContextManager[V]], type[AbstractAsyncContextManager[V]]
-        ] = None,
-        actx: Optional[type[AbstractAsyncContextManager[V]]] = None,
-    ) -> None:
-        self.ctx = ctx
-        self.actx = actx
-
-    def __call__(self) -> V:
-        return self.value
@@ -1,19 +1,22 @@
 from typing import Annotated

 from langgraph.managed.base import ManagedValue
+from langgraph.types import PregelScratchpad


 class IsLastStepManager(ManagedValue[bool]):
-    def __call__(self) -> bool:
-        return self.loop.step == self.loop.stop - 1
+    @staticmethod
+    def get(scratchpad: PregelScratchpad) -> bool:
+        return scratchpad.step == scratchpad.stop - 1


 IsLastStep = Annotated[bool, IsLastStepManager]


 class RemainingStepsManager(ManagedValue[int]):
-    def __call__(self) -> int:
-        return self.loop.stop - self.loop.step
+    @staticmethod
+    def get(scratchpad: PregelScratchpad) -> int:
+        return scratchpad.stop - scratchpad.step


 RemainingSteps = Annotated[int, RemainingStepsManager]
@@ -1,120 +0,0 @@
-import collections.abc
-from collections.abc import AsyncIterator, Iterator, Sequence
-from contextlib import asynccontextmanager, contextmanager
-from typing import (
-    Any,
-    Optional,
-)
-
-from typing_extensions import NotRequired, Required, Self
-
-from langgraph.constants import CONF
-from langgraph.errors import InvalidUpdateError
-from langgraph.managed.base import (
-    ChannelKeyPlaceholder,
-    ChannelTypePlaceholder,
-    ConfiguredManagedValue,
-    WritableManagedValue,
-)
-from langgraph.store.base import PutOp
-from langgraph.types import LoopProtocol
-
-V = dict[str, Any]
-
-
-Value = dict[str, V]
-Update = dict[str, Optional[V]]
-
-
-# Adapted from typing_extensions
-def _strip_extras(t):  # type: ignore[no-untyped-def]
-    """Strips Annotated, Required and NotRequired from a given type."""
-    if hasattr(t, "__origin__"):
-        return _strip_extras(t.__origin__)
-    if hasattr(t, "__origin__") and t.__origin__ in (Required, NotRequired):
-        return _strip_extras(t.__args__[0])
-
-    return t
-
-
-class SharedValue(WritableManagedValue[Value, Update]):
-    @staticmethod
-    def on(scope: str) -> ConfiguredManagedValue:
-        return ConfiguredManagedValue(
-            SharedValue,
-            {
-                "scope": scope,
-                "key": ChannelKeyPlaceholder,
-                "typ": ChannelTypePlaceholder,
-            },
-        )
-
-    @classmethod
-    @contextmanager
-    def enter(cls, loop: LoopProtocol, **kwargs: Any) -> Iterator[Self]:
-        with super().enter(loop, **kwargs) as value:
-            if loop.store is not None:
-                saved = loop.store.search(value.ns)
-                value.value = {it.key: it.value for it in saved}
-            yield value
-
-    @classmethod
-    @asynccontextmanager
-    async def aenter(cls, loop: LoopProtocol, **kwargs: Any) -> AsyncIterator[Self]:
-        async with super().aenter(loop, **kwargs) as value:
-            if loop.store is not None:
-                saved = await loop.store.asearch(value.ns)
-                value.value = {it.key: it.value for it in saved}
-            yield value
-
-    def __init__(
-        self, loop: LoopProtocol, *, typ: type[Any], scope: str, key: str
-    ) -> None:
-        super().__init__(loop)
-        if typ := _strip_extras(typ):
-            if typ not in (
-                dict,
-                collections.abc.Mapping,
-                collections.abc.MutableMapping,
-            ):
-                raise ValueError("SharedValue must be a dict")
-        self.scope = scope
-        self.value: Value = {}
-        if self.loop.store is None:
-            pass
-        elif scope_value := self.loop.config[CONF].get(self.scope):
-            self.ns = ("scoped", scope, key, scope_value)
-        else:
-            raise ValueError(
-                f"Scope {scope} for shared state key not in config.configurable"
-            )
-
-    def __call__(self) -> Value:
-        return self.value
-
-    def _process_update(self, values: Sequence[Update]) -> list[PutOp]:
-        writes: list[PutOp] = []
-        for vv in values:
-            for k, v in vv.items():
-                if v is None:
-                    if k in self.value:
-                        del self.value[k]
-                        writes.append(PutOp(self.ns, k, None))
-                elif not isinstance(v, dict):
-                    raise InvalidUpdateError("Received a non-dict value")
-                else:
-                    self.value[k] = v
-                    writes.append(PutOp(self.ns, k, v))
-        return writes
-
-    def update(self, values: Sequence[Update]) -> None:
-        if self.loop.store is None:
-            self._process_update(values)
-        else:
-            return self.loop.store.batch(self._process_update(values))
-
-    async def aupdate(self, writes: Sequence[Update]) -> None:
-        if self.loop.store is None:
-            self._process_update(writes)
-        else:
-            return await self.loop.store.abatch(self._process_update(writes))
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Nuno Campos	194c4c1d1c	cli 0.2.12	2025-06-04 15:50:07 -07:00
Florian Courtial GitHub Nuno Campos	1a76f6a92a	🐛 [CLI] Generate one `--build-context` argument for each dependency in the `docker build` command. (#4962 ) * Generate one `--build-context` for each dependency in the `docker build` command. * Try and fix test --------- Co-authored-by: Nuno Campos <nuno@langchain.dev>	2025-06-04 22:48:48 +00:00
Sydney RunkleandGitHub	aedf974dfd	docs: deploy from v0 branch for now (#4960 ) only deploy docs on v0	2025-06-04 13:27:48 -04:00
Sydney RunkleandGitHub	c0b6a85488	docs: format to allow for deploy (#4959 ) formatting	2025-06-04 17:20:19 +00:00
Sydney RunkleandGitHub	9fde14079a	docs: use `retry_policy` instead of `retry` in docs (#4958 )	2025-06-04 17:12:13 +00:00
Sydney RunkleandGitHub	02f3944e88	rename `retry` -> `retry_policy` (#4957 )	2025-06-04 14:55:06 +00:00
Nuno CamposandGitHub	1ea5812ed0	Fix step_timeout causing ParentCommand/GraphInterrupt exception to bubble up (#4950 )	2025-06-03 17:32:46 -07:00
Nuno Campos	314115513c	Fix step_timeout causing ParentCommand/GraphInterrupt exception to bubble up - We should not re-raise exceptions in commit() as that is now called in a future done callback - panic_or_proceed takes care of re-raising exceptions as needed anyway	2025-06-03 17:26:29 -07:00
Nuno Campos	161a1e3af6	Remove from ci matrix	2025-06-03 17:26:13 -07:00
Nuno Campos	f8995f234c	Remove scheduler-kafka library Superseded by upcoming distributed runner	2025-06-03 17:25:17 -07:00
Nuno CamposandGitHub	5f1d76ba3b	docs: fix a grammar issue in multiple files (#4935 )	2025-06-03 17:04:20 -07:00
Asamu DavidandGitHub	13ebe1ad8c	cli v0.2.11 release (#4941 )	2025-06-04 01:00:06 +01:00
David Asamu	ae2b766d0f	bump version	2025-06-04 00:53:56 +01:00
Nuno CamposandGitHub	b211d4e767	Make it possible to run test command without docker installed (#4948 )	2025-06-03 16:29:14 -07:00
Nuno CamposandGitHub	0b609646d5	Fix async callback manager tag handling (#4949 )	2025-06-03 16:23:22 -07:00
Nuno Campos	5ab5da767b	Pass tags when configuring async callback manager	2025-06-03 16:17:17 -07:00
Nuno CamposandGitHub	a534807ec9	Allow same-name channels and nodes in StateGraph (#4944 )	2025-06-03 16:15:07 -07:00
Nuno Campos	89e827bd15	Make it possible to run test command without docker installed	2025-06-03 16:14:46 -07:00
Nuno CamposandGitHub	d4018497a9	Fix ensure_config docstring (#4947 )	2025-06-03 16:02:44 -07:00
Nuno Campos	7080eaa79e	docs: fix ensure_config docstring	2025-06-03 15:56:41 -07:00
Nuno CamposandGitHub	9c6e8d5237	Fix duplicate article typos in docs (#4946 )	2025-06-03 15:56:04 -07:00
langchain-infraandGitHub	ae3c2e0b89	fix: use right deployment link (#4945 )	2025-06-03 18:55:05 -04:00
Nuno Campos	cd2847ee07	docs: fix double article typos	2025-06-03 15:53:19 -07:00
infra	7335b80586	fix: use right deployment link	2025-06-03 18:42:11 -04:00
Asamu DavidandGitHub	394ea1c12b	Merge branch 'main' into david/06-03/update-cli-version	2025-06-03 23:40:10 +01:00
Nuno CamposandGitHub	b00ea605e2	checkpoint-postgres: Use lock also for pipeline mode (#4942 )	2025-06-03 15:30:04 -07:00
Nuno Campos	bb990715f1	Allow same-name channels and nodes in StateGraph - We no longer auto create a same-name channel for each node, so this limitation can now be lifted	2025-06-03 15:29:49 -07:00
Asamu DavidandGitHub	fce7652b7c	Merge branch 'main' into david/06-03/update-cli-version	2025-06-03 23:26:42 +01:00
langchain-infraandGitHub	8837b8534f	fix: use right link (#4943 )	2025-06-03 18:15:51 -04:00
infra	4473db6361	fix: use right link	2025-06-03 18:12:41 -04:00
Nuno Campos	c31c940bbb	checkpoint-postgres: Use lock also for pipeline mode	2025-06-03 14:59:56 -07:00
Asamu DavidandGitHub	5c18123bb7	cli v0.2.11 release ## Description bump cli version to 0.2.11	2025-06-03 21:37:25 +01:00
Asamu DavidandGitHub	7ba5636200	add support for image_distro in config file (#4871 )	2025-06-03 20:59:06 +01:00
Sydney RunkleandGitHub	37c215b5a2	Improve type checking on graph `init` and `invoke`/`stream` (#4932 )	2025-06-03 15:21:51 -04:00
Asamu DavidandGitHub	f528f5ebd6	Merge branch 'main' into david/05-28/support-image-distro-config	2025-06-03 18:18:22 +01:00
Michael Li	cd28bef155	docs: fix a grammar issue in multiple files	2025-06-04 01:20:29 +10:00
David DuongandGitHub	3a74514bc3	fix(sdk-js): `uiMessageReducer` should handle `undefined` metadata (#4931 )	2025-06-03 15:21:25 +02:00
Tat Dat Duong	2b759a2a33	fix(sdk-js): `uiMessageReducer` should handle `undefined` metadata	2025-06-03 15:20:13 +02:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	723cff9001	build(deps): bump actions/setup-node from 3 to 4 (#4915 ) Bumps [actions/setup-node](https://github.com/actions/setup-node) from 3 to 4. - [Release notes](https://github.com/actions/setup-node/releases) - [Commits](https://github.com/actions/setup-node/compare/v3...v4) --- updated-dependencies: - dependency-name: actions/setup-node dependency-version: '4' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2025-06-03 08:37:31 -04:00
lc-arjunandGitHub	7d27d108fc	bump sdk (#4925 )	2025-06-02 18:16:08 -07:00
Arjun Natarajan	f4b480c4b8	bump sdk	2025-06-02 18:13:36 -07:00
lc-arjunandGitHub	8c560ef62a	feat: support custom tracer when creating run in ts sdk (#4924 )	2025-06-02 18:10:01 -07:00
David Asamu	1e324b681a	update dockerfile generation logic, fix pip removal in wolfi	2025-06-03 01:43:59 +01:00
Arjun Natarajan	6002aebec5	maybe nees the full lint rewrite	2025-06-02 17:35:12 -07:00
Arjun Natarajan	b438210a0d	maybe fixes lint	2025-06-02 17:33:29 -07:00
Arjun Natarajan	b908e96541	need to export it here too	2025-06-02 17:27:33 -07:00
Arjun Natarajan	70dc0323c1	i want this exported	2025-06-02 17:26:20 -07:00
Arjun Natarajan	a75e40192f	perhaps better typing	2025-06-02 17:18:59 -07:00
Arjun Natarajan	952017fbe2	feat: support custom tracer when creating run in ts sdk	2025-06-02 17:17:07 -07:00
Nuno CamposandGitHub	36bd88287f	Fix makefile command file for dev server (#4922 )	2025-06-02 16:41:59 -07:00
Nuno Campos	5e5840805e	Fix makefile command file for dev server - pidfile was always empty	2025-06-02 16:35:48 -07:00
Nuno CamposandGitHub	7828003958	Fix Command(graph=PARENT) when used together w checkpointer=True (#4921 )	2025-06-02 16:18:52 -07:00
Nuno Campos	e5e78e4192	Fix Command(graph=PARENT) when used together w checkpointer=True	2025-06-02 16:05:32 -07:00
William FHandGitHub	e2b14a9499	feat: Add onRequest callback to JS SDK (#4919 ) Add onRequest callback	2025-06-02 13:43:16 -07:00
Nuno CamposandGitHub	2563301f39	Remove unused deprecation decorator/warning (#4917 )	2025-06-02 15:53:26 -04:00
David Asamu	bbbadc3db9	regenerate schema + lint & format	2025-06-02 20:50:48 +01:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	b4cb4e72ab	build(deps): bump actions/configure-pages from 4 to 5 (#4916 ) Bumps [actions/configure-pages](https://github.com/actions/configure-pages) from 4 to 5. - [Release notes](https://github.com/actions/configure-pages/releases) - [Commits](https://github.com/actions/configure-pages/compare/v4...v5) --- updated-dependencies: - dependency-name: actions/configure-pages dependency-version: '5' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2025-06-02 15:36:18 -04:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	7e45a0530b	build(deps): bump actions/cache from 3 to 4 (#4913 ) Bumps [actions/cache](https://github.com/actions/cache) from 3 to 4. - [Release notes](https://github.com/actions/cache/releases) - [Changelog](https://github.com/actions/cache/blob/main/RELEASES.md) - [Commits](https://github.com/actions/cache/compare/v3...v4) --- updated-dependencies: - dependency-name: actions/cache dependency-version: '4' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2025-06-02 15:17:21 -04:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	dd059fea3c	build(deps): bump actions/checkout from 3 to 4 (#4912 ) Bumps [actions/checkout](https://github.com/actions/checkout) from 3 to 4. - [Release notes](https://github.com/actions/checkout/releases) - [Changelog](https://github.com/actions/checkout/blob/main/CHANGELOG.md) - [Commits](https://github.com/actions/checkout/compare/v3...v4) --- updated-dependencies: - dependency-name: actions/checkout dependency-version: '4' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2025-06-02 15:17:04 -04:00
Michael LiandGitHub	871b1dccd8	docs: fix grammar issues in deployment_options.md and faq.md (#4905 )	2025-06-02 15:04:03 -04:00
Daiki InoueandGitHub	f8b5e05c47	docs: fix a typo (#4908 )	2025-06-02 15:03:23 -04:00
Eugene YurtsevandGitHub	6cde3ee7a3	ci: add depandabot config for upgrading actions (#4911 ) Depandabot config for suggesting to update old actions	2025-06-02 15:03:03 -04:00
Michael LiandGitHub	d44581d754	docs: fix typos in langgraph_studio.md and low_level.md (#4909 )	2025-06-02 17:49:04 +00:00
Michael LiandGitHub	dc787b5aa6	docs: fix grammar issues in langgraph_data_plane.md and langgraph_platform.md (#4906 ) * docs: fix grammar issues in langgraph_data_plane.md and langgraph_platform.md * docs: further update the wording in langgraph_data_plane.md	2025-06-02 17:48:55 +00:00
Michael LiandGitHub	0d6af09187	docs: fix grammar issues in application_structure.md and assistants.md (#4904 ) * docs: fix a grammar error in mcp.md * docs: fix a grammar error in multi-agent.md * docs: fix grammar issues in application_structure.md and assistants.md	2025-06-02 13:48:50 -04:00
Michael LiandGitHub	469576f966	docs: fix grammar issues in langgraph_self_hosted_control_plane.md and langgraph_server.md (#4907 )	2025-06-02 17:47:16 +00:00
Michael LiandGitHub	f00b994692	docs: fix incorrect word in memory.md (#4902 )	2025-06-02 13:48:18 +00:00
Michael LiandGitHub	7d6243e6b4	docs: fix incorrect word in custom_docker.md (#4901 )	2025-06-02 13:48:06 +00:00
Michael LiandGitHub	8fd50a50d4	docs: fix grammar errors in mcp.md and multi-agent.md (#4903 ) * docs: fix a grammar error in mcp.md * docs: fix a grammar error in multi-agent.md	2025-06-02 13:46:29 +00:00
Sydney RunkleandGitHub	eaa1b37645	Require `state_schema` in `StateGraph.__init__` (#4897 )	2025-06-02 01:52:35 +00:00
David DuongandGitHub	21f762140a	docs(sdk-js): improve resumability docs (#4899 )	2025-06-02 02:10:16 +02:00
Tat Dat Duong	5973fcb0ae	docs(sdk-js): improve resumability docs	2025-06-02 02:09:43 +02:00
David DuongandGitHub	624c688013	feat(sdk-js): add docs for `reconnectOnMount`, expose run metadata (#4898 )	2025-06-02 01:53:46 +02:00
Tat Dat Duong	d05b323b88	feat(sdk-js): add docs for `reconnectOnMount`, expose run metadata	2025-06-02 01:52:40 +02:00
Ikpreet S BabraandGitHub	765bc3b9e0	docs: Update 2-add-tools.md (#4896 ) Typo in first sentence.	2025-06-01 10:47:35 +00:00
Nuno CamposandGitHub	78d3d8c802	Remove add_conditional_edge(..., then=) (#4893 ) - This is redundant with deferred nodes, and not documented	2025-06-01 06:46:51 -04:00
Nuno CamposandGitHub	3b85c83d51	Avoid repeated runtime calls to get_type_hints (#4888 )	2025-05-31 08:37:23 -07:00
Eugene YurtsevandGitHub	08f88ce8da	docs: add gtm (#4887 ) Add google tag manager	2025-05-30 21:49:22 -04:00
Nuno Campos	d661d528b2	Avoid repeated runtime calls to get_type_hints - Replace get_type_hints logic with much simpler implementation which only collects annotated keys (not the annotations themselves) - Cache annotations in a WeakKeyDictionary	2025-05-30 16:30:53 -07:00
Sydney RunkleandGitHub	5cbf31e8a4	Remove non-state Graph (#4872 )	2025-05-29 21:56:53 -04:00
Sydney Runkle GitHub Nuno Campos	a5e6223569	Remove `MessageGraph` (#4875 ) Remove MessageGraph Co-authored-by: Nuno Campos <nuno@langchain.dev>	2025-05-29 21:50:23 -04:00
Nuno CamposandSydney Runkle	afb83d2201	Remove non-state Graph - Not used in any examples/docs	2025-05-29 21:38:58 -04:00
David Asamu	1e8f5dd2b6	add warning when image distro is not configured as wolfi	2025-05-30 00:25:27 +01:00
David Asamu	574a9246a6	add unit tests for image_distro config	2025-05-29 23:30:27 +01:00
David Asamu	c46f7a4c3d	add support for image_distro in config file	2025-05-29 21:54:54 +01:00
langchain-infraandGitHub	8e82c7d1a0	docs: fix self hosted control plane concepts/docs (#4870 )	2025-05-29 16:16:02 -04:00
infra	2066e4c018	docs: fix self hosted control plane concepts/docs	2025-05-29 13:10:16 -07:00
Sydney RunkleandGitHub	f8fae30aa9	docs: fix pprint import (#4869 )	2025-05-29 19:37:43 +00:00
Nuno CamposandGitHub	2615c7046c	Remove UntrackedValue channel (#4868 )	2025-05-29 12:14:36 -07:00
Nuno Campos	05f3904d09	Remove UntrackedValue channel - This is incompatible with distributed execution modes, so needs to go	2025-05-29 12:07:51 -07:00
Nuno CamposandGitHub	583fe9fd84	Remove UntrackedValue channel (#4859 )	2025-05-29 11:52:10 -07:00
Nuno Campos	0afc4ebda3	Flip default for checkpoint_during - Now defaulting to False, ie. saving only the final checkpoint - All features other than time travel into an intermediate step are supported by checkpoint_during=False so this is a better default for almost all use cases	2025-05-29 11:45:54 -07:00
Nuno CamposandGitHub	d55cafad29	Remove Channel node builder (#4858 )	2025-05-28 14:30:12 -07:00
Nuno Campos	1b961f68b9	Remove Channel node builder - Replaced by NodeBuilder introduced in earlier PR	2025-05-28 14:14:47 -07:00
Nuno CamposandGitHub	045d6dfb82	Update managed value usage in local_read (#4854 )	2025-05-28 14:05:14 -07:00
Nuno CamposandGitHub	26d5fccfa0	Modify stream mode messages and custom to respect subgraphs=False (#4843 )	2025-05-28 14:05:03 -07:00
Nuno CamposandGitHub	76661f4c9c	Remove Checkpoint.writes (#4822 )	2025-05-28 14:04:45 -07:00
Nuno CamposandGitHub	9320bedd2a	Remove Checkpoint.pending_sends (#4820 )	2025-05-28 14:04:26 -07:00
Nuno CamposandGitHub	3737d886e7	Pregel: Add NodeBuilder class to replace Channel.subscribe_to (#4819 )	2025-05-28 14:04:08 -07:00
Nuno CamposandGitHub	8386eaceab	Remove dict subclasses used for values/updates stream chunks (#4816 )	2025-05-28 14:03:40 -07:00
Nuno CamposandGitHub	c32677c92d	Remove old checkpoint test fixtures (#4814 )	2025-05-28 14:03:12 -07:00
Nuno CamposandGitHub	e384ccacaf	Remove postgres shallow checkpointer (#4813 )	2025-05-28 14:02:58 -07:00
Nuno CamposandGitHub	dc95f95181	Remove ChannelsManager, ManagedValues are now static classes and never instantiated (#4812 )	2025-05-28 14:02:43 -07:00
Nuno CamposandGitHub	2e4645ce61	Remove Context channel / managed value, Remove SharedValue (#4857 )	2025-05-28 14:02:12 -07:00
Nuno Campos	197b01e985	Update managed value usage in local_read	2025-05-28 10:40:20 -07:00
Nuno Campos	bc404d14f6	Format	2025-05-27 16:07:54 -07:00
Nuno Campos	936176eb21	Remove Checkpoint.writes - This has been superseded by saving the individual writes of each task through put_writes() - Removing this speeds up checkpoint operations as it was duplicating data saved elsewhere already	2025-05-25 19:06:46 -07:00
Nuno Campos	4e8fbe4525	Remove Checkpoint.pending_sends - Instead store sends in a Topic channel, removing the need to fetch sends as writes against the parent checkpoint - Remove deprecated/unused functions in langgraph-checkpoint (will require bumping min range for langgraph-checkpoint in langgraph lib) - Implement migration of old pending sends in langgraph-checkpoint-postgres - Ensure parent config of `checkpoint_during=False` checkpoints always points to checkpoints that were also saved	2025-05-25 19:06:02 -07:00
Nuno Campos	85522ec6a6	Pregel: Add NodeBuilder class to replace Channel.subscribe_to - For now keeping Channel.subscribe_to given it was used in docs	2025-05-24 20:35:37 -07:00
Nuno Campos	0f8f5444bc	Lint	2025-05-24 14:54:57 -07:00
Nuno Campos	bc17c3522b	Remove old checkpoint test fixtures - Now all tests fully migrated to more recent sync_checkpointer and async_checkpointer fixtures for parametrising on checkpointer - Use sync/async_store fixtures where tests used only in memory store - Remove unused "should snapshot" check for older versions of langchain core no longer tested against	2025-05-24 14:27:13 -07:00
Nuno Campos	8c11c1155a	Remove postgres shallow checkpointer - This was deprecated, and superseded by checkpoint_during=False, which is available for all checkpointers	2025-05-24 12:39:50 -07:00
Nuno Campos	9c79d887a0	Remove ChannelsManager, ManagedValues are now static classes and never instantiated - This keeps the existing API for accessing IsLastValue, RemainingSteps, but simplifies the internal implementation - This lets us remove ChannelsManager completely, which no longer needs to be a context manager, now replaced w function channels_from_checkpoint - Managed values are now simple functions that return a value given a PregelScratchpad, which will be easier to implement in distributed runner	2025-05-24 12:15:11 -07:00
Nuno Campos	889b40e7a8	Remove Context channel / managed value, Remove SharedValue - Both have been deprecated long ago, have not been present in docs for quite a while - Removing these lets us delete some code paths that were dedicated to this, easing work on distributed runner	2025-05-24 11:05:15 -07:00