chore(delta-channel): remove supports_delta_channels flag

Rely on the runtime raise in DeltaChannel.from_checkpoint() instead of
a compile-time boolean flag. Savers that assemble DeltaChainValue inside
_load_blobs work transparently; savers that don't will pass through a raw
DeltaValue and hit a clear ValueError on first reload.

Removes: BaseCheckpointSaver.supports_delta_channels, the attribute on
InMemorySaver / PostgresSaver / AsyncPostgresSaver, the compile-time
UserWarning in StateGraph.compile(), and the associated test.

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

@@ -32,7 +32,6 @@ Conn = _internal.Conn  # For backward compatibility
 class PostgresSaver(BasePostgresSaver):
     """Checkpointer that stores checkpoints in a Postgres database."""
 
-    supports_delta_channels: bool = True
     lock: threading.RLock
 
     def __init__(

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

@@ -34,7 +34,6 @@ Conn = _ainternal.Conn  # For backward compatibility
 class AsyncPostgresSaver(BasePostgresSaver):
     """Asynchronous checkpointer that stores checkpoints in a Postgres database."""
 
-    supports_delta_channels: bool = True
     lock: asyncio.Lock
 
     def __init__(

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

@@ -475,16 +475,6 @@ class BaseCheckpointSaver(Generic[V]):
         """
         raise NotImplementedError
 
-    supports_delta_channels: bool = False
-    """True if this saver assembles DeltaChannel chains inside get_tuple.
-
-    Savers that set this to True (InMemorySaver, PostgresSaver) assemble the
-    full DeltaChainValue before returning from get_tuple, so the channel's
-    from_checkpoint method always receives a DeltaChainValue.  Savers that
-    leave this False will return a raw DeltaValue in channel_values, which
-    causes DeltaChannel.from_checkpoint to raise a clear error.
-    """
-
     def get_next_version(self, current: V | None, channel: None) -> V:
         """Generate the next version ID for a channel.
 

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

@@ -122,8 +122,6 @@ class InMemorySaver(
     ) -> bool | None:
         return self.stack.__exit__(__exc_type, __exc_value, __traceback)
 
-    supports_delta_channels: bool = True
-
     def _load_blobs(
         self,
         thread_id: str,

libs/langgraph/langgraph/channels/delta.py

@@ -125,7 +125,7 @@ class DeltaChannel(Generic[Value], BaseChannel[list[Value], Value, DeltaValue]):
         elif isinstance(checkpoint, DeltaValue):
             raise ValueError(
                 f"Channel '{self.key}' uses DeltaChannel but the checkpointer "
-                "does not support incremental storage (supports_delta_channels=False). "
+                "does not support incremental channel storage. "
                 "Use InMemorySaver or PostgresSaver, or remove DeltaChannel from your schema."
             )
         else:

libs/langgraph/langgraph/graph/state.py

@@ -1083,26 +1083,6 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         """
         checkpointer = ensure_valid_checkpointer(checkpointer)
 
-        # Warn early if DeltaChannel channels are paired with an incompatible checkpointer.
-        if checkpointer is not None and checkpointer is not False:
-            from langgraph.channels.delta import DeltaChannel
-
-            delta_keys = [
-                k for k, v in self.channels.items() if isinstance(v, DeltaChannel)
-            ]
-            if delta_keys and not getattr(
-                checkpointer, "supports_delta_channels", False
-            ):
-                warnings.warn(
-                    f"Channel(s) {delta_keys} use DeltaChannel but "
-                    f"{type(checkpointer).__name__} does not support incremental "
-                    "channel storage (supports_delta_channels=False). "
-                    "Loading the graph will raise a ValueError. "
-                    "Use InMemorySaver or PostgresSaver.",
-                    UserWarning,
-                    stacklevel=2,
-                )
-
         serde_allowlist: set[tuple[str, ...]] | None = None
         if _serde.STRICT_MSGPACK_ENABLED:
             schema_types: list[type[Any]] = [

libs/langgraph/tests/test_channels.py

@@ -233,7 +233,9 @@ def test_delta_channel_unsupported_saver_raises() -> None:
 
     spec = DeltaChannel(add_messages)
     raw = DeltaValue(delta=[{"type": "human", "content": "hello"}])
-    with pytest.raises(ValueError, match="supports_delta_channels"):
+    with pytest.raises(
+        ValueError, match="does not support incremental channel storage"
+    ):
         spec.from_checkpoint(raw)
 
 
@@ -523,45 +525,3 @@ def test_delta_channel_dict_reducer_with_deletions() -> None:
     spec = DeltaChannel(merge_files, dict)
     ch2 = spec.from_checkpoint(chain)
     assert ch2.get() == {"file2.py": "content2", "file3.py": "content3"}
-
-
-def test_delta_channel_compile_warns_on_incompatible_saver() -> None:
-    """compile() warns when DeltaChannel is used with a saver that lacks supports_delta_channels."""
-    import warnings
-    from typing import Annotated
-
-    from langgraph.checkpoint.base import BaseCheckpointSaver
-    from typing_extensions import TypedDict
-
-    from langgraph.channels.delta import DeltaChannel
-    from langgraph.graph import START, StateGraph
-    from langgraph.graph.message import add_messages
-
-    class FakeSaver(BaseCheckpointSaver):
-        # Third-party saver that has not opted into delta channel support.
-        supports_delta_channels = False
-
-        def get_tuple(self, config):
-            return None
-
-        def put(self, config, checkpoint, metadata, new_versions):
-            return config
-
-        def put_writes(self, config, writes, task_id, task_path=""):
-            pass
-
-        def list(self, config, *, filter=None, before=None, limit=None):
-            return iter([])
-
-    class State(TypedDict):
-        messages: Annotated[list, DeltaChannel(add_messages)]
-
-    builder = StateGraph(State)
-    builder.add_node("echo", lambda s: {})
-    builder.add_edge(START, "echo")
-
-    with warnings.catch_warnings(record=True) as w:
-        warnings.simplefilter("always")
-        builder.compile(checkpointer=FakeSaver())  # type: ignore[arg-type]
-
-    assert any("supports_delta_channels" in str(warning.message) for warning in w)